Ivar Zapata
12/03/2025, 11:28 AMMarvin
12/03/2025, 11:28 AMMarvin
12/03/2025, 11:31 AMparameters
- 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
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.