prefect.server.models.artifacts

count_artifacts(session, artifact_filter=None, flow_run_filter=None, task_run_filter=None, deployment_filter=None, flow_filter=None) async

Counts artifacts. Args: session: A database session artifact_filter: Only select artifacts matching this filter flow_run_filter: Only select artifacts whose flow runs matching this filter task_run_filter: Only select artifacts whose task runs matching this filter

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

async def count_artifacts(
    session: AsyncSession,
    artifact_filter: Optional[filters.ArtifactFilter] = None,
    flow_run_filter: Optional[filters.FlowRunFilter] = None,
    task_run_filter: Optional[filters.TaskRunFilter] = None,
    deployment_filter: Optional[filters.DeploymentFilter] = None,
    flow_filter: Optional[filters.FlowFilter] = None,
) -> int:
    """
    Counts artifacts.
    Args:
        session: A database session
        artifact_filter: Only select artifacts matching this filter
        flow_run_filter: Only select artifacts whose flow runs matching this filter
        task_run_filter: Only select artifacts whose task runs matching this filter
    """
    query = sa.select(sa.func.count(orm_models.Artifact.id))

    query = await _apply_artifact_filters(
        query,
        artifact_filter=artifact_filter,
        flow_run_filter=flow_run_filter,
        task_run_filter=task_run_filter,
        deployment_filter=deployment_filter,
        flow_filter=flow_filter,
    )

    result = await session.execute(query)
    return result.scalar_one()

count_latest_artifacts(session, artifact_filter=None, flow_run_filter=None, task_run_filter=None, deployment_filter=None, flow_filter=None) async

Counts artifacts. Args: session: A database session artifact_filter: Only select artifacts matching this filter flow_run_filter: Only select artifacts whose flow runs matching this filter task_run_filter: Only select artifacts whose task runs matching this filter

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

async def count_latest_artifacts(
    session: AsyncSession,
    artifact_filter: Optional[filters.ArtifactCollectionFilter] = None,
    flow_run_filter: Optional[filters.FlowRunFilter] = None,
    task_run_filter: Optional[filters.TaskRunFilter] = None,
    deployment_filter: Optional[filters.DeploymentFilter] = None,
    flow_filter: Optional[filters.FlowFilter] = None,
) -> int:
    """
    Counts artifacts.
    Args:
        session: A database session
        artifact_filter: Only select artifacts matching this filter
        flow_run_filter: Only select artifacts whose flow runs matching this filter
        task_run_filter: Only select artifacts whose task runs matching this filter
    """
    query = sa.select(sa.func.count(orm_models.ArtifactCollection.id))

    query = await _apply_artifact_collection_filters(
        query,
        artifact_filter=artifact_filter,
        flow_run_filter=flow_run_filter,
        task_run_filter=task_run_filter,
        deployment_filter=deployment_filter,
        flow_filter=flow_filter,
    )

    result = await session.execute(query)
    return result.scalar_one()

delete_artifact(session, artifact_id) async

Deletes an artifact by id.

The ArtifactCollection table is used to track the latest version of an artifact by key. If we are deleting the latest version of an artifact from the Artifact table, we need to first update the latest version referenced in ArtifactCollection so that it points to the next latest version of the artifact.

Example: If we have the following artifacts in Artifact: - key: "foo", id: 1, created: 2020-01-01 - key: "foo", id: 2, created: 2020-01-02 - key: "foo", id: 3, created: 2020-01-03

the ArtifactCollection table has the following entry: - key: "foo", latest_id: 3

If we delete the artifact with id 3, we need to update the latest version of the artifact with key "foo" to be the artifact with id 2.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
artifact_id	UUID	The artifact id to delete	required

Returns:

Name	Type	Description
bool	bool	True if the delete was successful, False otherwise

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

