prefect.server.models.flows

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

count_flows(session, flow_filter=None, flow_run_filter=None, task_run_filter=None, deployment_filter=None, work_pool_filter=None) async

Count flows.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
flow_filter	Union[FlowFilter, None]	only count flows that match these filters	None
flow_run_filter	Union[FlowRunFilter, None]	only count flows whose flow runs match these filters	None
task_run_filter	Union[TaskRunFilter, None]	only count flows whose task runs match these filters	None
deployment_filter	Union[DeploymentFilter, None]	only count flows whose deployments match these filters	None
work_pool_filter	Union[WorkPoolFilter, None]	only count flows whose work pools match these filters	None

Returns:

Name	Type	Description
int	int	count of flows

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

async def count_flows(
    session: AsyncSession,
    flow_filter: Union[schemas.filters.FlowFilter, None] = None,
    flow_run_filter: Union[schemas.filters.FlowRunFilter, None] = None,
    task_run_filter: Union[schemas.filters.TaskRunFilter, None] = None,
    deployment_filter: Union[schemas.filters.DeploymentFilter, None] = None,
    work_pool_filter: Union[schemas.filters.WorkPoolFilter, None] = None,
) -> int:
    """
    Count flows.

    Args:
        session: A database session
        flow_filter: only count flows that match these filters
        flow_run_filter: only count flows whose flow runs match these filters
        task_run_filter: only count flows whose task runs match these filters
        deployment_filter: only count flows whose deployments match these filters
        work_pool_filter: only count flows whose work pools match these filters

    Returns:
        int: count of flows
    """

    query = select(sa.func.count(sa.text("*"))).select_from(orm_models.Flow)

    query = await _apply_flow_filters(
        query,
        flow_filter=flow_filter,
        flow_run_filter=flow_run_filter,
        task_run_filter=task_run_filter,
        deployment_filter=deployment_filter,
        work_pool_filter=work_pool_filter,
    )

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

create_flow(db, session, flow) async

Creates a new flow.

If a flow with the same name already exists, the existing flow is returned.

Parameters:

Name	Type	Description	Default
session	AsyncSession	a database session	required
flow	Flow	a flow model	required

Returns:

Type	Description
Flow	orm_models.Flow: the newly-created or existing flow

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

@db_injector
async def create_flow(
    db: PrefectDBInterface, session: AsyncSession, flow: schemas.core.Flow
) -> orm_models.Flow:
    """
    Creates a new flow.

    If a flow with the same name already exists, the existing flow is returned.

    Args:
        session: a database session
        flow: a flow model

    Returns:
        orm_models.Flow: the newly-created or existing flow
    """

    insert_stmt = (
        db.insert(orm_models.Flow)
        .values(**flow.model_dump_for_orm(exclude_unset=True))
        .on_conflict_do_nothing(
            index_elements=db.flow_unique_upsert_columns,
        )
    )
    await session.execute(insert_stmt)

    query = (
        sa.select(orm_models.Flow)
        .where(
            orm_models.Flow.name == flow.name,
        )
        .limit(1)
        .execution_options(populate_existing=True)
    )
    result = await session.execute(query)
    model = result.scalar_one()
    return model

delete_flow(session, flow_id) async

Delete a flow by id.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
flow_id	UUID	a flow id	required

Returns:

Name	Type	Description
bool	bool	whether or not the flow was deleted

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

async def delete_flow(session: AsyncSession, flow_id: UUID) -> bool:
    """
    Delete a flow by id.

    Args:
        session: A database session
        flow_id: a flow id

    Returns:
        bool: whether or not the flow was deleted
    """

    result = await session.execute(
        delete(orm_models.Flow).where(orm_models.Flow.id == flow_id)
    )
    return result.rowcount > 0

read_flow(session, flow_id) async

Reads a flow by id.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
flow_id	UUID	a flow id	required

Returns:

Type	Description
Optional[Flow]	orm_models.Flow: the flow

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

async def read_flow(session: AsyncSession, flow_id: UUID) -> Optional[orm_models.Flow]:
    """
    Reads a flow by id.

    Args:
        session: A database session
        flow_id: a flow id

    Returns:
        orm_models.Flow: the flow
    """
    return await session.get(orm_models.Flow, flow_id)

