https://prefect.io
Join Slack
Menu
Home
Archive
previously I had something along the lines of ``` with Flow( "...", state_handlers=[utils...
# 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!
15 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