PostgreSQL

The postgres/sqlalchemy adapter provides a PostgreSQL-specific SQLAlchemy integration, including a concrete adapter, session manager, and session manager registry that extend the base SQLAlchemy components.

Session Managers

PostgreSQL-specific session manager handling connection pooling and lifecycle for PostgreSQL databases.

archipy.adapters.postgres.sqlalchemy.session_managers.PostgresSQlAlchemySessionManager

Bases: BaseSQLAlchemySessionManager[PostgresSQLAlchemyConfig]

Synchronous SQLAlchemy session manager for PostgreSQL.

Inherits from BaseSQLAlchemySessionManager to provide PostgreSQL-specific session management, including connection URL creation and engine configuration.

Parameters:

Name	Type	Description	Default
orm_config	PostgresSQLAlchemyConfig | None	PostgreSQL-specific configuration. If None, uses global config.	None

Source code in archipy/adapters/postgres/sqlalchemy/session_managers.py

class PostgresSQlAlchemySessionManager(BaseSQLAlchemySessionManager[PostgresSQLAlchemyConfig], metaclass=Singleton):
    """Synchronous SQLAlchemy session manager for PostgreSQL.

    Inherits from BaseSQLAlchemySessionManager to provide PostgreSQL-specific session
    management, including connection URL creation and engine configuration.

    Args:
        orm_config: PostgreSQL-specific configuration. If None, uses global config.
    """

    def __init__(self, orm_config: PostgresSQLAlchemyConfig | None = None) -> None:
        """Initialize the PostgreSQL session manager.

        Args:
            orm_config: PostgreSQL-specific configuration. If None, uses global config.
        """
        configs = BaseConfig.global_config().POSTGRES_SQLALCHEMY if orm_config is None else orm_config
        super().__init__(configs)

    @override
    def _expected_config_type(self) -> type[PostgresSQLAlchemyConfig]:
        """Return the expected configuration type for PostgreSQL.

        Returns:
            The PostgresSQLAlchemyConfig class.
        """
        return PostgresSQLAlchemyConfig

    @override
    def _get_database_name(self) -> str:
        """Return the name of the database being used.

        Returns:
            str: The name of the database ('postgresql').
        """
        return "postgresql"

    @override
    def _create_url(self, configs: PostgresSQLAlchemyConfig) -> URL:
        """Create a PostgreSQL connection URL.

        Args:
            configs: PostgreSQL configuration.

        Returns:
            A SQLAlchemy URL object for PostgreSQL.

        Raises:
            DatabaseConnectionError: If there's an error creating the URL.
        """
        try:
            return URL.create(
                drivername=configs.DRIVER_NAME,
                username=configs.USERNAME,
                password=configs.PASSWORD,
                host=configs.HOST,
                port=configs.PORT,
                database=configs.DATABASE,
            )
        except SQLAlchemyError as e:
            raise DatabaseConnectionError(
                database=self._get_database_name(),
            ) from e

archipy.adapters.postgres.sqlalchemy.session_managers.PostgresSQlAlchemySessionManager.engine instance-attribute

engine = _create_engine(orm_config)

archipy.adapters.postgres.sqlalchemy.session_managers.PostgresSQlAlchemySessionManager.get_session

get_session() -> Session

Retrieve a thread-safe SQLAlchemy session.

Returns:

Name	Type	Description
Session	Session	A SQLAlchemy session instance for database operations.

Raises:

Type	Description
DatabaseConnectionError	If there's an error creating the session.
DatabaseConfigurationError	If there's an error in the database configuration.

Source code in archipy/adapters/base/sqlalchemy/session_managers.py

@override
def get_session(self) -> Session:
    """Retrieve a thread-safe SQLAlchemy session.

    Returns:
        Session: A SQLAlchemy session instance for database operations.

    Raises:
        DatabaseConnectionError: If there's an error creating the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    try:
        return self._session_generator()
    except SQLAlchemyError as e:
        if "configuration" in str(e).lower():
            raise DatabaseConfigurationError(
                database=self._get_database_name(),
            ) from e
        raise DatabaseConnectionError(
            database=self._get_database_name(),
        ) from e

archipy.adapters.postgres.sqlalchemy.session_managers.PostgresSQlAlchemySessionManager.remove_session

remove_session() -> None

Remove the current session from the registry.

Cleans up the session to prevent resource leaks, typically called at the end of a request.

Raises:

Type	Description
DatabaseConnectionError	If there's an error removing the session.
DatabaseConfigurationError	If there's an error in the database configuration.

Source code in archipy/adapters/base/sqlalchemy/session_managers.py

@override
def remove_session(self) -> None:
    """Remove the current session from the registry.

    Cleans up the session to prevent resource leaks, typically called at the end
    of a request.

    Raises:
        DatabaseConnectionError: If there's an error removing the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    try:
        self._session_generator.remove()
    except SQLAlchemyError as e:
        if "configuration" in str(e).lower():
            raise DatabaseConfigurationError(
                database=self._get_database_name(),
            ) from e
        raise DatabaseConnectionError(
            database=self._get_database_name(),
        ) from e

archipy.adapters.postgres.sqlalchemy.session_managers.AsyncPostgresSQlAlchemySessionManager

Bases: AsyncBaseSQLAlchemySessionManager[PostgresSQLAlchemyConfig]

Asynchronous SQLAlchemy session manager for PostgreSQL.

Inherits from AsyncBaseSQLAlchemySessionManager to provide async PostgreSQL-specific session management, including connection URL creation and async engine configuration.

