https://prefect.io logo
Join Slack
Powered by
Does anyone know if there is a way to get the args...
# ask-community
d
Does anyone know if there is a way to get the args from a task inside a state change hook? I'm running a database update on completion and need to get the row_id from the task, passed in as positional argument
n
as long as you dont interfere with the first 3 positional arguments, you can partial whatever to your state hook
Copy code
def hook(task, task_run, state, row_id):
   # some hook
@flow
def foo():
   row_id = 42
   new_task = old_task.with_options(on_completion=[partial(hook, row_id=row_id)])
   new_task(some_args, row_id)
d
Nice that works great, I guess those will be merged with any configuration on the flow decorator, not replaced?
n
as is, it would be replaced, but you could access the current value and += it i.e.
Copy code
all_on_completion_hooks = old_task.on_completion + [partial(hook, row_id=row_id)]
new_task = old_task.with_options(on_completion=all_on_completion_hooks)
d
One edge case I can think of would be passing extra args to the entrypoint flow state change callback? I would have to nest it in another flow to be able to add options is that right?
n
correct
i'd suggest there's likely a different way to solve that problem, when do you need to partial kwargs to an entrypoints' hooks?
d
if the flow fails I need to update my db, maybe easier I could wrap the entry point in its own try catch. Its not as clean as using the state change hooks
n
if you're running the flow as a deployment, you can't really wrap the entrypoint flow in logic (besides the hooks on the flow object when the flow object gets loaded at run time) can you write a function for a hook that discovers what it needs so you don't need to dynamically modify hooks on the flow object? for example, 
prefect.runtime.flow_run.parameters
contains the params passed to the current flow run which would be accessible in a static hook you could implement
d
Copy code
def handle_fail(flow: Flow, flow_run: FlowRun, state: State):
    row_id = flow_run.parameters.get('row_id')
    with Session(get_engine()) as session:
        row_prompt = session.query(Prompt).filter_by(id=row_id).first()
        row_prompt.status = 'FAILED'
Thats a perfect solution in my case
n
yeah sweet that should work then 👍
d
I notice TaskRun doesn't have the same parameter property
Would be great to have similar and would eliminate having add options to tasks
n
ah well, the tricky thing there is that we intentionally don't store task run parameters in the API for a couple reasons • they're not always serializable (dask + ray inputs are sent to other processes, db connection / client objects etc) • security
d
Ahh okay, good to know!
45 Views
Open in Slack
PreviousNext
Powered by