RecapClient

Primary entry point for interacting with a RECAP provenance database.

Usage

RecapClient(
    connection_state,
    *,
    namespace_path=None,
    namespace_context=None,
)

RecapClient wraps a SQLAlchemy session and exposes factory methods for creating and loading the core domain objects: namespaces, resources, resource templates, process templates, and process runs.

Use from_sqlite() for local SQLite databases and from_url() for remote recap servers. These are the canonical initialization methods.

The client can be used as a context manager, which closes the underlying engine on exit:

with RecapClient.from_sqlite() as client:
    client.create_namespace("projects")
    client.create_namespace("projects/my-project")

Methods

Name	Description
__enter__()	Return the client itself when used as a context manager.
__exit__()	Close the client when leaving the `with` block.
__init__()	Initialise common empty client state.
build_process_run()	Open a builder for a `~recap.dsl.process_builder.ProcessRunBuilder`.
build_process_template()	Open a builder for a
build_resource()	Open a builder for a `~recap.dsl.resource_builder.ResourceBuilder`.
build_resource_template()	Open a builder for a
close()	Close the underlying session and engine to release SQLite locks.
copy_process_run()	Copy process run into current namespace with fresh aggregate identity.
copy_resource()	Copy resource across namespaces and commit or roll back atomically.
create_namespace()	Create a namespace and make it active for subsequent writes.
create_resource()	Create a resource instance from a template with default values.
from_sqlite()	Create or upgrade a local SQLite database and return a connected client.
from_url()	Connect to a recap webserver.
get_namespace()	Get existing namespace information
get_resource()	Load a single resource by name and template.
list_namespaces()	Return relative names of direct child namespaces.
namespace()	Return a view scoped to an additive namespace path.
permissions()	Return typed effective permissions for this client’s namespace.
query_maker()	Return a query DSL scoped to this client’s namespace.
update_namespace()	Apply namespace metadata/status update and make result active.

enter()

Return the client itself when used as a context manager.

Usage

Source

__enter__()

exit()

Close the client when leaving the with block.

Usage

Source

__exit__(exc_type, exc, tb)

init()

Initialise common empty client state.

Usage

Source

__init__(self, connection_state: ConnectionState, namespace_path: str)
__init__(self, connection_state: ConnectionState, namespace_context: NamespaceContext)

Use from_sqlite() for local clients and from_url() for remote clients. These are the canonical initialization methods.

build_process_run()

Open a builder for a ~recap.dsl.process_builder.ProcessRunBuilder.

Usage

Source

build_process_run(self, name: str, description: str, template_name: str, version: str) -> ProcessRunBuilder
build_process_run(self, process_run_id: UUID) -> ProcessRunBuilder

Namespace context must be active before calling this method with new run arguments.

Call this method in two mutually exclusive ways:

Create a new run — pass all four positional arguments::

with client.build_process_run(
    "Run 001", "First run", "MX Data Collection", "1.0"
) as run:
    run.assign_resource(plate, "crystal_plate")

Load an existing run by ID::

with client.build_process_run(process_run_id=uuid) as run:
    ...

Parameters

name: Display name for this run (positional).
description: Free-text description of this run (positional).
template_name: Name of the ProcessTemplate to instantiate (positional).
version: Version of the template (positional).
process_run_id: UUID of an existing run to load. When supplied, positional arguments must not be provided.
on_existing: Controls behavior when run already exists: "warn" (default), "raise", or "silent".

Returns

A ~recap.dsl.process_builder.ProcessRunBuilder context manager.

Raises

RuntimeError: If the backend has not been initialised.
ValueError: If no namespace context is set when creating a new run.
TypeError: On invalid argument combinations.

build_process_template()

Open a builder for a

Usage

Source

build_process_template(self, name: str, version: str) -> ProcessTemplateBuilder
build_process_template(self, process_template_id: UUID) -> ProcessTemplateBuilder

~recap.dsl.process_builder.ProcessTemplateBuilder.

Call this method in two mutually exclusive ways:

Create or update by name and version — pass positional arguments name and version::

with client.build_process_template("MX Data Collection", "1.0") as pt:
    pt.add_step("Mount", order=1)

Load an existing template by ID — pass the keyword argument process_template_id::

with client.build_process_template(
    process_template_id=uuid
) as pt:
    ...

Parameters

name: Human-readable name of the process template (positional).
version: Version string, e.g. "1.0" (positional).
process_template_id: UUID of an existing template to load. When supplied, name and version must not be provided.
on_existing: Controls behavior when template already exists: "warn" (default), "raise", or "silent".

Returns

