jobflow_remote.jobs.data module

class jobflow_remote.jobs.data.BatchDoc(*, batch_uid: str, process_id: str, batch_state: BatchState, worker: str, jobs: list = <factory>, created_on: datetime = <factory>, updated_on: datetime = <factory>, start_time: datetime | None = None, end_time: datetime | None = None, last_ping_time: datetime | None = None)

Bases: BaseModel

Model for the standard representation of a batch process in the batch database.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

as_db_dict() → dict

Generate a dict representation suitable to be inserted in the database.

Returns:

The dict representing the BatchDoc.

Return type:

dict

batch_state: BatchState

batch_uid: str

created_on: datetime

end_time: datetime | None

jobs: list

last_ping_time: datetime | None

model_config = {}

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

process_id: str

start_time: datetime | None

updated_on: datetime

worker: str

class jobflow_remote.jobs.data.DbCollection(value)

Bases: Enum

AUX = 'aux'

FLOWS = 'flows'

JOBS = 'jobs'

class jobflow_remote.jobs.data.DynamicResponseType(value)

Bases: Enum

Types of dynamic responses in jobflow.

ADDITION = 'addition'

DETOUR = 'detour'

REPLACE = 'replace'

class jobflow_remote.jobs.data.FlowDoc(*, uuid: str, jobs: list[str], state: ~jobflow_remote.jobs.state.FlowState, name: str, lock_id: str | None = None, lock_time: ~datetime.datetime | None = None, created_on: ~datetime.datetime = <factory>, updated_on: ~datetime.datetime = <factory>, metadata: dict = <factory>, parents: dict[str, dict[str, list[str]]] = <factory>, ids: list[tuple[str, str, int]] = <factory>, jobstore: str | None = None)

Bases: BaseModel

Model for the standard representation of a Flow in the queue database.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

as_db_dict() → dict

Generate a dict representation suitable to be inserted in the database.

Returns:

The dict representing the FlowDoc.

Return type:

dict

property children: dict[str, list[tuple[str, int]]]

created_on: datetime

descendants(job_uuid: str) → list[tuple[str, int]]

ids: list[tuple[str, str, int]]

property ids_mapping: dict[str, dict[int, str]]

property int_index_parents

jobs: list[str]

jobstore: str | None

lock_id: str | None

lock_time: datetime | None

metadata: dict

model_config = {}

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

name: str

parents: dict[str, dict[str, list[str]]]

state: FlowState

updated_on: datetime

uuid: str

class jobflow_remote.jobs.data.FlowInfo(*, db_ids: list[str], job_ids: list[str], job_indexes: list[int], flow_id: str, state: FlowState, name: str, created_on: datetime, updated_on: datetime, workers: list[str], job_states: list[JobState], job_names: list[str], parents: list[list[str]], hosts: list[list[str]], flow_metadata: dict, jobs_info: list[JobInfo] | None = None)

Bases: BaseModel

Model with information extracted from a FlowDoc. Mainly for visualization purposes.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

created_on: datetime

db_ids: list[str]

flow_id: str

flow_metadata: dict

classmethod from_query_dict(d) → FlowInfo

hosts: list[list[str]]

property ids_mapping: dict[str, dict[int, str]]

iter_job_prop()

job_ids: list[str]

job_indexes: list[int]

job_names: list[str]

job_states: list[JobState]

jobs_info: list[JobInfo] | None

model_config = {}

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

name: str

parents: list[list[str]]

state: FlowState

updated_on: datetime

workers: list[str]

class jobflow_remote.jobs.data.JobDoc(*, job: Job, uuid: str, index: int, db_id: str, worker: str, state: JobState, remote: RemoteInfo = RemoteInfo(step_attempts=0, queue_state=None, process_id=None, retry_time_limit=None, error=None, prerun_cleanup=False, queue_out=None, queue_err=None), parents: list[str] | None = None, previous_state: JobState | None = None, error: str | None = None, lock_id: str | None = None, lock_time: datetime | None = None, run_dir: str | None = None, start_time: datetime | None = None, end_time: datetime | None = None, created_on: datetime = <factory>, updated_on: datetime = <factory>, priority: int = 0, exec_config: ExecutionConfig | str | None = None, resources: QResources | dict | None = None, stored_data: dict | None = None)

Bases: BaseModel

Model for the standard representation of a Job in the queue database.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

as_db_dict() → dict

Generate a dict representation suitable to be inserted in the database.

Returns:

The dict representing the JobDoc.

Return type:

dict

created_on: datetime

db_id: str

end_time: datetime | None

error: str | None

exec_config: ExecutionConfig | str | None

index: int

job: Job

lock_id: str | None

lock_time: datetime | None

model_config = {}

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

