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.

Prefect Community

Anyone who has worked with Prefect 3 Events can you query or subscribe to event in a flow_run?

I have an upstream process that will be emitting events to prefect and creating flow that process files, I then get a trailing or caboose event in which I want to trigger another flow, but only once all the previous flow have finished in either completed or failed(max retry)

`from prefect.events import get_events_subscriber`
it takes an `EventFilter` as input and you can use it as a context manager. it's an infinite loop so you'll need to add your own logic to exit it once you've seen the events you're looking for

This is awesome and what I was really hoping we could do with Prefect 3

```seen_events = set()
filtered_events = set(event_filter.event)

async with get_events_subscriber(filter=event_filter) as subscriber:
    async for event in subscriber:
        logger.info(f"📬 Received event {event.event!r}")
        seen_events.add(event)
            if seen_events.issuperset(filtered_events)):
                return```

Awesome I am gonna do some testing in my local env

<@U05TQMN8RUP> Does this only catch events from the time when the subscriber is started, or does is scan all events

you can specify a start time (including in the past) at which the subscriber should look for events as part of the `EventFilter`

it's a little nested in there but if you dig through the data models you'll find it

OK cool, if i don't add the filter does it scan all events?
Or is there a default

```since: DateTime = Field(
        default_factory=lambda: cast(
            DateTime,
            pendulum.now("UTC").start_of("day").subtract(days=180),
        ),
        description="Only include events after this time (inclusive)",
    )
    until: DateTime = Field(
        default_factory=lambda: cast(DateTime, pendulum.now("UTC")),
        description="Only include events prior to this time (inclusive)",
    )```