prefect.server.models.variables

count_variables(session, variable_filter=None) async

Count variables, applying filters.

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

async def count_variables(
    session: AsyncSession, variable_filter: Optional[filters.VariableFilter] = None
) -> int:
    """
    Count variables, applying filters.
    """

    query = sa.select(sa.func.count()).select_from(orm_models.Variable)

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

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

create_variable(session, variable) async

Create a variable

Parameters:

Name	Type	Description	Default
session	AsyncSession	async database session	required
variable	VariableCreate	variable to create	required

Returns:

Type	Description
Variable	orm_models.Variable

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

async def create_variable(
    session: AsyncSession, variable: VariableCreate
) -> orm_models.Variable:
    """
    Create a variable

    Args:
        session: async database session
        variable: variable to create

    Returns:
        orm_models.Variable
    """
    model = orm_models.Variable(**variable.model_dump())
    session.add(model)
    await session.flush()

    return model

delete_variable(session, variable_id) async

Delete a variable by id.

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

async def delete_variable(session: AsyncSession, variable_id: UUID) -> bool:
    """
    Delete a variable by id.
    """

    query = sa.delete(orm_models.Variable).where(orm_models.Variable.id == variable_id)

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

delete_variable_by_name(session, name) async

Delete a variable by name.

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

async def delete_variable_by_name(session: AsyncSession, name: str) -> bool:
    """
    Delete a variable by name.
    """

    query = sa.delete(orm_models.Variable).where(orm_models.Variable.name == name)

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

read_variable(session, variable_id) async

Reads a variable by id.

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

async def read_variable(
    session: AsyncSession, variable_id: UUID
) -> Optional[orm_models.Variable]:
    """
    Reads a variable by id.
    """

    query = sa.select(orm_models.Variable).where(orm_models.Variable.id == variable_id)

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

read_variable_by_name(session, name) async

Reads a variable by name.

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

async def read_variable_by_name(
    session: AsyncSession, name: str
) -> Optional[orm_models.Variable]:
    """
    Reads a variable by name.
    """

    query = sa.select(orm_models.Variable).where(orm_models.Variable.name == name)

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

read_variables(session, variable_filter=None, sort=sorting.VariableSort.NAME_ASC, offset=None, limit=None) async

Read variables, applying filers.

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

async def read_variables(
    session: AsyncSession,
    variable_filter: Optional[filters.VariableFilter] = None,
    sort: sorting.VariableSort = sorting.VariableSort.NAME_ASC,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
) -> Sequence[orm_models.Variable]:
    """
    Read variables, applying filers.
    """
    query = sa.select(orm_models.Variable).order_by(sort.as_sql_sort())

    if variable_filter:
        query = query.where(variable_filter.as_sql_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_variable(session, variable_id, variable) async

Updates a variable by id.

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

async def update_variable(
    session: AsyncSession, variable_id: UUID, variable: VariableUpdate
) -> bool:
    """
    Updates a variable by id.
    """
    query = (
        sa.update(orm_models.Variable)
        .where(orm_models.Variable.id == variable_id)
        .values(**variable.model_dump_for_orm(exclude_unset=True))
    )

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

update_variable_by_name(session, name, variable) async

Updates a variable by name.

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

async def update_variable_by_name(
    session: AsyncSession, name: str, variable: VariableUpdate
) -> bool:
    """
    Updates a variable by name.
    """
    query = (
        sa.update(orm_models.Variable)
        .where(orm_models.Variable.name == name)
        .values(**variable.model_dump_for_orm(exclude_unset=True))
    )

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