TaskSpec model reference

sheppy.models.TaskSpec

Bases: BaseModel

Task specification.

ATTRIBUTE	DESCRIPTION
func	Fully qualified function name, e.g. my_module.my_submodule:my_function TYPE: str
args	Positional arguments to be passed to the function. TYPE: tuple[Any, ...]
kwargs	Keyword arguments to be passed to the function. TYPE: dict[str, Any]
middleware	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. TYPE: list[str] | None

Note

• You should not create TaskSpec instances directly. Instead, use the @task decorator to define a task function, and then call that function to create a Task instance.
• args and kwargs 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"

func

func: str

str: Fully qualified function name, e.g. my_module.my_submodule:my_function

args

args: tuple[Any, ...] = Field(default_factory=tuple)

tuple[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.

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.