manifest package

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

manifest.manifest module

Manifest class for managing the manifest file.

class pis.manifest.manifest.Manifest[source]

Manifest class for managing the manifest file.

complete()[source]: Close the manifest and save it.

is_completed() → bool[source]

Return whether the manifest is completed.

Note a successful run is defined as all steps in the manifest being in validated state if the run is local, or in complete state if the run has a remote URI.

Returns:: Whether the manifest is completed.
Return type:: bool

run_ok() → bool[source]

Return whether the run was successful.

Note a successful run is defined as the step being in validated state if the run is local, or in complete state if the run has a remote URI.

Returns:: Whether the run was successful.
Return type:: bool

update_step(step: Step)[source]

Update the manifest with the step.

Parameters:: step (Step) – The step to update the manifest with.

manifest.models module

Manifest data models.

pydantic model pis.manifest.models.Resource[source]

Bases: BaseModel

Resource model.

A resource is the resulting file or files from a task. All tasks must produce a resource, and it will be logged into the manifest for tracking.

Show JSON schema

{
   "title": "Resource",
   "description": "Resource model.\n\nA resource is the resulting file or files from a task. All tasks must produce a\nresource, and it will be logged into the manifest for tracking.",
   "type": "object",
   "properties": {
      "source": {
         "title": "Source",
         "type": "string"
      },
      "destination": {
         "title": "Destination",
         "type": "string"
      }
   },
   "required": [
      "source",
      "destination"
   ]
}

Fields:

destination (str)
source (str)

field destination: str [Required]

field source: str [Required]

make_absolute() → Resource[source]: Make the destination path absolute.

