perago.TaskDefinition#

class perago.TaskDefinition(name, owner_email, fn, params_model, output_model, description=None, workspace=None, controls=<factory>)[源代码]#

Validated contract extracted from a single Perago task module.

A task definition is created by the task() decorator after Perago validates the function signature, Pydantic contract models, workspace declaration, and runtime controls. Task authors normally receive this object through load_module_task() or from the function's __perago_task__ attribute; they do not construct it directly.

Parameters:
namestr

Conductor task definition name generated by Perago.

owner_emailstr

Owner email written to the generated Conductor TaskDef.

fncollections.abc.Callable

Synchronous Python function that implements the task body.

params_modeltype[pydantic.BaseModel]

Pydantic model used to validate the params input object.

output_modeltype[pydantic.BaseModel]

Pydantic model used to validate the task result object.

descriptionstr or None, default=None

Optional human-readable description copied into the TaskDef.

workspaceWorkspaceSpec or None, default=None

Workspace declaration for workspace tasks, or None for workspace-free tasks.

controlsTaskControls, optional

Retry, timeout, execution limit, and publish budget controls.

Attributes:
has_workspacebool

Whether this task expects an injected attempt-local workspace path.

参数:

参见

task

Declare and validate a task definition.

load_module_task

Import a module and retrieve its single task definition.

Notes

Perago supports one task definition per Python module. Modules that register zero or multiple tasks are rejected by load_module_task().

Examples

>>> definition = load_module_task("app.workers.features_build")
>>> definition.name
'features.build'
__init__(name, owner_email, fn, params_model, output_model, description=None, workspace=None, controls=<factory>)#
参数:
返回类型:

None

Methods

__init__(name, owner_email, fn, ...[, ...])

Attributes

description

has_workspace

Whether this task expects an injected attempt-local workspace path.

workspace

name

owner_email

fn

params_model

output_model

controls