Task API

pab.task.TaskList

Type for an explicit list of Tasks.

alias of list[Task]

pab.task.RawTasksData

Type for an explicit list of task data dictionaries.

alias of List[dict]

class pab.task.Task(id_: int, strat: BaseStrategy, next_at: int, repeat_every: Optional[dict] = None)

Container for a strategy to be executed in the future

RUN_ASAP: int = -10

Constant. Means job should be rescheduled to run ASAP.

RUN_NEVER: int = -20

Constant. Means job should’t be rescheduled.

id

Internal Task ID

strategy: BaseStrategy

Strategy object

next_at: int

Next execution time as timestamp

repeat_every: dict[str, int] | None

Repetition data. A dict that functions as kwargs for datetime.timedelta

last_start: int

Last execution start time as timestamp

reschedule() → None

Calculates next execution if applies and calls schedule_for()

repeats() → bool

True if Task has repetition data.

next_repetition_time() → int

Returns next repetition time based on last_start and repeat_every.

schedule_for(next_at: int) → None

Updates self.next_at for a specific time. Will disable job if value is Task.RUN_NEVER.

is_ready() → bool

Returns True if job is ready to run based on next_at.

process() → None

Calls _process() and handles pab.strategy.RescheduleError.

_process() → None

Runs strategy and updates schedule.

class pab.task.TaskFileParser(root: Path, blockchain: Blockchain, strategies: dict[str, type['BaseStrategy']])

Parses a tasks file and loads a TaskList.

REQUIRED_TASK_FIELDS: list[str] = ['name', 'strategy']

Fields that must be declared in all tasks.

root: Path

Root of the project.

blockchain: Blockchain

Blockchain used by tasks.

strategies: dict[str, type['BaseStrategy']]

Strategies dictionary.

load() → TaskList

Loads TaskList from tasks file.

_load_tasks_json_or_exception(fhandle: TextIO) → RawTasksData

Parses TextIO input as JSON, validates and returns raw data. May raise TasksFileParseError.

_validate_raw_data(data: Any) → None

Validates raw tasks data format. May raise TasksFileParseError.

_create_tasklist(tasks: RawTasksData) → TaskList

Creates a list of Task objects from raw data. May raise TaskLoadError.

_create_strat_from_data(data: dict) → BaseStrategy

Creates a single Task object from raw data. May raise TaskLoadError.

_find_strat_by_name(name: str) → type[pab.strategy.BaseStrategy]

Finds a strategy by name. May raise UnkownStrategyError.

exception pab.task.TasksFileParseError

Error while parsing tasks.json

exception pab.task.TaskLoadError

Error while loading a task

exception pab.task.UnkownStrategyError

A strategy required by a task could not be found.