https://prefect.io logo
Join Slack
Powered by
previously I had something along the lines of ``` ...
# ask-community
p
previously I had something along the lines of
Copy code
with Flow(
    "...",
    state_handlers=[utils.notify_completion, utils.notify_running],...
with 
def notify_running(flow: Flow, old_state, new_state) -> State:
And I could ask what the new state is doing. eg if
Copy code
if new_state.is_running()
I'm reading https://docs.prefect.io/concepts/states/to try to work out how to capture these flow level state changes and am struggling. Any pointers?
✅ 1
z
We haven’t added hooks for notifications on state transitions client-side yet, I think we’ll probably add a callback in our internal engine that works similar to 
state_handlers
.
Are you planning on using Cloud or the open source server?
p
I have to use the OS server.
z
You could add an orchestration rule that fires on every state transition then and send the notifications server-side
p
(Well, for production I have to.)
z
That’d be the “proper” way to do it
We haven’t really exposed dynamic adding of user orchestration rules yet either though, it’s a more advanced use-case.
p
i guess... i could just put a task at the top and bottom of the flow to kinda get the job done for now.
z
😄 that’s one way to do it
p
yeah, i have to just hammer out this move to Prefect2. OK...
z
You can also decorate your function with start / end notifications?
p
decorate the flow function?
oh, as in write my own decorator to do something at the start and end of a flow? Hmm.
z
Like…
Copy code
from prefect import flow
from functools import wraps 


def notify(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        print("Starting!")
        result = fn()
        print("Ending!")
        return result
    return wrapper


@flow
@notify
def foo():
    return "hi!"

if __name__ == "__main__":
    foo()
Copy code
❯ PREFECT_DEBUG_MODE=1 python example.py     
15:39:01.803 | DEBUG   | prefect.client - Using ephemeral application with database at sqlite+aiosqlite:////Users/mz/.prefect/orion.db
15:39:01.853 | INFO    | prefect.engine - Created flow run 'gorgeous-waxbill' for flow 'foo'
15:39:01.854 | DEBUG   | Flow run 'gorgeous-waxbill' - Starting 'ConcurrentTaskRunner'; submitted tasks will be run concurrently...
15:39:01.855 | DEBUG   | prefect.task_runner.concurrent - Starting task runner...
15:39:01.855 | DEBUG   | prefect.client - Using ephemeral application with database at sqlite+aiosqlite:////Users/mz/.prefect/orion.db
15:39:01.987 | DEBUG   | Flow run 'gorgeous-waxbill' - Executing flow 'foo' for flow run 'gorgeous-waxbill'...
15:39:01.987 | DEBUG   | Flow run 'gorgeous-waxbill' - Executing foo()
Starting!
Ending!
15:39:02.000 | DEBUG   | prefect.task_runner.concurrent - Shutting down task runner...
15:39:02.000 | INFO    | Flow run 'gorgeous-waxbill' - Finished in state Completed(message=None, type=COMPLETED, result='hi!')
15:39:02.001 | DEBUG   | prefect.client - Using ephemeral application with database at sqlite+aiosqlite:////Users/mz/.prefect/orion.db
p
right right.
z
That’ll be called multiple times on retry though
p
yeah, that's reasonably elegant.
Not stressed about retry ATM really
z
👍 Yeah I think that’s the cleanest path for now
p
as in, when the flow is retried in its entirety, right?
z
Yep
I’m guessing we’ll have hooks for callbacks on state changes in the next couple months as well
p
yeah, that's not a big thing for us right now. Thanks Michael!
16 Views
Open in Slack
PreviousNext
Powered by