prefect.server.api.artifacts

Routes for interacting with artifact objects.

count_artifacts(artifacts=None, flow_runs=None, task_runs=None, flows=None, deployments=None, db=Depends(provide_database_interface)) async

Count artifacts from the database.

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

@router.post("/count")
async def count_artifacts(
    artifacts: filters.ArtifactFilter = None,
    flow_runs: filters.FlowRunFilter = None,
    task_runs: filters.TaskRunFilter = None,
    flows: filters.FlowFilter = None,
    deployments: filters.DeploymentFilter = None,
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> int:
    """
    Count artifacts from the database.
    """
    async with db.session_context() as session:
        return await models.artifacts.count_artifacts(
            session=session,
            artifact_filter=artifacts,
            flow_run_filter=flow_runs,
            task_run_filter=task_runs,
            flow_filter=flows,
            deployment_filter=deployments,
        )

count_latest_artifacts(artifacts=None, flow_runs=None, task_runs=None, flows=None, deployments=None, db=Depends(provide_database_interface)) async

Count artifacts from the database.

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

@router.post("/latest/count")
async def count_latest_artifacts(
    artifacts: filters.ArtifactCollectionFilter = None,
    flow_runs: filters.FlowRunFilter = None,
    task_runs: filters.TaskRunFilter = None,
    flows: filters.FlowFilter = None,
    deployments: filters.DeploymentFilter = None,
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> int:
    """
    Count artifacts from the database.
    """
    async with db.session_context() as session:
        return await models.artifacts.count_latest_artifacts(
            session=session,
            artifact_filter=artifacts,
            flow_run_filter=flow_runs,
            task_run_filter=task_runs,
            flow_filter=flows,
            deployment_filter=deployments,
        )

delete_artifact(artifact_id=Path(..., description='The ID of the artifact to delete.', alias='id'), db=Depends(provide_database_interface)) async

Delete an artifact from the database.

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

@router.delete("/{id}", status_code=204)
async def delete_artifact(
    artifact_id: UUID = Path(
        ..., description="The ID of the artifact to delete.", alias="id"
    ),
    db: PrefectDBInterface = Depends(provide_database_interface),
):
    """
    Delete an artifact from the database.
    """
    async with db.session_context(begin_transaction=True) as session:
        result = await models.artifacts.delete_artifact(
            session=session,
            artifact_id=artifact_id,
        )
    if not result:
        raise HTTPException(status_code=404, detail="Artifact not found.")

read_artifact(artifact_id=Path(..., description='The ID of the artifact to retrieve.', alias='id'), db=Depends(provide_database_interface)) async

Retrieve an artifact from the database.

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

@router.get("/{id}")
async def read_artifact(
    artifact_id: UUID = Path(
        ..., description="The ID of the artifact to retrieve.", alias="id"
    ),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> core.Artifact:
    """
    Retrieve an artifact from the database.
    """
    async with db.session_context() as session:
        artifact = await models.artifacts.read_artifact(
            session=session, artifact_id=artifact_id
        )

    if artifact is None:
        raise HTTPException(status_code=404, detail="Artifact not found.")
    return artifact

read_artifacts(sort=Body(sorting.ArtifactSort.ID_DESC), limit=dependencies.LimitBody(), offset=Body(0, ge=0), artifacts=None, flow_runs=None, task_runs=None, flows=None, deployments=None, db=Depends(provide_database_interface)) async

Retrieve artifacts from the database.

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

@router.post("/filter")
async def read_artifacts(
    sort: sorting.ArtifactSort = Body(sorting.ArtifactSort.ID_DESC),
    limit: int = dependencies.LimitBody(),
    offset: int = Body(0, ge=0),
    artifacts: filters.ArtifactFilter = None,
    flow_runs: filters.FlowRunFilter = None,
    task_runs: filters.TaskRunFilter = None,
    flows: filters.FlowFilter = None,
    deployments: filters.DeploymentFilter = None,
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> List[core.Artifact]:
    """
    Retrieve artifacts from the database.
    """
    async with db.session_context() as session:
        return await models.artifacts.read_artifacts(
            session=session,
            artifact_filter=artifacts,
            flow_run_filter=flow_runs,
            task_run_filter=task_runs,
            flow_filter=flows,
            deployment_filter=deployments,
            offset=offset,
            limit=limit,
            sort=sort,
        )

read_latest_artifact(key=Path(..., description='The key of the artifact to retrieve.'), db=Depends(provide_database_interface)) async

Retrieve the latest artifact from the artifact table.

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

@router.get("/{key}/latest")
async def read_latest_artifact(
    key: str = Path(
        ...,
        description="The key of the artifact to retrieve.",
    ),
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> core.Artifact:
    """
    Retrieve the latest artifact from the artifact table.
    """
    async with db.session_context() as session:
        artifact = await models.artifacts.read_latest_artifact(session=session, key=key)

    if artifact is None:
        raise HTTPException(status_code=404, detail="Artifact not found.")
    return artifact

read_latest_artifacts(sort=Body(sorting.ArtifactCollectionSort.ID_DESC), limit=dependencies.LimitBody(), offset=Body(0, ge=0), artifacts=None, flow_runs=None, task_runs=None, flows=None, deployments=None, db=Depends(provide_database_interface)) async

Retrieve artifacts from the database.

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

@router.post("/latest/filter")
async def read_latest_artifacts(
    sort: sorting.ArtifactCollectionSort = Body(sorting.ArtifactCollectionSort.ID_DESC),
    limit: int = dependencies.LimitBody(),
    offset: int = Body(0, ge=0),
    artifacts: filters.ArtifactCollectionFilter = None,
    flow_runs: filters.FlowRunFilter = None,
    task_runs: filters.TaskRunFilter = None,
    flows: filters.FlowFilter = None,
    deployments: filters.DeploymentFilter = None,
    db: PrefectDBInterface = Depends(provide_database_interface),
) -> List[core.ArtifactCollection]:
    """
    Retrieve artifacts from the database.
    """
    async with db.session_context() as session:
        return await models.artifacts.read_latest_artifacts(
            session=session,
            artifact_filter=artifacts,
            flow_run_filter=flow_runs,
            task_run_filter=task_runs,
            flow_filter=flows,
            deployment_filter=deployments,
            offset=offset,
            limit=limit,
            sort=sort,
        )

update_artifact(artifact, artifact_id=Path(..., description='The ID of the artifact to update.', alias='id'), db=Depends(provide_database_interface)) async

Update an artifact in the database.

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

@router.patch("/{id}", status_code=204)
async def update_artifact(
    artifact: actions.ArtifactUpdate,
    artifact_id: UUID = Path(
        ..., description="The ID of the artifact to update.", alias="id"
    ),
    db: PrefectDBInterface = Depends(provide_database_interface),
):
    """
    Update an artifact in the database.
    """
    async with db.session_context(begin_transaction=True) as session:
        result = await models.artifacts.update_artifact(
            session=session,
            artifact_id=artifact_id,
            artifact=artifact,
        )
    if not result:
        raise HTTPException(status_code=404, detail="Artifact not found.")