prefect.server.api.logs

Routes for interacting with log objects.

create_logs(logs, db=Depends(provide_database_interface)) async

Create new logs from the provided schema.

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

@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_logs(
    logs: List[schemas.actions.LogCreate],
    db: PrefectDBInterface = Depends(provide_database_interface),
):
    """Create new logs from the provided schema."""
    for batch in models.logs.split_logs_into_batches(logs):
        async with db.session_context(begin_transaction=True) as session:
            await models.logs.create_logs(session=session, logs=batch)

read_logs(limit=dependencies.LimitBody(), offset=Body(0, ge=0), logs=None, sort=Body(schemas.sorting.LogSort.TIMESTAMP_ASC), db=Depends(provide_database_interface)) async

Query for logs.

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

@router.post("/filter")
async def read_logs(
    limit: int = dependencies.LimitBody(),
    offset: int = Body(0, ge=0),
    logs: schemas.filters.LogFilter = None,
    sort: schemas.sorting.LogSort = Body(schemas.sorting.LogSort.TIMESTAMP_ASC),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> List[schemas.core.Log]:
    """
    Query for logs.
    """
    async with db.session_context() as session:
        return await models.logs.read_logs(
            session=session, log_filter=logs, offset=offset, limit=limit, sort=sort
        )