API reference¶

This is the public, developer-facing API of the oclp package. It covers the decorators, runtime helpers, record models, local storage, and validation tools that an application may intentionally use. Import from oclp unless a section explicitly names another module.

This page is a practical SDK contract, not a restatement of the protocol. Core record field semantics remain defined by the OCLP specification.

Choose an API¶

Core records¶

The SDK re-exports immutable Pydantic models for all Core record kinds:

from oclp import Artifact, ArtifactSet, Computation, Evidence, Event, Execution
from oclp.models import (
    ArtifactSetMember,
    Digest,
    GitSource,
    OpaqueSource,
    PortDefinition,
    RecordReference,
    ServiceSource,
)

Construct these directly when the application already has a durable fact to record. Otherwise prefer the decorators and runtime below, which create UUIDs, references, Event chronology, and Artifact bindings consistently. Supporting Core models include Diagnostic, ParameterDefinition, GitCheckout, and ArtifactSource; see records and canonicalization.

name, description, and annotations¶

name is a concise application-owned display label. description is optional plain-text explanation intended to help a human understand a durable record; it is not Markdown or HTML. annotations remain application-defined, structured JSON metadata. All three are immutable record fields when supplied.

Where one callable declares one Core record, @computation, @evidence, and Artifact acquisition decorators accept either description="..." or the explicit opt-in description_from_docstring=True. The SDK uses inspect.getdoc() only when that flag is set. An explicit description wins; when the flag is set without an explicit value, the callable must have a non-empty docstring or declaration fails. The SDK never treats ordinary # comments as record metadata.

Artifact boundaries¶

ArtifactType¶

ArtifactType declares a durable representation. Pass a concrete class to a Computation input and a named concrete instance to an output:

from oclp import CsvArtifact, JsonArtifact, computation


@computation(
    name="Summarize demand",
    inputs={"features": CsvArtifact},
    outputs={"summary": JsonArtifact(name="Demand summary")},
)
def summarize_demand(features) -> dict[str, object]:
    return {"rows": len(features)}

Input classes validate an Artifact media type. Output instances additionally carry a required application name and may contain an explicit description, annotations, schema_uri, and serialization options. For multi-output Computations, set each output's description on its own ArtifactType instance; a function docstring is not shared across several Artifact records.

For output metadata that depends on one call's ordinary parameters, provide an annotation_factory. It declares the parameters it needs by name; the SDK matches those names to the decorated Computation call and passes their actual values when it materializes that output. The factory returns JSON-compatible annotations for that one immutable Artifact record. It need not declare every Computation argument. Static annotations remain in place and the factory's values are added to them; a factory value wins if both maps use the same key:

def fold_annotations(*, fold_number: int) -> dict[str, int | str]:
    return {
        "fold_number": fold_number,
        "split_strategy": "temporal",
    }


@computation(
    name="Train temporal fold",
    outputs={
        "model": CatBoostModelArtifact(
            name="Temporal fold model",
            annotations={"model_family": "catboost"},
            annotation_factory=fold_annotations,
        ),
    },
)
def train_fold(*, fold_number: int): ...

For train_fold(fold_number=2), the SDK calls fold_annotations(fold_number=2) and persists this model Artifact metadata:

{
  "model_family": "catboost",
  "fold_number": 2,
  "split_strategy": "temporal"
}

outputs determines how a callable result becomes persisted output ports. With one declared output, the entire return value is materialized at that port. The returned {"rows": ...} in this example therefore becomes the JSON Artifact bound to outputs["summary"]; the return mapping does not need a "summary" key. With multiple declared outputs, the return value must be a mapping or object with one field for each output-port name:

@computation(
    name="Score demand",
    outputs={
        "predictions": JsonArtifact(name="Demand predictions"),
        "metrics": JsonArtifact(name="Demand metrics"),
    },
)
def score_demand(...) -> dict[str, object]:
    return {
        "predictions": {"values": [12.0, 14.0]},
        "metrics": {"mae": 1.2},
    }

Persistence occurs only inside an active OclpRun; outside it, the decorated function returns its ordinary Python value.

Acquisition decorators¶

Use these when a function obtains an external value that must become an Artifact before it reaches a Computation. They require an active OclpRun and return an ArtifactHandle.

from oclp import json_artifact


@json_artifact(name="Training plan", description_from_docstring=True)
def read_training_plan() -> dict[str, object]:
    """Declared temporal-fold configuration used to prepare model inputs."""

    return {"folds": 3}

The corresponding JsonArtifact, CsvArtifact, ParquetArtifact, ArrowIpcArtifact, NpyArtifact, NpzArtifact, YamlArtifact, TomlArtifact, and XmlArtifact classes are output declarations. The SDK also supplies BytesArtifact, FileArtifact, CatBoostModelArtifact, XGBoostModelArtifact, LightGBMModelArtifact, and SklearnModelArtifact. See artifact formats and integrations for value types, media types, options, and package extras.

