prefect.server.models.logs

Functions for interacting with log ORM objects. Intended for internal use by the Prefect REST API.

create_logs(session, db, logs) async

Creates new logs

Parameters:

Name	Type	Description	Default
session	AsyncSession	a database session	required
logs	List[Log]	a list of log schemas	required

Returns:

Type	Description
	None

Source code in src/prefect/server/models/logs.py

@inject_db
async def create_logs(
    session: AsyncSession, db: PrefectDBInterface, logs: List[schemas.core.Log]
):
    """
    Creates new logs

    Args:
        session: a database session
        logs: a list of log schemas

    Returns:
        None
    """
    try:
        await session.execute(
            db.insert(db.Log).values([log.model_dump() for log in logs])
        )
    except RuntimeError as exc:
        if "can't create new thread at interpreter shutdown" in str(exc):
            # Background logs sometimes fail to write when the interpreter is shutting down.
            # This is a known issue in Python 3.12.2 that can be ignored and is fixed in Python 3.12.3.
            # see e.g. https://github.com/python/cpython/issues/113964
            logger.debug("Received event during interpreter shutdown, ignoring")
        else:
            raise

read_logs(session, db, log_filter, offset=None, limit=None, sort=schemas.sorting.LogSort.TIMESTAMP_ASC) async

Read logs.

Parameters:

Name	Type	Description	Default
session	AsyncSession	a database session	required
db	PrefectDBInterface	the database interface	required
log_filter	LogFilter	only select logs that match these filters	required
offset	Optional[int]	Query offset	None
limit	Optional[int]	Query limit	None
sort	LogSort	Query sort	TIMESTAMP_ASC

Returns:

Type	Description
	List[db.Log]: the matching logs

Source code in src/prefect/server/models/logs.py

@inject_db
async def read_logs(
    session: AsyncSession,
    db: PrefectDBInterface,
    log_filter: schemas.filters.LogFilter,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
    sort: schemas.sorting.LogSort = schemas.sorting.LogSort.TIMESTAMP_ASC,
):
    """
    Read logs.

    Args:
        session: a database session
        db: the database interface
        log_filter: only select logs that match these filters
        offset: Query offset
        limit: Query limit
        sort: Query sort

    Returns:
        List[db.Log]: the matching logs
    """
    query = select(db.Log).order_by(sort.as_sql_sort()).offset(offset).limit(limit)

    if log_filter:
        query = query.where(log_filter.as_sql_filter())

    result = await session.execute(query)
    return result.scalars().unique().all()