from prefect import flow, serve
from prefect.events import DeploymentCompoundTrigger, DeploymentEventTrigger


@flow
def upstream_flow():
    print("upstream flow")


@flow
def other_upstream_flow():
    print("other upstream flow")


@flow
def downstream_flow():
    print("downstream flow")


if __name__ == "__main__":
    upstream_deployment = upstream_flow.to_deployment(name="upstream_deployment_a")
    other_upstream_deployment = other_upstream_flow.to_deployment(
        name="upstream_deployment_b", tags=["foo"]
    )
    downstream_deployment = downstream_flow.to_deployment(
        name="downstream_deployment",
        triggers=[
            DeploymentCompoundTrigger(
                require="all",
                triggers=[
                    DeploymentEventTrigger(
                        expect={"prefect.flow-run.Completed"},
                        match_related={
                            "prefect.resource.name": "upstream_deployment_*",
                        },
                    ),
                    DeploymentEventTrigger(
                        expect={"prefect.flow-run.Completed"},
                        match_related={
                            "prefect.resource.id": "prefect.tag.foo",
                        },
                    ),
                ],
            )
        ],
    )

    serve(upstream_deployment, other_upstream_deployment, downstream_deployment)