task_dao

CheckMate task data access object (DAO) for the EleanorAI Framework backend.

Note

CheckMate tasks are global to the entire framework and are not considered namespaced resources.

Attributes

Classes

TaskDao

TaskDao(session: Session)

Bases: BaseDao

Task data access object (DAO)

Attributes

model_class property

model_class: Type[Task]

Functions

get_cleanable_tasks

get_cleanable_tasks() -> List[Task]

Retrieve tasks that are candidates for cleanup.

Returns:

• List[Task] –

  List[Task]: A list of tasks that are not cleaned and are in an inactive state.

Examples:

dao = TaskDao(session)
cleanable_tasks = dao.get_cleanable_tasks()

get_filtered_tasks

get_filtered_tasks(
    cleaned: bool | None = None,
    status: (
        Union[StageStatus, List[StageStatus]] | None
    ) = None,
    since: datetime | None = None,
    sort: str = "asc",
    max_results: int | None = None,
    queue: str | None = None,
    tags: List[str] | None = None,
) -> list[Task]

Retrieve tasks from the database based on provided filters.

Parameters:

• cleaned (bool | None, default: None ) –

  Filter tasks based on whether they are cleaned.

• status (Union[StageStatus, List[StageStatus]] | None, default: None ) –

  Filter tasks based on their status.

• since (datetime | None, default: None ) –

  Filter tasks created or updated since this datetime.

• sort (str, default: 'asc' ) –

  Sort order of the results by update_ts.

• max_results (int | None, default: None ) –

  Maximum number of results to return.

• queue (str | None, default: None ) –

  Filter tasks based on their queue.

Returns:

• list[Task] –

  list[Task]: A list of tasks matching the provided filters.

Examples:

dao = TaskDao(session)
tasks = dao.get_filtered_tasks(
    cleaned=True,
    status=StageStatus.COMPLETED,
    since=datetime.now() - timedelta(days=1),
    sort="asc",
    max_results=50,
    queue="default",
)

get_tasks_by_pkids

get_tasks_by_pkids(task_ids: List[str]) -> list[Task]

Retrieve tasks from the database based on their primary key identifiers.

Parameters:

• task_ids (List[str]) –

  A list of task primary key identifiers.

Returns:

• list[Task] –

  list[Task]: A list of tasks matching the provided identifiers.

Examples:

dao = TaskDao(session)
tasks = dao.get_tasks_by_pkids(["id1", "id2", "id3"])

list_queues

list_queues() -> List[str]

Retrieve a list of unique queue names from the database.

Returns:

• List[str] –

  List[str]: A list of unique queue names.

Examples:

dao = TaskDao(session)
queues = dao.list_queues()