Parameters:

Name	Type	Description	Default
orm_config	PostgresSQLAlchemyConfig | None	PostgreSQL-specific configuration. If None, uses global config.	None

Source code in archipy/adapters/postgres/sqlalchemy/session_managers.py

class AsyncPostgresSQlAlchemySessionManager(
    AsyncBaseSQLAlchemySessionManager[PostgresSQLAlchemyConfig],
    metaclass=Singleton,
):
    """Asynchronous SQLAlchemy session manager for PostgreSQL.

    Inherits from AsyncBaseSQLAlchemySessionManager to provide async PostgreSQL-specific
    session management, including connection URL creation and async engine configuration.

    Args:
        orm_config: PostgreSQL-specific configuration. If None, uses global config.
    """

    def __init__(self, orm_config: PostgresSQLAlchemyConfig | None = None) -> None:
        """Initialize the async PostgreSQL session manager.

        Args:
            orm_config: PostgreSQL-specific configuration. If None, uses global config.
        """
        configs = BaseConfig.global_config().POSTGRES_SQLALCHEMY if orm_config is None else orm_config
        super().__init__(configs)

    @override
    def _expected_config_type(self) -> type[PostgresSQLAlchemyConfig]:
        """Return the expected configuration type for PostgreSQL.

        Returns:
            The PostgresSQLAlchemyConfig class.
        """
        return PostgresSQLAlchemyConfig

    @override
    def _get_database_name(self) -> str:
        """Return the name of the database being used.

        Returns:
            str: The name of the database ('postgresql').
        """
        return "postgresql"

    @override
    def _create_url(self, configs: PostgresSQLAlchemyConfig) -> URL:
        """Create an async PostgreSQL connection URL.

        Args:
            configs: PostgreSQL configuration.

        Returns:
            A SQLAlchemy URL object for PostgreSQL.

        Raises:
            DatabaseConnectionError: If there's an error creating the URL.
        """
        try:
            return URL.create(
                drivername=configs.DRIVER_NAME,
                username=configs.USERNAME,
                password=configs.PASSWORD,
                host=configs.HOST,
                port=configs.PORT,
                database=configs.DATABASE,
            )
        except SQLAlchemyError as e:
            raise DatabaseConnectionError(
                database=self._get_database_name(),
            ) from e

archipy.adapters.postgres.sqlalchemy.session_managers.AsyncPostgresSQlAlchemySessionManager.engine instance-attribute

engine = _create_async_engine(orm_config)

archipy.adapters.postgres.sqlalchemy.session_managers.AsyncPostgresSQlAlchemySessionManager.get_session

get_session() -> AsyncSession

Retrieve a task-safe async SQLAlchemy session.

Returns:

Name	Type	Description
AsyncSession	AsyncSession	An async SQLAlchemy session instance for database operations.

Raises:

Type	Description
DatabaseConnectionError	If there's an error creating the session.
DatabaseConfigurationError	If there's an error in the database configuration.

Source code in archipy/adapters/base/sqlalchemy/session_managers.py

@override
def get_session(self) -> AsyncSession:
    """Retrieve a task-safe async SQLAlchemy session.

    Returns:
        AsyncSession: An async SQLAlchemy session instance for database operations.

    Raises:
        DatabaseConnectionError: If there's an error creating the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    try:
        return self._session_generator()
    except SQLAlchemyError as e:
        if "configuration" in str(e).lower():
            raise DatabaseConfigurationError(
                database=self._get_database_name(),
            ) from e
        raise DatabaseConnectionError(
            database=self._get_database_name(),
        ) from e

archipy.adapters.postgres.sqlalchemy.session_managers.AsyncPostgresSQlAlchemySessionManager.remove_session async

remove_session() -> None

Remove the current session from the registry.

Cleans up the session to prevent resource leaks, typically called at the end of a request.

Raises:

Type	Description
DatabaseConnectionError	If there's an error removing the session.
DatabaseConfigurationError	If there's an error in the database configuration.

Source code in archipy/adapters/base/sqlalchemy/session_managers.py

@override
async def remove_session(self) -> None:
    """Remove the current session from the registry.

    Cleans up the session to prevent resource leaks, typically called at the end
    of a request.

    Raises:
        DatabaseConnectionError: If there's an error removing the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    try:
        await self._session_generator.remove()
    except SQLAlchemyError as e:
        if "configuration" in str(e).lower():
            raise DatabaseConfigurationError(
                database=self._get_database_name(),
            ) from e
        raise DatabaseConnectionError(
            database=self._get_database_name(),
        ) from e

options: show_root_toc_entry: false heading_level: 3

Session Manager Registry

Registry for PostgreSQL session manager instances.

archipy.adapters.postgres.sqlalchemy.session_manager_registry.PostgresSessionManagerRegistry

Bases: SessionManagerRegistry

Registry for PostgreSQL SQLAlchemy session managers.

This registry provides a centralized access point for both synchronous and asynchronous PostgreSQL session managers, implementing the Service Locator pattern. It lazily initializes the appropriate session manager when first requested.

The registry maintains singleton instances of: - A synchronous session manager (PostgresSQlAlchemySessionManager) - An asynchronous session manager (AsyncPostgresSQlAlchemySessionManager)

Source code in archipy/adapters/postgres/sqlalchemy/session_manager_registry.py

