maestro.api package#

Submodules#

maestro.api.index module#

maestro.api.index.ansi2html_filter(value: str) → str[source]#

Convert ANSI escape sequences to HTML.

Parameters:: value – String containing ANSI escape sequences
Returns:: HTML formatted string

async maestro.api.index.get_index(request: Request) → HTMLResponse[source]#

async maestro.api.index.get_job(request: Request, neo4j_tx: Annotated[Neo4jTransaction, Depends(provide_neo4j_transaction, use_cache=False)], job_maestro_id: Annotated[UUID, UuidVersion(uuid_version=4)]) → HTMLResponse[source]#

Get details for a specific job including its input artefacts.

Parameters:

request – FastAPI request object
neo4j_tx – Neo4j transaction dependency
job_maestro_id – maestro_id (UUID) of the job to fetch

Returns:

HTML response with job details template

async maestro.api.index.get_version() → PlainTextResponse[source]#

async maestro.api.index.get_workflow_job_count(request: Request, neo4j_tx: Annotated[Neo4jTransaction, Depends(provide_neo4j_transaction, use_cache=False)], search: Annotated[Search, Depends(from_string)]) → HTMLResponse[source]#

async maestro.api.index.get_workflow_step_artefacts(request: Request, neo4j_tx: Annotated[Neo4jTransaction, Depends(provide_neo4j_transaction, use_cache=False)], workflow_step_maestro_id: Annotated[UUID, UuidVersion(uuid_version=4)], job_name: str, job_status: Annotated[str, 'JobStatus.FAILED'], job_maestro_id: Annotated[UUID, UuidVersion(uuid_version=4)]) → HTMLResponse[source]#

async maestro.api.index.get_workflows(request: Request, neo4j_tx: Annotated[Neo4jTransaction, Depends(provide_neo4j_transaction, use_cache=False)], search: Annotated[Search, Depends(from_string)], withArtefacts: bool = False, page: int = 0, per_page: int = 100) → HTMLResponse[source]#

maestro.api.main module#

maestro.api.main.create_api(settings: Settings | None = None) → FastAPI[source]#

maestro.api.main.hot_reload_lifespan(app: FastAPI) → AsyncGenerator[None, None][source]#

maestro.api.main.pre_load_models(files: list[Path]) → None[source]#

Load BaseArtefactModel subclasses from files

This is required to register any models that are not imported in the API code, and can be used to register arbitrary models to show custom HTML for them in the UI.

Note: This runs the code in the supplied files, so they should not contain any code that has side effects

maestro.api.main.start() → None[source]#

maestro.api.post_install module#

Functions to download and compile static files after installation.

maestro.api.post_install.compile_tailwind() → None[source]#

Compile tailwind css.

If production is True, the tailwind binary will be run with the –minify flag. If production is False, the tailwind binary will be run with the –watch=always flag.

On inital run, the tailwind binary will be downloaded from the tailwindlabs/tailwindcss releases.

maestro.api.post_install.download_static_files() → None[source]#

maestro.api.post_install.maestro_post_install() → None[source]#

maestro.api.post_install.md5_from_file(file: Path) → str[source]#

maestro.api.schema module#

pydantic model maestro.api.schema.JobWithWorkflowStep[source]#

Bases: Job

Show JSON schema