All acquisition decorators accept description= and description_from_docstring=. Those fields affect only the durable Artifact record; the wrapped callable remains an ordinary function outside an active OclpRun.

Handles and adapters¶

An ArtifactHandle exposes:

handle.artifact               # Core Artifact
handle.reference              # RecordReference
handle.path                   # local payload path
handle.read_verified_bytes()  # SHA-256 verification, then bytes

ArtifactAdapterRegistry converts verified payloads into downstream Python types. DEFAULT_ARTIFACT_ADAPTERS provides shipped adapters:

frame = DEFAULT_ARTIFACT_ADAPTERS.load(handle, pandas.DataFrame)

Pass a custom ArtifactAdapterRegistry as artifact_adapters= to OclpRun or observe_run(...). Subclass ArtifactAdapter for application formats by implementing supports(handle, target_type) and load(handle, target_type). artifact_handle(published) converts a PublishedArtifact to its handle, and artifact_type(function) returns the representation bound to an acquisition callable. Representation and integrity failures raise ArtifactAdapterError or its ArtifactIntegrityError subtype.

Computations¶

@computation(...)¶

@computation(
    name="...",
    description=None,
    description_from_docstring=False,
    input_ports=(),
    inputs=None,
    output_ports=(),
    outputs=None,
    artifact_set=None,
    requires=None,
    profiles=None,
    annotations=None,
)
def work(...): ...

name is required and application-owned. description may be explicit or, when description_from_docstring=True, be the normalized callable docstring. inputs maps parameter/port names to an ArtifactType, many(ArtifactType), or artifact_set_input(...). outputs maps output-port names to named ArtifactType instances. In an active runtime, the SDK emits one source-bound Computation per callable per observed run and an Execution for each call. Outside a runtime, the callable retains normal Python behavior.

Within that same active runtime, passing an exact raw value returned from one decorated Computation to another decorated Computation automatically reuses the published Artifact binding for the downstream input. Passing an ArtifactHandle instead requests an explicit verified payload load through the consuming parameter's adapter. Handles are required once a process or run boundary removes in-memory object identity.

Use input_ports and output_ports directly only to declare portable interfaces that do not need SDK-managed payload serialization. requires contains evaluators decorated with @evidence.

ComputationArtifactSet¶

Use this when outputs from one computation also form one package. It adds an ArtifactSet output to the same real Execution; it does not create a synthetic packaging computation.

@computation(
    name="Train candidate",
    outputs={
        "model": CatBoostModelArtifact(name="Candidate model"),
        "metrics": JsonArtifact(name="Candidate metrics"),
    },
    artifact_set=ComputationArtifactSet(
        name="Candidate release",
        members={"model": ("model", "model"), "metrics": ("metrics", "metrics")},
        port="release",
    ),
)
def train_candidate(...): ...

members maps a set member name to (output_port, optional_role). port defaults to "artifact_set".

Evidence¶

@evidence(
    name="Candidate quality",
    description=None,
    description_from_docstring=False,
    profiles=None,
    annotations=None,
)
def candidate_quality(result) -> Literal["pass", "fail", "error"]:
    return "pass"

Attach it with @computation(..., requires=(candidate_quality,)). Required evidence is evaluated for each Execution; a successful Execution requires all gates to pass.

Evaluator exceptions and invalid outcomes become outcome: "error" Evidence with a Diagnostic rather than disappearing as logs.

Runs and release sets¶

@run(...) and observe_run(...)¶

@run(
    name="Daily training",
    adapters=(),
    required_evidence_policy="continue",
)
def train(*, observed): ...


with observe_run(train, publisher=publisher, source=source) as observed:
    train(observed=observed)

@run declares an application workflow; it is not a Core record or parent Execution. observe_run creates a fresh run UUID, activates the runtime, and applies the same profiles.run binding to the real child Executions. Supply run_id= only when the application already owns that concrete UUID. run_template(workflow) returns the static RunTemplate declaration attached by @run. adapters holds optional SDK integration objects. It can also be supplied to observe_run(...) for bootstrap-time configuration.

Set required_evidence_policy="raise" when a workflow must stop after a real Execution publishes failed required Evidence. The default, "continue", preserves the failed Execution and leaves application control flow unchanged. With "raise", the runtime raises RequiredEvidenceFailedError only after outputs, every Evidence record, and the failed terminal Event have been published. Its execution and evidence attributes identify those records.

An adapter declared on @run is a reusable configuration template, not its active per-run session. The runtime calls an adapter's optional for_run() method to obtain a fresh active instance for each observation. Access that active instance from an OclpRun by its class:

with observe_run(workflow, publisher=publisher, source=source) as observed:
    workflow()
    mlflow = observed.adapter(MlflowAdapter)
    mlflow.log_metrics({"validation_rmse": 42.1})

