Manifest

The otter.manifest package contains the manifest models and utilities. The manifest file is a way to provide tracking inforamtion on a run of otter.

See also

Refer to the Manifest Specification for more information, although the schema has changed slightly since the blog post.

Useful JQ queries

TODO

manifest.model module

Manifest data models.

class otter.manifest.model.Result(value)

Bases: StrEnum

Result enumeration.

The result of a Task, Step or the whole set of steps. Used in the manifest

to track the status of the run.

See also

TaskManifest, StepManifest and:class:RootManifest.

PENDING = 'pending'

SUCCESS = 'success'

FAILURE = 'failure'

ABORTED = 'aborted'

class otter.manifest.model.Artifact(*, source: str | list[str], destination: str | list[str])

Bases: BaseModel

Artifact model.

An Artifact is the resulting product of a task. Tasks can produce none, one or more artifacts. It is not necesarily a file, it can be a folder or a database.

The Artifact class can be subclassed to add additional fields that better describe things.

Artifacts must have a source and a destination, which can be used to track them through the flow of the pipeline.

source: str | list[str]

The sources of the resource.

destination: str | list[str]

The destinations of the resource.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class otter.manifest.model.TaskManifest(*, name: str, result: Result = Result.PENDING, started_run_at: datetime | None = None, finished_run_at: datetime | None = None, started_validation_at: datetime | None = None, finished_validation_at: datetime | None = None, log: list[str] = [], artifacts: list[Artifact] = [], failure_reason: str | None = None, **extra_data: Any)

Bases: BaseModel

Model representing a task in a step of the manifest.

name: str

result: Result

started_run_at: datetime | None

finished_run_at: datetime | None

started_validation_at: datetime | None

finished_validation_at: datetime | None

log: list[str]

artifacts: list[Artifact]

failure_reason: str | None

property run_elapsed: float | None

Calculate the elapsed time for the run.

property validation_elapsed: float | None

Calculate the elapsed time for the validation.

property elapsed: float | None

Calculate the elapsed time.

model_config = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class otter.manifest.model.StepManifest(*, name: str, result: Result = Result.PENDING, started_run_at: datetime | None = None, finished_run_at: datetime | None = None, log: list[str] = [], tasks: list[TaskManifest] = [], artifacts: list[Artifact] = [], failure_reason: str | None = None)

Bases: BaseModel

Model representing a step in the manifest.

name: str

result: Result

started_run_at: datetime | None

finished_run_at: datetime | None

log: list[str]

tasks: list[TaskManifest]

artifacts: list[Artifact]

failure_reason: str | None

property elapsed: float | None

Calculate the elapsed time.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class otter.manifest.model.RootManifest(*, result: Result = Result.PENDING, started_at: datetime = datetime.datetime(2026, 2, 23, 13, 17, 43, 478063, tzinfo=datetime.timezone.utc), modified_at: datetime = datetime.datetime(2026, 2, 23, 13, 17, 43, 478069, tzinfo=datetime.timezone.utc), log: list[str] = [], steps: dict[str, StepManifest] = {})

Bases: BaseModel

Model representing the root of the manifest.

result: Result

started_at: datetime

modified_at: datetime

log: list[str]

steps: dict[str, StepManifest]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class otter.manifest.model.Manifest(config: Config)

Bases: object

Manifest class that handles operations.

This class wraps a RootManifest and provides atomic read-modify-write operations with optimistic locking for concurrent access.

update(step_manifest: StepManifest) → None

Update manifest with the given step and save it.

Parameters:

step_manifest (StepManifest) – The otter.manifest.model.StepManifest to update in the manifest.

Raises:

ManifestError – If an error occurs during the update

Module contents

Manifest module.