A ~recap.dsl.process_builder.ProcessTemplateBuilder context manager that commits on clean exit and rolls back on exception.

Raises

RuntimeError: If the backend has not been initialised.
TypeError: On invalid argument combinations.

build_resource()

Open a builder for a ~recap.dsl.resource_builder.ResourceBuilder.

Usage

Source

build_resource(self, name: str, template_name: str, template_version: str = '1.0', parent: ResourceSchema | UUID | None = None) -> ResourceBuilder
build_resource(self, resource_id: UUID) -> ResourceBuilder

Use this when you need to inspect or modify a resource’s property values before (or after) persisting them. For simple creation with default values prefer create_resource().

Call this method in two mutually exclusive ways:

Create or update by name and template — pass positional arguments name and template_name::

with client.build_resource("Plate A", "Library Plate") as rb:
    model = rb.get_model()
    model.children["A01"].properties.status.used = True
    rb.set_model(model)

Load an existing resource by ID::

with client.build_resource(resource_id=uuid) as rb:
    ...

Parameters

name: Display name for the resource (positional).
template_name: Name of the ResourceTemplate to instantiate from (positional).
template_version: Version of the resource template. Defaults to "1.0" (keyword only).
resource_id: UUID of an existing resource to load. When supplied, positional arguments must not be provided.
on_existing: Controls behavior when resource already exists: "warn" (default), "raise", "silent", or "create".
parent: Optional parent resource for nesting the new resource. Accepts a ~recap.schemas.resource.ResourceSchema or a ~uuid.UUID (which will be resolved to a schema via backend query). Cannot be combined with resource_id.

Returns

A ~recap.dsl.resource_builder.ResourceBuilder context manager that commits on clean exit and rolls back on exception.

Raises

RuntimeError: If the backend has not been initialised.
TypeError: On invalid argument combinations.

build_resource_template()

Open a builder for a

Usage

Source

build_resource_template(self, name: str, type_names: list[str], version: str = '1.0') -> ResourceTemplateBuilder
build_resource_template(self, resource_template_id: UUID) -> ResourceTemplateBuilder

~recap.dsl.resource_builder.ResourceTemplateBuilder.

A ~recap.schemas.resource.ResourceTemplateSchema is the blueprint for a ~recap.schemas.resource.ResourceSchema. This method supports two mutually exclusive call patterns:

Create or update a template by name — the most common usage::

with client.build_resource_template(
    name="Library Plate",
    type_names=["container", "plate", "library_plate"],
) as tb:
    tb.add_properties(
        {"dimensions": [{"name": "rows", "type": "int", "default": 8}]}
    )

Load an existing template by ID::

with client.build_resource_template(resource_template_id=uuid) as tb:
    ...

Parameters

name: Unique human-readable name of the template. Required when not supplying resource_template_id.
type_names: List of type tag strings (e.g. ["container", "plate"]). Required when not supplying resource_template_id.
version: Schema version string. Defaults to "1.0".
resource_template_id: UUID of an existing template to load. When supplied, name and type_names must not be provided.
on_existing: Controls behavior when template already exists: "warn" (default), "raise", or "silent".

Returns

A ~recap.dsl.resource_builder.ResourceTemplateBuilder context manager.

Raises

RuntimeError: If the backend has not been initialised.
TypeError: If type_names is a string, contains non-string items, or if conflicting arguments are provided.

close()

Close the underlying session and engine to release SQLite locks.

Usage

Source

close()

Safe to call multiple times. After calling this method the client should no longer be used.

copy_process_run()

Copy process run into current namespace with fresh aggregate identity.

Usage

Source

copy_process_run(source_process_run_id, options=None)

copy_resource()

Copy resource across namespaces and commit or roll back atomically.

Usage

Source

copy_resource(source_resource_id, options=None)

Destination namespace comes from this client’s scope. Returns persisted full schema and propagates backend validation or authorization errors.

create_namespace()

Create a namespace and make it active for subsequent writes.

Usage

Source

create_namespace(path, metadata=None, as_current=False)

create_resource()

Create a resource instance from a template with default values.

Usage

Source

create_resource(
    name,
    template_name,
    template_version="1.0",
    parent=None,
    on_existing="create"
)

This is the convenience shortcut when you do not need to override any property values before saving. Child resources defined by the template are created automatically and all properties are populated with their declared defaults.

For more control over property values before persisting, use build_resource() instead.

Examples

:
plate = client.create_resource("Plate A", "Library Plate")
plate.children["A01"].properties.status.used.value  # False

Parameters