OclpRun.adapter(AdapterClass) returns the exactly one active adapter that is an instance of AdapterClass. It raises when no adapter or more than one adapter matches; this makes an ambiguous integration configuration visible. The returned object is useful for application-selected external context only; the adapter itself automatically mirrors OCLP records.

MlflowAdapter (optional)¶

Install the optional extra first:

pip install 'oclp[mlflow]'

Then attach one adapter to an observed run:

from oclp import MlflowAdapter, observe_run, run

@run(
    name="Daily training",
    adapters=(
        MlflowAdapter(
            experiment_name="daily-training",
            tracking_uri="sqlite:///mlflow.db",
        ),
    ),
)
def workflow(): ...

with observe_run(workflow, publisher=publisher, source=source):
    workflow()

The runtime calls for_run() when available, so an adapter declaration receives a fresh active mirror instance for each observation. When neither tracking_uri nor artifact_location is set, the local SDK publisher default is a SQLite MLflow store and artifact directory beside the OCLP record directory.

The adapter mirrors canonical OCLP record JSON, UUID tags, typed Execution parameters, and top-level numeric Evidence details. Execution parameter keys include their OCLP Execution UUID because MLflow parameters are immutable within one MLflow run while a Computation may run repeatedly with different arguments. It mirrors model payloads by default. OCLP is always authoritative: the adapter never reads MLflow to create, validate, or query OCLP provenance. Each mirrored payload is stored beneath its OCLP Artifact UUID, so repeated same-named outputs (for example, temporal-fold models) remain distinct in the MLflow artifact store.

@mlflow and MlflowMetrics¶

Use @mlflow outside @computation to keep MLflow presentation policy next to the output it projects:

from oclp import JsonArtifact, MlflowMetrics, computation, mlflow

@mlflow(
    metrics=(
        MlflowMetrics(
            output_port="metrics",
            prefix="fold",
            dimensions=("fold_number",),
        ),
    ),
)
@computation(
    name="Train fold",
    outputs={"metrics": JsonArtifact(name="Fold metrics")},
)
def train_fold(*, fold_number: int) -> dict[str, object]:
    return {"metrics": {"rmse": 42.1}}

The selected output must be an application/json Artifact. The adapter reads its verified bytes and logs only top-level numeric scalar fields. dimensions must name declared scalar Execution parameters and prevent repeated invocations from producing the same MLflow key. An attempted collision is an adapter failure: it follows the normal Diagnostic behavior unless strict=True.

@mlflow(payloads=("report",)) similarly mirrors a named non-model output payload. Payload names are exact output ports, not artifact display names. Model payloads always mirror; canonical OCLP record JSON always mirrors. The declaration is inert when the active run has no MlflowAdapter, so the same Computation works unchanged in environments without MLflow. Execution parameters already mirror automatically, so the decorator is needed only for metric projections and additional payloads.

Workflow-level MLflow parameters¶

Use the same decorator outside @run when a small, explicit set of workflow arguments should also appear in MLflow. The mapping is MLflow parameter name to workflow argument name:

from oclp import MlflowAdapter, mlflow, run

@mlflow(
    run_parameters={
        "release_id": "release_id",
        "temporal_fold_count": "fold_count",
    },
)
@run(
    name="Daily training",
    adapters=(MlflowAdapter(experiment_name="daily-training"),),
)
def train(*, release_id: str, fold_count: int) -> None:
    ...

The adapter logs these as workflow.release_id and workflow.temporal_fold_count, and tags them with oclp.mlflow.workflow_parameter.<name>=true. This namespace keeps them separate from the automatically mirrored, Execution-UUID-scoped OCLP parameters. It also makes their provenance clear: they are application-chosen workflow context, not Core Execution parameters.

Every selected argument must be JSON-compatible. The decorator rejects unknown argument names when the module loads. A conflicting value for the same workflow.* key within one MLflow run follows the normal adapter policy: it becomes an integration Diagnostic by default and raises when MlflowAdapter(strict=True) is configured. The declaration is inert without an MLflow adapter and never creates an OCLP record, Execution, or derivation edge.

run_parameters belongs only on @mlflow outside @run; metrics and payloads belong only on @mlflow outside @computation.

After the adapter is declared, an application can retrieve its active session with observed.adapter(MlflowAdapter) and deliberately add domain-specific comparison values without recreating OCLP reference wiring:

mlflow = observed.adapter(MlflowAdapter)
mlflow.log_metrics({"validation_rmse": 42.1})
mlflow.log_parameters({"candidate_family": "CatBoostRegressor"})

Use this only for values whose MLflow presentation is an application decision; the adapter already mirrors OCLP Execution parameters and Evidence records.

