class openfe.TaskStatusDB(engine: Engine)#

Database for managing execution and orchestration of tasks.

This implementation is built on SQLAlchemy. For simple usage, the recommendation is to use the from_filename() method to create this object, rather than its __init__. The __init__ method takes a SQLAlchemy engine, which provides much more flexibility in choice of backend.

property tasks_table#

SQLAlchemy table for tasks

property dependencies_table#

SQLAlchemy table for depedencies

get_all_tasks() → Iterable[Row]#

Yield current row for all tasks.

This is mainly intended for debug and development usage; a more standardized variant will likely become part of the main API when we want dashboards, etc.

classmethod from_filename(filename: PathLike, *, overwrite: bool = False, **kwargs)#

Create an sqlite dialect database from a filename.

Parameters:

  • filename (os.PathLike) – Filename for the database

  • overwrite (bool) – If True, all tables will be dropped from the existing database

  • kwargs – Additional keyword arguments will be passed to sqlalchemy.create_engine. Particularly useful for debugging is echo=True, which will output the SQL statements being generated internally.

add_task(taskid: str, requirements: Iterable[str], max_tries: int, task_type: str = '')#

Add a task to the database.

Parameters:

  • taskid (str) – the task to add to the database

  • requirements (Iterable[str]) – taskids that directly block the task to be added (typically, whose outputs are inputs to the task)

  • max_tries (int) – the maximum number of trials for this task (this is total tries, so retries + 1)

  • task_type (str) – application-defined task type metadata for downstream routing

add_task_network(taskid_network: DiGraph, max_tries: int, task_types: Mapping[str, str] | None = None)#

Add a network of tasks to the database.

Parameters:

  • taskid_network (nx.Digraph) – A network with taskids (str) as nodes. Edges in this graph follow the direction of time/flow of information: from earlier tasks to later tasks; from requirements to subsequent.

  • max_tries (int) – the maximum number of trials for these tasks (this is total tries, so retries + 1)

  • task_types (Optional[Mapping[str, str]]) – optional application-defined task type metadata keyed by taskid

check_out_task()#

Select a task to be run.

This should include any internal updates to the task status database to indicate that the given task has been checked out by a worker.

Returns:

The taskid of the task to run; None if there are no available tasks

Return type:

str | None

mark_task_completed(taskid: str, success: bool)#

Update the database when a task has completed.

Parameters:

  • taskid (str) – the taskd of the completed task

  • success (bool) – True if the task completed successfully, False if there was a failure