name: Display name for the new resource.
template_name: Name of the ResourceTemplate to instantiate.
template_version: Version of the resource template. Defaults to "1.0".
parent: Optional parent ~recap.schemas.resource.ResourceSchema when the new resource should be nested inside an existing one.
on_existing: Controls behavior when a resource with the same name, parent, and template already exists: - "create" (default): always create a new resource. Resource names are NOT globally unique — multiple resources with the same name can coexist (e.g., for different namespaces). - "silent": reuse the existing resource silently. - "warn": reuse the existing resource and emit a warning. - "raise": raise ExistingResourceError.

Returns

A ~recap.schemas.resource.ResourceSchema representing the persisted resource, including any auto-created children.

from_sqlite()

Create or upgrade a local SQLite database and return a connected client.

Usage

Source

from_sqlite(path=None, echo=False, *, namespace=None)

This is the canonical way to create a RecapClient. The method creates the database file (and any missing parent directories) if it does not already exist, then runs any pending Alembic migrations so the schema is always up to date.

Examples

:
# Temporary database (auto-generated filename in the system temp dir)
client = RecapClient.from_sqlite()

# Persistent database at a specific path
client = RecapClient.from_sqlite("/data/my_experiment.db")

Parameters

path: Filesystem path for the SQLite database. Accepts a str or pathlib.Path. When omitted a new file named recap-<uuid>.db is created in the system temp directory.
echo: Forward all SQL statements to the Python logging infrastructure. Useful for debugging. Defaults to False.

Returns

A fully initialised RecapClient connected to path.

Raises

ValueError: If path points to an existing directory rather than a file.

from_url()

Connect to a recap webserver.

Usage

Source

from_url(url, *, api_key, timeout=30.0, namespace=None, unscoped=False)

Uses ~recap.adapter.rest.RESTAdapter for reads and writes. The client does not require access to the server’s database filesystem.

Parameters

url: str: Base URL of the recap server, e.g. "http://localhost:8000".
api_key: str: API key used to authenticate requests.
timeout: float = 30.0: HTTP request timeout in seconds.
namespace: str = None: Optional namespace to initialize client with

Returns

RecapClient: Fully initialized client with REST reads and writes.

Raises

RecapConnectionError: If the server is unreachable.
RecapProtocolError: If the server returns a malformed response.
RecapRequestError: If the server returns an API error response.

get_namespace()

Get existing namespace information

Usage

Source

get_namespace(path)

get_resource()

Load a single resource by name and template.

Usage

Source

get_resource(self, name: str, template_name: str, template_version: str | None = '1.0', expand: Literal[False] = False) -> ResourceRef
get_resource(self, name: str, template_name: str, template_version: str | None = '1.0', expand: Literal[True]) -> ResourceSchema

Looks up the active resource with the given name whose template matches template_name and template_version. Raises if no such resource exists.

Examples

:
plate = client.get_resource("Plate A", "Library Plate", expand=True)
plate.children["A01"].properties.status.used.value  # False

Parameters

name: Name of the resource to load.
template_name: Name of the ResourceTemplate the resource was instantiated from.
template_version: Version of the resource template. Defaults to "1.0".
expand: When True, eagerly hydrate the full resource subtree (template, properties, and the entire child hierarchy) and return a ~recap.schemas.resource.ResourceSchema. When False (default), return a lightweight ~recap.schemas.resource.ResourceRef.

Returns

A ~recap.schemas.resource.ResourceRef when expand=False, or a fully hydrated ~recap.schemas.resource.ResourceSchema when expand=True.

list_namespaces()

Return relative names of direct child namespaces.

Usage

Source

list_namespaces()

namespace()

Return a view scoped to an additive namespace path.

Usage

Source

namespace(path)

permissions()

Return typed effective permissions for this client’s namespace.

Usage

Source

permissions()

query_maker()

Return a query DSL scoped to this client’s namespace.

Usage

Source

query_maker(*, on_unloaded="warn")

update_namespace()

Apply namespace metadata/status update and make result active.

Usage

Source

update_namespace(
    namespace_id=None, *, expected_revision=None, metadata=None, status=None
)

Methods

__enter__()

__exit__()

__init__()

build_process_run()

Parameters

Returns

Raises

build_process_template()

Parameters

Returns

Raises

build_resource()

Parameters

Returns

Raises

build_resource_template()

Parameters

Returns

Raises

close()

copy_process_run()

copy_resource()

create_namespace()

create_resource()

Examples

Parameters

Returns

from_sqlite()

Examples

Parameters

Returns

Raises

from_url()

Parameters

Returns

Raises

get_namespace()

get_resource()

Examples

Parameters

Returns

list_namespaces()

namespace()

permissions()

query_maker()

update_namespace()

enter()

exit()

init()