monkey_wrench.task.base module

The module providing base models for tasks.

class monkey_wrench.task.base.Context(*values)

Bases: str, Enum

Enum for all possible task contexts.

product_ids = 'ids'

product_files = 'files'

chimp = 'chimp'

class monkey_wrench.task.base.Action(*values)

Bases: str, Enum

Enum for all possible task actions.

fetch = 'fetch'

verify = 'verify'

retrieve = 'retrieve'

monkey_wrench.task.base._fold_collections(specifications: Model, folding_threshold: Annotated[int, Gt(gt=0)] = 10, number_of_items_to_show_at_each_end: Annotated[int, Gt(gt=0)] = 2) → dict

Fold the collections inside the given specifications if they have too many items.

This helps with better log messages.

class monkey_wrench.task.base.TaskBase(*, verbose: bool = False, context: Context, action: Action, specifications: type[Model])

Bases: BaseModel

Pydantic base model for a task.

verbose: bool

Whether the long collections must be folded or not.

context: Context

action: Action

specifications: type[Model]

static log(func: Callable[[...], dict[str, Any]]) → Callable[[...], dict[str, Any]]

Decorator to log the details of the given task as well as the returned result.

perform() → dict[str, Any] | None

Perform the action using the given arguments.