parents: list[str] | None

previous_state: JobState | None

classmethod previous_state_validator(value)

priority: int

remote: RemoteInfo

resources: QResources | dict | None

run_dir: str | None

start_time: datetime | None

state: JobState

classmethod state_validator(value)

stored_data: dict | None

updated_on: datetime

uuid: str

worker: str

class jobflow_remote.jobs.data.JobInfo(*, uuid: str, index: int, db_id: str, worker: str, name: str, state: JobState, created_on: datetime, updated_on: datetime, remote: RemoteInfo = RemoteInfo(step_attempts=0, queue_state=None, process_id=None, retry_time_limit=None, error=None, prerun_cleanup=False, queue_out=None, queue_err=None), parents: list[str] | None = None, previous_state: JobState | None = None, error: str | None = None, lock_id: str | None = None, lock_time: datetime | None = None, run_dir: str | None = None, start_time: datetime | None = None, end_time: datetime | None = None, priority: int = 0, metadata: dict | None = None, stored_data: dict | None = None, hosts: list[str] | None = None)

Bases: BaseModel

Model with information extracted from a JobDoc. Mainly for visualization purposes.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

created_on: datetime

db_id: str

end_time: datetime | None

error: str | None

property estimated_run_time: float | None

Estimate the current run time based on the start time and the current time.

Returns:

The estimated run time in seconds.

Return type:

float

classmethod from_query_output(d) → JobInfo

Generate an instance from the output of a query to the JobDoc collection.

Parameters:

d – The dictionary with the queried data.

Returns:

The instance of JobInfo based on the data

Return type:

JobInfo

hosts: list[str] | None

index: int

property is_locked: bool

lock_id: str | None

lock_time: datetime | None

metadata: dict | None

model_config = {}

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

name: str

parents: list[str] | None

previous_state: JobState | None

classmethod previous_state_validator(value)

priority: int

remote: RemoteInfo

run_dir: str | None

property run_time: float | None

Calculate the run time based on start and end time.

Returns:

The run time in seconds

Return type:

float

start_time: datetime | None

state: JobState

classmethod state_validator(value)

stored_data: dict | None

updated_on: datetime

uuid: str

worker: str

exception jobflow_remote.jobs.data.RemoteError(msg, no_retry=False)

Bases: RuntimeError

An exception signaling errors during the update of the remote states.

class jobflow_remote.jobs.data.RemoteInfo(*, step_attempts: int = 0, queue_state: QState | None = None, process_id: str | None = None, retry_time_limit: datetime | None = None, error: str | None = None, prerun_cleanup: bool = False, queue_out: str | None = None, queue_err: str | None = None)

Bases: BaseModel

Model with data describing the remote state of a Job.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

error: str | None

model_config = {}

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

prerun_cleanup: bool

process_id: str | None

queue_err: str | None

queue_out: str | None

queue_state: QState | None

retry_time_limit: datetime | None

step_attempts: int

jobflow_remote.jobs.data.get_initial_batch_doc_dict(batch_uid, process_id, worker)

Generate a serialized FlowDoc for initial insertion in the DB.

Parameters:

• flow – The Flow used to generate the FlowDoc.

• job_dicts – The dictionaries of the Jobs composing the Flow.

• jobstore – The name of the JobStore used for the output of the submitted Flow. If None the default is used.

Returns:

A serialized version of a new FlowDoc.

Return type:

dict

jobflow_remote.jobs.data.get_initial_flow_doc_dict(flow: Flow, job_dicts: list[dict], jobstore: str | None = None) → dict

Generate a serialized FlowDoc for initial insertion in the DB.

Parameters:

• flow – The Flow used to generate the FlowDoc.

• job_dicts – The dictionaries of the Jobs composing the Flow.

• jobstore – The name of the JobStore used for the output of the submitted Flow. If None the default is used.

Returns:

A serialized version of a new FlowDoc.

Return type:

dict

jobflow_remote.jobs.data.get_initial_job_doc_dict(job: Job, parents: list[str] | None, db_id: str, worker: str, exec_config: ExecutionConfig | None, resources: dict | QResources | None, priority: int) → dict

Generate an instance of JobDoc for initial insertion in the DB.

Parameters:

• job – The Job of the JobDoc.

• parents – The parents of the Job.

• db_id – The db_id.

• worker – The worker where the Job should be executed.

• exec_config – The ExecutionConfig used for execution.

• resources – The resources used to run the Job.

• priority – The priority of the Job.

Returns:

A new JobDoc.

Return type:

JobDoc

jobflow_remote.jobs.data.get_reset_job_base_dict() → dict

Generate a dictionary with the basic properties to update in case of reset.

Returns:

Data to be reset.

Return type:

dict