Interfaces (Protocol APIs)¶

A seam is a place you plug your own code in. Ubunye has five, for anyone who wants to keep models, run records, deploys or credentials somewhere Ubunye does not ship.

Each seam is a Python typing.Protocol: any class with the right methods fits, with no base class to inherit. Your class is found through an entry point (a name your package registers in its pyproject.toml).

Protocol	What you plug in	Entry point group	Chosen by
`DeployAdapter`	A deploy adapter: sends a task to a remote platform	`ubunye.deploy_adapters`	The deploy command
`RegistryBackend`	A registry backend: where model versions and their stages live	`ubunye.registry_backends`	Name, then `UBUNYE_REGISTRY_BACKEND`, then `filesystem`
`RunRecordStore`	A run record store: where run records go	`ubunye.lineage_backends`	`--lineage-backend`, then `UBUNYE_LINEAGE_BACKEND`, then `filesystem`
`LineageBackend`	The older lineage backend contract, kept for code that uses it	`ubunye.lineage_backends`	As for the run record store
`AuthBackend`	An auth backend: finds and checks Databricks credentials	`ubunye.auth_backends`	The credentials that are set
`RunIdentityProvider`	A run identity provider: who and where ran a task, and the job parameters a platform passes	`ubunye.run_identity`	All are asked, by `priority`; the first that finds its platform wins

These are not the same as an execution backend (spark, pandas, the engine a task runs on). That seam is a base class, not a protocol, and has its own page: Execution backends. Readers, writers and transforms are plugins too: Writing a plugin.

Design principles¶

Three decisions shape how every seam behaves.

Decision 1: Non-blocking metadata writes¶

record(), register(), and promote() send their metadata updates to a background worker thread with a bounded queue (default 1 000, UBUNYE_METADATA_QUEUE_SIZE).

Queue overflow: drop the oldest pending write, log a warning.
Shutdown: flush() drains the queue, with a timeout (default 30 s, UBUNYE_METADATA_FLUSH_TIMEOUT).
Artifact writes (model files) stay synchronous: the method returns only after the model is safely stored.

It uses a worker thread, not async/await, because async fights Spark's threading model in Python.

Decision 2: Graceful degradation¶

When a metadata write fails, the run goes on and the record is kept in a local file to replay later. See Fallback manifests.

Auth is excluded: authentication failures always stop the run at once. A pipeline must not run with invalid credentials.

Decision 3: Schema evolution¶

Every dataclass that crosses a seam separates strict core fields from a flexible metadata: Dict[str, str]:

Core fields never change without a major engine version bump.
New data goes into metadata: old clients ignore unknown keys, new clients write new keys without touching the schema.
Every record stamps engine_version so readers can filter by writer version.

Protocols¶

DeployAdapter¶

A deploy adapter sends a task's config and artifacts to a remote environment.

from ubunye.interfaces import DeployAdapter, DeployContext, DeployResult

Bases: Protocol

Structural interface for deployment backends.

Each adapter is responsible for:

Validating that the target dict contains the fields it needs.
Preparing artifacts (notebooks, config bundles) for the target.
Uploading artifacts and creating/updating the remote job.
Returning a :class:DeployResult with job metadata.

name `property` ¶

name

Short identifier for this adapter (e.g. "databricks").

deploy ¶

deploy(ctx, target, *, dry_run=False)

Deploy a task to the target environment.

Parameters¶

ctx: Task metadata, config dict, and filesystem paths. target: Adapter-specific target configuration (validated by the adapter via :meth:validate_target). dry_run: If True, build and return the job spec without making any remote API calls.

Returns¶

DeployResult Job metadata including name, id, spec, and workspace path.

get_job_url ¶

get_job_url(job_id, target)

Return a browser URL to the deployed job.

Parameters¶

job_id: The job identifier returned by :meth:deploy. target: The same target dict, which typically contains the host URL.

validate_target ¶

validate_target(target)

Validate adapter-specific target config.

Raises¶

ubunye.core.errors.TargetNotFoundError When required keys are missing or values are invalid.

