prefect.server.events.storage.database

raw_count_events(db, session, events_filter) async

Count events from the database with the given filter.

Only returns the count and does not return any addition metadata. For additional metadata, use count_events.

Parameters:

Name	Type	Description	Default
session	AsyncSession	a database session	required
events_filter	EventFilter	filter criteria for events	required

Returns:

Type	Description
int	The count of events in the database that match the filter criteria.

Source code in src/prefect/server/events/storage/database.py

@db_injector
async def raw_count_events(
    db: PrefectDBInterface,
    session: AsyncSession,
    events_filter: EventFilter,
) -> int:
    """
    Count events from the database with the given filter.

    Only returns the count and does not return any addition metadata. For additional
    metadata, use `count_events`.

    Args:
        session: a database session
        events_filter: filter criteria for events

    Returns:
        The count of events in the database that match the filter criteria.
    """
    # start with sa.func.count(), don't sa.select
    select_events_query = sa.select(sa.func.count()).select_from(db.Event)

    if distinct_fields := build_distinct_queries(events_filter):
        select_events_query = sa.select(
            sa.func.count(sa.distinct(*distinct_fields))
        ).select_from(db.Event)

    select_events_query_result = await session.execute(
        select_events_query.where(sa.and_(*events_filter.build_where_clauses()))
    )
    return select_events_query_result.scalar() or 0

read_events(db, session, events_filter, limit=None, offset=None) async

Read events from the Postgres database.

Parameters:

Name	Type	Description	Default
session	AsyncSession	a Postgres events session.	required
filter		filter criteria for events.	required
limit	Optional[int]	limit for the query.	None
offset	Optional[int]	offset for the query.	None

Returns:

Type	Description
Sequence[ORMEvent]	A list of events ORM objects.

Source code in src/prefect/server/events/storage/database.py

@db_injector
async def read_events(
    db: PrefectDBInterface,
    session: AsyncSession,
    events_filter: EventFilter,
    limit: Optional[int] = None,
    offset: Optional[int] = None,
) -> Sequence["ORMEvent"]:
    """
    Read events from the Postgres database.

    Args:
        session: a Postgres events session.
        filter: filter criteria for events.
        limit: limit for the query.
        offset: offset for the query.

    Returns:
        A list of events ORM objects.
    """
    # Always order by occurred timestamp, with placeholder for order direction
    order = sa.desc if events_filter.order == EventOrder.DESC else sa.asc

    # Check if distinct fields are provided
    if distinct_fields := build_distinct_queries(events_filter):
        # Define window function
        window_function = (
            sa.func.row_number()
            .over(partition_by=distinct_fields, order_by=order(db.Event.occurred))
            .label("row_number")
        )
        # Create a subquery with the window function
        subquery = (
            sa.select(db.Event, window_function)
            .where(
                sa.and_(
                    *events_filter.build_where_clauses()
                )  # Ensure the same filters are applied here
            )
            .subquery()
        )

        # Alias the subquery for easier column references
        aliased_table = aliased(db.Event, subquery)

        # Create the final query from the subquery, filtering to get only rows with row_number = 1
        select_events_query = sa.select(aliased_table).where(subquery.c.row_number == 1)

        # Order by the occurred timestamp
        select_events_query = select_events_query.order_by(order(subquery.c.occurred))

    else:
        # If no distinct fields are provided, create a query for all events
        select_events_query = sa.select(db.Event).where(
            sa.and_(*events_filter.build_where_clauses())
        )
        # Order by the occurred timestamp
        select_events_query = select_events_query.order_by(order(db.Event.occurred))

    if limit is not None:
        limit = max(0, min(limit, events_filter.logical_limit))
        select_events_query = select_events_query.limit(limit=limit)
    if offset is not None:
        select_events_query = select_events_query.offset(offset=offset)

    logger.debug("Running PostgreSQL query: %s", select_events_query)

    select_events_query_result = await session.execute(select_events_query)
    return select_events_query_result.scalars().unique().all()

write_events(session, events) async

Write events to the database.

Parameters:

Name	Type	Description	Default
session	AsyncSession	a database session	required
events	List[ReceivedEvent]	the events to insert	required

Source code in src/prefect/server/events/storage/database.py

async def write_events(session: AsyncSession, events: List[ReceivedEvent]) -> None:
    """
    Write events to the database.

    Args:
        session: a database session
        events: the events to insert
    """
    if events:
        dialect = get_dialect(PREFECT_API_DATABASE_CONNECTION_URL.value())
        if dialect.name == "postgresql":
            await _write_postgres_events(session, events)
        else:
            await _write_sqlite_events(session, events)