Skip to content

prefect.events.filters

AutomationFilterCreated

Bases: PrefectBaseModel

Filter by Automation.created.

Source code in src/prefect/events/filters.py 
14
15
16
17
18
19
20
class AutomationFilterCreated(PrefectBaseModel):
    """Filter by `Automation.created`."""

    before_: Optional[DateTime] = Field(
        default=None,
        description="Only include automations created before this datetime",
    )

AutomationFilterName

Bases: PrefectBaseModel

Filter by Automation.created.

Source code in src/prefect/events/filters.py 
23
24
25
26
27
28
29
class AutomationFilterName(PrefectBaseModel):
    """Filter by `Automation.created`."""

    any_: Optional[List[str]] = Field(
        default=None,
        description="Only include automations with names that match any of these strings",
    )

EventDataFilter

Bases: PrefectBaseModel

A base class for filtering event data.

Source code in src/prefect/events/filters.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class EventDataFilter(PrefectBaseModel, extra="forbid"):  # type: ignore[call-arg]
    """A base class for filtering event data."""

    _top_level_filter: Optional["EventFilter"] = PrivateAttr(None)

    def get_filters(self) -> List["EventDataFilter"]:
        filters: List["EventDataFilter"] = [
            filter
            for filter in [
                getattr(self, name) for name, field in self.model_fields.items()
            ]
            if isinstance(filter, EventDataFilter)
        ]
        for filter in filters:
            filter._top_level_filter = self._top_level_filter
        return filters

    def includes(self, event: Event) -> bool:
        """Does the given event match the criteria of this filter?"""
        return all(filter.includes(event) for filter in self.get_filters())

    def excludes(self, event: Event) -> bool:
        """Would the given filter exclude this event?"""
        return not self.includes(event)

excludes(event)

Would the given filter exclude this event?

Source code in src/prefect/events/filters.py 
62
63
64
def excludes(self, event: Event) -> bool:
    """Would the given filter exclude this event?"""
    return not self.includes(event)

includes(event)

Does the given event match the criteria of this filter?

Source code in src/prefect/events/filters.py 
58
59
60
def includes(self, event: Event) -> bool:
    """Does the given event match the criteria of this filter?"""
    return all(filter.includes(event) for filter in self.get_filters())