ubunye.interfaces.deploy.DeployContext `dataclass` ¶

Structured metadata for a deploy operation.

Callers construct this from CLI flags and the parsed config; the adapter never derives task identity from filesystem paths.

ubunye.interfaces.deploy.DeployResult `dataclass` ¶

Returned by :meth:DeployAdapter.deploy.

metadata carries adapter-specific extras (notebook source, cluster id, etc.) without polluting the core fields.

Built-in: DatabricksDeployAdapter, entry point ubunye.deploy_adapters:databricks.

RegistryBackend¶

A registry backend manages model versions: register, promote, demote, delete.

from ubunye.interfaces import RegistryBackend, ModelVersionInfo

Bases: Protocol

Structural interface for model registries.

Concrete implementations: FilesystemRegistryBackend (built-in), MLflowModelRegistryBackend (mlflow, ADR 014).

delete ¶

delete(*, use_case, model_name, version)

Remove a version and its artifacts.

demote ¶

demote(*, use_case, model_name, version, to_stage)

Demote a version to a lower stage.

Never move a version up: refuse (RegistryError) a move to production, or to staging from below. Call :func:ubunye.models.lifecycle.check_demote first.

get ¶

get(*, use_case, model_name, version=None, stage=None)

Retrieve a model version by exact version or active stage.

Returns¶

tuple of (artifact_path, ModelVersionInfo) artifact_path is an opaque string — it may be a local path, an MLflow artifact URI, or a cloud storage URL. Callers pass it to ModelClass.load(artifact_path).

Raises¶

ubunye.core.errors.VersionNotFoundError When no matching version exists.

list_versions ¶

list_versions(*, use_case, model_name)

List all registered versions, newest first.

promote ¶

promote(*, use_case, model_name, version, to_stage, gates=None, promoted_by=None)

Promote a version to a higher lifecycle stage.

If gates are provided, all checks must pass before the transition. On gate failure, raises :class:~ubunye.core.errors.PromotionBlockedError — this is NOT caught by the graceful degradation layer (auth and gate failures must propagate).

Metadata update is non-blocking per Decision 1.

register ¶

register(*, use_case, model_name, version, model, metrics, metadata=None, lineage_run_id=None)

Register a trained model as a new version.

Artifact writes are synchronous — the method returns only after the model is safely persisted. Index/metadata updates are non-blocking per Decision 1.

Parameters¶

use_case: Logical namespace (e.g. "fraud_detection"). model_name: Model identifier within the use case. version: Explicit semver string, or None for auto-increment. model: A trained model instance whose save()/metadata() methods will be called by the backend. metrics: Training metrics dict. metadata: Optional flexible key-value pairs (Decision 3). lineage_run_id: Optional lineage run ID for cross-referencing.

ubunye.interfaces.registry.ModelVersionInfo `dataclass` ¶

Cross-boundary model version record (Decision 3 compliant).

Core fields are strict; everything else goes into metadata. Every record stamps engine_version so downstream consumers can filter by writer version.

Built-in:

FilesystemRegistryBackend, entry point ubunye.registry_backends:filesystem. Keeps models and their stages in a folder.
MLflowModelRegistryBackend, entry point ubunye.registry_backends:mlflow. Keeps them in the MLflow Model Registry: each Ubunye version is a registered model version, and staging and production are MLflow aliases. Needs pip install 'ubunye-engine[ml]'. See ADR 014.

MLflowRegistryBackend (a folder, with runs logged to MLflow on top, best effort, never blocking the pipeline) is deprecated since 0.9 and still works; use MLflowModelRegistryBackend.

