Skip to content

checkmate

CheckMate service module

Attributes

Classes

CheckMateService 

CheckMateService(
    sa_session: SASession, neo4j_session: NJSession
)

Bases: BaseService

CheckMate task maintenance service layer

Attributes

task_dao instance-attribute 
task_dao = TaskDao(session=sa_session)

Functions

cancel 
cancel(
    task_ids: Union[List[str], str] | None = None,
    since: datetime | None = None,
    sort: Literal["asc", "desc"] = "desc",
    max_results: int | None = 100,
    queue: str | None = None,
    dry_run: bool = True,
    db_transact: bool = True,
) -> Tuple[
    List[CheckMateTaskBatchResult],
    List[CheckMateTaskBatchResult],
]

Submit a task cancellation request by task ID.

This method updates the task status to StageStatus.CANCELED in the RDBMS. If the task is already in a terminal state (e.g., COMPLETED, CANCELED, ERROR), it will not change the status.

Parameters:

  • task_ids (Union[List[str], str], default: None ) –

    The unique identifier(s) of the task to cancel.

Raises:

  • ServiceError –

    If the task does not exist or an error occurs during cancellation.

Example 
with service_context(CheckMateService) as (checkmate_service,):
    checkmate_service.cancel(task_id="123e4567-e89b-12d3-a456-426614174000")
clean 
clean(
    dry_run: bool = True, db_transact: bool = True
) -> Tuple[
    List[CheckMateTaskBatchResult],
    List[CheckMateTaskBatchResult],
]

Removes task object metadata from persistence storage for tasks that are determined as cleanable.

If there is a problem loading the task from the persistence store, the corresponding RDBMS record will be marked as clean.

Returns:

Raises:

  • ServiceError –

    If a service-related error occurs.

  • Exception –

    If any other error occurs, it is wrapped in a ServiceError.

Example 
with service_context(CheckMateService) as (checkmate_service,):
    checkmate_service.clean()
get 
get(
    task_id: str,
    include_metadata: bool = True,
    db_transact: bool = True,
) -> CheckMateTask

Retrieve a task by its ID.

Parameters:

  • task_id (str) –

    The unique identifier of the task to retrieve.

Returns:

  • CheckMateTask –

    Tuple[orm.Task, List[Dict[str, Any]], bool, StageStatus]: A tuple containing the task object, the task metadata, a boolean indicating if a lock is detected, and the current stage status.

Example 
with service_context(CheckMateService) as (checkmate_service,):
    task_ref, stage_metadata, lock_detected, status = checkmate_service.get(
        task_id="123e4567-e89b-12d3-a456-426614174000"
    )

Raises:

  • ServiceError –

    If a service-related error occurs.

  • Exception –

    If any other error occurs, it is wrapped in a ServiceError.

invoke 
invoke(task_id: str, db_transact: bool = True) -> None

Invoke / Execute an existing CheckMate task.

Parameters:

  • task_id (str) –

    The unique identifier of the task to invoke.

Raises:

  • ServiceError –

    If a service-related error occurs.

  • Exception –

    If any other error occurs, it is wrapped in a ServiceError.

Example

```python with service_context(CheckMateService) as (checkmate_service,): checkmate_service.invoke(task_id=”123e4567-e89b-12d3-a456-426614174000”)

list 
list(
    *,
    cleaned: bool | None = None,
    status: StageStatus | None = None,
    queue: str | None = None,
    tags: List[str] | None = None,
    since: datetime | None = None,
    sort: Literal["asc", "desc"] = "desc",
    max_results: int | None = 100,
    include_metadata: bool = True,
    db_transact: bool = True
) -> List[CheckMateTask]

Retrieve a list of CheckMate managed task records from the RDBMS.

Parameters:

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

    Filter tasks based on whether they are cleaned.

  • status (StageStatus | None, default: None ) –

    Filter tasks based on their status.

  • since (datetime, default: None ) –

    Filter tasks created since this datetime.

  • sort (Literal['asc', 'desc'], default: 'desc' ) –

    Sort order of the results.

  • max_results (int, default: 100 ) –

    Maximum number of results to return.

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

    Filter tasks based on their queue.

Returns:

  • List[CheckMateTask] –

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

Example 
with service_context(CheckMateService) as (checkmate_service,):
    tasks = checkmate_service.list(
        cleaned=True,
        status="COMPLETED",
        since=datetime.now() - timedelta(days=1),
        sort="asc",
        max_results=50,
        queue="default",
    )
unlock 
unlock(
    task_ids: Union[List[str], str] | None = None,
    status: StageStatus | None = None,
    since: datetime | None = None,
    sort: Literal["asc", "desc"] = "desc",
    max_results: int | None = 100,
    queue: str | None = None,
    dry_run: bool = True,
    db_transact: bool = True,
) -> Tuple[
    List[CheckMateTaskBatchResult],
    List[CheckMateTaskBatchResult],
]

Unlock a task by removing its lock file.

Parameters:

  • task_id (str) –

    The unique identifier of the task to unlock.

Raises:

  • ServiceError –

    If a service-related error occurs.

  • Exception –

    If any other error occurs, it is wrapped in a ServiceError.

Example 
with service_context(CheckMateService) as (checkmate_service,):
    checkmate_service.unlock(task_id="123e4567-e89b-12d3-a456-426614174000")

CheckMateTask

Bases: BaseModel

CheckMate RDBMS model reference and persistence metadata

Attributes

lock_detected class-attribute instance-attribute 
lock_detected: bool = Field(default=False)
model_config class-attribute instance-attribute 
model_config = ConfigDict(
    populate_by_name=True,
    use_enum_values=True,
    extra="forbid",
    strict=False,
    arbitrary_types_allowed=True,
)
stage_metadata class-attribute instance-attribute 
stage_metadata: List[Dict[str, Any]] = Field(
    default_factory=list
)
task_ref class-attribute instance-attribute 
task_ref: Task = Field(...)

CheckMateTaskBatchResult

Bases: BaseDataModel

CheckMate task batch result

Used for service operations that return a list of task results, particularly when a mixture of success / failures is possible.

Attributes

error class-attribute instance-attribute 
error: str | None = Field(
    None,
    title="Error",
    description="Error message if an error occurred during processing",
)
status class-attribute instance-attribute 
status: StageStatus = Field(
    ...,
    title="Status",
    description="The current status of the task",
)
task_id class-attribute instance-attribute 
task_id: str = Field(
    ...,
    title="TaskID",
    description="The unique identifier of the task",
)

Functions