class pis.manifest.models.Result(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: StrEnum

Result enumeration.

The result of a task or step. See TaskManifest and StepManifest.

ABORTED = 'aborted'

COMPLETED = 'completed'

FAILED = 'failed'

PENDING = 'pending'

STAGED = 'staged'

VALIDATED = 'validated'

pydantic model pis.manifest.models.RootManifest[source]

Bases: BaseModel

Model for the root of the manifest.

Show JSON schema

{
   "title": "RootManifest",
   "description": "Model for the root of the manifest.",
   "type": "object",
   "properties": {
      "result": {
         "$ref": "#/$defs/Result",
         "default": "pending"
      },
      "created": {
         "default": "2025-01-07T10:19:49.581222Z",
         "format": "date-time",
         "title": "Created",
         "type": "string"
      },
      "modified": {
         "default": "2025-01-07T10:19:49.581224Z",
         "format": "date-time",
         "title": "Modified",
         "type": "string"
      },
      "log": {
         "default": [],
         "items": {
            "type": "string"
         },
         "title": "Log",
         "type": "array"
      },
      "steps": {
         "additionalProperties": {
            "$ref": "#/$defs/StepManifest"
         },
         "default": {},
         "title": "Steps",
         "type": "object"
      }
   },
   "$defs": {
      "Resource": {
         "description": "Resource model.\n\nA resource is the resulting file or files from a task. All tasks must produce a\nresource, and it will be logged into the manifest for tracking.",
         "properties": {
            "source": {
               "title": "Source",
               "type": "string"
            },
            "destination": {
               "title": "Destination",
               "type": "string"
            }
         },
         "required": [
            "source",
            "destination"
         ],
         "title": "Resource",
         "type": "object"
      },
      "Result": {
         "description": "Result enumeration.\n\nThe result of a task or step. See :class:`TaskManifest` and :class:`StepManifest`.",
         "enum": [
            "pending",
            "staged",
            "validated",
            "completed",
            "failed",
            "aborted"
         ],
         "title": "Result",
         "type": "string"
      },
      "StepManifest": {
         "description": "Model for a step in the manifest.",
         "properties": {
            "name": {
               "title": "Name",
               "type": "string"
            },
            "result": {
               "$ref": "#/$defs/Result",
               "default": "pending"
            },
            "created": {
               "default": "2025-01-07T10:19:49.579770Z",
               "format": "date-time",
               "title": "Created",
               "type": "string"
            },
            "completed": {
               "anyOf": [
                  {
                     "format": "date-time",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Completed"
            },
            "elapsed": {
               "default": 0.0,
               "title": "Elapsed",
               "type": "number"
            },
            "log": {
               "default": [],
               "items": {
                  "type": "string"
               },
               "title": "Log",
               "type": "array"
            },
            "tasks": {
               "default": [],
               "items": {
                  "$ref": "#/$defs/TaskManifest"
               },
               "title": "Tasks",
               "type": "array"
            },
            "resources": {
               "default": [],
               "items": {
                  "$ref": "#/$defs/Resource"
               },
               "title": "Resources",
               "type": "array"
            }
         },
         "required": [
            "name"
         ],
         "title": "StepManifest",
         "type": "object"
      },
      "TaskManifest": {
         "additionalProperties": true,
         "description": "Model for a task in a step of the manifest.",
         "properties": {
            "name": {
               "title": "Name",
               "type": "string"
            },
            "result": {
               "$ref": "#/$defs/Result",
               "default": "pending"
            },
            "created": {
               "default": "2025-01-07T10:19:49.578446Z",
               "format": "date-time",
               "title": "Created",
               "type": "string"
            },
            "staged": {
               "default": "2025-01-07T10:19:49.578455Z",
               "format": "date-time",
               "title": "Staged",
               "type": "string"
            },
            "elapsed": {
               "default": 0.0,
               "title": "Elapsed",
               "type": "number"
            },
            "log": {
               "default": [],
               "items": {
                  "type": "string"
               },
               "title": "Log",
               "type": "array"
            },
            "definition": {
               "default": {},
               "title": "Definition",
               "type": "object"
            }
         },
         "required": [
            "name"
         ],
         "title": "TaskManifest",
         "type": "object"
      }
   }
}

Fields:

created (datetime.datetime)
log (list[str])
modified (datetime.datetime)
result (pis.manifest.models.Result)
steps (dict[str, pis.manifest.models.StepManifest])

field created: datetime = datetime.datetime(2025, 1, 7, 10, 19, 49, 581222, tzinfo=datetime.timezone.utc)

field log: list[str] = []

field modified: datetime = datetime.datetime(2025, 1, 7, 10, 19, 49, 581224, tzinfo=datetime.timezone.utc)

field result: Result = Result.PENDING

field steps: dict[str, StepManifest] = {}

pydantic model pis.manifest.models.StepManifest[source]

Bases: BaseModel

Model for a step in the manifest.

Show JSON schema

{
   "title": "StepManifest",
   "description": "Model for a step in the manifest.",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "result": {
         "$ref": "#/$defs/Result",
         "default": "pending"
      },
      "created": {
         "default": "2025-01-07T10:19:49.579770Z",
         "format": "date-time",
         "title": "Created",
         "type": "string"
      },
      "completed": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Completed"
      },
      "elapsed": {
         "default": 0.0,
         "title": "Elapsed",
         "type": "number"
      },
      "log": {
         "default": [],
         "items": {
            "type": "string"
         },
         "title": "Log",
         "type": "array"
      },
      "tasks": {
         "default": [],
         "items": {
            "$ref": "#/$defs/TaskManifest"
         },
         "title": "Tasks",
         "type": "array"
      },
      "resources": {
         "default": [],
         "items": {
            "$ref": "#/$defs/Resource"
         },
         "title": "Resources",
         "type": "array"
      }
   },
   "$defs": {
      "Resource": {
         "description": "Resource model.\n\nA resource is the resulting file or files from a task. All tasks must produce a\nresource, and it will be logged into the manifest for tracking.",
         "properties": {
            "source": {
               "title": "Source",
               "type": "string"
            },
            "destination": {
               "title": "Destination",
               "type": "string"
            }
         },
         "required": [
            "source",
            "destination"
         ],
         "title": "Resource",
         "type": "object"
      },
      "Result": {
         "description": "Result enumeration.\n\nThe result of a task or step. See :class:`TaskManifest` and :class:`StepManifest`.",
         "enum": [
            "pending",
            "staged",
            "validated",
            "completed",
            "failed",
            "aborted"
         ],
         "title": "Result",
         "type": "string"
      },
      "TaskManifest": {
         "additionalProperties": true,
         "description": "Model for a task in a step of the manifest.",
         "properties": {
            "name": {
               "title": "Name",
               "type": "string"
            },
            "result": {
               "$ref": "#/$defs/Result",
               "default": "pending"
            },
            "created": {
               "default": "2025-01-07T10:19:49.578446Z",
               "format": "date-time",
               "title": "Created",
               "type": "string"
            },
            "staged": {
               "default": "2025-01-07T10:19:49.578455Z",
               "format": "date-time",
               "title": "Staged",
               "type": "string"
            },
            "elapsed": {
               "default": 0.0,
               "title": "Elapsed",
               "type": "number"
            },
            "log": {
               "default": [],
               "items": {
                  "type": "string"
               },
               "title": "Log",
               "type": "array"
            },
            "definition": {
               "default": {},
               "title": "Definition",
               "type": "object"
            }
         },
         "required": [
            "name"
         ],
         "title": "TaskManifest",
         "type": "object"
      }
   },
   "required": [
      "name"
   ]
}

Fields:

completed (datetime.datetime | None)
created (datetime.datetime)
elapsed (float)
log (list[str])
name (str)
resources (list[pis.manifest.models.Resource])
result (pis.manifest.models.Result)
tasks (list[pis.manifest.models.TaskManifest])

field completed: datetime | None = None

field created: datetime = datetime.datetime(2025, 1, 7, 10, 19, 49, 579770, tzinfo=datetime.timezone.utc)

field elapsed: float = 0.0

field log: list[str] = []

field name: str [Required]

field resources: list[Resource] = []

field result: Result = Result.PENDING

field tasks: list[TaskManifest] = []

pydantic model pis.manifest.models.TaskManifest[source]

Bases: BaseModel

Model for a task in a step of the manifest.

Show JSON schema

{
   "title": "TaskManifest",
   "description": "Model for a task in a step of the manifest.",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "result": {
         "$ref": "#/$defs/Result",
         "default": "pending"
      },
      "created": {
         "default": "2025-01-07T10:19:49.578446Z",
         "format": "date-time",
         "title": "Created",
         "type": "string"
      },
      "staged": {
         "default": "2025-01-07T10:19:49.578455Z",
         "format": "date-time",
         "title": "Staged",
         "type": "string"
      },
      "elapsed": {
         "default": 0.0,
         "title": "Elapsed",
         "type": "number"
      },
      "log": {
         "default": [],
         "items": {
            "type": "string"
         },
         "title": "Log",
         "type": "array"
      },
      "definition": {
         "default": {},
         "title": "Definition",
         "type": "object"
      }
   },
   "$defs": {
      "Result": {
         "description": "Result enumeration.\n\nThe result of a task or step. See :class:`TaskManifest` and :class:`StepManifest`.",
         "enum": [
            "pending",
            "staged",
            "validated",
            "completed",
            "failed",
            "aborted"
         ],
         "title": "Result",
         "type": "string"
      }
   },
   "additionalProperties": true,
   "required": [
      "name"
   ]
}