{
   "title": "JobWithWorkflowStep",
   "type": "object",
   "properties": {
      "invalidated_at": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Invalidated At"
      },
      "maestro_id": {
         "description": "Unique identifier",
         "format": "uuid4",
         "title": "Maestro Id",
         "type": "string"
      },
      "additional_labels": {
         "default": [],
         "description": "Additional labels to be used in Neo4j",
         "items": {
            "type": "string"
         },
         "title": "Additional Labels",
         "type": "array"
      },
      "workflow_step": {
         "title": "Workflow Step",
         "type": "string"
      },
      "instance_id": {
         "default": "default",
         "title": "Instance Id",
         "type": "string"
      },
      "display_name": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Display Name"
      },
      "status": {
         "$ref": "#/$defs/JobStatus",
         "default": "pending"
      },
      "priority": {
         "$ref": "#/$defs/JobPriorityEnum",
         "default": 1,
         "description": "priority level for job in int"
      },
      "inputs": {
         "additionalProperties": {
            "maxItems": 2,
            "minItems": 2,
            "prefixItems": [
               {
                  "additionalProperties": true,
                  "properties": {
                     "invalidated_at": {
                        "anyOf": [
                           {
                              "format": "date-time",
                              "type": "string"
                           },
                           {
                              "type": "null"
                           }
                        ],
                        "default": null,
                        "title": "Invalidated At"
                     },
                     "maestro_id": {
                        "description": "Unique identifier",
                        "format": "uuid4",
                        "title": "Maestro Id",
                        "type": "string"
                     },
                     "additional_labels": {
                        "default": [],
                        "description": "Additional labels to be used in Neo4j",
                        "items": {
                           "type": "string"
                        },
                        "title": "Additional Labels",
                        "type": "array"
                     }
                  },
                  "title": "BaseArtefactModel",
                  "type": "object"
               },
               {
                  "type": "object"
               }
            ],
            "type": "array"
         },
         "title": "Inputs",
         "type": "object"
      },
      "params": {
         "title": "Params",
         "type": "object"
      },
      "search_metadata": {
         "items": {},
         "title": "Search Metadata",
         "type": "array"
      },
      "pending_timeout": {
         "default": 86400,
         "description": "max amount of time in seconds for which job can be in pending state",
         "title": "Pending Timeout",
         "type": "integer"
      },
      "running_timeout": {
         "default": 86400,
         "description": "max amount of time in seconds for which job can be in running state",
         "title": "Running Timeout",
         "type": "integer"
      },
      "created_at": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Created At"
      },
      "updated_at": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Updated At"
      },
      "revision": {
         "default": 0,
         "minimum": 0,
         "title": "Revision",
         "type": "integer"
      },
      "workflow_step_maestro_id": {
         "anyOf": [
            {
               "format": "uuid4",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Workflow Step Maestro Id"
      },
      "workflow_step_invalidated_at": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Workflow Step Invalidated At"
      }
   },
   "$defs": {
      "JobPriorityEnum": {
         "enum": [
            1,
            2,
            3
         ],
         "title": "JobPriorityEnum",
         "type": "integer"
      },
      "JobStatus": {
         "enum": [
            "pending",
            "queued",
            "running",
            "completed",
            "failed",
            "cancelled",
            "halted"
         ],
         "title": "JobStatus",
         "type": "string"
      }
   },
   "required": [
      "workflow_step"
   ]
}

Config:

populate_by_name: bool = True

Fields:

additional_labels ()
created_at ()
display_name ()
inputs ()
instance_id ()
invalidated_at ()
maestro_id ()
params ()
pending_timeout ()
priority ()
revision ()
running_timeout ()
search_metadata ()
status ()
updated_at ()
workflow_step ()
workflow_step_invalidated_at (datetime.datetime | None)
workflow_step_maestro_id (Annotated[uuid.UUID, pydantic.types.UuidVersion(uuid_version=4)] | None)

Validators:

deserialize_collections » all fields
neo4j_datetime_to_native » all fields
validate_input_properties » inputs
validate_workflow_step » workflow_step
validate_workflow_step_inputs » all fields

field additional_labels: list[str] = []#

Additional labels to be used in Neo4j

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field created_at: datetime | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field display_name: str | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field inputs: dict[str, tuple[SerializableBaseArtefactModelType, dict]] [Optional]#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_input_properties
validate_workflow_step_inputs

field instance_id: str = 'default'#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field invalidated_at: datetime | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field maestro_id: UUID4 [Optional]#

Unique identifier

Constraints:

uuid_version = 4

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field params: dict [Optional]#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field pending_timeout: int = 86400#

max amount of time in seconds for which job can be in pending state

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field priority: JobPriorityEnum = JobPriorityEnum.NORMAL#

priority level for job in int

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field revision: NonNegativeInt = 0#

Constraints:

ge = 0

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field running_timeout: int = 86400#

max amount of time in seconds for which job can be in running state

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field search_metadata: list [Optional]#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field status: JobStatus = JobStatus.PENDING#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field updated_at: datetime | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field workflow_step: str [Required]#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step
validate_workflow_step_inputs

field workflow_step_invalidated_at: datetime | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field workflow_step_maestro_id: Annotated[UUID, UuidVersion(uuid_version=4)] | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

classmethod class_labels() → list[str]#: List of labels to be used in Neo4j. Includes the class name, superclasses class names.

validator deserialize_collections » all fields#

Collections coming from the database are stored as JSON strings, serialized in model_dump_primitive.

This function attempts to run normal pydantic validation on all values, and if it fails, it tries to deserialize the value from a JSON string.

get_nested() → dict[str, MaestroBase]#

Returns a dictionary {field_name: value} of MaestroBase objects in the model.

These are represented in the database as relationships, e.g.

class Nested(MaestroBase):: pass
class Root(MaestroBase):: HAS: Nested

nested = Nested() obj = Root(nested=nested) obj.get_nested() -> {“nested”: nested}

This will be represented in the database as a relationship of type HAS between obj and nested.

Field alias takes precedence over field name

get_nested_iterables() → dict[str, list[MaestroBase] | tuple[MaestroBase] | set[MaestroBase]]#

