Does anyone else chain flows together using a run_deployment(...) inside a @flow(on_completion=...) hook? Anyone got a good example? I have been trying to get it to work using code from Marvin but it does not work and marvin has suggesting methods that do not exist

hi @Alastair - you can do that, something like this

def trigger_on_complete(flow, flow_run, state):
    print(
        f"hello from {flow_run.name}'s completion hook |"
        f" the return value was {(r := state.result())!r}"
    )
    run_deployment(name="triggered-flow/triggered", parameters=dict(prev_result=r))

@flow(
    persist_result=True,
    result_storage=S3Bucket.load("my-result-storage"),
    on_completion=[trigger_on_complete],
)
def foobar() -> str:
    return "foobar"

@flow(log_prints=True)
def triggered_flow(prev_result: str) -> str:
    print(f"got {prev_result=!r}")

but I would check out events for this https://docs.prefect.io/v3/automate/events/automations-triggers#trigger-a-downstream-deployment-with-an-event

Thanks @Nate I have code that looks much like that but it is not working - the trigger_on_complete fires but the deployment never actually gets run. Prefect 3.1.7 Not using persist_results - is that required? I have also tried adding

timeout=0

to run_deployment but either way the deployment does not trigger which is odd