API#
Command line interface#
To extend pytask’s command line interface and set the right types for your options, pytask offers the following functionalities.
Classes#
- class pytask.ColoredCommand(name: str | None, context_settings: MutableMapping[str, Any] | None = None, callback: Callable[[...], Any] | None = None, params: List[Parameter] | None = None, help: str | None = None, epilog: str | None = None, short_help: str | None = None, options_metavar: str | None = '[OPTIONS]', add_help_option: bool = True, no_args_is_help: bool = False, hidden: bool = False, deprecated: bool = False)[source]#
A command with colored help pages.
- class pytask.EnumChoice(enum_type: type[Enum], case_sensitive: bool = True)[source]#
An enum-based choice type.
The implementation is copied from https://github.com/pallets/click/pull/2210 and related discussion can be found in https://github.com/pallets/click/issues/605.
In contrast to using
click.Choice
, using this type ensures that the error message does not show the enum members.In contrast to the proposed implementation in the PR, this implementation does not use the members than rather the values of the enum.
Compatibility#
- pytask.check_for_optional_program(name: str, extra: str = '', errors: str = 'raise', caller: str = 'pytask') bool | None [source]#
Check whether an optional program exists.
- pytask.import_optional_dependency(name: str, extra: str = '', errors: str = 'raise', min_version: str | None = None, caller: str = 'pytask') types.ModuleType | None [source]#
Import an optional dependency.
By default, if a dependency is missing an ImportError with a nice message will be raised. If a dependency is present, but too old, we raise.
- Parameters:
name – The module name.
extra – Additional text to include in the ImportError message.
errors –
What to do when a dependency is not found or its version is too old.
raise : Raise an ImportError
warn : Only applicable when a module’s version is to old. Warns that the version is too old and returns None
ignore: If the module is not installed, return None, otherwise, return the module, even if the version is too old. It’s expected that users validate the version locally when using
errors="ignore"
(see.io/html.py
)
min_version – Specify a minimum version that is different from the global pandas minimum version required.
caller – The caller of the function.
- Returns:
The imported module, when found and the version is correct. None is returned when the package is not found and errors is False, or when the package’s version is too old and errors is
'warn'
.- Return type:
types.ModuleType | None
Console#
To write to the terminal, use pytask’s console.
- class pytask.console#
Marks#
pytask uses marks to attach additional information to task functions which is processed by the host or by plugins. The following marks are available by default.
Marks#
- pytask.mark.depends_on(objects: Any | Iterable[Any] | dict[Any, Any])#
Specify dependencies for a task.
- Parameters:
objects (Any | Iterable[Any] | dict[Any, Any]) – Can be any valid Python object or an iterable of any Python objects. To be valid, it must be parsed by some hook implementation for the
_pytask.hookspecs.pytask_collect_node()
entry-point.
- pytask.mark.persist()#
A marker for a task which should be persisted.
- pytask.mark.produces(objects: Any | Iterable[Any] | dict[Any, Any])#
Specify products of a task.
- Parameters:
objects (Any | Iterable[Any] | dict[Any, Any]) – Can be any valid Python object or an iterable of any Python objects. To be valid, it must be parsed by some hook implementation for the
_pytask.hookspecs.pytask_collect_node()
entry-point.
- pytask.mark.skipif(condition: bool, *, reason: str)#
Skip a task based on a condition and provide a necessary reason.
- pytask.mark.skip_ancestor_failed(reason: str = 'No reason provided')#
An internal marker for a task which is skipped because an ancestor failed.
- Parameters:
reason (str) – A reason why the task is skipped.
- pytask.mark.skip_unchanged()#
An internal marker for a task which is skipped because nothing has changed.
- Parameters:
reason (str) – A reason why the task is skipped.
- pytask.mark.skip()#
Skip a task.
- pytask.mark.task(name, *, id, kwargs)#
The task decorator allows to mark any task function regardless of its name as a task or assigns a new task name.
It also allows to repeat tasks in for-loops by adding a specific
id
or keyword arguments viakwargs
.Deprecated since version 0.4.0: Will be removed in v0.5.0. Use
task()
instead.
- pytask.mark.try_first()#
Indicate that the task should be executed as soon as possible.
This indicator is a soft measure to influence the execution order of pytask.
Important
This indicator is not intended for general use to influence the build order and to overcome misspecification of task dependencies and products.
It should only be applied to situations where it is hard to define all dependencies and products and automatic inference may be incomplete like with pytask-latex and latex-dependency-scanner.
- pytask.mark.try_last()#
Indicate that the task should be executed as late as possible.
This indicator is a soft measure to influence the execution order of pytask.
Important
This indicator is not intended for general use to influence the build order and to overcome misspecification of task dependencies and products.
It should only be applied to situations where it is hard to define all dependencies and products and automatic inference may be incomplete like with pytask-latex and latex-dependency-scanner.
Custom marks#
Marks are created dynamically using the factory object pytask.mark
and applied
as a decorator.
For example:
@pytask.mark.timeout(10, "slow", method="thread")
def task_function():
...
Will create and attach a Mark
object to the collected
Task
to the markers
attribute. The mark
object will have the
following attributes:
mark.args == (10, "slow")
mark.kwargs == {"method": "thread"}
Example for using multiple custom markers:
@pytask.mark.timeout(10, "slow", method="thread")
@pytask.mark.slow
def task_function():
...
Classes#
- class pytask.Mark(name: str, args: tuple[Any, ...], kwargs: Mapping[str, Any])[source]#
A class for a mark containing the name, positional and keyword arguments.
- pytask.mark#
alias of <_pytask.mark.structures.MarkGenerator object>
- class pytask.MarkDecorator(mark: Mark)[source]#
A decorator for applying a mark on task function.
Decorators are created with
pytask.mark
.mark1 = pytask.mark.NAME # Simple MarkDecorator mark2 = pytask.mark.NAME(name1=value) # Parametrized MarkDecorator
and can then be applied as decorators to task functions
@mark2 def task_function(): pass
When a
MarkDecorator
is called it does the following:If called with a single function as its only positional argument and no additional keyword arguments, it attaches the mark to the function, containing all the arguments already stored internally in the
MarkDecorator
.When called in any other case, it returns a new
MarkDecorator
instance with the originalMarkDecorator
’s content updated with the arguments passed to this call.
Notes
The rules above prevent decorators from storing only a single function or class reference as their positional argument with no additional keyword or positional arguments. You can work around this by using
MarkDecorator.with_args()
.
- class pytask.MarkGenerator[source]#
Factory for
MarkDecorator
objects.Exposed as a
pytask.mark
singleton instance.Example
>>> import pytask
>>> @pytask.mark.skip ... def task_function(): ... pass
applies a ‘skip’
Mark
ontask_function
.
Functions#
These functions help you to handle marks.
- pytask.get_all_marks(obj_or_task: Any | PTask) list[Mark] [source]#
Get all marks from a callable or task.
- pytask.get_marks(obj_or_task: Any | PTask, marker_name: str) list[Mark] [source]#
Get marks from callable or task.
- pytask.has_mark(obj_or_task: Any | PTask, marker_name: str) bool [source]#
Test if callable or task has a certain mark.
Exceptions#
Exceptions all inherit from
- class pytask.PytaskError[source]#
Base exception for pytask which should be inherited by all other exceptions.
The following exceptions can be used to interrupt pytask’s flow, emit reduced tracebacks and return the correct exit codes.
The remaining exceptions convey specific errors.
General classes#
- class pytask.Session(*, config: dict[str, Any] = _Nothing.NOTHING, collection_reports: list[CollectionReport] = _Nothing.NOTHING, dag: nx.DiGraph = _Nothing.NOTHING, hook: HookRelay = _Nothing.NOTHING, tasks: list[PTask] = _Nothing.NOTHING, dag_reports: DagReport | None = None, execution_reports: list[ExecutionReport] = _Nothing.NOTHING, exit_code: ExitCode = ExitCode.OK, collection_start: float = inf, collection_end: float = inf, execution_start: float = inf, execution_end: float = inf, n_tasks_failed: int = 0, scheduler: Any = None, should_stop: bool = False, warnings: list[WarningReport] = _Nothing.NOTHING)[source]#
The session of pytask.
- Parameters:
collection_reports (list[CollectionReport]) – Reports for collected items.
dag (nx.DiGraph) – The DAG of the project.
hook (HookRelay) – Holds all hooks collected by pytask.
tasks (list[PTask]) – List of collected tasks.
resolving_dependencies_reports – Reports for resolving dependencies failed.
execution_reports (list[ExecutionReport]) – Reports for executed tasks.
n_tasks_failed (int) – Number of tests which have failed.
should_stop (bool) – Indicates whether the session should be stopped.
warnings (list[WarningReport]) – A list of warnings captured during the run.
Nodes#
Nodes are the interface for different kinds of dependencies or products. They inherit
from pytask.MetaNode
.
- class pytask.MetaNode(*args, **kwargs)[source]#
Protocol for an intersection between nodes and tasks.
Then, different kinds of nodes can be implemented.
- class pytask.PathNode(*, name: str, path: Path)[source]#
The class for a node which is a path.
- classmethod from_path(path: Path) PathNode [source]#
Instantiate class from path to file.
The lru_cache decorator ensures that the same object is not collected twice.
- path: Path#
The path to the file.
- class pytask.PythonNode(*, name: str = '', value: Any = None, hash: bool | Callable[[Any], bool] = False)[source]#
The class for a node which is a Python object.
- state() str | None [source]#
Calculate state of the node.
If
hash = False
, the function returns"0"
, a constant hash value, so thePythonNode
is ignored when checking for a changed state of the task.If
hash
is a callable, then use this function to calculate a hash.If
hash = True
, the builtinhash()
function (link) is used for all types except strings.The hash for strings and bytes is calculated using hashlib because
hash("asd")
returns a different value every invocation since the hash of strings is salted with a random integer and it would confuse users. See {meth}`object.__hash__` for more information.
To parse dependencies and products from nodes, use the following functions.
- pytask.depends_on(objects: PyTree[Any]) PyTree[Any] [source]#
Specify dependencies for a task.
- Parameters:
objects – Can be any valid Python object or an iterable of any Python objects. To be valid, it must be parsed by some hook implementation for the
_pytask.hookspecs.pytask_collect_node()
entry-point.
- pytask.parse_dependencies_from_task_function(session: Session, task_path: Path | None, task_name: str, node_path: Path, obj: Any) dict[str, Any] [source]#
Parse dependencies from task function.
- pytask.parse_products_from_task_function(session: Session, task_path: Path | None, task_name: str, node_path: Path, obj: Any) dict[str, Any] [source]#
Parse products from task function.
- Raises:
NodeNotCollectedError – If multiple ways were used to specify products.
- pytask.produces(objects: PyTree[Any]) PyTree[Any] [source]#
Specify products of a task.
- Parameters:
objects – Can be any valid Python object or an iterable of any Python objects. To be valid, it must be parsed by some hook implementation for the
_pytask.hookspecs.pytask_collect_node()
entry-point.
Tasks#
To mark any callable as a task use
- pytask.task(name: str | None = None, *, id: str | None = None, kwargs: dict[Any, Any] | None = None, produces: PyTree[Any] | None = None) Callable[..., Callable[..., Any]] [source]#
Decorate a task function.
This decorator declares every callable as a pytask task.
The function also attaches some metadata to the function like parsed kwargs and markers.
- Parameters:
name – Use it to override the name of the task that is, by default, the name of the callable.
id – An id for the task if it is part of a parametrization. Otherwise, an automatic id will be generated. See this tutorial for more information.
kwargs – A dictionary containing keyword arguments which are passed to the task when it is executed.
produces – Definition of products to parse the function returns and store them. See this how-to guide for more information.
Examples
To mark a function without the
task_
prefix as a task, attach the decorator.from typing_extensions import Annotated from pytask import task @task def create_text_file() -> Annotated[str, Path("file.txt")]: return "Hello, World!"
Task are currently represented by the following class:
- class pytask.Task(*, base_name: str, path: Path, function: Callable[..., Any], depends_on: dict[str, PyTree[PNode]] = _Nothing.NOTHING, produces: dict[str, PyTree[PNode]] = _Nothing.NOTHING, markers: list[Mark] = _Nothing.NOTHING, report_sections: list[tuple[str, str, str]] = _Nothing.NOTHING, attributes: dict[Any, Any] = _Nothing.NOTHING)[source]#
The class for tasks which are Python functions.
Currently, there are no different types of tasks since changing the .function
attribute with a custom callable proved to be sufficient.
To carry over information from user-defined tasks like task functions to
pytask.Task
objects, use a metadata object that is stored in an .pytask_meta
attribute of the task function.
Outcomes#
The exit code of pytask is determined by
- class pytask.ExitCode(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#
Exit codes for pytask.
- OK = 0#
Tasks were executed successfully.
- FAILED = 1#
Failed while executing tasks.
- COLLECTION_FAILED = 3#
Failed while collecting tasks.
- DAG_FAILED = 4#
Failed while building the DAG.
Collected items can have the following outcomes
- class pytask.CollectionOutcome(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#
Outcomes of collected files or tasks.
- FAIL#
Outcome for failed collected files or tasks.
- SUCCESS#
Outcome for task which was executed successfully.
Tasks can have the following outcomes
- class pytask.TaskOutcome(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#
Outcomes of tasks.
- FAIL#
Outcome for failed tasks.
- PERSISTENCE#
Outcome for tasks which should persist. Even if dependencies or products have changed, skip the task, update all hashes to the new ones, mark it as successful.
- SKIP#
Outcome for skipped tasks.
- SKIP_PREVIOUS_FAILED#
Outcome for tasks where a necessary preceding task has failed and, thus, this task could not have been executed.
- SKIP_UNCHANGED#
Outcome for tasks which do not need to be executed since all dependencies, source files and products have not changed.
- SUCCESS#
Outcome for task which was executed successfully.
The following exceptions are used to abort the execution of a task with an appropriate outcome.
- class pytask.Exit(msg: str = 'unknown reason', returncode: int | None = None)[source]#
Raised for immediate program exits (no tracebacks/summaries).
Functions#
- pytask.count_outcomes(reports: Sequence[CollectionReport | ExecutionReport], outcome_enum: type[CollectionOutcome | TaskOutcome]) dict[Enum, int] [source]#
Count how often an outcome occurred.
Examples
>>> from _pytask.outcomes import CollectionOutcome, TaskOutcome >>> count_outcomes([], CollectionOutcome) {<CollectionOutcome.SUCCESS: 1>: 0, <CollectionOutcome.FAIL: 2>: 0}
Programmatic Interfaces#
- pytask.build_dag(raw_config: dict[str, Any]) DiGraph [source]#
Build the DAG.
This function is the programmatic interface to
pytask dag
and returns a preprocessedpygraphviz.AGraph
which makes plotting easier than with matplotlib.To change the style of the graph, it might be easier to convert the graph back to networkx, set attributes, and convert back to pygraphviz.
- Parameters:
raw_config (Dict[str, Any]) – The configuration usually received from the CLI. For example, use
{"paths": "example-directory/"}
to collect tasks from a directory.- Returns:
A preprocessed graph which can be customized and exported.
- Return type:
- pytask.build(*, capture: Literal['fd', 'no', 'sys', 'tee-sys'] | CaptureMethod = CaptureMethod.NO, check_casing_of_paths: bool = True, config: Path | None = None, database_url: str = '', debug_pytask: bool = False, disable_warnings: bool = False, dry_run: bool = False, editor_url_scheme: Literal['no_link', 'file', 'vscode', 'pycharm'] | str = 'file', expression: str = '', force: bool = False, ignore: Iterable[str] = (), marker_expression: str = '', max_failures: float = inf, n_entries_in_table: int = 15, paths: str | Path | Iterable[str | Path] = (), pdb: bool = False, pdb_cls: str = '', s: bool = False, show_capture: bool = True, show_errors_immediately: bool = False, show_locals: bool = False, show_traceback: bool = True, sort_table: bool = True, stop_after_first_failure: bool = False, strict_markers: bool = False, tasks: Callable[..., Any] | PTask | Iterable[Callable[..., Any] | PTask] = (), task_files: str | Iterable[str] = 'task_*.py', trace: bool = False, verbose: int = 1, **kwargs: Any) Session [source]#
Run pytask.
This is the main command to run pytask which usually receives kwargs from the command line interface. It can also be used to run pytask interactively. Pass configuration in a dictionary.
- Parameters:
capture – The capture method for stdout and stderr.
check_casing_of_paths – Whether errors should be raised when file names have different casings.
config – A path to the configuration file.
database_url – An URL to the database that tracks the status of tasks.
debug_pytask – Whether debug information should be shown.
disable_warnings – Whether warnings should be disabled and not displayed.
dry_run – Whether a dry-run should be performed that shows which tasks need to be rerun.
editor_url_scheme – An url scheme that allows to click on task names, node names and filenames and jump right into you preferred editor to the right line.
expression – Same as
-k
on the command line. Select tasks via expressions on task ids.force – Run tasks even though they would be skipped since nothing has changed.
ignore – A pattern to ignore files or directories. Refer to
pathlib.Path.match
for more info.marker_expression – Same as
-m
on the command line. Select tasks via marker expressions.max_failures – Stop after some failures.
n_entries_in_table – How many entries to display in the table during the execution. Tasks which are running are always displayed.
paths – A path or collection of paths where pytask looks for the configuration and tasks.
pdb – Start the interactive debugger on errors.
pdb_cls – Start a custom debugger on errors. For example:
--pdbcls=IPython.terminal.debugger:TerminalPdb
s – Shortcut for
pytask.build(capture"no")
.show_capture – Choose which captured output should be shown for failed tasks.
show_errors_immediately – Show errors with tracebacks as soon as the task fails.
show_locals – Show local variables in tracebacks.
show_traceback – Choose whether tracebacks should be displayed or not.
sort_table – Sort the table of tasks at the end of the execution.
stop_after_first_failure – Stop after the first failure.
strict_markers – Raise errors for unknown markers.
tasks – A task or a collection of tasks that is passed to
pytask.build(tasks=...)
.task_files – A pattern to describe modules that contain tasks.
trace – Enter debugger in the beginning of each task.
verbose – Make pytask verbose (>= 0) or quiet (= 0).
- Returns:
session – The session captures all the information of the current run.
- Return type:
Reports#
There are some classes to handle different kinds of reports.
- class pytask.CollectionReport(outcome: CollectionOutcome, node: MetaNode | None = None, exc_info: OptionalExceptionInfo | None = None)[source]#
A collection report for a task.
- class pytask.ExecutionReport(task: PTask, outcome: TaskOutcome, exc_info: OptionalExceptionInfo | None = None, sections: list[tuple[str, str, str]] = _Nothing.NOTHING)[source]#
A report for an executed task.
- class pytask.DagReport(exc_info: Tuple[Type[BaseException], BaseException, TracebackType | None] | Tuple[None, None, None])[source]#
A report for an error during the creation of the DAG.
Typing#
Tracebacks#
- pytask.format_exception_without_traceback(exc_info: Tuple[Type[BaseException], BaseException, TracebackType | None]) str [source]#
Format an exception without displaying the traceback.
- pytask.remove_internal_traceback_frames_from_exc_info(exc_info: Tuple[Type[BaseException], BaseException, TracebackType | None] | Tuple[None, None, None]) Tuple[Type[BaseException], BaseException, TracebackType | None] | Tuple[None, None, None] [source]#
Remove internal traceback frames from exception info.
If a non-internal traceback frame is found, return the traceback from the first occurrence downwards.
- pytask.remove_traceback_from_exc_info(exc_info: Tuple[Type[BaseException], BaseException, TracebackType | None] | Tuple[None, None, None]) Tuple[Type[BaseException], BaseException, TracebackType | None] | Tuple[None, None, None] [source]#
Remove traceback from exception.
- pytask.render_exc_info(exc_type: type[BaseException], exc_value: BaseException, traceback: str | TracebackType, *, show_locals: bool = False) str | Traceback [source]#
Render an exception info.
Warnings#
Classes#
Functions#
- pytask.parse_warning_filter(arg: str, *, escape: bool) tuple[warnings._ActionKind, str, type[Warning], str, int] [source]#
Parse a warnings filter string.
This is copied from warnings._setoption with the following changes:
Does not apply the filter.
Escaping is optional.
Raises UsageError so we get nice error messages on failure.