class PostgresSessionManagerRegistry(SessionManagerRegistry, metaclass=Singleton):
    """Registry for PostgreSQL SQLAlchemy session managers.

    This registry provides a centralized access point for both synchronous and
    asynchronous PostgreSQL session managers, implementing the Service Locator pattern.
    It lazily initializes the appropriate session manager when first requested.

    The registry maintains singleton instances of:
    - A synchronous session manager (PostgresSQlAlchemySessionManager)
    - An asynchronous session manager (AsyncPostgresSQlAlchemySessionManager)
    """

    @classmethod
    def get_sync_manager(cls) -> SessionManagerPort:
        """Get the synchronous PostgreSQL session manager instance.

        Lazily initializes a default PostgresSQlAlchemySessionManager if none has been set.

        Returns:
            SessionManagerPort: The registered synchronous session manager

        Raises:
            DatabaseConnectionError: If there's an error initializing the session manager
        """
        if cls._sync_instance is None:
            try:
                from archipy.adapters.postgres.sqlalchemy.session_managers import PostgresSQlAlchemySessionManager

                cls._sync_instance = PostgresSQlAlchemySessionManager()
            except Exception as e:
                raise DatabaseConnectionError(
                    database="postgresql",
                ) from e
        return cls._sync_instance

    @classmethod
    def set_sync_manager(cls, manager: SessionManagerPort) -> None:
        """Set a custom synchronous session manager.

        Args:
            manager: An instance implementing SessionManagerPort

        Raises:
            InvalidArgumentError: If the manager is None or doesn't implement SessionManagerPort
        """
        if manager is None:
            raise InvalidArgumentError("PostgreSQL session manager cannot be None")
        from archipy.adapters.base.sqlalchemy.session_manager_ports import SessionManagerPort

        if not isinstance(manager, SessionManagerPort):
            raise InvalidArgumentError(f"Manager must implement SessionManagerPort, got {type(manager).__name__}")
        cls._sync_instance = manager

    @classmethod
    def get_async_manager(cls) -> AsyncSessionManagerPort:
        """Get the asynchronous PostgreSQL session manager instance.

        Lazily initializes a default AsyncPostgresSQlAlchemySessionManager if none has been set.

        Returns:
            AsyncSessionManagerPort: The registered asynchronous session manager

        Raises:
            DatabaseConnectionError: If there's an error initializing the session manager
        """
        if cls._async_instance is None:
            try:
                from archipy.adapters.postgres.sqlalchemy.session_managers import AsyncPostgresSQlAlchemySessionManager

                cls._async_instance = AsyncPostgresSQlAlchemySessionManager()
            except Exception as e:
                raise DatabaseConnectionError(
                    database="postgresql",
                ) from e
        return cls._async_instance

    @classmethod
    def set_async_manager(cls, manager: AsyncSessionManagerPort) -> None:
        """Set a custom asynchronous session manager.

        Args:
            manager: An instance implementing AsyncSessionManagerPort

        Raises:
            InvalidArgumentError: If the manager is None or doesn't implement AsyncSessionManagerPort
        """
        if manager is None:
            raise InvalidArgumentError("PostgreSQL async session manager cannot be None")
        from archipy.adapters.base.sqlalchemy.session_manager_ports import AsyncSessionManagerPort

        if not isinstance(manager, AsyncSessionManagerPort):
            raise InvalidArgumentError(f"Manager must implement AsyncSessionManagerPort, got {type(manager).__name__}")
        cls._async_instance = manager

    @classmethod
    def reset(cls) -> None:
        """Reset the registry to its initial state.

        This method clears both registered managers, useful for testing.
        """
        cls._sync_instance = None
        cls._async_instance = None

archipy.adapters.postgres.sqlalchemy.session_manager_registry.PostgresSessionManagerRegistry.get_sync_manager classmethod

get_sync_manager() -> SessionManagerPort

Get the synchronous PostgreSQL session manager instance.

Lazily initializes a default PostgresSQlAlchemySessionManager if none has been set.

Returns:

Name	Type	Description
SessionManagerPort	SessionManagerPort	The registered synchronous session manager

Raises:

Type	Description
DatabaseConnectionError	If there's an error initializing the session manager

Source code in archipy/adapters/postgres/sqlalchemy/session_manager_registry.py

@classmethod
def get_sync_manager(cls) -> SessionManagerPort:
    """Get the synchronous PostgreSQL session manager instance.

    Lazily initializes a default PostgresSQlAlchemySessionManager if none has been set.

    Returns:
        SessionManagerPort: The registered synchronous session manager

    Raises:
        DatabaseConnectionError: If there's an error initializing the session manager
    """
    if cls._sync_instance is None:
        try:
            from archipy.adapters.postgres.sqlalchemy.session_managers import PostgresSQlAlchemySessionManager

            cls._sync_instance = PostgresSQlAlchemySessionManager()
        except Exception as e:
            raise DatabaseConnectionError(
                database="postgresql",
            ) from e
    return cls._sync_instance

archipy.adapters.postgres.sqlalchemy.session_manager_registry.PostgresSessionManagerRegistry.set_sync_manager classmethod

set_sync_manager(manager: SessionManagerPort) -> None

Set a custom synchronous session manager.

Parameters:

Name	Type	Description	Default
manager	SessionManagerPort	An instance implementing SessionManagerPort	required

Raises:

Type	Description
InvalidArgumentError	If the manager is None or doesn't implement SessionManagerPort

Source code in archipy/adapters/postgres/sqlalchemy/session_manager_registry.py