Returns a dictionary {field_name: value} of lists or tuples consisting of MaestroBase object in the model.

These are represented in the database as relationships, e.g.

class Nested(MaestroBase):: pass
class Root(MaestroBase):: HAS: list[Nested]

nested1 = Nested() nested2 = Nested()

obj = Root(nested=[nested1, nested2]) obj.get_nested_iterables() -> {“nested”: [nested1, nested2]}

This will be represented in the database as a relationship of type HAS between obj and nested1, and between obj and nested2.

Field alias takes precedence over field name

classmethod get_subclass_matching_labels(labels: list[str]) → type[MaestroBase] | None#: Returns the subclass of cls that has the same labels as the list of labels

model_dump_primitive() → dict[str, str]#

Dump all ‘primitive’ fields - meaning fields not containing MaestroBase objects or iterables of MaestroBase objects.

Used for dumping the model to the database.

Field alias takes precedence over field name

validator neo4j_datetime_to_native » all fields#: Converts Neo4j DateTime objects to native Python datetime objects.

classmethod nested_fields() → Generator[tuple[str, type[MaestroBase]], None, None]#

Best effort to get which fields contain MaestroBase objects.

Supports at least the following constructions:

class Test(MaestroBase):: pass
class TestNested(MaestroBase):: test: Test optional_test1: Test | None optional_test2: Optional[Test] tuple_test1: tuple[Test, Test] tuple_test2: tuple[Test, …] list_test1: list[Test] optional_tuple_test1: Optional[tuple[Test, Test]] optional_tuple_test2: Optional[tuple[Test, …]] alias_test: Test = Field(alias=”alias_field”) -> “alias_field”, Test

Yields tuples of (alias or field_name, Test)

validator validate_input_properties » inputs#: Check that properties are valid for the artefact class

validator validate_workflow_step » workflow_step#

Check that the workflow step is a valid subclass of WorkflowStepBase

TODO: This should no longer be necessary if Job.workflow_step can be annotated as type[WorkflowStepBase] (similar to SerializableBaseArtefactModelType)

validator validate_workflow_step_inputs » all fields#: Check that all requested inputs are valid for the workflow step’s process-method

property hash: int#

Used when filtering out existing and valid jobs from workflow orders.

Computed from the attributes inputs, params, and workflow_step.

property labels: list[str]#: List of labels to be used in Neo4j. Includes the class name, superclasses class names, and additional labels if supplied.

property name: str#

property name_and_id: str#

property type: str#: Returns the class name of the object - or last additional label if supplied

property workflow_step_class: type[WorkflowStepBase]#: Utility method to get the workflow step class - since Job.workflow_step is a string

pydantic model maestro.api.schema.JobWithInputs[source]#

Bases: Job

Show JSON schema

