TaskSpec
model reference¶
sheppy.models.Spec
¶
Bases: BaseModel
Task specification.
ATTRIBUTE | DESCRIPTION |
---|---|
func |
Fully qualified function name, e.g.
TYPE:
|
args |
Positional arguments to be passed to the function.
TYPE:
|
kwargs |
Keyword arguments to be passed to the function.
TYPE:
|
return_type |
Fully qualified return type name, e.g.
TYPE:
|
middleware |
List of fully qualified middleware function names to be applied to the task, e.g.
TYPE:
|
Note
- You should not create Spec instances directly. Instead, use the
@task
decorator to define a task function, and then call that function to create a Task instance. args
andkwargs
must be JSON serializable.
Example
from sheppy import task
@task
def my_task(x: int, y: str) -> str:
return f"Received {x} and {y}"
t = my_task(42, "hello") # returns a Task instance, it is NOT executed yet
print(t.spec.func) # e.g. "my_module:my_task"
print(t.spec.args) # [42, "hello"]
print(t.spec.return_type) # "builtins.str"
args
¶
args: list[Any] = Field(default_factory=list)
list[Any]: Positional arguments to be passed to the function.
kwargs
¶
kwargs: dict[str, Any] = Field(default_factory=dict)
dict[str, Any]: Keyword arguments to be passed to the function.
return_type
¶
return_type: str | None = None
str|None: Fully qualified return type name, e.g. my_module.submodule:MyPydanticModel
. This is used to reconstruct the return value if it's a pydantic model.
middleware
¶
middleware: list[str] | None = None
list[str]|None: List of fully qualified middleware function names to be applied to the task, e.g. ['my_module.submodule:my_middleware']
. Middleware will be applied in the order they are listed.