https://prefect.io
Join Slack
Does anyone know if there is a way to get the args from a task inside a state change hook? I'm runni...
# 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!
49 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by