{
   "title": "JobWithInputs",
   "type": "object",
   "properties": {
      "invalidated_at": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Invalidated At"
      },
      "maestro_id": {
         "description": "Unique identifier",
         "format": "uuid4",
         "title": "Maestro Id",
         "type": "string"
      },
      "additional_labels": {
         "default": [],
         "description": "Additional labels to be used in Neo4j",
         "items": {
            "type": "string"
         },
         "title": "Additional Labels",
         "type": "array"
      },
      "workflow_step": {
         "title": "Workflow Step",
         "type": "string"
      },
      "instance_id": {
         "default": "default",
         "title": "Instance Id",
         "type": "string"
      },
      "display_name": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Display Name"
      },
      "status": {
         "$ref": "#/$defs/JobStatus",
         "default": "pending"
      },
      "priority": {
         "$ref": "#/$defs/JobPriorityEnum",
         "default": 1,
         "description": "priority level for job in int"
      },
      "inputs": {
         "additionalProperties": {
            "maxItems": 2,
            "minItems": 2,
            "prefixItems": [
               {
                  "additionalProperties": true,
                  "properties": {
                     "invalidated_at": {
                        "anyOf": [
                           {
                              "format": "date-time",
                              "type": "string"
                           },
                           {
                              "type": "null"
                           }
                        ],
                        "default": null,
                        "title": "Invalidated At"
                     },
                     "maestro_id": {
                        "description": "Unique identifier",
                        "format": "uuid4",
                        "title": "Maestro Id",
                        "type": "string"
                     },
                     "additional_labels": {
                        "default": [],
                        "description": "Additional labels to be used in Neo4j",
                        "items": {
                           "type": "string"
                        },
                        "title": "Additional Labels",
                        "type": "array"
                     }
                  },
                  "title": "BaseArtefactModel",
                  "type": "object"
               },
               {
                  "type": "object"
               }
            ],
            "type": "array"
         },
         "title": "Inputs",
         "type": "object"
      },
      "params": {
         "title": "Params",
         "type": "object"
      },
      "search_metadata": {
         "items": {},
         "title": "Search Metadata",
         "type": "array"
      },
      "pending_timeout": {
         "default": 86400,
         "description": "max amount of time in seconds for which job can be in pending state",
         "title": "Pending Timeout",
         "type": "integer"
      },
      "running_timeout": {
         "default": 86400,
         "description": "max amount of time in seconds for which job can be in running state",
         "title": "Running Timeout",
         "type": "integer"
      },
      "created_at": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Created At"
      },
      "updated_at": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Updated At"
      },
      "revision": {
         "default": 0,
         "minimum": 0,
         "title": "Revision",
         "type": "integer"
      },
      "workflow_name": {
         "title": "Workflow Name",
         "type": "string"
      },
      "created": {
         "anyOf": [
            {
               "items": {
                  "$ref": "#/$defs/BaseArtefactModel"
               },
               "type": "array"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Created"
      },
      "formatted_pending": {
         "anyOf": [
            {
               "items": {
                  "type": "object"
               },
               "type": "array"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Formatted Pending"
      }
   },
   "$defs": {
      "BaseArtefactModel": {
         "additionalProperties": true,
         "properties": {
            "invalidated_at": {
               "anyOf": [
                  {
                     "format": "date-time",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Invalidated At"
            },
            "maestro_id": {
               "description": "Unique identifier",
               "format": "uuid4",
               "title": "Maestro Id",
               "type": "string"
            },
            "additional_labels": {
               "default": [],
               "description": "Additional labels to be used in Neo4j",
               "items": {
                  "type": "string"
               },
               "title": "Additional Labels",
               "type": "array"
            }
         },
         "title": "BaseArtefactModel",
         "type": "object"
      },
      "JobPriorityEnum": {
         "enum": [
            1,
            2,
            3
         ],
         "title": "JobPriorityEnum",
         "type": "integer"
      },
      "JobStatus": {
         "enum": [
            "pending",
            "queued",
            "running",
            "completed",
            "failed",
            "cancelled",
            "halted"
         ],
         "title": "JobStatus",
         "type": "string"
      }
   },
   "required": [
      "workflow_step",
      "workflow_name"
   ]
}

Config:

populate_by_name: bool = True

Fields:

additional_labels ()
created (list[maestro.models.BaseArtefactModel] | None)
created_at ()
display_name ()
formatted_pending (list[dict[str, Any]] | None)
inputs ()
instance_id ()
invalidated_at ()
maestro_id ()
params ()
pending_timeout ()
priority ()
revision ()
running_timeout ()
search_metadata ()
status ()
updated_at ()
workflow_name (str)
workflow_step ()

Validators:

deserialize_collections » all fields
neo4j_datetime_to_native » all fields
validate_input_properties » inputs
validate_workflow_step » workflow_step
validate_workflow_step_inputs » all fields

field additional_labels: list[str] = []#

Additional labels to be used in Neo4j

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field created: list[BaseArtefactModel] | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field created_at: datetime | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field display_name: str | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field formatted_pending: list[dict[str, Any]] | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field inputs: dict[str, tuple[SerializableBaseArtefactModelType, dict]] [Optional]#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_input_properties
validate_workflow_step_inputs

field instance_id: str = 'default'#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field invalidated_at: datetime | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field maestro_id: UUID4 [Optional]#

Unique identifier

Constraints:

uuid_version = 4

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field params: dict [Optional]#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field pending_timeout: int = 86400#

max amount of time in seconds for which job can be in pending state

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field priority: JobPriorityEnum = JobPriorityEnum.NORMAL#

priority level for job in int

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field revision: NonNegativeInt = 0#

Constraints:

ge = 0

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field running_timeout: int = 86400#

max amount of time in seconds for which job can be in running state

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field search_metadata: list [Optional]#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field status: JobStatus = JobStatus.PENDING#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field updated_at: datetime | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field workflow_name: str [Required]#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step_inputs

field workflow_step: str [Required]#

Validated by:

deserialize_collections
neo4j_datetime_to_native
validate_workflow_step
validate_workflow_step_inputs

classmethod class_labels() → list[str]#: List of labels to be used in Neo4j. Includes the class name, superclasses class names.

validator deserialize_collections » all fields#

Collections coming from the database are stored as JSON strings, serialized in model_dump_primitive.

This function attempts to run normal pydantic validation on all values, and if it fails, it tries to deserialize the value from a JSON string.

format_pending(input_class_name: str, input_filter: dict | None = None) → dict[str, Any][source]#

Format pending input details into a standardized dictionary format.

Parameters:

input_class_name – Name of the input class that is pending
input_filter – Filter criteria for the pending input

Returns:

Dictionary containing the input type and identifier criteria

get_nested() → dict[str, MaestroBase]#

Returns a dictionary {field_name: value} of MaestroBase objects in the model.

These are represented in the database as relationships, e.g.

class Nested(MaestroBase):: pass
class Root(MaestroBase):: HAS: Nested

nested = Nested() obj = Root(nested=nested) obj.get_nested() -> {“nested”: nested}

This will be represented in the database as a relationship of type HAS between obj and nested.

Field alias takes precedence over field name

get_nested_iterables() → dict[str, list[MaestroBase] | tuple[MaestroBase] | set[MaestroBase]]#

Returns a dictionary {field_name: value} of lists or tuples consisting of MaestroBase object in the model.

These are represented in the database as relationships, e.g.

class Nested(MaestroBase):: pass
class Root(MaestroBase):: HAS: list[Nested]

nested1 = Nested() nested2 = Nested()

obj = Root(nested=[nested1, nested2]) obj.get_nested_iterables() -> {“nested”: [nested1, nested2]}

This will be represented in the database as a relationship of type HAS between obj and nested1, and between obj and nested2.

Field alias takes precedence over field name

classmethod get_subclass_matching_labels(labels: list[str]) → type[MaestroBase] | None#: Returns the subclass of cls that has the same labels as the list of labels

model_dump_primitive() → dict[str, str]#

Dump all ‘primitive’ fields - meaning fields not containing MaestroBase objects or iterables of MaestroBase objects.

Used for dumping the model to the database.

Field alias takes precedence over field name

validator neo4j_datetime_to_native » all fields#: Converts Neo4j DateTime objects to native Python datetime objects.

classmethod nested_fields() → Generator[tuple[str, type[MaestroBase]], None, None]#

Best effort to get which fields contain MaestroBase objects.

Supports at least the following constructions:

class Test(MaestroBase):: pass
class TestNested(MaestroBase):: test: Test optional_test1: Test | None optional_test2: Optional[Test] tuple_test1: tuple[Test, Test] tuple_test2: tuple[Test, …] list_test1: list[Test] optional_tuple_test1: Optional[tuple[Test, Test]] optional_tuple_test2: Optional[tuple[Test, …]] alias_test: Test = Field(alias=”alias_field”) -> “alias_field”, Test

Yields tuples of (alias or field_name, Test)

validator validate_input_properties » inputs#: Check that properties are valid for the artefact class

validator validate_workflow_step » workflow_step#

Check that the workflow step is a valid subclass of WorkflowStepBase

TODO: This should no longer be necessary if Job.workflow_step can be annotated as type[WorkflowStepBase] (similar to SerializableBaseArtefactModelType)

validator validate_workflow_step_inputs » all fields#: Check that all requested inputs are valid for the workflow step’s process-method

property hash: int#

Used when filtering out existing and valid jobs from workflow orders.

Computed from the attributes inputs, params, and workflow_step.

property labels: list[str]#: List of labels to be used in Neo4j. Includes the class name, superclasses class names, and additional labels if supplied.

property name: str#

property name_and_id: str#

property type: str#: Returns the class name of the object - or last additional label if supplied

property workflow_step_class: type[WorkflowStepBase]#: Utility method to get the workflow step class - since Job.workflow_step is a string

pydantic model maestro.api.schema.WorkflowModelWithDetails[source]#

Bases: MaestroBase

Show JSON schema

{
   "title": "WorkflowModelWithDetails",
   "type": "object",
   "properties": {
      "invalidated_at": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Invalidated At"
      },
      "maestro_id": {
         "description": "Unique identifier",
         "format": "uuid4",
         "title": "Maestro Id",
         "type": "string"
      },
      "additional_labels": {
         "default": [],
         "description": "Additional labels to be used in Neo4j",
         "items": {
            "type": "string"
         },
         "title": "Additional Labels",
         "type": "array"
      },
      "workflow_name": {
         "title": "Workflow Name",
         "type": "string"
      },
      "jobs": {
         "items": {
            "$ref": "#/$defs/JobWithWorkflowStep"
         },
         "title": "Jobs",
         "type": "array",
         "uniqueItems": true
      },
      "search_metadata": {
         "items": {},
         "title": "Search Metadata",
         "type": "array"
      }
   },
   "$defs": {
      "JobPriorityEnum": {
         "enum": [
            1,
            2,
            3
         ],
         "title": "JobPriorityEnum",
         "type": "integer"
      },
      "JobStatus": {
         "enum": [
            "pending",
            "queued",
            "running",
            "completed",
            "failed",
            "cancelled",
            "halted"
         ],
         "title": "JobStatus",
         "type": "string"
      },
      "JobWithWorkflowStep": {
         "properties": {
            "invalidated_at": {
               "anyOf": [
                  {
                     "format": "date-time",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Invalidated At"
            },
            "maestro_id": {
               "description": "Unique identifier",
               "format": "uuid4",
               "title": "Maestro Id",
               "type": "string"
            },
            "additional_labels": {
               "default": [],
               "description": "Additional labels to be used in Neo4j",
               "items": {
                  "type": "string"
               },
               "title": "Additional Labels",
               "type": "array"
            },
            "workflow_step": {
               "title": "Workflow Step",
               "type": "string"
            },
            "instance_id": {
               "default": "default",
               "title": "Instance Id",
               "type": "string"
            },
            "display_name": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Display Name"
            },
            "status": {
               "$ref": "#/$defs/JobStatus",
               "default": "pending"
            },
            "priority": {
               "$ref": "#/$defs/JobPriorityEnum",
               "default": 1,
               "description": "priority level for job in int"
            },
            "inputs": {
               "additionalProperties": {
                  "maxItems": 2,
                  "minItems": 2,
                  "prefixItems": [
                     {
                        "additionalProperties": true,
                        "properties": {
                           "invalidated_at": {
                              "anyOf": [
                                 {
                                    "format": "date-time",
                                    "type": "string"
                                 },
                                 {
                                    "type": "null"
                                 }
                              ],
                              "default": null,
                              "title": "Invalidated At"
                           },
                           "maestro_id": {
                              "description": "Unique identifier",
                              "format": "uuid4",
                              "title": "Maestro Id",
                              "type": "string"
                           },
                           "additional_labels": {
                              "default": [],
                              "description": "Additional labels to be used in Neo4j",
                              "items": {
                                 "type": "string"
                              },
                              "title": "Additional Labels",
                              "type": "array"
                           }
                        },
                        "title": "BaseArtefactModel",
                        "type": "object"
                     },
                     {
                        "type": "object"
                     }
                  ],
                  "type": "array"
               },
               "title": "Inputs",
               "type": "object"
            },
            "params": {
               "title": "Params",
               "type": "object"
            },
            "search_metadata": {
               "items": {},
               "title": "Search Metadata",
               "type": "array"
            },
            "pending_timeout": {
               "default": 86400,
               "description": "max amount of time in seconds for which job can be in pending state",
               "title": "Pending Timeout",
               "type": "integer"
            },
            "running_timeout": {
               "default": 86400,
               "description": "max amount of time in seconds for which job can be in running state",
               "title": "Running Timeout",
               "type": "integer"
            },
            "created_at": {
               "anyOf": [
                  {
                     "format": "date-time",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Created At"
            },
            "updated_at": {
               "anyOf": [
                  {
                     "format": "date-time",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Updated At"
            },
            "revision": {
               "default": 0,
               "minimum": 0,
               "title": "Revision",
               "type": "integer"
            },
            "workflow_step_maestro_id": {
               "anyOf": [
                  {
                     "format": "uuid4",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Workflow Step Maestro Id"
            },
            "workflow_step_invalidated_at": {
               "anyOf": [
                  {
                     "format": "date-time",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Workflow Step Invalidated At"
            }
         },
         "required": [
            "workflow_step"
         ],
         "title": "JobWithWorkflowStep",
         "type": "object"
      }
   },
   "required": [
      "workflow_name",
      "jobs",
      "search_metadata"
   ]
}

Config:

populate_by_name: bool = True

Fields:

additional_labels ()
invalidated_at ()
jobs (set[maestro.api.schema.JobWithWorkflowStep])
maestro_id ()
search_metadata (list)
workflow_name (str)

Validators:

deserialize_collections » all fields
neo4j_datetime_to_native » all fields

field additional_labels: list[str] = []#

Additional labels to be used in Neo4j

Validated by:

deserialize_collections
neo4j_datetime_to_native

field invalidated_at: datetime | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field jobs: set[JobWithWorkflowStep] [Required]#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field maestro_id: UUID4 [Optional]#

Unique identifier

Constraints:

uuid_version = 4

Validated by:

deserialize_collections
neo4j_datetime_to_native

field search_metadata: list [Required]#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field workflow_name: str [Required]#

Validated by:

deserialize_collections
neo4j_datetime_to_native

classmethod class_labels() → list[str]#: List of labels to be used in Neo4j. Includes the class name, superclasses class names.

validator deserialize_collections » all fields#

Collections coming from the database are stored as JSON strings, serialized in model_dump_primitive.

This function attempts to run normal pydantic validation on all values, and if it fails, it tries to deserialize the value from a JSON string.

get_nested() → dict[str, MaestroBase]#

Returns a dictionary {field_name: value} of MaestroBase objects in the model.

These are represented in the database as relationships, e.g.

class Nested(MaestroBase):: pass
class Root(MaestroBase):: HAS: Nested

nested = Nested() obj = Root(nested=nested) obj.get_nested() -> {“nested”: nested}

This will be represented in the database as a relationship of type HAS between obj and nested.

Field alias takes precedence over field name

get_nested_iterables() → dict[str, list[MaestroBase] | tuple[MaestroBase] | set[MaestroBase]]#

Returns a dictionary {field_name: value} of lists or tuples consisting of MaestroBase object in the model.

These are represented in the database as relationships, e.g.

class Nested(MaestroBase):: pass
class Root(MaestroBase):: HAS: list[Nested]

nested1 = Nested() nested2 = Nested()

obj = Root(nested=[nested1, nested2]) obj.get_nested_iterables() -> {“nested”: [nested1, nested2]}

This will be represented in the database as a relationship of type HAS between obj and nested1, and between obj and nested2.

Field alias takes precedence over field name

classmethod get_subclass_matching_labels(labels: list[str]) → type[MaestroBase] | None#: Returns the subclass of cls that has the same labels as the list of labels

model_dump_primitive() → dict[str, str]#

Dump all ‘primitive’ fields - meaning fields not containing MaestroBase objects or iterables of MaestroBase objects.

Used for dumping the model to the database.

Field alias takes precedence over field name

validator neo4j_datetime_to_native » all fields#: Converts Neo4j DateTime objects to native Python datetime objects.

classmethod nested_fields() → Generator[tuple[str, type[MaestroBase]], None, None]#

Best effort to get which fields contain MaestroBase objects.

Supports at least the following constructions:

class Test(MaestroBase):: pass
class TestNested(MaestroBase):: test: Test optional_test1: Test | None optional_test2: Optional[Test] tuple_test1: tuple[Test, Test] tuple_test2: tuple[Test, …] list_test1: list[Test] optional_tuple_test1: Optional[tuple[Test, Test]] optional_tuple_test2: Optional[tuple[Test, …]] alias_test: Test = Field(alias=”alias_field”) -> “alias_field”, Test

Yields tuples of (alias or field_name, Test)

property labels: list[str]#: List of labels to be used in Neo4j. Includes the class name, superclasses class names, and additional labels if supplied.

property type: str#: Returns the class name of the object - or last additional label if supplied

pydantic model maestro.api.schema.WorkflowCounts[source]#

Bases: MaestroBase

Show JSON schema

{
   "title": "WorkflowCounts",
   "type": "object",
   "properties": {
      "invalidated_at": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Invalidated At"
      },
      "maestro_id": {
         "description": "Unique identifier",
         "format": "uuid4",
         "title": "Maestro Id",
         "type": "string"
      },
      "additional_labels": {
         "default": [],
         "description": "Additional labels to be used in Neo4j",
         "items": {
            "type": "string"
         },
         "title": "Additional Labels",
         "type": "array"
      },
      "pending": {
         "default": 0,
         "title": "Pending",
         "type": "integer"
      },
      "queued": {
         "default": 0,
         "title": "Queued",
         "type": "integer"
      },
      "running": {
         "default": 0,
         "title": "Running",
         "type": "integer"
      },
      "completed": {
         "default": 0,
         "title": "Completed",
         "type": "integer"
      },
      "cancelled": {
         "default": 0,
         "title": "Cancelled",
         "type": "integer"
      },
      "failed": {
         "default": 0,
         "title": "Failed",
         "type": "integer"
      },
      "halted": {
         "default": 0,
         "title": "Halted",
         "type": "integer"
      }
   }
}

Config:

populate_by_name: bool = True

Fields:

additional_labels ()
cancelled (int)
completed (int)
failed (int)
halted (int)
invalidated_at ()
maestro_id ()
pending (int)
queued (int)
running (int)

Validators:

deserialize_collections » all fields
neo4j_datetime_to_native » all fields

field additional_labels: list[str] = []#

Additional labels to be used in Neo4j

Validated by:

deserialize_collections
neo4j_datetime_to_native

field cancelled: int = 0#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field completed: int = 0#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field failed: int = 0#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field halted: int = 0#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field invalidated_at: datetime | None = None#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field maestro_id: UUID4 [Optional]#

Unique identifier

Constraints:

uuid_version = 4

Validated by:

deserialize_collections
neo4j_datetime_to_native

field pending: int = 0#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field queued: int = 0#

Validated by:

deserialize_collections
neo4j_datetime_to_native

field running: int = 0#

Validated by:

deserialize_collections
neo4j_datetime_to_native

classmethod class_labels() → list[str]#: List of labels to be used in Neo4j. Includes the class name, superclasses class names.

validator deserialize_collections » all fields#

Collections coming from the database are stored as JSON strings, serialized in model_dump_primitive.

This function attempts to run normal pydantic validation on all values, and if it fails, it tries to deserialize the value from a JSON string.

get_nested() → dict[str, MaestroBase]#

Returns a dictionary {field_name: value} of MaestroBase objects in the model.

These are represented in the database as relationships, e.g.

class Nested(MaestroBase):: pass
class Root(MaestroBase):: HAS: Nested

nested = Nested() obj = Root(nested=nested) obj.get_nested() -> {“nested”: nested}

This will be represented in the database as a relationship of type HAS between obj and nested.

Field alias takes precedence over field name

get_nested_iterables() → dict[str, list[MaestroBase] | tuple[MaestroBase] | set[MaestroBase]]#

Returns a dictionary {field_name: value} of lists or tuples consisting of MaestroBase object in the model.

These are represented in the database as relationships, e.g.

class Nested(MaestroBase):: pass
class Root(MaestroBase):: HAS: list[Nested]

nested1 = Nested() nested2 = Nested()

obj = Root(nested=[nested1, nested2]) obj.get_nested_iterables() -> {“nested”: [nested1, nested2]}

This will be represented in the database as a relationship of type HAS between obj and nested1, and between obj and nested2.

Field alias takes precedence over field name

classmethod get_subclass_matching_labels(labels: list[str]) → type[MaestroBase] | None#: Returns the subclass of cls that has the same labels as the list of labels

model_dump_primitive() → dict[str, str]#

Dump all ‘primitive’ fields - meaning fields not containing MaestroBase objects or iterables of MaestroBase objects.

Used for dumping the model to the database.

Field alias takes precedence over field name

validator neo4j_datetime_to_native » all fields#: Converts Neo4j DateTime objects to native Python datetime objects.

classmethod nested_fields() → Generator[tuple[str, type[MaestroBase]], None, None]#

Best effort to get which fields contain MaestroBase objects.

Supports at least the following constructions:

class Test(MaestroBase):: pass
class TestNested(MaestroBase):: test: Test optional_test1: Test | None optional_test2: Optional[Test] tuple_test1: tuple[Test, Test] tuple_test2: tuple[Test, …] list_test1: list[Test] optional_tuple_test1: Optional[tuple[Test, Test]] optional_tuple_test2: Optional[tuple[Test, …]] alias_test: Test = Field(alias=”alias_field”) -> “alias_field”, Test

Yields tuples of (alias or field_name, Test)

property all: int#

property labels: list[str]#: List of labels to be used in Neo4j. Includes the class name, superclasses class names, and additional labels if supplied.

property type: str#: Returns the class name of the object - or last additional label if supplied

pydantic model maestro.api.schema.Search[source]#

Bases: BaseModel

Show JSON schema

{
   "title": "Search",
   "type": "object",
   "properties": {
      "statuses": {
         "default": [],
         "items": {
            "$ref": "#/$defs/JobStatus"
         },
         "title": "Statuses",
         "type": "array",
         "uniqueItems": true
      },
      "priorities": {
         "default": [],
         "items": {
            "$ref": "#/$defs/JobPriorityEnum"
         },
         "title": "Priorities",
         "type": "array",
         "uniqueItems": true
      },
      "search_terms": {
         "default": [],
         "items": {
            "type": "string"
         },
         "title": "Search Terms",
         "type": "array"
      },
      "negate_terms": {
         "default": [],
         "items": {
            "type": "string"
         },
         "title": "Negate Terms",
         "type": "array"
      }
   },
   "$defs": {
      "JobPriorityEnum": {
         "enum": [
            1,
            2,
            3
         ],
         "title": "JobPriorityEnum",
         "type": "integer"
      },
      "JobStatus": {
         "enum": [
            "pending",
            "queued",
            "running",
            "completed",
            "failed",
            "cancelled",
            "halted"
         ],
         "title": "JobStatus",
         "type": "string"
      }
   }
}

Fields:

negate_terms (list[str])
priorities (set[maestro.models.JobPriorityEnum])
search_terms (list[str])
statuses (set[maestro.models.JobStatus])

Validators:

parse_priorities » priorities
parse_statuses » statuses

field negate_terms: list[str] = []#

field priorities: set[JobPriorityEnum] = {}#

Validated by:

parse_priorities

field search_terms: list[str] = []#

field statuses: set[JobStatus] = {}#

Validated by:

parse_statuses

classmethod from_string(search: str = '') → Search[source]#: Build a Search object from a string, by using the SEARCH_REGEX

validator parse_priorities » priorities[source]#

validator parse_statuses » statuses[source]#

maestro.api package#

Subpackages#

Submodules#

maestro.api.index module#

maestro.api.main module#

maestro.api.post_install module#

maestro.api.schema module#