@classmethod
def set_sync_manager(cls, manager: SessionManagerPort) -> None:
    """Set a custom synchronous session manager.

    Args:
        manager: An instance implementing SessionManagerPort

    Raises:
        InvalidArgumentError: If the manager is None or doesn't implement SessionManagerPort
    """
    if manager is None:
        raise InvalidArgumentError("PostgreSQL session manager cannot be None")
    from archipy.adapters.base.sqlalchemy.session_manager_ports import SessionManagerPort

    if not isinstance(manager, SessionManagerPort):
        raise InvalidArgumentError(f"Manager must implement SessionManagerPort, got {type(manager).__name__}")
    cls._sync_instance = manager

archipy.adapters.postgres.sqlalchemy.session_manager_registry.PostgresSessionManagerRegistry.get_async_manager classmethod

get_async_manager() -> AsyncSessionManagerPort

Get the asynchronous PostgreSQL session manager instance.

Lazily initializes a default AsyncPostgresSQlAlchemySessionManager if none has been set.

Returns:

Name	Type	Description
AsyncSessionManagerPort	AsyncSessionManagerPort	The registered asynchronous session manager

Raises:

Type	Description
DatabaseConnectionError	If there's an error initializing the session manager

Source code in archipy/adapters/postgres/sqlalchemy/session_manager_registry.py

@classmethod
def get_async_manager(cls) -> AsyncSessionManagerPort:
    """Get the asynchronous PostgreSQL session manager instance.

    Lazily initializes a default AsyncPostgresSQlAlchemySessionManager if none has been set.

    Returns:
        AsyncSessionManagerPort: The registered asynchronous session manager

    Raises:
        DatabaseConnectionError: If there's an error initializing the session manager
    """
    if cls._async_instance is None:
        try:
            from archipy.adapters.postgres.sqlalchemy.session_managers import AsyncPostgresSQlAlchemySessionManager

            cls._async_instance = AsyncPostgresSQlAlchemySessionManager()
        except Exception as e:
            raise DatabaseConnectionError(
                database="postgresql",
            ) from e
    return cls._async_instance

archipy.adapters.postgres.sqlalchemy.session_manager_registry.PostgresSessionManagerRegistry.set_async_manager classmethod

set_async_manager(manager: AsyncSessionManagerPort) -> None

Set a custom asynchronous session manager.

Parameters:

Name	Type	Description	Default
manager	AsyncSessionManagerPort	An instance implementing AsyncSessionManagerPort	required

Raises:

Type	Description
InvalidArgumentError	If the manager is None or doesn't implement AsyncSessionManagerPort

Source code in archipy/adapters/postgres/sqlalchemy/session_manager_registry.py

@classmethod
def set_async_manager(cls, manager: AsyncSessionManagerPort) -> None:
    """Set a custom asynchronous session manager.

    Args:
        manager: An instance implementing AsyncSessionManagerPort

    Raises:
        InvalidArgumentError: If the manager is None or doesn't implement AsyncSessionManagerPort
    """
    if manager is None:
        raise InvalidArgumentError("PostgreSQL async session manager cannot be None")
    from archipy.adapters.base.sqlalchemy.session_manager_ports import AsyncSessionManagerPort

    if not isinstance(manager, AsyncSessionManagerPort):
        raise InvalidArgumentError(f"Manager must implement AsyncSessionManagerPort, got {type(manager).__name__}")
    cls._async_instance = manager

archipy.adapters.postgres.sqlalchemy.session_manager_registry.PostgresSessionManagerRegistry.reset classmethod

reset() -> None

Reset the registry to its initial state.

This method clears both registered managers, useful for testing.

Source code in archipy/adapters/postgres/sqlalchemy/session_manager_registry.py

@classmethod
def reset(cls) -> None:
    """Reset the registry to its initial state.

    This method clears both registered managers, useful for testing.
    """
    cls._sync_instance = None
    cls._async_instance = None

options: show_root_toc_entry: false heading_level: 3

Adapters

Concrete PostgreSQL adapter built on top of the base SQLAlchemy adapter with PostgreSQL-specific configuration.

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter

Bases: BaseSQLAlchemyAdapter[PostgresSQLAlchemyConfig]

Synchronous SQLAlchemy adapter for PostgreSQL.

Inherits from BaseSQLAlchemyAdapter to provide PostgreSQL-specific session management and database operations.

Parameters:

Name	Type	Description	Default
orm_config	PostgresSQLAlchemyConfig | None	PostgreSQL-specific configuration. If None, uses global config.	None

Source code in archipy/adapters/postgres/sqlalchemy/adapters.py

class PostgresSQLAlchemyAdapter(BaseSQLAlchemyAdapter[PostgresSQLAlchemyConfig]):
    """Synchronous SQLAlchemy adapter for PostgreSQL.

    Inherits from BaseSQLAlchemyAdapter to provide PostgreSQL-specific session management
    and database operations.

    Args:
        orm_config: PostgreSQL-specific configuration. If None, uses global config.
    """

    def __init__(self, orm_config: PostgresSQLAlchemyConfig | None = None) -> None:
        """Initialize the PostgreSQL adapter with a session manager.

        Args:
            orm_config: PostgreSQL-specific configuration. If None, uses global config.
        """
        configs = BaseConfig.global_config().POSTGRES_SQLALCHEMY if orm_config is None else orm_config
        super().__init__(configs)

    @override
    def _create_session_manager(self, configs: PostgresSQLAlchemyConfig) -> PostgresSQlAlchemySessionManager:
        """Create a PostgreSQL-specific session manager.

        Args:
            configs: PostgreSQL configuration.

        Returns:
            A PostgreSQL session manager instance.
        """
        return PostgresSQlAlchemySessionManager(configs)

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.session_manager instance-attribute

