SQLAlchemy

append_filter_to_statement

append_filter_to_statement(
    statement: Select[_T], model: type[_Model], filter_: _Filter
) -> Select[_T]

Append filtering to statement.

PARAMETER	DESCRIPTION
statement	Some select statement. TYPE: Select[_T]
model	Declaratively defined model. TYPE: type[_Model]
filter_	Filter object. TYPE: _Filter

RAISES	DESCRIPTION
AttributeNotFoundSaDriverError	Attribute not found
RelationshipNotFoundSaDriverError	Relationship not found

Source code in pydantic_filters/drivers/sqlalchemy/_main.py

def append_filter_to_statement(
        statement: sa.Select[_T],
        model: type[_Model],
        filter_: _Filter,
) -> sa.Select[_T]:
    """
    Append filtering to statement.

    Args:
        statement: Some select statement.
        model: Declaratively defined model.
        filter_: Filter object.

    Raises:
        AttributeNotFoundSaDriverError: Attribute not found
        RelationshipNotFoundSaDriverError:  Relationship not found
    """

    join_targets = filter_to_join_targets(filter_, model)
    for target in join_targets:
        statement = statement.join(
            target=target.target,
            onclause=target.on_clause,
        )

    clauses = filter_to_column_clauses(filter_, model)
    if clauses:
        statement = statement.where(*clauses)

    return statement

append_pagination_to_statement

append_pagination_to_statement(
    statement: Select[_T], pagination: _Pagination
) -> Select[_T]

Append pagination to statement.

PARAMETER	DESCRIPTION
statement	Some select statement. TYPE: Select[_T]
pagination	Pagination object. TYPE: _Pagination

RAISES	DESCRIPTION
AttributeNotFoundSaDriverError	Attribute not found

Source code in pydantic_filters/drivers/sqlalchemy/_main.py

def append_pagination_to_statement(
        statement: sa.Select[_T],
        pagination: _Pagination,
) -> sa.Select[_T]:
    """
    Append pagination to statement.

    Args:
        statement: Some select statement.
        pagination: Pagination object.

    Raises:
        AttributeNotFoundSaDriverError: Attribute not found
    """

    return (
        statement
        .limit(pagination.get_limit())
        .offset(pagination.get_offset())
    )

append_sort_to_statement

append_sort_to_statement(
    statement: Select[_T], model: type[_Model], sort: _Sort
) -> Select[_T]

Append sorting to statement.

PARAMETER	DESCRIPTION
statement	Some select statement. TYPE: Select[_T]
model	Declaratively defined model. TYPE: type[_Model]
sort	Sort object. TYPE: _Sort

RAISES	DESCRIPTION
AttributeNotFoundSaDriverError	Attribute not found

Source code in pydantic_filters/drivers/sqlalchemy/_main.py

def append_sort_to_statement(
        statement: sa.Select[_T],
        model: type[_Model],
        sort: _Sort,
) -> sa.Select[_T]:
    """
    Append sorting to statement.

    Args:
        statement: Some select statement.
        model: Declaratively defined model.
        sort: Sort object.

    Raises:
        AttributeNotFoundSaDriverError: Attribute not found
    """

    if sort.sort_by is None:
        return statement

    try:
        column: sa.ColumnElement = getattr(model, sort.sort_by)
    except AttributeError as e:
        raise AttributeNotFoundSaDriverError(
            f"{sort.__class__.__name__}.sort_by: "
            f"Column {model.__name__}.{sort.sort_by} not found",
        ) from e

    return statement.order_by(
        sa.desc(column) if sort.sort_by_order == SortByOrder.desc else sa.asc(column),
    )

append_to_statement

append_to_statement(
    statement: Select[_T],
    model: type[_Model],
    *,
    filter_: _Filter | None = None,
    sort: _Sort | None = None,
    pagination: _Pagination | None = None
) -> Select[_T]

All in one function.

PARAMETER	DESCRIPTION
statement	Some select statement. TYPE: Select[_T]
model	Declaratively defined model. TYPE: type[_Model]
filter_	Filter object. TYPE: _Filter | None DEFAULT: None
sort	Sort object. TYPE: _Sort | None DEFAULT: None
pagination	Pagination object. TYPE: _Pagination | None DEFAULT: None

RAISES	DESCRIPTION
AttributeNotFoundSaDriverError	Attribute not found
RelationshipNotFoundSaDriverError	Relationship not found

Source code in pydantic_filters/drivers/sqlalchemy/_main.py

def append_to_statement(
        statement: sa.Select[_T],
        model: type[_Model],
        *,
        filter_: _Filter | None = None,
        sort: _Sort | None = None,
        pagination: _Pagination | None = None,
) -> sa.Select[_T]:
    """
    All in one function.

    Args:
        statement: Some select statement.
        model: Declaratively defined model.
        filter_: Filter object.
        sort: Sort object.
        pagination: Pagination object.

    Raises:
        AttributeNotFoundSaDriverError: Attribute not found
        RelationshipNotFoundSaDriverError:  Relationship not found
    """

    if filter_ is not None:
        statement = append_filter_to_statement(statement=statement, model=model, filter_=filter_)
    if sort is not None:
        statement = append_sort_to_statement(statement=statement, model=model, sort=sort)
    if pagination is not None:
        statement = append_pagination_to_statement(statement=statement, pagination=pagination)

    return statement

get_count_statement

get_count_statement(
    model: type[_Model], filter_: _Filter
) -> Select[tuple[int]]

Get count statement.

PARAMETER	DESCRIPTION
model	Declaratively defined model. TYPE: type[_Model]
filter_	Filter object. TYPE: _Filter

RAISES	DESCRIPTION
AttributeNotFoundSaDriverError	Attribute not found.
RelationshipNotFoundSaDriverError	Relationship not found.
SupportSaDriverError	Composite primary keys are not supported.

Source code in pydantic_filters/drivers/sqlalchemy/_main.py

def get_count_statement(
        model: type[_Model],
        filter_: _Filter,
) -> sa.Select[tuple[int]]:
    """
    Get count statement.

    Args:
        model: Declaratively defined model.
        filter_: Filter object.

    Raises:
        AttributeNotFoundSaDriverError: Attribute not found.
        RelationshipNotFoundSaDriverError:  Relationship not found.
        SupportSaDriverError: Composite primary keys are not supported.
    """

    primary_key: tuple[sa.ColumnElement, ...] = sa.inspect(model).primary_key
    if len(primary_key) > 1:
        raise SupportSaDriverError("Composite primary keys are not supported")

    statement = sa.select(
        sa.func.count(
            sa.distinct(primary_key[0]),
        ),
    )

    return append_filter_to_statement(statement, model, filter_)

BaseSaDriverError

Bases: Exception

Base SQLAlchemy driver error

AttributeNotFoundSaDriverError

Bases: BaseSaDriverError

RelationshipNotFoundSaDriverError

Bases: BaseSaDriverError

SupportSaDriverError

Bases: BaseSaDriverError

Driver support error