sqlalchemy_async_repositories.base ================================== .. py:module:: sqlalchemy_async_repositories.base .. autoapi-nested-parse:: Base repository - orchestrates use case handlers. Architecture: - use_cases/ - Query/Command handlers organized by use case - pagination/ - Strategy Pattern for pagination backends Attributes ---------- .. autoapisummary:: sqlalchemy_async_repositories.base.T Classes ------- .. autoapisummary:: sqlalchemy_async_repositories.base.BaseRepository Module Contents --------------- .. py:data:: T .. py:class:: BaseRepository(db_session: sqlalchemy.ext.asyncio.AsyncSession, model_class: Type[T]) Bases: :py:obj:`sqlalchemy_async_repositories.interfaces.IRepository`\ [\ :py:obj:`T`\ ], :py:obj:`Generic`\ [\ :py:obj:`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 .. py:attribute:: db .. py:attribute:: model_class .. py:method:: get_by_id(entity_id: Any) -> Optional[T] :async: Query: Get entity by ID. .. py:method:: get_all(limit: Optional[int] = None, offset: int = 0) -> List[T] :async: Query: Get all entities. .. py:method:: exists(entity_id: Any) -> bool :async: Query: Check if entity exists. .. py:method:: count() -> int :async: Query: Count total entities. .. py:method:: create(entity: T) -> T :async: Command: Create new entity. .. py:method:: update(entity: T) -> T :async: Command: Update existing entity. .. py:method:: delete(entity_id: Any) -> bool :async: Command: Delete entity. .. py:method:: find_paginated(page: int = 1, page_size: int = 10, filters: Optional[List[sqlalchemy_async_repositories.pagination.models.FilterSpec]] = None, sort: Optional[List[sqlalchemy_async_repositories.pagination.models.SortSpec]] = None) -> sqlalchemy_async_repositories.pagination.models.PaginatedResult[T] :async: Query: Paginated results using Strategy Pattern. .. py:method:: 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"