session_manager: BaseSQLAlchemySessionManager[ConfigT] = (
    _create_session_manager(configs)
)

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.get_session

get_session() -> Session

Get a database session.

Returns:

Name	Type	Description
Session	Session	A SQLAlchemy session.

Raises:

Type	Description
DatabaseConnectionError	If there's an error getting the session.
DatabaseConfigurationError	If there's an error in the database configuration.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def get_session(self) -> Session:
    """Get a database session.

    Returns:
        Session: A SQLAlchemy session.

    Raises:
        DatabaseConnectionError: If there's an error getting the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    return self.session_manager.get_session()

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.execute_search_query

execute_search_query(
    entity: type[T],
    query: Select,
    pagination: PaginationDTO | None = None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
) -> tuple[list[T], int]

Execute a search query with pagination and sorting.

Parameters:

Name	Type	Description	Default
entity	type[T]	The entity class to query.	required
query	Select	The SQLAlchemy SELECT query.	required
pagination	PaginationDTO | None	Optional pagination settings.	None
sort_info	SortDTO | None	Optional sorting information.	None
has_multiple_entities	bool	Optional bool.	False

Returns:

Type	Description
tuple[list[T], int]	Tuple of the list of entities and the total count.

Raises:

Type	Description
DatabaseQueryError	If the database query fails.
DatabaseTimeoutError	If the query times out.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def execute_search_query(
    self,
    entity: type[T],
    query: Select,
    pagination: PaginationDTO | None = None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
) -> tuple[list[T], int]:
    """Execute a search query with pagination and sorting.

    Args:
        entity: The entity class to query.
        query: The SQLAlchemy SELECT query.
        pagination: Optional pagination settings.
        sort_info: Optional sorting information.
        has_multiple_entities: Optional bool.

    Returns:
        Tuple of the list of entities and the total count.

    Raises:
        DatabaseQueryError: If the database query fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        sort_info = sort_info or SortDTO.default()
        session = self.get_session()
        sorted_query = self._apply_sorting(entity, query, sort_info)
        paginated_query = self._apply_pagination(sorted_query, pagination)
        result_set = session.execute(paginated_query)
        if has_multiple_entities:
            # For multiple entities, fetchall returns list of Row objects
            raw_results = list(result_set.fetchall())
            # Convert to list[T] - each row contains entities of type T
            # Use tuple unpacking to access the first element
            results: list[T] = []
            for row in raw_results:
                if row:
                    # Row supports indexing and tuple unpacking
                    row_tuple = tuple(row)
                    if row_tuple:
                        first_entity = row_tuple[0]
                        # first_entity is T (entity type), verify it's an instance
                        if isinstance(first_entity, entity):
                            results.append(first_entity)
        else:
            # For single entity, scalars() returns list[T] directly
            results = list(result_set.scalars().all())
        count_query = select(func.count()).select_from(query.subquery())
        total_count = session.execute(count_query).scalar_one()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        # Type: results is list[T] where T extends BaseEntity, total_count is int
        return results, total_count

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.create

create(entity: T) -> T | None

Create a new entity in the database.

Parameters:

Name	Type	Description	Default
entity	T	The entity to create.	required

Returns:

Type	Description
T | None	The created entity with updated attributes, preserving the original type.

Raises:

Type	Description
InvalidEntityTypeError	If the entity type is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseIntegrityError	If there's an integrity constraint violation.
DatabaseConstraintError	If there's a constraint violation.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def create(self, entity: T) -> T | None:
    """Create a new entity in the database.

    Args:
        entity: The entity to create.

    Returns:
        The created entity with updated attributes, preserving the original type.

    Raises:
        InvalidEntityTypeError: If the entity type is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not isinstance(entity, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {type(entity).__name__}",
            expected_type="BaseEntity",
            actual_type=type(entity).__name__,
        )

    try:
        session = self.get_session()
        session.add(entity)
        session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return entity

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.bulk_create

bulk_create(entities: list[T]) -> list[T] | None

Creates multiple entities in a single database operation.

Parameters:

Name	Type	Description	Default
entities	list[T]	List of entities to create.	required

Returns:

Type	Description
list[T] | None	List of created entities with updated attributes, preserving original types.

Raises:

Type	Description
InvalidEntityTypeError	If any entity is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseIntegrityError	If there's an integrity constraint violation.
DatabaseConstraintError	If there's a constraint violation.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def bulk_create(self, entities: list[T]) -> list[T] | None:
    """Creates multiple entities in a single database operation.

    Args:
        entities: List of entities to create.

    Returns:
        List of created entities with updated attributes, preserving original types.

    Raises:
        InvalidEntityTypeError: If any entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not all(isinstance(entity, BaseEntity) for entity in entities):
        raise InvalidEntityTypeError(
            message="All entities must be BaseEntity subclasses",
            expected_type="BaseEntity",
            actual_type="mixed",
        )

    try:
        session = self.get_session()
        session.add_all(entities)
        session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return entities

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.get_by_uuid

get_by_uuid(
    entity_type: type[T], entity_uuid: UUID
) -> BaseEntity | None

Retrieve an entity by its UUID.

Parameters:

Name	Type	Description	Default
entity_type	type[T]	The type of entity to retrieve.	required
entity_uuid	UUID	The UUID of the entity.	required

Returns:

Type	Description
BaseEntity | None	The entity if found, None otherwise.

Raises:

Type	Description
InvalidEntityTypeError	If the entity type is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseTimeoutError	If the query times out.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def get_by_uuid(self, entity_type: type[T], entity_uuid: UUID) -> BaseEntity | None:
    """Retrieve an entity by its UUID.

    Args:
        entity_type: The type of entity to retrieve.
        entity_uuid: The UUID of the entity.

    Returns:
        The entity if found, None otherwise.

    Raises:
        InvalidEntityTypeError: If the entity type is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not issubclass(entity_type, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {entity_type.__name__}",
            expected_type="BaseEntity",
            actual_type=entity_type.__name__,
        )

    try:
        session = self.get_session()
        result = session.get(entity_type, entity_uuid)
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        # result is T | None where T extends BaseEntity, compatible with BaseEntity | None
        # The type checker needs explicit type annotation to understand the relationship
        typed_result: BaseEntity | None = result
        return typed_result

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.delete

delete(entity: T) -> None

Delete an entity from the database.

Parameters:

Name	Type	Description	Default
entity	T	The entity to delete.	required

Raises:

Type	Description
InvalidEntityTypeError	If the entity is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseIntegrityError	If there's an integrity constraint violation.
DatabaseConstraintError	If there's a constraint violation.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def delete(self, entity: T) -> None:
    """Delete an entity from the database.

    Args:
        entity: The entity to delete.

    Raises:
        InvalidEntityTypeError: If the entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not isinstance(entity, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {type(entity).__name__}",
            expected_type="BaseEntity",
            actual_type=type(entity).__name__,
        )

    try:
        session = self.get_session()
        session.delete(entity)
        session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.bulk_delete

bulk_delete(entities: list[T]) -> None

Delete multiple entities from the database.

Parameters:

Name	Type	Description	Default
entities	list[T]	List of entities to delete.	required

Raises:

Type	Description
InvalidEntityTypeError	If any entity is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseIntegrityError	If there's an integrity constraint violation.
DatabaseConstraintError	If there's a constraint violation.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def bulk_delete(self, entities: list[T]) -> None:
    """Delete multiple entities from the database.

    Args:
        entities: List of entities to delete.

    Raises:
        InvalidEntityTypeError: If any entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not all(isinstance(entity, BaseEntity) for entity in entities):
        raise InvalidEntityTypeError(
            message="All entities must be BaseEntity subclasses",
            expected_type="BaseEntity",
            actual_type="mixed",
        )

    try:
        session = self.get_session()
        for entity in entities:
            session.delete(entity)
        session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.execute

execute(
    statement: Executable,
    params: AnyExecuteParams | None = None,
) -> Result[Any]

Execute a SQLAlchemy statement.

Parameters:

Name	Type	Description	Default
statement	Executable	The SQLAlchemy statement to execute.	required
params	AnyExecuteParams | None	Optional parameters for the statement.	None

Returns:

Type	Description
Result[Any]	The result of the execution.

Raises:

Type	Description
DatabaseQueryError	If the database operation fails.
DatabaseTimeoutError	If the query times out.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def execute(self, statement: Executable, params: AnyExecuteParams | None = None) -> Result[Any]:
    """Execute a SQLAlchemy statement.

    Args:
        statement: The SQLAlchemy statement to execute.
        params: Optional parameters for the statement.

    Returns:
        The result of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = session.execute(statement, params or {})
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

archipy.adapters.postgres.sqlalchemy.adapters.PostgresSQLAlchemyAdapter.scalars

scalars(
    statement: Executable,
    params: AnyExecuteParams | None = None,
) -> ScalarResult[Any]

Execute a SQLAlchemy statement and return scalar results.

Parameters:

Name	Type	Description	Default
statement	Executable	The SQLAlchemy statement to execute.	required
params	AnyExecuteParams | None	Optional parameters for the statement.	None

Returns:

Type	Description
ScalarResult[Any]	The scalar results of the execution.

Raises:

Type	Description
DatabaseQueryError	If the database operation fails.
DatabaseTimeoutError	If the query times out.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def scalars(self, statement: Executable, params: AnyExecuteParams | None = None) -> ScalarResult[Any]:
    """Execute a SQLAlchemy statement and return scalar results.

    Args:
        statement: The SQLAlchemy statement to execute.
        params: Optional parameters for the statement.

    Returns:
        The scalar results of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = session.scalars(statement, params or {})
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter

Bases: AsyncBaseSQLAlchemyAdapter[PostgresSQLAlchemyConfig]

Asynchronous SQLAlchemy adapter for PostgreSQL.

Inherits from AsyncBaseSQLAlchemyAdapter to provide async PostgreSQL-specific session management and database operations.

Parameters:

Name	Type	Description	Default
orm_config	PostgresSQLAlchemyConfig | None	PostgreSQL-specific configuration. If None, uses global config.	None

Source code in archipy/adapters/postgres/sqlalchemy/adapters.py

class AsyncPostgresSQLAlchemyAdapter(AsyncBaseSQLAlchemyAdapter[PostgresSQLAlchemyConfig]):
    """Asynchronous SQLAlchemy adapter for PostgreSQL.

    Inherits from AsyncBaseSQLAlchemyAdapter to provide async PostgreSQL-specific session
    management and database operations.

    Args:
        orm_config: PostgreSQL-specific configuration. If None, uses global config.
    """

    def __init__(self, orm_config: PostgresSQLAlchemyConfig | None = None) -> None:
        """Initialize the async PostgreSQL adapter with a session manager.

        Args:
            orm_config: PostgreSQL-specific configuration. If None, uses global config.
        """
        configs = BaseConfig.global_config().POSTGRES_SQLALCHEMY if orm_config is None else orm_config
        super().__init__(configs)

    @override
    def _create_async_session_manager(self, configs: PostgresSQLAlchemyConfig) -> AsyncPostgresSQlAlchemySessionManager:
        """Create an async PostgreSQL-specific session manager.

        Args:
            configs: PostgreSQL configuration.

        Returns:
            An async PostgreSQL session manager instance.
        """
        return AsyncPostgresSQlAlchemySessionManager(configs)

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.session_manager instance-attribute

session_manager: AsyncBaseSQLAlchemySessionManager[
    ConfigT
] = _create_async_session_manager(configs)

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.get_session

get_session() -> AsyncSession

Get a database session.

Returns:

Name	Type	Description
AsyncSession	AsyncSession	A SQLAlchemy async session.

Raises:

Type	Description
DatabaseConnectionError	If there's an error getting the session.
DatabaseConfigurationError	If there's an error in the database configuration.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
def get_session(self) -> AsyncSession:
    """Get a database session.

    Returns:
        AsyncSession: A SQLAlchemy async session.

    Raises:
        DatabaseConnectionError: If there's an error getting the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    return self.session_manager.get_session()

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.execute_search_query async