Which one is used: the name you give (ubunye models --backend, or backend: in the config's registry block), else UBUNYE_REGISTRY_BACKEND, else filesystem. Ubunye does not guess from the host: on Databricks, set UBUNYE_REGISTRY_BACKEND=mlflow if you want MLflow. Why: ADR 010.

RunRecordStore¶

A run record store keeps run records (the whole RunContext, what lineage show prints). Every store in ubunye.lineage_backends provides it. No base class needed.

from ubunye.interfaces import RunRecordStore

Bases: Protocol

Where run records live (ADR 011). Four methods, no base class needed.

A task is named by its task_path, "usecase/package/task". Newest first means the record written last comes first. A store is built with one argument, its location (a directory, a table name, a URL), given unchanged.

list_runs ¶

list_runs(task_path, n=20)

The n newest records of one task, newest first.

load ¶

load(task_path, run_id)

One record, every field as saved.

Raises :class:ubunye.core.errors.LineageRecordNotFoundError when the task has no run with that id.

save ¶

save(ctx)

Create or replace the record of ctx.run_id for its task.

search ¶

search(task_path=None, status=None, since=None, limit=None)

Records newest first, filtered by task (or its first parts), status and started_at >= since, at most limit of them.

Built-in:

FileSystemLineageStore, entry point ubunye.lineage_backends:filesystem. One JSON file per run. Uses a run_id to path index for fast lookups and an in-memory RunContext cache to avoid parsing JSON again on search.
DeltaLineageBackend, entry point ubunye.lineage_backends:delta. Stores each run record whole (JSON plus index columns) in a Delta table or path, through the active SparkSession or a local one it starts. Like every store, a record it cannot write goes to a fallback manifest, with a warning.

Which one is used: --lineage-backend (or lineage_backend= in Python), else UBUNYE_LINEAGE_BACKEND, else filesystem. Where each keeps its records: Where run records are stored. Why: ADR 011.

Prove a store keeps the contract with the shipped suite:

import pytest

from ubunye.testing.lineage_conformance import LineageStoreConformance


class TestMyStore(LineageStoreConformance):
    @pytest.fixture
    def store(self, tmp_path):
        return MyStore(str(tmp_path / "runs"))

LineageBackend¶

The older lineage contract: record, search, compact, over a flat LineageRecord. Kept for code that uses it; no run record is stored through it, since a LineageRecord keeps only a few fields of one.

from ubunye.interfaces import LineageBackend, LineageRecord

Bases: Protocol

Structural interface for lineage storage backends.

Concrete implementations: FilesystemLineageBackend (built-in), DeltaLineageBackend (Phase 2).

compact ¶

compact()

Optimize the underlying storage.

For filesystem backends this is a no-op. For Delta backends this triggers OPTIMIZE on the lineage table.

get_run ¶

get_run(run_id)

Load a single run record by ID.

Raises¶

ubunye.core.errors.LineageRecordNotFoundError When no record exists for run_id.

record ¶

record(record)

Persist a lineage record (non-blocking per Decision 1).

Creates or overwrites the record for the given run_id. Typically called twice per task: once at start (status="running") and once at end (status="success" or "error").

The actual write is dispatched to a background worker thread. On failure, the record is written to a fallback manifest (Decision 2).

Parameters¶

record: The lineage record to persist. Callers construct this with the current engine version and task metadata.

search ¶

search(*, task=None, usecase=None, pipeline=None, status=None, since=None, limit=100)

Search across recorded runs with optional filters.

Parameters¶

task: Filter by task name. usecase: Filter by use case. pipeline: Filter by pipeline. status: Filter by status ("success", "error", "running"). since: ISO-8601 datetime; only return runs recorded at or after. limit: Maximum number of records to return.

ubunye.interfaces.lineage.LineageRecord `dataclass` ¶

Cross-boundary lineage record (Decision 3 compliant).

Core columns are strict; everything else goes into metadata. Every record stamps engine_version so analysts can filter by writer version.

Built-in: the same two classes as the run record store. The Delta one's older LineageRecord methods still use JSON files when no SparkSession existed when it was built, and log that.

A store's save raises when it cannot write. The recorder then keeps the record in the fallback manifest, with a warning naming the ubunye sync lineage command that replays it. This is the same for every store.

AuthBackend¶

An auth backend finds and checks Databricks credentials.

from ubunye.interfaces import AuthBackend, Credentials

Bases: Protocol

Structural interface for authentication backends.

Concrete implementations: TokenAuthBackend (built-in), ServicePrincipalAuthBackend (Phase 2).

resolve ¶

resolve(workspace_url, **kwargs)

Resolve credentials for the given workspace.

Parameters¶

workspace_url: The Databricks (or other) workspace host URL. **kwargs: Backend-specific hints (e.g. token_env="MY_TOKEN").

Returns¶

Credentials Populated credentials ready for client construction.

Raises¶

ubunye.core.errors.AuthNotFoundError When credentials cannot be located (missing env var, etc.).

validate ¶

validate(credentials)

Validate that credentials are accepted by the workspace.

Returns True on success.

Raises¶

ubunye.core.errors.AuthInvalidError When the workspace rejects the credentials.

ubunye.interfaces.auth.Credentials `dataclass` ¶

Resolved authentication credentials.

The auth_type field tells callers which fields are populated. metadata carries backend-specific extras (e.g. token expiry, scope, tenant ID) without polluting the core fields.

Built-in:

TokenAuthBackend, entry point ubunye.auth_backends:token. Reads the token from DATABRICKS_TOKEN (or the env var named by token_env in targets.yaml).
ServicePrincipalAuthBackend, entry point ubunye.auth_backends:service_principal. Uses DATABRICKS_CLIENT_ID + DATABRICKS_CLIENT_SECRET for OAuth M2M auth.

Which one is used:

Set in the environment	Auth backend
`DATABRICKS_CLIENT_ID` and `DATABRICKS_CLIENT_SECRET`	`service_principal` (wins when both kinds are set)
`DATABRICKS_TOKEN`	`token`
Neither	Raises `AuthNotFoundError`

See Databricks authentication.

RunIdentityProvider¶

Says who and where ran a task (platform, workspace, job, job run, task, user, git commit and branch, deploy target), and reads the job parameters its platform passes (on Databricks, the notebook widgets). The result goes on every run record as environment["identity"]. See ADR 015.

Bases: Protocol

What a ubunye.run_identity plugin provides (see the module docstring).

A provider is built with no arguments. Both methods return {} when it is not on its platform; one that raises is logged and skipped, never failing a run. The first provider (lowest priority) that returns a platform owns the run fields; only git_commit and git_branch may come from a later one.

Built-in:

DatabricksIdentity (databricks, priority 10): job parameters named like a field (git_commit, git_branch, deploy_target, job_run_id, ...), then the notebook context. Imports nothing from Databricks until called.
CiIdentity (ci, priority 20): GitHub Actions, GitLab CI, Azure Pipelines and a generic CI=true, from the variables they set.
LocalIdentity (local, priority 90): user, host name, and git rev-parse in the task folder.

The job settings (ubunye.core.settings.job_settings) read their job parameters through these providers. See Databricks: job parameters.

Entry point groups¶

Every group the engine reads, with what it ships. Third-party packages add their own classes under these groups in their pyproject.toml.

Group	What it holds	Built-in names
`ubunye.deploy_adapters`	Deploy adapters	`databricks` (`DatabricksDeployAdapter`)
`ubunye.registry_backends`	Registry backends	`filesystem` (`FilesystemRegistryBackend`), `mlflow` (`MLflowModelRegistryBackend`)
`ubunye.lineage_backends`	Run record stores	`filesystem` (`FileSystemLineageStore`), `delta` (`DeltaLineageBackend`)
`ubunye.auth_backends`	Auth backends	`token` (`TokenAuthBackend`), `service_principal` (`ServicePrincipalAuthBackend`)
`ubunye.run_identity`	Run identity providers	`databricks` (`DatabricksIdentity`), `ci` (`CiIdentity`), `local` (`LocalIdentity`)
`ubunye.artifact_stores`	Where model files go, by URL scheme	`file` (`LocalArtifactStore`); a third party adds a scheme here
`ubunye.backends`	Execution backends (Execution backends)	`spark`, `databricks`, `pandas`
`ubunye.readers`	Readers (`format:` of an input)	`hive`, `jdbc`, `s3`, `delta`, `binary`, `unity`, `rest_api`
`ubunye.writers`	Writers (`format:` of an output)	`s3`, `delta`, `hive`, `jdbc`, `unity`, `rest_api`
`ubunye.transforms`	Transforms	`noop`, `model`
`ubunye.ml`	Model wrappers	`sklearn`, `sparkml`
`ubunye.monitors`	Monitors (older than hooks)	`mlflow`, `lineage`
`ubunye.hooks`	Hooks, always on	`events`, `otel`, `prometheus`
`ubunye.secrets`	Secret providers for `secret://` (Secrets)	`env`, `file`, `databricks`, `aws-sm`, `gcp-sm`, `azure-kv`
`ubunye.llm_backends`	Model call providers (LLM calls)	`anthropic`, `openai_compatible`, `databricks_serving`

How to write and register readers, writers and transforms: Writing a plugin.

Fallback manifests and `ubunye sync`¶

When a metadata write fails (a run record, or a model registry update), the run goes on, and the record is appended to a local fallback manifest:

~/.ubunye/fallback/{run_id}/{kind}.jsonl

kind is lineage or registry. Each line is a JSON object with all record fields plus a _fallback_at timestamp. The run logs a warning that says so.

Replay them once the store is back:

ubunye sync lineage                    # replay all run record manifests
ubunye sync lineage --run-id abc123    # replay one run
ubunye sync lineage --lineage-backend delta --lineage-dir main.ops.ubunye_runs
ubunye sync registry                   # replay all registry manifests

Replay skips records already written (the key is run_id + task + recorded_at), so running it twice is safe. Replayed manifests are moved to ~/.ubunye/fallback/synced/. Every flag: ubunye sync.

Writing your own¶

Implement the protocol methods (no base class needed).

Register the entry point in your pyproject.toml:

[project.entry-points."ubunye.lineage_backends"]
my_store = "my_package.lineage:MyRunRecordStore"

Install the package (pip install -e .).
The engine finds it by name. A run record store is then used with --lineage-backend my_store (or UBUNYE_LINEAGE_BACKEND=my_store), and built with one argument, --lineage-dir as given:
```
from ubunye.lineage import stores

store = stores.open_store("my-location", "my_store")
```

A registry backend is used with ubunye models --backend my_registry (or UBUNYE_REGISTRY_BACKEND=my_registry).

Interfaces (Protocol APIs)¶

Design principles¶

Decision 1: Non-blocking metadata writes¶

Decision 2: Graceful degradation¶

Decision 3: Schema evolution¶

Protocols¶

DeployAdapter¶

name property ¶

deploy ¶

Parameters¶

Returns¶

get_job_url ¶

Parameters¶

validate_target ¶

Raises¶

ubunye.interfaces.deploy.DeployContext dataclass ¶

ubunye.interfaces.deploy.DeployResult dataclass ¶

RegistryBackend¶

delete ¶

demote ¶

get ¶

Returns¶

Raises¶

list_versions ¶

promote ¶

register ¶

Parameters¶

ubunye.interfaces.registry.ModelVersionInfo dataclass ¶

RunRecordStore¶

list_runs ¶

load ¶

save ¶

search ¶

LineageBackend¶

compact ¶

get_run ¶

Raises¶

record ¶

Parameters¶

search ¶

Parameters¶

ubunye.interfaces.lineage.LineageRecord dataclass ¶

AuthBackend¶

resolve ¶

Parameters¶

Returns¶

Raises¶

validate ¶

Raises¶

ubunye.interfaces.auth.Credentials dataclass ¶

RunIdentityProvider¶

Entry point groups¶

Fallback manifests and ubunye sync¶

Writing your own¶

name `property` ¶

ubunye.interfaces.deploy.DeployContext `dataclass` ¶

ubunye.interfaces.deploy.DeployResult `dataclass` ¶

ubunye.interfaces.registry.ModelVersionInfo `dataclass` ¶

ubunye.interfaces.lineage.LineageRecord `dataclass` ¶

ubunye.interfaces.auth.Credentials `dataclass` ¶

Fallback manifests and `ubunye sync`¶