Skip to main content

Type safety filter for SQLAlchemy

Project description

SQLAlchemy Filter

Package for convenient filtering functionality in SQLAlchemy

Quite often, optional filtering functionality is required. To facilitate the implementation of filtering functionality, this package was written

Example:

# db/models.py
class Review(Base):
    __tablename__ = "review"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    content: Mapped[str]
    book_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("book.id"))
    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("user.id"))


class Author(Base):
    __tablename__ = "author"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)


class BookAuthor(Base):
    __tablename__ = "book__author"

    book_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("book.id"), primary_key=True)
    author_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey("author.id"),
        primary_key=True,
    )


class Book(Base):
    __tablename__ = "book"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    authors: Mapped[list[Author]] = relationship(secondary=BookAuthor.__table__)
    reviews: Mapped[list[Review]] = relationship()
    created_at: Mapped[datetime]

This is how it was before:

# core/domain/book/dto.py
@dataclass(frozen=True, slots=True)
class BookFilter:
    ident: UUID | None = None
    created_at_from: datetime | None = None
    created_at_to: datetime | None = None

    author_ids: Sequence[UUID] | None = None
    review_ids: Sequence[UUID] | None = None

    review_content_contains: str | None = None


# core/domain/book/repository.py
class BookRepository:
    ...

    async def get_list(self, filter_: BookFilter) -> Sequence[Book]:
        stmt = select(Book)

        if filter_.ident is not None:
            stmt = stmt.where(Book.id == filter_.ident)
        if filter_.created_at_from is not None:
            stmt = stmt.where(Book.created_at >= filter_.created_at_from)
        if filter_.created_at_to is not None:
            stmt = stmt.where(Book.created_at <= filter_.created_at_to)
        if filter_.author_ids is not None:
            stmt = stmt.join(Book.authors).where(Author.id.in_(filter_.author_ids))
        if filter_.review_ids is not None:
            stmt = stmt.join(Book.reviews).where(Review.id.in_(filter_.review_ids))

        return (await self._session.scalars(stmt)).all()

And here's how after using the package:

# core/domain/book/dto.py
from sqlalchemy.sql.operators import eq, ge, icontains_op, in_op, le
from sqla_filter import (
    BaseFilter,
    Unset,
    UNSET,
    FilterField,
    RelationshipInfo,
)


class BookFilter(BaseFilter):
    ident: Annotated[UUID | Unset, FilterField(Book.id, operator=eq)] = UNSET

    created_at_from: Annotated[
        datetime | Unset,
        FilterField(Book.created_at, operator=ge),
    ] = UNSET
    created_at_to: Annotated[
        datetime | Unset,
        FilterField(Book.created_at, operator=le),
    ] = UNSET

    author_ids: Annotated[
        Sequence[UUID] | Unset,
        FilterField(
            Author.id,
            operator=in_op,
            relationship=RelationshipInfo(field=Book.authors),
        ),
    ] = UNSET
    review_ids: Annotated[
        list[UUID] | Unset,
        FilterField(
            Review.id,
            operator=in_op,
            relationship=RelationshipInfo(field=Book.reviews),
        ),
    ] = UNSET
    review_content_contains: Annotated[
        str | Unset,
        FilterField(
            Review.content,
            operator=icontains_op,
            relationship=RelationshipInfo(field=Book.reviews),
        ),
    ] = UNSET
    author_user_id: Annotated[
        UUID | Unset,
        FilterField(
            User.id,
            operator=eq,
            relationships=[
                RelationshipInfo(field=Book.authors),
                RelationshipInfo(field=Author.user),
            ],
        ),
    ] = UNSET


# core/domain/book/repository.py
class BookRepository:
    ...

    async def get_list(self, filter_: BookFilter) -> Sequence[Book]:
        stmt = select(Book)
        stmt = filter_.apply(stmt)
        return (await self._session.scalars(stmt)).all()

If the field is set to "NOT_SET", the filter will not be applied.

The entire example is available at this link

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

sqla_filter-0.2.0.tar.gz (7.0 kB view details)

Uploaded Source

Built Distribution

sqla_filter-0.2.0-py3-none-any.whl (6.3 kB view details)

Uploaded Python 3

File details

Details for the file sqla_filter-0.2.0.tar.gz.

File metadata

  • Download URL: sqla_filter-0.2.0.tar.gz
  • Upload date:
  • Size: 7.0 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: pdm/2.18.1 CPython/3.11.9 Windows/10

File hashes

Hashes for sqla_filter-0.2.0.tar.gz
Algorithm Hash digest
SHA256 85dc945585a975066e01824eba1ed1685ddadb19ef1f49d5ef5e652bb1ce2fb9
MD5 dabf5f4f902ad96a3cb19d861ef30a1f
BLAKE2b-256 c4b75d896bc5a5a562e5faff3f809748cf1fe11041e4e199942dc67f0e286cf1

See more details on using hashes here.

File details

Details for the file sqla_filter-0.2.0-py3-none-any.whl.

File metadata

  • Download URL: sqla_filter-0.2.0-py3-none-any.whl
  • Upload date:
  • Size: 6.3 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: pdm/2.18.1 CPython/3.11.9 Windows/10

File hashes

Hashes for sqla_filter-0.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 5ab5fbb62c39a368cbd71869c06ddf433308b6fa4fd1de5eb35d9086a68dd55a
MD5 0fbb8b0669d73bceadfea4cfa591f07b
BLAKE2b-256 3730b26eb44d5e6ee77d2cbf21f21b1190a6d5c2d6a322db1fc2dc9ed7454104

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page