execute_search_query(
    entity: type[T],
    query: Select,
    pagination: PaginationDTO | None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
) -> tuple[list[T], int]

Execute a search query with pagination and sorting.

Parameters:

Name	Type	Description	Default
entity	type[T]	The entity class to query.	required
query	Select	The SQLAlchemy SELECT query.	required
pagination	PaginationDTO | None	Optional pagination settings.	required
sort_info	SortDTO | None	Optional sorting information.	None
has_multiple_entities	bool	Optional bool	False

Returns:

Type	Description
tuple[list[T], int]	Tuple of the list of entities and the total count.

Raises:

Type	Description
DatabaseQueryError	If the database query fails.
DatabaseTimeoutError	If the query times out.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
async def execute_search_query(
    self,
    entity: type[T],
    query: Select,
    pagination: PaginationDTO | None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
) -> tuple[list[T], int]:
    """Execute a search query with pagination and sorting.

    Args:
        entity: The entity class to query.
        query: The SQLAlchemy SELECT query.
        pagination: Optional pagination settings.
        sort_info: Optional sorting information.
        has_multiple_entities: Optional bool

    Returns:
        Tuple of the list of entities and the total count.

    Raises:
        DatabaseQueryError: If the database query fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        sort_info = sort_info or SortDTO.default()
        session = self.get_session()
        sorted_query = self._apply_sorting(entity, query, sort_info)
        paginated_query = self._apply_pagination(sorted_query, pagination)
        result_set = await session.execute(paginated_query)
        if has_multiple_entities:
            # For multiple entities, fetchall returns list of Row objects
            raw_results = list(result_set.fetchall())
            # Convert to list[T] - each row contains entities of type T
            # Use tuple unpacking to access the first element
            results: list[T] = []
            for row in raw_results:
                if row:
                    # Row supports indexing and tuple unpacking
                    row_tuple = tuple(row)
                    if row_tuple:
                        first_entity = row_tuple[0]
                        # first_entity is T (entity type), verify it's an instance
                        if isinstance(first_entity, entity):
                            results.append(first_entity)
        else:
            # For single entity, scalars() returns list[T] directly
            results = list(result_set.scalars().all())
        count_query = select(func.count()).select_from(query.subquery())
        total_count_result = await session.execute(count_query)
        total_count = total_count_result.scalar_one()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        # Type: results is list[T] where T extends BaseEntity, total_count is int
        return results, total_count

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.create async

create(entity: T) -> T | None

Create a new entity in the database.

Parameters:

Name	Type	Description	Default
entity	T	The entity to create.	required

Returns:

Type	Description
T | None	The created entity with updated attributes, preserving the original type.

Raises:

Type	Description
InvalidEntityTypeError	If the entity type is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseIntegrityError	If there's an integrity constraint violation.
DatabaseConstraintError	If there's a constraint violation.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
async def create(self, entity: T) -> T | None:
    """Create a new entity in the database.

    Args:
        entity: The entity to create.

    Returns:
        The created entity with updated attributes, preserving the original type.

    Raises:
        InvalidEntityTypeError: If the entity type is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not isinstance(entity, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {type(entity).__name__}",
            expected_type="BaseEntity",
            actual_type=type(entity).__name__,
        )

    try:
        session = self.get_session()
        session.add(entity)
        await session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return entity

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.bulk_create async

bulk_create(entities: list[T]) -> list[T] | None

Creates multiple entities in a single database operation.

Parameters:

Name	Type	Description	Default
entities	list[T]	List of entities to create.	required

Returns:

Type	Description
list[T] | None	List of created entities with updated attributes, preserving original types.

Raises:

Type	Description
InvalidEntityTypeError	If any entity is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseIntegrityError	If there's an integrity constraint violation.
DatabaseConstraintError	If there's a constraint violation.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
async def bulk_create(self, entities: list[T]) -> list[T] | None:
    """Creates multiple entities in a single database operation.

    Args:
        entities: List of entities to create.

    Returns:
        List of created entities with updated attributes, preserving original types.

    Raises:
        InvalidEntityTypeError: If any entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not all(isinstance(entity, BaseEntity) for entity in entities):
        raise InvalidEntityTypeError(
            message="All entities must be BaseEntity subclasses",
            expected_type="BaseEntity",
            actual_type="mixed",
        )

    try:
        session = self.get_session()
        session.add_all(entities)
        await session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return entities

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.get_by_uuid async

get_by_uuid(
    entity_type: type[T], entity_uuid: UUID
) -> BaseEntity | None

Retrieve an entity by its UUID.

Parameters:

Name	Type	Description	Default
entity_type	type[T]	The type of entity to retrieve.	required
entity_uuid	UUID	The UUID of the entity.	required

Returns:

Type	Description
BaseEntity | None	The entity if found, None otherwise.

Raises:

Type	Description
InvalidEntityTypeError	If the entity type is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseTimeoutError	If the query times out.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
async def get_by_uuid(self, entity_type: type[T], entity_uuid: UUID) -> BaseEntity | None:
    """Retrieve an entity by its UUID.

    Args:
        entity_type: The type of entity to retrieve.
        entity_uuid: The UUID of the entity.

    Returns:
        The entity if found, None otherwise.

    Raises:
        InvalidEntityTypeError: If the entity type is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not issubclass(entity_type, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {entity_type.__name__}",
            expected_type="BaseEntity",
            actual_type=entity_type.__name__,
        )

    try:
        session = self.get_session()
        result = await session.get(entity_type, entity_uuid)
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        # result is T | None where T extends BaseEntity, compatible with BaseEntity | None
        # The type checker needs explicit type annotation to understand the relationship
        typed_result: BaseEntity | None = result
        return typed_result

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.delete async

delete(entity: BaseEntity) -> None

Delete an entity from the database.

Parameters:

Name	Type	Description	Default
entity	BaseEntity	The entity to delete.	required

Raises:

Type	Description
InvalidEntityTypeError	If the entity is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseIntegrityError	If there's an integrity constraint violation.
DatabaseConstraintError	If there's a constraint violation.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
async def delete(self, entity: BaseEntity) -> None:
    """Delete an entity from the database.

    Args:
        entity: The entity to delete.

    Raises:
        InvalidEntityTypeError: If the entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not isinstance(entity, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {type(entity).__name__}",
            expected_type="BaseEntity",
            actual_type=type(entity).__name__,
        )

    try:
        session = self.get_session()
        await session.delete(entity)
        await session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.bulk_delete async

bulk_delete(entities: list[T]) -> None

Delete multiple entities from the database.

Parameters:

Name	Type	Description	Default
entities	list[T]	List of entities to delete.	required

Raises:

Type	Description
InvalidEntityTypeError	If any entity is not a valid SQLAlchemy model.
DatabaseQueryError	If the database operation fails.
DatabaseIntegrityError	If there's an integrity constraint violation.
DatabaseConstraintError	If there's a constraint violation.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
async def bulk_delete(self, entities: list[T]) -> None:
    """Delete multiple entities from the database.

    Args:
        entities: List of entities to delete.

    Raises:
        InvalidEntityTypeError: If any entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not all(isinstance(entity, BaseEntity) for entity in entities):
        raise InvalidEntityTypeError(
            message="All entities must be BaseEntity subclasses",
            expected_type="BaseEntity",
            actual_type="mixed",
        )

    try:
        session = self.get_session()
        for entity in entities:
            await session.delete(entity)
        await session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.execute async

execute(
    statement: Executable,
    params: AnyExecuteParams | None = None,
) -> Result[Any]

Execute a SQLAlchemy statement.

Parameters:

Name	Type	Description	Default
statement	Executable	The SQLAlchemy statement to execute.	required
params	AnyExecuteParams | None	Optional parameters for the statement.	None

Returns:

Type	Description
Result[Any]	The result of the execution.

Raises:

Type	Description
DatabaseQueryError	If the database operation fails.
DatabaseTimeoutError	If the query times out.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
async def execute(self, statement: Executable, params: AnyExecuteParams | None = None) -> Result[Any]:
    """Execute a SQLAlchemy statement.

    Args:
        statement: The SQLAlchemy statement to execute.
        params: Optional parameters for the statement.

    Returns:
        The result of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = await session.execute(statement, params or {})
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

archipy.adapters.postgres.sqlalchemy.adapters.AsyncPostgresSQLAlchemyAdapter.scalars async

scalars(
    statement: Executable,
    params: AnyExecuteParams | None = None,
) -> ScalarResult[Any]

Execute a SQLAlchemy statement and return scalar results.

Parameters:

Name	Type	Description	Default
statement	Executable	The SQLAlchemy statement to execute.	required
params	AnyExecuteParams | None	Optional parameters for the statement.	None

Returns:

Type	Description
ScalarResult[Any]	The scalar results of the execution.

Raises:

Type	Description
DatabaseQueryError	If the database operation fails.
DatabaseTimeoutError	If the query times out.
DatabaseConnectionError	If there's a connection error.
DatabaseTransactionError	If there's a transaction error.

Source code in archipy/adapters/base/sqlalchemy/adapters.py

@override
async def scalars(self, statement: Executable, params: AnyExecuteParams | None = None) -> ScalarResult[Any]:
    """Execute a SQLAlchemy statement and return scalar results.

    Args:
        statement: The SQLAlchemy statement to execute.
        params: Optional parameters for the statement.

    Returns:
        The scalar results of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = await session.scalars(statement, params or {})
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

options: show_root_toc_entry: false heading_level: 3