By default, an MLflow error does not undo OCLP publication. The runtime emits an adapter-failed Event with an integration Diagnostic on the next real Execution. Use strict=True to make the mirror failure fail the application workflow. For explicit registry publication, nominate the exact Artifact by its application-owned name:

from oclp import MlflowAdapter, MlflowModelRegistration

adapter = MlflowAdapter(
    experiment_name="daily-training",
    model_registration=MlflowModelRegistration(
        artifact_name="Validated model",
        registered_model_name="demand-model",
    ),
)

Registration is opt-in; the adapter does not infer promotion policy.

For scoped observation that is not a batch run, use:

with OclpRun(publisher=publisher, source=source) as observed:
    ...

active_run() returns that context or None.

@artifact_set¶

Use @artifact_set for a package assembled from outputs of one or several child computations. Apply it outside the @computation that emits the members. A single members mapping keeps related outputs from one computation together:

@artifact_set(
    name="Validated release",
    members={
        "model": ("model", "model"),
        "metrics": ("metrics", "validation-report"),
    },
)
@computation(
    name="Train model",
    outputs={
        "model": CatBoostModelArtifact(name="Validated model"),
        "metrics": JsonArtifact(name="Validation metrics"),
    },
)
def train_model(...): ...


@artifact_set(
    name="Validated release",
    output_port="features",
    role="training-data",
)
@computation(
    name="Prepare features",
    outputs={"features": CsvArtifact(name="Training features")},
)
def prepare_features(...): ...


@run(
    name="Daily training",
)
def train(...): ...

Decorator order¶

@computation must be closest to the function. Both @artifact_set and @mlflow read its declared output ports, so they must be applied outside it. When a Computation uses both, their order relative to one another does not matter:

@artifact_set(name="Validated release", output_port="metrics")
@mlflow(metrics=(MlflowMetrics(output_port="metrics", prefix="validation"),))
@computation(...)
def evaluate_model(...): ...

Python applies decorators from the bottom up. Therefore placing @computation outside either declaration decorator fails because the declaration would run before a Computation template exists.

The SDK resolves the local declarations only after a successful observed context. Retrieve the resulting ArtifactSetHandle with observed.artifact_set("Validated release"). Each member must materialize exactly once. The members keys are the member names; the concise output_port= form uses the output port as its default member name, with member_name= available for a one-output override. Use the dynamic API below when members cannot be known beforehand. The SDK creates a durable package representation automatically, using the same name as the ArtifactSet.

ArtifactSetHandle.member(name) returns a member handle; load_member(name, target_type) verifies and adapts it. Use load_release_manifest(path) for a materialized release manifest.

output_artifact_id(port)¶

Within an observed computation body, use this only when a payload must include the UUID of the Artifact that will describe it:

return {"response_id": output_artifact_id("response"), "score": score}

The SDK reserves that UUID before executing the body and publishes the output with exactly that identity. Most outputs do not need this helper.

Sources¶

source = source_from_git_checkout(project_root, path="src/model")

This returns GitSource, including dirty: true when applicable. To capture the exact dirty diff as a reproducible source basis, call:

source = capture_git_source_overlay(
    project_root,
    source=source,
    publisher=publisher,
    name="Source overlay",
    relative_path="source-overlays/current",
    untracked_files=(),
)

The helper publishes git diff HEAD as a patch Artifact and binds its ArtifactSet through GitSource.overlay. It refuses to capture untracked files until the application explicitly selects them.

Use Core ServiceSource for a deployed service rather than a Git checkout, and OpaqueSource only when no portable source model applies.

Publishing and local catalog¶

LocalArtifactPublisher¶

Import it from oclp.publishing:

publisher = LocalArtifactPublisher(
    catalog_path=Path("data/oclp.duckdb"),
    record_root=Path("data/oclp/record"),
    payload_root=Path("data/runs"),
)

PublishedArtifact contains artifact, path, and reference. utc_now() returns a timestamp suitable for record fields.

DuckdbCatalog¶

Import DuckdbCatalog from oclp.catalog.duckdb. It is a local, rebuildable index and resolver—not the authoritative record store.

The catalog raises RecordNotFoundError, AmbiguousRecordReferenceError, CatalogIntegrityError, and CatalogResolutionError for resolution failures.

Canonicalization and validation¶

from oclp import (
    canonical_json_bytes,
    parse_record,
    record_digest,
    validate_derivation_graph,
    validate_execution_acceptance,
    validate_execution_hierarchy,
)

Validation raises DerivationValidationError, OrchestrationValidationError, AcceptanceValidationError, or ParameterValidationError.

Compatibility boundary¶

Do not depend on names prefixed with _, decorator attributes such as __oclp_computation_template__, local record-file paths, or DuckDB table layouts. They are implementation details. Depend on the APIs documented here, Core records, and the explicit oclp.catalog.duckdb.DuckdbCatalog integration point.