read_flow_by_name(session, name) async

Reads a flow by name.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
name	str	a flow name	required

Returns:

Type	Description
Optional[Flow]	orm_models.Flow: the flow

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

async def read_flow_by_name(
    session: AsyncSession, name: str
) -> Optional[orm_models.Flow]:
    """
    Reads a flow by name.

    Args:
        session: A database session
        name: a flow name

    Returns:
        orm_models.Flow: the flow
    """

    result = await session.execute(select(orm_models.Flow).filter_by(name=name))
    return result.scalar()

read_flows(session, flow_filter=None, flow_run_filter=None, task_run_filter=None, deployment_filter=None, work_pool_filter=None, sort=schemas.sorting.FlowSort.NAME_ASC, offset=None, limit=None) async

Read multiple flows.

Parameters:

Name	Type	Description	Default
session	AsyncSession	A database session	required
flow_filter	Union[FlowFilter, None]	only select flows that match these filters	None
flow_run_filter	Union[FlowRunFilter, None]	only select flows whose flow runs match these filters	None
task_run_filter	Union[TaskRunFilter, None]	only select flows whose task runs match these filters	None
deployment_filter	Union[DeploymentFilter, None]	only select flows whose deployments match these filters	None
work_pool_filter	Union[WorkPoolFilter, None]	only select flows whose work pools match these filters	None
offset	Union[int, None]	Query offset	None
limit	Union[int, None]	Query limit	None

Returns:

Type	Description
Sequence[Flow]	List[orm_models.Flow]: flows

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

async def read_flows(
    session: AsyncSession,
    flow_filter: Union[schemas.filters.FlowFilter, None] = None,
    flow_run_filter: Union[schemas.filters.FlowRunFilter, None] = None,
    task_run_filter: Union[schemas.filters.TaskRunFilter, None] = None,
    deployment_filter: Union[schemas.filters.DeploymentFilter, None] = None,
    work_pool_filter: Union[schemas.filters.WorkPoolFilter, None] = None,
    sort: schemas.sorting.FlowSort = schemas.sorting.FlowSort.NAME_ASC,
    offset: Union[int, None] = None,
    limit: Union[int, None] = None,
) -> Sequence[orm_models.Flow]:
    """
    Read multiple flows.

    Args:
        session: A database session
        flow_filter: only select flows that match these filters
        flow_run_filter: only select flows whose flow runs match these filters
        task_run_filter: only select flows whose task runs match these filters
        deployment_filter: only select flows whose deployments match these filters
        work_pool_filter: only select flows whose work pools match these filters
        offset: Query offset
        limit: Query limit

    Returns:
        List[orm_models.Flow]: flows
    """

    query = select(orm_models.Flow).order_by(sort.as_sql_sort())

    query = await _apply_flow_filters(
        query,
        flow_filter=flow_filter,
        flow_run_filter=flow_run_filter,
        task_run_filter=task_run_filter,
        deployment_filter=deployment_filter,
        work_pool_filter=work_pool_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_flow(session, flow_id, flow) async

Updates a flow.

Parameters:

Name	Type	Description	Default
session	AsyncSession	a database session	required
flow_id	UUID	the flow id to update	required
flow	FlowUpdate	a flow update model	required

Returns:

Name	Type	Description
bool	bool	whether or not matching rows were found to update

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

async def update_flow(
    session: AsyncSession,
    flow_id: UUID,
    flow: schemas.actions.FlowUpdate,
) -> bool:
    """
    Updates a flow.

    Args:
        session: a database session
        flow_id: the flow id to update
        flow: a flow update model

    Returns:
        bool: whether or not matching rows were found to update
    """
    update_stmt = (
        sa.update(orm_models.Flow)
        .where(orm_models.Flow.id == flow_id)
        # exclude_unset=True allows us to only update values provided by
        # the user, ignoring any defaults on the model
        .values(**flow.model_dump_for_orm(exclude_unset=True))
    )
    result = await session.execute(update_stmt)
    return result.rowcount > 0