Config:

extra: str = allow

Fields:

created (datetime.datetime)
definition (dict[str, Any])
elapsed (float)
log (list[str])
name (str)
result (pis.manifest.models.Result)
staged (datetime.datetime)

field created: datetime = datetime.datetime(2025, 1, 7, 10, 19, 49, 578446, tzinfo=datetime.timezone.utc)

field definition: dict[str, Any] = {}

field elapsed: float = 0.0

field log: list[str] = []

field name: str [Required]

field result: Result = Result.PENDING

field staged: datetime = datetime.datetime(2025, 1, 7, 10, 19, 49, 578455, tzinfo=datetime.timezone.utc)

manifest.step_reporter module

StepReporter class and report decorator for logging and updating steps in the manifest.

class pis.manifest.step_reporter.StepReporter(name: str)[source]

Class for logging and updating steps in the manifest.

attach_manifest(task: Task)[source]: Attach a task manifest to the step manifest.

completed(log: str)[source]: Set the step result to COMPLETED.

failed(log: str)[source]: Set the step result to FAILED.

staged(log: str)[source]: Set the step result to STAGED.

upsert_task_manifests(tasks: list[Task])[source]: Update the step manifest with new task manifests.

validated(log: str)[source]: Set the step result to VALIDATED.

pis.manifest.step_reporter.report(func)[source]: Decorator for logging and updating steps in the manifest.

manifest.task_reporter module

TaskReporter class and report decorator for logging and updating tasks in the manifest.

class pis.manifest.task_reporter.TaskReporter(name: str)[source]

Class for logging and updating tasks in the manifest.

aborted()[source]: Set the task result to ABORTED.

completed(log: str, resource: Resource | None = None)[source]: Set the task result to COMPLETED.

failed(error: Exception, where: str)[source]: Set the task result to FAILED.

staged(log: str)[source]: Set the task result to STAGED.

validated(log: str, resource: Resource | None = None)[source]: Set the task result to VALIDATED.

pis.manifest.task_reporter.report(func)[source]: Decorator for logging and updating tasks in the manifest.

Module contents

Manifest module.