perago.task#

perago.task(*, name, owner_email, description=None, workspace=None, controls=None, **unsupported)[源代码]#

Declare the single Perago task exported by a Python module.

The decorator validates task metadata and the decorated function's type hints at import time. A workspace task must have the exact signature (workspace: pathlib.Path, params: ParamsModel) -> OutputModel and must pass workspace=WorkspaceSpec(...). A workspace-free task must have the exact signature (params: ParamsModel) -> OutputModel and must not pass a workspace declaration.

Parameters:
namestr

Required Conductor task name. Path separators are rejected.

owner_emailstr

Required owner email copied into the generated TaskDef.

descriptionstr or None, default=None

Optional TaskDef description.

workspaceWorkspaceSpec or None, default=None

Workspace declaration for workspace tasks. Leave as None for workspace-free tasks.

controlsTaskControls or None, default=None

Optional retry, timeout, execution limit, and publish budget controls. None uses default controls.

**unsupportedobject

Any extra decorator fields. Perago rejects them so that the task function signature remains the only business contract source.

Returns:
collections.abc.Callable

Decorator that returns the original task function after attaching a validated __perago_task__ definition.

Raises:
TaskDefinitionError

If metadata, controls, workspace declaration, function shape, or type hints violate the Perago task contract.

参数:
返回类型:

Callable[[Callable[[...], BaseModel]], Callable[[...], BaseModel]]

参见

TaskDefinition

Validated task contract created by this decorator.

load_module_task

Load the decorated task from a module target.

WorkspaceSpec

Declare the workspace prefix and guardrails for workspace tasks.

Examples

>>> from pathlib import Path
>>> from pydantic import BaseModel
>>> class Params(BaseModel):
...     source: str
>>> class Output(BaseModel):
...     rows: int
>>> @task(name="features.build", owner_email="data@example.com", workspace=WorkspaceSpec())
... def build_features(workspace: Path, params: Params) -> Output:
...     return Output(rows=0)