<@ULVA73B9P> is there a way that the task decorato...
# ask-marvin
i
@Marvin is there a way that the task decorator access arguments of the decorated function?
m
thought for 181.0 seconds
Short answer: Yes. In Prefect 3.x, the @task decorator can “see” the function’s arguments via a few callback options and templates. Options that can access a task’s arguments - task_run_name - Accepts: - A static string - A format template like "{user_id}-{action}" - A callable that may accept a single keyword argument named
parameters
- Signature:
(parameters: dict[str, Any]) -> str
or
() -> str
- cache_key_fn - Signature:
(context: TaskRunContext, parameters: dict[str, Any]) -> Optional[str]
- The
parameters
dict contains the resolved function arguments - retry_condition_fn - Signature:
(task: Task, task_run: TaskRun, state: State) -> bool | Awaitable[bool]
- You can inspect the failure via
state.result()
and access inputs via
task_run.task_inputs
- result_storage_key - Accepts a template string that can reference runtime fields, including parameters - Examples:
"results/{task_run.id}"
,
"user-{parameters[user_id]}/run-{task_run.id}"
Example: name and cache based on arguments
Copy code
from datetime import timedelta
from prefect import flow, task
from prefect.context import TaskRunContext
from prefect.tasks import task_input_hash

def run_name(parameters: dict[str, Any]) -> str:
    return f"{parameters.get('action')}-user-{parameters.get('user_id')}"

def cache_key(context: TaskRunContext, parameters: dict[str, Any]) -> str:
    # Cache only by user_id (ignores other args)
    return f"user:{parameters.get('user_id')}"

@task(
    task_run_name=run_name,                         # callable gets parameters
    cache_key_fn=cache_key,                         # gets context + parameters
    cache_expiration=timedelta(hours=1),
    result_storage_key="users/{parameters[user_id]}/{task_run.id}",  # template sees parameters
    persist_result=True
)
def do_work(user_id: int, action: str):
    return {"user_id": user_id, "action": action}

@flow
def demo():
    a = do_work(42, "delete")
    b = do_work(42, "delete")  # cache hit due to same user_id
    return a, b
If you have a specific behavior in mind (e.g., naming, caching, retries, or storage paths), tell me what you want to compute and I’ll tailor the callback/template for it.