============ Repositories ============ Advanced Alchemy's repository pattern provides a clean, consistent interface for database operations. This pattern abstracts away the complexity of SQLAlchemy sessions and query-building while providing type-safe operations. Understanding Repositories -------------------------- A repository acts as a collection-like interface to your database models, providing: - Type-safe CRUD operations - Filtering and pagination - Bulk operations - Transaction management - Specialized repository types for common patterns Base Repository Types --------------------- .. list-table:: Repository Types :header-rows: 1 :widths: 30 70 * - Repository Class - Features * - ``SQLAlchemyAsyncRepository`` - | - Async session support | - Basic CRUD operations | - Filtering and pagination | - Bulk operations * - ``SQLAlchemyAsyncSlugRepository`` - | - Async session support | - All base repository features | - Slug-based lookups | - URL-friendly operations * - ``SQLAlchemyAsyncQueryRepository`` - | - Async session support | - Custom query execution | - Complex aggregations | - Raw SQL support * - ``SQLAlchemySyncRepository`` - | - Sync session support | - Basic CRUD operations | - Filtering and pagination | - Bulk operations * - ``SQLAlchemySyncSlugRepository`` - | - Sync session support | - All base repository features | - Slug-based lookups | - URL-friendly operations * - ``SQLAlchemySyncQueryRepository`` - | - Sync session support | - Custom query execution | - Complex aggregations | - Raw SQL support Basic Repository Usage ---------------------- .. note:: The following examples assumes the existence of the ``Post`` model defined in :ref:`many_to_many_relationships` and the ``Tag`` model defined in :ref:`using_unique_mixin`. Let's implement a basic repository for our blog post model: .. code-block:: python from advanced_alchemy.repository import SQLAlchemyAsyncRepository from sqlalchemy.ext.asyncio import AsyncSession from uuid import UUID class PostRepository(SQLAlchemyAsyncRepository[Post]): """Repository for managing blog posts.""" model_type = Post async def create_post(db_session: AsyncSession, title: str, content: str) -> Post: repository = PostRepository(session=db_session) return await repository.add(Post(title=title, content=content), auto_commit=True) Filtering and Querying ---------------------- Advanced Alchemy provides powerful filtering capabilities: .. code-block:: python import datetime from sqlalchemy.ext.asyncio import AsyncSession async def get_recent_posts(db_session: AsyncSession) -> list[Post]: repository = PostRepository(session=db_session) # Create filter for posts from last week return await repository.list( Post.published.is_(True), Post.created_at > (datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta(days=7)) ) Pagination ---------- `list_and_count` enables us to quickly create paginated queries that include a total count of rows. .. code-block:: python from advanced_alchemy.filters import LimitOffset from sqlalchemy.ext.asyncio import AsyncSession async def get_paginated_posts( db_session: AsyncSession, page: int = 1, page_size: int = 20 ) -> tuple[list[Post], int]: repository = PostRepository(session=db_session) offset = (page - 1) * page_size # Get page of results and total count results, total = await repository.list_and_count( LimitOffset(offset=offset, limit=page_size) ) return results, total Bulk Operations --------------- Repositories support efficient bulk operations: Add Many -------- .. code-block:: python from collections.abc import Sequence from sqlalchemy.ext.asyncio import AsyncSession async def create_posts(db_session: AsyncSession, data: list[tuple[str, str]]) -> Sequence[Post]: repository = PostRepository(session=db_session) # Create posts return await repository.add_many( [Post(title=title, content=content) for title, content in data], auto_commit=True, ) Update Many ----------- .. code-block:: python from sqlalchemy.ext.asyncio import AsyncSession async def publish_posts(db_session: AsyncSession, post_ids: list[int]) -> list[Post]: repository = PostRepository(session=db_session) # Fetch posts to update posts = await repository.list(Post.id.in_(post_ids), published=False) # Update all posts for post in posts: post.published = True return await repository.update_many(posts) Delete Many ----------- .. code-block:: python from collections.abc import Sequence from sqlalchemy.ext.asyncio import AsyncSession async def delete_posts(db_session: AsyncSession, post_ids: list[int]) -> Sequence[Post]: repository = PostRepository(session=db_session) return await repository.delete_many(post_ids) Delete Where ------------- .. code-block:: python from collections.abc import Sequence from sqlalchemy.ext.asyncio import AsyncSession async def delete_unpublished_posts(db_session: AsyncSession) -> Sequence[Post]: repository = PostRepository(session=db_session) return await repository.delete_where(Post.published.is_(False)) Transaction Management ---------------------- .. code-block:: python async def create_post_with_tags( db_session: AsyncSession, title: str, content: str, tag_names: list[str] ) -> Post: # Both repositories share the same transaction post_repo = PostRepository(session=db_session) tag_repo = TagRepository(session=db_session) async with db_session.begin(): # Create or get existing tags tags = [] for name in tag_names: tag = await tag_repo.get_one_or_none(name=name) if not tag: tag = await tag_repo.add(Tag(name=name, slug=slugify(name))) tags.append(tag) # Create post with tags return await post_repo.add(Post(title=title, content=content, tags=tags)) .. seealso:: This is just to illustrate the concept. In practice, :class:`UniqueMixin` should be used to handle this lookup even more easily. See :ref:`using_unique_mixin`. Specialized Repositories ------------------------ Advanced Alchemy provides specialized repositories for common patterns. Slug Repository ~~~~~~~~~~~~~~~ For models using the :class:`SlugKey` mixin, there is a specialized Slug repository that adds a ``get_by_slug`` method: .. code-block:: python from advanced_alchemy.repository import SQLAlchemyAsyncSlugRepository from sqlalchemy.ext.asyncio import AsyncSession class TagRepository(SQLAlchemyAsyncSlugRepository[Tag]): """Repository for tags with slug-based lookups.""" model_type = Tag async def get_tag_by_slug(db_session: AsyncSession, slug: str) -> Tag: repository = TagRepository(session=db_session) return await repository.get_by_slug(slug) Query Repository ~~~~~~~~~~~~~~~~ For complex custom queries: .. code-block:: python from typing import Any from advanced_alchemy.repository import SQLAlchemyAsyncQueryRepository from sqlalchemy import select, func, Row from sqlalchemy.ext.asyncio import AsyncSession async def get_posts_count_by_status(db_session: AsyncSession) -> list[Row[Any]]: repository = SQLAlchemyAsyncQueryRepository(session=db_session) return await repository.list(select(Post.published, func.count(Post.id)).group_by(Post.published)) This covers the core functionality of repositories. The next section will explore services, which build upon repositories to provide higher-level business logic and data transformation.