async def delete_artifact(
    session: AsyncSession,
    artifact_id: UUID,
) -> bool:
    """
    Deletes an artifact by id.

    The ArtifactCollection table is used to track the latest version of an artifact
    by key. If we are deleting the latest version of an artifact from the Artifact
    table, we need to first update the latest version referenced in ArtifactCollection
    so that it points to the next latest version of the artifact.

    Example:
    If we have the following artifacts in Artifact:
    - key: "foo", id: 1, created: 2020-01-01
    - key: "foo", id: 2, created: 2020-01-02
    - key: "foo", id: 3, created: 2020-01-03

    the ArtifactCollection table has the following entry:
    - key: "foo", latest_id: 3

    If we delete the artifact with id 3, we need to update the latest version of the
    artifact with key "foo" to be the artifact with id 2.

    Args:
        session: A database session
        artifact_id (UUID): The artifact id to delete

    Returns:
        bool: True if the delete was successful, False otherwise
    """
    artifact = await session.get(orm_models.Artifact, artifact_id)
    if artifact is None:
        return False

    is_latest_version = (
        await session.execute(
            sa.select(orm_models.ArtifactCollection)
            .where(orm_models.ArtifactCollection.key == artifact.key)
            .where(orm_models.ArtifactCollection.latest_id == artifact_id)
        )
    ).scalar_one_or_none() is not None

    if is_latest_version:
        next_latest_version = (
            await session.execute(
                sa.select(orm_models.Artifact)
                .where(orm_models.Artifact.key == artifact.key)
                .where(orm_models.Artifact.id != artifact_id)
                .order_by(orm_models.Artifact.created.desc())
                .limit(1)
            )
        ).scalar_one_or_none()

        if next_latest_version is not None:
            set_next_latest_version = (
                sa.update(orm_models.ArtifactCollection)
                .where(orm_models.ArtifactCollection.key == artifact.key)
                .values(
                    latest_id=next_latest_version.id,
                    data=next_latest_version.data,
                    description=next_latest_version.description,
                    type=next_latest_version.type,
                    created=next_latest_version.created,
                    updated=next_latest_version.updated,
                    flow_run_id=next_latest_version.flow_run_id,
                    task_run_id=next_latest_version.task_run_id,
                    metadata_=next_latest_version.metadata_,
                )
            )
            await session.execute(set_next_latest_version)

        else:
            await session.execute(
                sa.delete(orm_models.ArtifactCollection)
                .where(orm_models.ArtifactCollection.key == artifact.key)
                .where(orm_models.ArtifactCollection.latest_id == artifact_id)
            )

    delete_stmt = sa.delete(orm_models.Artifact).where(
        orm_models.Artifact.id == artifact_id
    )

    result = await session.execute(delete_stmt)
    return result.rowcount > 0

read_artifact(session, artifact_id) async

Reads an artifact by id.

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

async def read_artifact(
    session: AsyncSession,
    artifact_id: UUID,
) -> Union[orm_models.Artifact, None]:
    """
    Reads an artifact by id.
    """

    query = sa.select(orm_models.Artifact).where(orm_models.Artifact.id == artifact_id)

    result = await session.execute(query)
    return result.scalar()

read_artifacts(session, offset=None, limit=None, artifact_filter=None, flow_run_filter=None, task_run_filter=None, deployment_filter=None, flow_filter=None, sort=sorting.ArtifactSort.ID_DESC) async

Reads artifacts.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
offset	Optional[int]	Query offset	None
limit	Optional[int]	Query limit	None
artifact_filter	Optional[ArtifactFilter]	Only select artifacts matching this filter	None
flow_run_filter	Optional[FlowRunFilter]	Only select artifacts whose flow runs matching this filter	None
task_run_filter	Optional[TaskRunFilter]	Only select artifacts whose task runs matching this filter	None
deployment_filter	Optional[DeploymentFilter]	Only select artifacts whose flow runs belong to deployments matching this filter	None
flow_filter	Optional[FlowFilter]	Only select artifacts whose flow runs belong to flows matching this filter	None
work_pool_filter		Only select artifacts whose flow runs belong to work pools matching this filter	required

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

async def read_artifacts(
    session: AsyncSession,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
    artifact_filter: Optional[filters.ArtifactFilter] = None,
    flow_run_filter: Optional[filters.FlowRunFilter] = None,
    task_run_filter: Optional[filters.TaskRunFilter] = None,
    deployment_filter: Optional[filters.DeploymentFilter] = None,
    flow_filter: Optional[filters.FlowFilter] = None,
    sort: sorting.ArtifactSort = sorting.ArtifactSort.ID_DESC,
) -> Sequence[orm_models.Artifact]:
    """
    Reads artifacts.

    Args:
        session: A database session
        offset: Query offset
        limit: Query limit
        artifact_filter: Only select artifacts matching this filter
        flow_run_filter: Only select artifacts whose flow runs matching this filter
        task_run_filter: Only select artifacts whose task runs matching this filter
        deployment_filter: Only select artifacts whose flow runs belong to deployments matching this filter
        flow_filter: Only select artifacts whose flow runs belong to flows matching this filter
        work_pool_filter: Only select artifacts whose flow runs belong to work pools matching this filter
    """
    query = sa.select(orm_models.Artifact).order_by(sort.as_sql_sort())

    query = await _apply_artifact_filters(
        query,
        artifact_filter=artifact_filter,
        flow_run_filter=flow_run_filter,
        task_run_filter=task_run_filter,
        deployment_filter=deployment_filter,
        flow_filter=flow_filter,
    )

    if offset is not None:
        query = query.offset(offset)
    if limit is not None:
        query = query.limit(limit)

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

