lc_task.task module¶

lc_task.task.StatePropagationSource¶

Types that can be merged into Task (see Task.merge_object()).

alias of Union[Dict[str, Any], TaskResult, None, Namespace, Task]

class lc_task.task.Task(raise_exceptions: bool = False)¶

Bases: object

Base task.

Cannot be used directly. Child classes must implement Task._perform_task().

Examples:

Create a task to count the number of lines in a file:

>>> import tempfile
>>> from pathlib import Path
>>> from lc_task import Task, TaskResult, taskclass
>>> @taskclass
... class CountLinesTaskResult(TaskResult):
...     num_lines: int = 0
...
>>> @taskclass
... class CountLinesTask(Task):
...     _result_cls = CountLinesTaskResult
...     input_path: typing.Optional[str] = None
...     def _perform_task(self) -> None:
...         input_path: Path = Path(self.input_path or "")
...         if not input_path.is_file():
...             raise FileNotFoundError(self.input_path)
...         num_lines = 0
...         with input_path.open() as input_file:
...             for _ in input_file:
...                 num_lines += 1
...         self.result.num_lines = num_lines
...
>>> with tempfile.NamedTemporaryFile() as tmp_file:
...     # Write ten lines to file
...     for _ in range(10):
...         __ = tmp_file.write(b"\n")
...     # Flush writes
...     tmp_file.flush()
...     result = CountLinesTask(input_path=tmp_file.name).run()
...     assert result.num_lines == 10

_perform_task() → None¶

Perform the task.

Child classes should set attributes on Task.result in this function.

_postamble() → None¶

Perform cleanup steps after task is completed.

Should not raise an exception, as this method always runs even if task fails.

_preamble() → None¶

Perform setup steps before running the task.

Should not raise an exception, as this method always runs even if task fails.

_result_cls¶

Type of result from running the task (see: Task.result).

alias of TaskResult

merge_object(obj: lc_task.task.StatePropagationSource, include: Set[str] | None = None, exclude: Set[str] | None = None, overwrite: Dict[str, Any] | None = None) → _TTask¶

Merge object into this task.

Only attributes defined on the class are merged. include and exclude can be used to further control merge behavior. If both include and exclude are specified, exclude takes priority and include is ignored. For Namespace, all kwargs are merged. For Task and TaskResult, all variables in __slots__ are merged.

Parameters:

obj – object to merge.
include – attributes/values to merge.
exclude – attributes/values not to merge.
overwrite – overwrite values after merging the object.

Returns:

self.

raise_exceptions: bool¶: Raise any exceptions in Task._perform_task() instead of capturing them in the task result.

result: TaskResult¶: Result from running the task.

run() → TaskResult¶

Run the task.

Returns:: The task result (as defined on Task._result_cls).
Raises:: Exception – Any exceptions raised in Task._perform_task() if Task.raise_exceptions.

class lc_task.task.TaskResult(err: Exception | None = None)¶

Bases: object

Result of running a task.

Use to pass result data to other compatible tasks. By default, the Exception raised from running the task (if occurred) is passed. Fields should be typing.Optional so Task can instantiate the result class without parameters (via Task._result_cls).

err: Exception | None¶: Exception raised during task execution, if any. Will be None if no exceptionn raised for Task.raise_exceptions is True.

lc_task.task.taskclass(cls, **kwargs)¶

Task class decorator.

All TaskResult and Task classes are dataclasses with the same options set. This convenience decorator sets those options by default, which include:

eq=True
slots=True

All options supported by that dataclasses.dataclass() are supported, with one caveat. The slots feature is only supported in dataclasses in Python >=3.10, but it’s supported in all Python versions supported by this library.

Examples:

Create a TaskResult child class:

>>> import dataclasses
>>> import sys
>>> from lc_task import TaskResult, taskclass
>>> # Using taskclass
>>> @taskclass
... class SomeTaskResult(TaskResult):
...     some_result: typing.Optional[str] = None
>>> assert hasattr(SomeTaskResult, "__slots__")
>>> # Using dataclass
>>> # The ``slots`` feature in dataclasses is only available in Python >=3.10,
>>> # but is available in taskclass in all supported Python versions
>>> if sys.version_info.major == 3 and sys.version_info.minor >= 10:
...     @dataclasses.dataclass(eq=True, slots=True)
...     class SomeOtherTaskResult(TaskResult):
...         some_result: typing.Optional[str] = None
...     # Both classes define __slots__ with the field some_result
...     assert SomeTaskResult.__slots__ == SomeOtherTaskResult.__slots__
... else:
...     @dataclasses.dataclass(eq=True)
...     class SomeOtherTaskResult(TaskResult):
...         some_result: typing.Optional[str] = None
>>> # Both classes define an __eq__ method
>>> assert hasattr(SomeTaskResult, "__eq__")
>>> assert hasattr(SomeOtherTaskResult, "__eq__")