sqlalchemy_async_repositories.base¶

Base repository - orchestrates use case handlers.

Architecture: - use_cases/ - Query/Command handlers organized by use case - pagination/ - Strategy Pattern for pagination backends

Attributes¶

T

Classes¶

BaseRepository

Base repository orchestrating use case handlers.

Module Contents¶

sqlalchemy_async_repositories.base.T¶

class sqlalchemy_async_repositories.base.BaseRepository(db_session: sqlalchemy.ext.asyncio.AsyncSession, model_class: Type[T])¶

Bases: sqlalchemy_async_repositories.interfaces.IRepository[T], Generic[T]

Base repository orchestrating use case handlers.

Structure mirrors GridFlow conventions: - use_cases/{use_case_name}/query_handler.py - use_cases/{use_case_name}/command_handler.py - use_cases/{use_case_name}/query_model.py - use_cases/{use_case_name}/command_model.py

db¶

model_class¶

async get_by_id(entity_id: Any) → T | None¶: Query: Get entity by ID.

async get_all(limit: int | None = None, offset: int = 0) → List[T]¶: Query: Get all entities.

async exists(entity_id: Any) → bool¶: Query: Check if entity exists.

async count() → int¶: Query: Count total entities.

async create(entity: T) → T¶: Command: Create new entity.

async update(entity: T) → T¶: Command: Update existing entity.

async delete(entity_id: Any) → bool¶: Command: Delete entity.

async find_paginated(page: int = 1, page_size: int = 10, filters: List[sqlalchemy_async_repositories.pagination.models.FilterSpec] | None = None, sort: List[sqlalchemy_async_repositories.pagination.models.SortSpec] | None = None) → sqlalchemy_async_repositories.pagination.models.PaginatedResult[T]¶: Query: Paginated results using Strategy Pattern.

get_pagination_backend_info() → dict¶

Get current pagination backend info (debugging/monitoring).

Returns:

dict: Backend metadata (FastCRUD or Native, availability)

Example:

>>> repo = UserRepository(session, UserORM)
>>> info = repo.get_pagination_backend_info()
>>> print(info["backend"])  # "FastCRUD" or "Native"