read_latest_artifact(session, key) async

Reads the latest artifact by key. Args: session: A database session key: The artifact key Returns: Artifact: The latest artifact

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

async def read_latest_artifact(
    session: AsyncSession,
    key: str,
) -> Union[orm_models.ArtifactCollection, None]:
    """
    Reads the latest artifact by key.
    Args:
        session: A database session
        key: The artifact key
    Returns:
        Artifact: The latest artifact
    """
    latest_artifact_query = sa.select(orm_models.ArtifactCollection).where(
        orm_models.ArtifactCollection.key == key
    )
    result = await session.execute(latest_artifact_query)
    return result.scalar()

read_latest_artifacts(session, offset=None, limit=None, artifact_filter=None, flow_run_filter=None, task_run_filter=None, deployment_filter=None, flow_filter=None, sort=sorting.ArtifactCollectionSort.ID_DESC) async

Reads artifacts.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
offset	Optional[int]	Query offset	None
limit	Optional[int]	Query limit	None
artifact_filter	Optional[ArtifactCollectionFilter]	Only select artifacts matching this filter	None
flow_run_filter	Optional[FlowRunFilter]	Only select artifacts whose flow runs matching this filter	None
task_run_filter	Optional[TaskRunFilter]	Only select artifacts whose task runs matching this filter	None
deployment_filter	Optional[DeploymentFilter]	Only select artifacts whose flow runs belong to deployments matching this filter	None
flow_filter	Optional[FlowFilter]	Only select artifacts whose flow runs belong to flows matching this filter	None
work_pool_filter		Only select artifacts whose flow runs belong to work pools matching this filter	required

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

async def read_latest_artifacts(
    session: AsyncSession,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
    artifact_filter: Optional[filters.ArtifactCollectionFilter] = None,
    flow_run_filter: Optional[filters.FlowRunFilter] = None,
    task_run_filter: Optional[filters.TaskRunFilter] = None,
    deployment_filter: Optional[filters.DeploymentFilter] = None,
    flow_filter: Optional[filters.FlowFilter] = None,
    sort: sorting.ArtifactCollectionSort = sorting.ArtifactCollectionSort.ID_DESC,
) -> Sequence[orm_models.ArtifactCollection]:
    """
    Reads artifacts.

    Args:
        session: A database session
        offset: Query offset
        limit: Query limit
        artifact_filter: Only select artifacts matching this filter
        flow_run_filter: Only select artifacts whose flow runs matching this filter
        task_run_filter: Only select artifacts whose task runs matching this filter
        deployment_filter: Only select artifacts whose flow runs belong to deployments matching this filter
        flow_filter: Only select artifacts whose flow runs belong to flows matching this filter
        work_pool_filter: Only select artifacts whose flow runs belong to work pools matching this filter
    """
    query = sa.select(orm_models.ArtifactCollection).order_by(sort.as_sql_sort())
    query = await _apply_artifact_collection_filters(
        query,
        artifact_filter=artifact_filter,
        flow_run_filter=flow_run_filter,
        task_run_filter=task_run_filter,
        deployment_filter=deployment_filter,
        flow_filter=flow_filter,
    )

    if offset is not None:
        query = query.offset(offset)
    if limit is not None:
        query = query.limit(limit)

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

update_artifact(session, artifact_id, artifact) async

Updates an artifact by id.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
artifact_id	UUID	The artifact id to update	required
artifact	ArtifactUpdate	An artifact model	required

Returns:

Name	Type	Description
bool	bool	True if the update was successful, False otherwise

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

async def update_artifact(
    session: AsyncSession,
    artifact_id: UUID,
    artifact: actions.ArtifactUpdate,
) -> bool:
    """
    Updates an artifact by id.

    Args:
        session: A database session
        artifact_id (UUID): The artifact id to update
        artifact: An artifact model

    Returns:
        bool: True if the update was successful, False otherwise
    """
    update_artifact_data = artifact.model_dump_for_orm(exclude_unset=True)

    update_artifact_stmt = (
        sa.update(orm_models.Artifact)
        .where(orm_models.Artifact.id == artifact_id)
        .values(**update_artifact_data)
    )

    artifact_result = await session.execute(update_artifact_stmt)

    update_artifact_collection_data = artifact.model_dump_for_orm(exclude_unset=True)
    update_artifact_collection_stmt = (
        sa.update(orm_models.ArtifactCollection)
        .where(orm_models.ArtifactCollection.latest_id == artifact_id)
        .values(**update_artifact_collection_data)
    )
    collection_result = await session.execute(update_artifact_collection_stmt)

    return artifact_result.rowcount + collection_result.rowcount > 0