Base SQLAlchemy

The base/sqlalchemy subpackage provides the foundational SQLAlchemy components shared across all relational database adapters, including the abstract port interface, session managers, and session manager registries.

Ports

Abstract port interface defining the contract all SQLAlchemy-based adapters must fulfil.

Base SQLAlchemy port interfaces.

archipy.adapters.base.sqlalchemy.ports.AnyExecuteParams module-attribute

AnyExecuteParams = (
    _CoreMultiExecuteParams | _CoreSingleExecuteParams
)

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort

Abstract interface defining synchronous SQLAlchemy database operations.

This interface defines the contract that all synchronous SQLAlchemy adapters must implement, providing standard methods for database operations like create, read, update, delete (CRUD), as well as search and transaction management.

Implementations of this interface are responsible for handling the specific details of database interactions and connection management.

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

class SQLAlchemyPort:
    """Abstract interface defining synchronous SQLAlchemy database operations.

    This interface defines the contract that all synchronous SQLAlchemy adapters must
    implement, providing standard methods for database operations like create,
    read, update, delete (CRUD), as well as search and transaction management.

    Implementations of this interface are responsible for handling the specific
    details of database interactions and connection management.
    """

    @abstractmethod
    def get_session(self) -> Session:
        """Retrieves a SQLAlchemy session for database operations.

        Returns:
            Session: A SQLAlchemy session object
        """
        raise NotImplementedError

    @abstractmethod
    def execute_search_query(  # noqa: PLR0913
        self,
        entity: type[BaseEntity],
        query: Select,
        pagination: PaginationDTO | None = None,
        sort_info: SortDTO | None = None,
        has_multiple_entities: bool = False,
        *,
        count_threshold: int | None = None,
    ) -> tuple[list[BaseEntity], int]:
        """Executes 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.
            count_threshold: When set, the total is exact up to this many rows and an estimate above it
                (PostgreSQL only; other databases count exactly)

        Returns:
            A tuple containing the list of entities and the total count
        """
        raise NotImplementedError

    @abstractmethod
    def create(self, entity: BaseEntity) -> BaseEntity | None:
        """Creates a new entity in the database.

        Args:
            entity: The entity to create

        Returns:
            The created entity (with updated attributes) or None if creation failed
        """
        raise NotImplementedError

    @abstractmethod
    def bulk_create(self, entities: list[BaseEntity]) -> list[BaseEntity] | None:
        """Creates multiple entities in the database.

        Args:
            entities: List of entities to create

        Returns:
            The list of created entities or None if creation failed
        """
        raise NotImplementedError

    @abstractmethod
    def get_by_uuid(self, entity_type: type[BaseEntity], entity_uuid: UUID) -> BaseEntity | None:
        """Retrieves 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
        """
        raise NotImplementedError

    @abstractmethod
    def delete(self, entity: BaseEntity) -> None:
        """Deletes an entity from the database.

        Args:
            entity: The entity to delete
        """
        raise NotImplementedError

    @abstractmethod
    def bulk_delete(self, entities: list[BaseEntity]) -> None:
        """Deletes multiple entities from the database.

        Args:
            entities: List of entities to delete
        """
        raise NotImplementedError

    @abstractmethod
    def execute(self, statement: Executable, params: AnyExecuteParams | None = None) -> Result[Any]:
        """Executes a raw SQL statement.

        Args:
            statement: The SQL statement to execute
            params: Optional parameters for the SQL statement

        Returns:
            The result of the execution
        """
        raise NotImplementedError

    @abstractmethod
    def scalars(self, statement: Executable, params: AnyExecuteParams | None = None) -> ScalarResult[Any]:
        """Executes a statement and returns the scalar result.

        Args:
            statement: The SQL statement to execute
            params: Optional parameters for the SQL statement

        Returns:
            The scalar result of the execution
        """
        raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort.get_session abstractmethod

get_session() -> Session

Retrieves a SQLAlchemy session for database operations.

Returns:

Name	Type	Description
Session	Session	A SQLAlchemy session object

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

@abstractmethod
def get_session(self) -> Session:
    """Retrieves a SQLAlchemy session for database operations.

    Returns:
        Session: A SQLAlchemy session object
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort.execute_search_query abstractmethod

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

Executes a search query with pagination and sorting.

Parameters:

Name	Type	Description	Default
entity	type[BaseEntity]	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
count_threshold	int | None	When set, the total is exact up to this many rows and an estimate above it (PostgreSQL only; other databases count exactly)	None

Returns:

Type	Description
tuple[list[BaseEntity], int]	A tuple containing the list of entities and the total count

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

@abstractmethod
def execute_search_query(  # noqa: PLR0913
    self,
    entity: type[BaseEntity],
    query: Select,
    pagination: PaginationDTO | None = None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
    *,
    count_threshold: int | None = None,
) -> tuple[list[BaseEntity], int]:
    """Executes 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.
        count_threshold: When set, the total is exact up to this many rows and an estimate above it
            (PostgreSQL only; other databases count exactly)

    Returns:
        A tuple containing the list of entities and the total count
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort.create abstractmethod

create(entity: BaseEntity) -> BaseEntity | None

Creates a new entity in the database.

Parameters:

Name	Type	Description	Default
entity	BaseEntity	The entity to create	required

Returns:

Type	Description
BaseEntity | None	The created entity (with updated attributes) or None if creation failed

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

@abstractmethod
def create(self, entity: BaseEntity) -> BaseEntity | None:
    """Creates a new entity in the database.

    Args:
        entity: The entity to create

    Returns:
        The created entity (with updated attributes) or None if creation failed
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort.bulk_create abstractmethod

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

Creates multiple entities in the database.

Parameters:

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

Returns:

Type	Description
list[BaseEntity] | None	The list of created entities or None if creation failed

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

@abstractmethod
def bulk_create(self, entities: list[BaseEntity]) -> list[BaseEntity] | None:
    """Creates multiple entities in the database.

    Args:
        entities: List of entities to create

    Returns:
        The list of created entities or None if creation failed
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort.get_by_uuid abstractmethod

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

Retrieves an entity by its UUID.

Parameters:

Name	Type	Description	Default
entity_type	type[BaseEntity]	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

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

@abstractmethod
def get_by_uuid(self, entity_type: type[BaseEntity], entity_uuid: UUID) -> BaseEntity | None:
    """Retrieves 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
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort.delete abstractmethod

delete(entity: BaseEntity) -> None

Deletes an entity from the database.

Parameters:

Name	Type	Description	Default
entity	BaseEntity	The entity to delete	required

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

@abstractmethod
def delete(self, entity: BaseEntity) -> None:
    """Deletes an entity from the database.

    Args:
        entity: The entity to delete
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort.bulk_delete abstractmethod

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

Deletes multiple entities from the database.

Parameters:

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

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

@abstractmethod
def bulk_delete(self, entities: list[BaseEntity]) -> None:
    """Deletes multiple entities from the database.

    Args:
        entities: List of entities to delete
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort.execute abstractmethod

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

Executes a raw SQL statement.

Parameters:

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

Returns:

Type	Description
Result[Any]	The result of the execution

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

@abstractmethod
def execute(self, statement: Executable, params: AnyExecuteParams | None = None) -> Result[Any]:
    """Executes a raw SQL statement.

    Args:
        statement: The SQL statement to execute
        params: Optional parameters for the SQL statement

    Returns:
        The result of the execution
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.SQLAlchemyPort.scalars abstractmethod

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

Executes a statement and returns the scalar result.

Parameters:

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

Returns:

Type	Description
ScalarResult[Any]	The scalar result of the execution

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

@abstractmethod
def scalars(self, statement: Executable, params: AnyExecuteParams | None = None) -> ScalarResult[Any]:
    """Executes a statement and returns the scalar result.

    Args:
        statement: The SQL statement to execute
        params: Optional parameters for the SQL statement

    Returns:
        The scalar result of the execution
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort

Abstract interface defining asynchronous SQLAlchemy database operations.

This interface defines the contract that all asynchronous SQLAlchemy adapters must implement, providing standard methods for database operations like create, read, update, delete (CRUD), as well as search and transaction management.

Implementations of this interface are responsible for handling the specific details of asynchronous database interactions and connection management.

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

class AsyncSQLAlchemyPort:
    """Abstract interface defining asynchronous SQLAlchemy database operations.

    This interface defines the contract that all asynchronous SQLAlchemy adapters must
    implement, providing standard methods for database operations like create,
    read, update, delete (CRUD), as well as search and transaction management.

    Implementations of this interface are responsible for handling the specific
    details of asynchronous database interactions and connection management.
    """

    @abstractmethod
    def get_session(self) -> AsyncSession:
        """Retrieves an asynchronous SQLAlchemy session for database operations.

        Returns:
            AsyncSession: An asynchronous SQLAlchemy session object
        """
        raise NotImplementedError

    @abstractmethod
    async def execute_search_query(  # noqa: PLR0913
        self,
        entity: type[BaseEntity],
        query: Select,
        pagination: PaginationDTO | None,
        sort_info: SortDTO | None = None,
        has_multiple_entities: bool = False,
        *,
        count_threshold: int | None = None,
    ) -> tuple[list[BaseEntity], int]:
        """Executes a search query with pagination and sorting asynchronously.

        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
            count_threshold: When set, the total is exact up to this many rows and an estimate above it
                (PostgreSQL only; other databases count exactly)

        Returns:
            A tuple containing the list of entities and the total count
        """
        raise NotImplementedError

    @abstractmethod
    async def create(self, entity: BaseEntity) -> BaseEntity | None:
        """Creates a new entity in the database asynchronously.

        Args:
            entity: The entity to create

        Returns:
            The created entity (with updated attributes) or None if creation failed
        """
        raise NotImplementedError

    @abstractmethod
    async def bulk_create(self, entities: list[BaseEntity]) -> list[BaseEntity] | None:
        """Creates multiple entities in the database asynchronously.

        Args:
            entities: List of entities to create

        Returns:
            The list of created entities or None if creation failed
        """
        raise NotImplementedError

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

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

        Returns:
            The entity if found, None otherwise
        """
        raise NotImplementedError

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

        Args:
            entity: The entity to delete
        """
        raise NotImplementedError

    @abstractmethod
    async def bulk_delete(self, entities: list[BaseEntity]) -> None:
        """Deletes multiple entities from the database asynchronously.

        Args:
            entities: List of entities to delete
        """
        raise NotImplementedError

    @abstractmethod
    async def execute(self, statement: Executable, params: AnyExecuteParams | None = None) -> Result[Any]:
        """Executes a raw SQL statement asynchronously.

        Args:
            statement: The SQL statement to execute
            params: Optional parameters for the SQL statement

        Returns:
            The result of the execution
        """
        raise NotImplementedError

    @abstractmethod
    async def scalars(self, statement: Executable, params: AnyExecuteParams | None = None) -> ScalarResult[Any]:
        """Executes a statement and returns the scalar result asynchronously.

        Args:
            statement: The SQL statement to execute
            params: Optional parameters for the SQL statement

        Returns:
            The scalar result of the execution
        """
        raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort.get_session abstractmethod

get_session() -> AsyncSession

Retrieves an asynchronous SQLAlchemy session for database operations.

Returns:

Name	Type	Description
AsyncSession	AsyncSession	An asynchronous SQLAlchemy session object

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

@abstractmethod
def get_session(self) -> AsyncSession:
    """Retrieves an asynchronous SQLAlchemy session for database operations.

    Returns:
        AsyncSession: An asynchronous SQLAlchemy session object
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort.execute_search_query abstractmethod async

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

Executes a search query with pagination and sorting asynchronously.

Parameters:

Name	Type	Description	Default
entity	type[BaseEntity]	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
count_threshold	int | None	When set, the total is exact up to this many rows and an estimate above it (PostgreSQL only; other databases count exactly)	None

Returns:

Type	Description
tuple[list[BaseEntity], int]	A tuple containing the list of entities and the total count

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

@abstractmethod
async def execute_search_query(  # noqa: PLR0913
    self,
    entity: type[BaseEntity],
    query: Select,
    pagination: PaginationDTO | None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
    *,
    count_threshold: int | None = None,
) -> tuple[list[BaseEntity], int]:
    """Executes a search query with pagination and sorting asynchronously.

    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
        count_threshold: When set, the total is exact up to this many rows and an estimate above it
            (PostgreSQL only; other databases count exactly)

    Returns:
        A tuple containing the list of entities and the total count
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort.create abstractmethod async

create(entity: BaseEntity) -> BaseEntity | None

Creates a new entity in the database asynchronously.

Parameters:

Name	Type	Description	Default
entity	BaseEntity	The entity to create	required

Returns:

Type	Description
BaseEntity | None	The created entity (with updated attributes) or None if creation failed

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

@abstractmethod
async def create(self, entity: BaseEntity) -> BaseEntity | None:
    """Creates a new entity in the database asynchronously.

    Args:
        entity: The entity to create

    Returns:
        The created entity (with updated attributes) or None if creation failed
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort.bulk_create abstractmethod async

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

Creates multiple entities in the database asynchronously.

Parameters:

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

Returns:

Type	Description
list[BaseEntity] | None	The list of created entities or None if creation failed

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

@abstractmethod
async def bulk_create(self, entities: list[BaseEntity]) -> list[BaseEntity] | None:
    """Creates multiple entities in the database asynchronously.

    Args:
        entities: List of entities to create

    Returns:
        The list of created entities or None if creation failed
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort.get_by_uuid abstractmethod async

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

Retrieves an entity by its UUID asynchronously.

Parameters:

Name	Type	Description	Default
entity_type	type[BaseEntity]	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

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

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

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

    Returns:
        The entity if found, None otherwise
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort.delete abstractmethod async

delete(entity: BaseEntity) -> None

Deletes an entity from the database asynchronously.

Parameters:

Name	Type	Description	Default
entity	BaseEntity	The entity to delete	required

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

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

    Args:
        entity: The entity to delete
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort.bulk_delete abstractmethod async

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

Deletes multiple entities from the database asynchronously.

Parameters:

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

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

@abstractmethod
async def bulk_delete(self, entities: list[BaseEntity]) -> None:
    """Deletes multiple entities from the database asynchronously.

    Args:
        entities: List of entities to delete
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort.execute abstractmethod async

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

Executes a raw SQL statement asynchronously.

Parameters:

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

Returns:

Type	Description
Result[Any]	The result of the execution

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

@abstractmethod
async def execute(self, statement: Executable, params: AnyExecuteParams | None = None) -> Result[Any]:
    """Executes a raw SQL statement asynchronously.

    Args:
        statement: The SQL statement to execute
        params: Optional parameters for the SQL statement

    Returns:
        The result of the execution
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.ports.AsyncSQLAlchemyPort.scalars abstractmethod async

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

Executes a statement and returns the scalar result asynchronously.

Parameters:

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

Returns:

Type	Description
ScalarResult[Any]	The scalar result of the execution

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

@abstractmethod
async def scalars(self, statement: Executable, params: AnyExecuteParams | None = None) -> ScalarResult[Any]:
    """Executes a statement and returns the scalar result asynchronously.

    Args:
        statement: The SQL statement to execute
        params: Optional parameters for the SQL statement

    Returns:
        The scalar result of the execution
    """
    raise NotImplementedError

options: show_root_toc_entry: false heading_level: 3

Session Manager Ports

Abstract interface for SQLAlchemy session managers, decoupling session lifecycle from adapter logic.

Port interfaces for SQLAlchemy session managers.

archipy.adapters.base.sqlalchemy.session_manager_ports.SessionManagerPort

Interface for SQLAlchemy session management operations.

This interface defines the contract for session management adapters, providing methods for retrieving and managing database sessions in a synchronous context.

Implementing classes must provide mechanisms to: 1. Retrieve a properly configured SQLAlchemy session 2. Release/remove sessions when they're no longer needed

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

class SessionManagerPort:
    """Interface for SQLAlchemy session management operations.

    This interface defines the contract for session management adapters,
    providing methods for retrieving and managing database sessions
    in a synchronous context.

    Implementing classes must provide mechanisms to:
    1. Retrieve a properly configured SQLAlchemy session
    2. Release/remove sessions when they're no longer needed
    """

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

        This method provides a database session that can be used for
        querying, creating, updating, and deleting data.

        Returns:
            Session: A SQLAlchemy session object

        Examples:
            >>> session = session_manager.get_session()
            >>> results = session.query(User).all()
        """
        raise NotImplementedError

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

        This method should be called to clean up the session when it's
        no longer needed, helping to prevent resource leaks and ensure
        proper session management.
        """
        raise NotImplementedError

archipy.adapters.base.sqlalchemy.session_manager_ports.SessionManagerPort.get_session abstractmethod

get_session() -> Session

Retrieve a SQLAlchemy session.

This method provides a database session that can be used for querying, creating, updating, and deleting data.

Returns:

Name	Type	Description
Session	Session	A SQLAlchemy session object

Examples:

>>> session = session_manager.get_session()
>>> results = session.query(User).all()

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

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

    This method provides a database session that can be used for
    querying, creating, updating, and deleting data.

    Returns:
        Session: A SQLAlchemy session object

    Examples:
        >>> session = session_manager.get_session()
        >>> results = session.query(User).all()
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.session_manager_ports.SessionManagerPort.remove_session abstractmethod

remove_session() -> None

Remove the current session from the registry.

This method should be called to clean up the session when it's no longer needed, helping to prevent resource leaks and ensure proper session management.

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

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

    This method should be called to clean up the session when it's
    no longer needed, helping to prevent resource leaks and ensure
    proper session management.
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.session_manager_ports.AsyncSessionManagerPort

Interface for asynchronous SQLAlchemy session management operations.

This interface defines the contract for asynchronous session management adapters, providing methods for retrieving and managing database sessions in an asynchronous context using SQLAlchemy's async capabilities.

Implementing classes must provide mechanisms to: 1. Retrieve a properly configured asynchronous SQLAlchemy session 2. Release/remove sessions asynchronously when they're no longer needed

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

class AsyncSessionManagerPort:
    """Interface for asynchronous SQLAlchemy session management operations.

    This interface defines the contract for asynchronous session management adapters,
    providing methods for retrieving and managing database sessions in an
    asynchronous context using SQLAlchemy's async capabilities.

    Implementing classes must provide mechanisms to:
    1. Retrieve a properly configured asynchronous SQLAlchemy session
    2. Release/remove sessions asynchronously when they're no longer needed
    """

    @abstractmethod
    def get_session(self) -> AsyncSession:
        """Retrieve an asynchronous SQLAlchemy session.

        This method provides an async database session that can be used for
        asynchronous querying, creating, updating, and deleting data.

        Returns:
            AsyncSession: An asynchronous SQLAlchemy session object
        """
        raise NotImplementedError

    @abstractmethod
    async def remove_session(self) -> None:
        """Asynchronously remove the current session from the registry.

        This method should be called to clean up the session when it's
        no longer needed, helping to prevent resource leaks and ensure
        proper session management in async contexts.
        """
        raise NotImplementedError

archipy.adapters.base.sqlalchemy.session_manager_ports.AsyncSessionManagerPort.get_session abstractmethod

get_session() -> AsyncSession

Retrieve an asynchronous SQLAlchemy session.

This method provides an async database session that can be used for asynchronous querying, creating, updating, and deleting data.

Returns:

Name	Type	Description
AsyncSession	AsyncSession	An asynchronous SQLAlchemy session object

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

@abstractmethod
def get_session(self) -> AsyncSession:
    """Retrieve an asynchronous SQLAlchemy session.

    This method provides an async database session that can be used for
    asynchronous querying, creating, updating, and deleting data.

    Returns:
        AsyncSession: An asynchronous SQLAlchemy session object
    """
    raise NotImplementedError

archipy.adapters.base.sqlalchemy.session_manager_ports.AsyncSessionManagerPort.remove_session abstractmethod async

remove_session() -> None

Asynchronously remove the current session from the registry.

This method should be called to clean up the session when it's no longer needed, helping to prevent resource leaks and ensure proper session management in async contexts.

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

@abstractmethod
async def remove_session(self) -> None:
    """Asynchronously remove the current session from the registry.

    This method should be called to clean up the session when it's
    no longer needed, helping to prevent resource leaks and ensure
    proper session management in async contexts.
    """
    raise NotImplementedError

options: show_root_toc_entry: false heading_level: 3

Session Manager Registry

Registry for tracking and resolving active session manager instances.

Base SQLAlchemy session manager registry for ArchiPy adapters.

archipy.adapters.base.sqlalchemy.session_manager_registry.SessionManagerRegistry

Registry for SQLAlchemy session managers.

This registry provides a centralized access point for both synchronous and asynchronous session managers, implementing the Service Locator pattern.

Subclasses should override get_sync_manager and get_async_manager to provide concrete session managers, or use set_sync_manager and set_async_manager to register managers manually.

Examples:

>>> from archipy.adapters.postgres.sqlalchemy.session_manager_registry import PostgresSessionManagerRegistry
>>> sync_manager = PostgresSessionManagerRegistry.get_sync_manager()
>>> session = sync_manager.get_session()

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

class SessionManagerRegistry:
    """Registry for SQLAlchemy session managers.

    This registry provides a centralized access point for both synchronous and
    asynchronous session managers, implementing the Service Locator pattern.

    Subclasses should override get_sync_manager and get_async_manager to provide
    concrete session managers, or use set_sync_manager and set_async_manager to
    register managers manually.

    Examples:
        >>> from archipy.adapters.postgres.sqlalchemy.session_manager_registry import PostgresSessionManagerRegistry
        >>> sync_manager = PostgresSessionManagerRegistry.get_sync_manager()
        >>> session = sync_manager.get_session()
    """

    _sync_instance: ClassVar[SessionManagerPort | None] = None
    _async_instance: ClassVar[AsyncSessionManagerPort | None] = None

    def __init_subclass__(cls, **kwargs: object) -> None:
        """Give every subclass its own manager slots instead of reading the parent's.

        A subclass of a concrete registry, such as ``class Analytics(PostgresSessionManagerRegistry)``, can then hold
        the session manager of another database of the same type.
        """
        super().__init_subclass__(**kwargs)
        cls._sync_instance = None
        cls._async_instance = None

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

        Returns:
            SessionManagerPort: The registered synchronous session manager

        Raises:
            InternalError: If no synchronous session manager is set
            DatabaseConnectionError: If there's an error initializing the session manager
        """
        if cls._sync_instance is None:
            raise InternalError(error_code="SYNC_SESSION_MANAGER_NOT_INITIALIZED")
        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
            InvalidEntityTypeError: If the manager doesn't implement SessionManagerPort
        """
        if manager is None:
            raise InvalidArgumentError(argument_name="session_manager")

        if not isinstance(manager, SessionManagerPort):
            raise InvalidEntityTypeError(
                expected_type="SessionManagerPort",
                actual_type=type(manager).__name__,
            )
        cls._sync_instance = manager

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

        Returns:
            AsyncSessionManagerPort: The registered asynchronous session manager

        Raises:
            InternalError: If no asynchronous session manager is set
            DatabaseConnectionError: If there's an error initializing the session manager
        """
        if cls._async_instance is None:
            raise InternalError(error_code="ASYNC_SESSION_MANAGER_NOT_INITIALIZED")
        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
            InvalidEntityTypeError: If the manager doesn't implement AsyncSessionManagerPort
        """
        if manager is None:
            raise InvalidArgumentError(argument_name="session_manager")

        if not isinstance(manager, AsyncSessionManagerPort):
            raise InvalidEntityTypeError(
                expected_type="AsyncSessionManagerPort",
                actual_type=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.base.sqlalchemy.session_manager_registry.SessionManagerRegistry.get_sync_manager classmethod

get_sync_manager() -> SessionManagerPort

Get the synchronous session manager instance.

Returns:

Name	Type	Description
SessionManagerPort	SessionManagerPort	The registered synchronous session manager

Raises:

Type	Description
InternalError	If no synchronous session manager is set
DatabaseConnectionError	If there's an error initializing the session manager

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

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

    Returns:
        SessionManagerPort: The registered synchronous session manager

    Raises:
        InternalError: If no synchronous session manager is set
        DatabaseConnectionError: If there's an error initializing the session manager
    """
    if cls._sync_instance is None:
        raise InternalError(error_code="SYNC_SESSION_MANAGER_NOT_INITIALIZED")
    return cls._sync_instance

archipy.adapters.base.sqlalchemy.session_manager_registry.SessionManagerRegistry.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
InvalidEntityTypeError	If the manager doesn't implement SessionManagerPort

Source code in archipy/adapters/base/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
        InvalidEntityTypeError: If the manager doesn't implement SessionManagerPort
    """
    if manager is None:
        raise InvalidArgumentError(argument_name="session_manager")

    if not isinstance(manager, SessionManagerPort):
        raise InvalidEntityTypeError(
            expected_type="SessionManagerPort",
            actual_type=type(manager).__name__,
        )
    cls._sync_instance = manager

archipy.adapters.base.sqlalchemy.session_manager_registry.SessionManagerRegistry.get_async_manager classmethod

get_async_manager() -> AsyncSessionManagerPort

Get the asynchronous session manager instance.

Returns:

Name	Type	Description
AsyncSessionManagerPort	AsyncSessionManagerPort	The registered asynchronous session manager

Raises:

Type	Description
InternalError	If no asynchronous session manager is set
DatabaseConnectionError	If there's an error initializing the session manager

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

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

    Returns:
        AsyncSessionManagerPort: The registered asynchronous session manager

    Raises:
        InternalError: If no asynchronous session manager is set
        DatabaseConnectionError: If there's an error initializing the session manager
    """
    if cls._async_instance is None:
        raise InternalError(error_code="ASYNC_SESSION_MANAGER_NOT_INITIALIZED")
    return cls._async_instance

archipy.adapters.base.sqlalchemy.session_manager_registry.SessionManagerRegistry.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
InvalidEntityTypeError	If the manager doesn't implement AsyncSessionManagerPort

Source code in archipy/adapters/base/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
        InvalidEntityTypeError: If the manager doesn't implement AsyncSessionManagerPort
    """
    if manager is None:
        raise InvalidArgumentError(argument_name="session_manager")

    if not isinstance(manager, AsyncSessionManagerPort):
        raise InvalidEntityTypeError(
            expected_type="AsyncSessionManagerPort",
            actual_type=type(manager).__name__,
        )
    cls._async_instance = manager

archipy.adapters.base.sqlalchemy.session_manager_registry.SessionManagerRegistry.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/base/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

Session Managers

Concrete session manager implementations that handle SQLAlchemy session creation, scoping, and teardown.

Base SQLAlchemy session manager implementations.

archipy.adapters.base.sqlalchemy.session_managers.ConfigT module-attribute

ConfigT = TypeVar('ConfigT', bound=SQLAlchemyConfig)

archipy.adapters.base.sqlalchemy.session_managers.BaseSQLAlchemySessionManager

Bases: SessionManagerPort

Base synchronous SQLAlchemy session manager.

Implements the SessionManagerPort interface to provide session management for synchronous database operations. Database-specific session managers should inherit from this class and implement database-specific engine creation.

Parameters:

Name	Type	Description	Default
orm_config	ConfigT	SQLAlchemy configuration. Must match the expected config type for the database.	required

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

class BaseSQLAlchemySessionManager[ConfigT: SQLAlchemyConfig](SessionManagerPort):
    """Base synchronous SQLAlchemy session manager.

    Implements the SessionManagerPort interface to provide session management for
    synchronous database operations. Database-specific session managers should inherit
    from this class and implement database-specific engine creation.

    Args:
        orm_config: SQLAlchemy configuration. Must match the expected config type for the database.
    """

    @classmethod
    def _default_config(cls) -> ConfigT:
        """Return the configuration used when none is passed, normally the global config's section.

        Returns:
            The default configuration for this database.

        Raises:
            NotImplementedError: If the subclass does not define it.
        """
        raise NotImplementedError("Subclasses must implement _default_config")

    @classmethod
    def _singleton_key(cls, orm_config: ConfigT | None = None) -> str:
        """Key the ``Singleton`` metaclass uses: one manager per distinct configuration.

        Args:
            orm_config: The configuration passed to the constructor, or None for the default.

        Returns:
            The configuration serialised to JSON.
        """
        return (orm_config or cls._default_config()).model_dump_json()

    def __init__(self, orm_config: ConfigT) -> None:
        """Initialize the base session manager.

        Args:
            orm_config: SQLAlchemy configuration.

        Raises:
            InvalidArgumentError: If the configuration type is invalid.
            DatabaseConnectionError: If there's an error creating the database connection.
            DatabaseConfigurationError: If there's an error in the database configuration.
        """
        if not isinstance(orm_config, self._expected_config_type()):
            raise InvalidArgumentError(
                argument_name="orm_config",
                additional_data={
                    "expected": self._expected_config_type().__name__,
                    "got": type(orm_config).__name__,
                },
            )
        try:
            self.engine = self._create_engine(orm_config)
            self._session_generator = self._get_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

    @abstractmethod
    def _expected_config_type(self) -> type[SQLAlchemyConfig]:
        """Return the expected configuration type for the database.

        Returns:
            The SQLAlchemy configuration class expected by this session manager.
        """

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

        Returns:
            str: The name of the database (e.g., 'postgresql', 'sqlite', 'starrocks').
        """

    @abstractmethod
    def _create_url(self, configs: ConfigT) -> URL:
        """Create a database connection URL.

        Args:
            configs: Database-specific configuration.

        Returns:
            A SQLAlchemy URL object for the database.

        Raises:
            DatabaseConnectionError: If there's an error creating the URL.
        """

    def _create_engine(self, configs: ConfigT) -> Engine:
        """Create a SQLAlchemy engine with common configuration.

        Args:
            configs: SQLAlchemy configuration.

        Returns:
            A configured SQLAlchemy engine.

        Raises:
            DatabaseConnectionError: If there's an error creating the engine.
            DatabaseConfigurationError: If there's an error in the database configuration.
        """
        try:
            url = self._create_url(configs)
            # Looked up on the module at call time: OpenTelemetry's SQLAlchemy instrumentor replaces
            # ``sqlalchemy.create_engine``, so a name bound at import time would skip the instrumentation.
            return sqlalchemy.create_engine(
                url,
                isolation_level=configs.ISOLATION_LEVEL,
                echo=configs.ECHO,
                echo_pool=configs.ECHO_POOL,
                enable_from_linting=configs.ENABLE_FROM_LINTING,
                hide_parameters=configs.HIDE_PARAMETERS,
                query_cache_size=configs.QUERY_CACHE_SIZE,
                connect_args=self._get_connect_args(),
                **self._get_pool_kwargs(configs),
            )
        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

    def _get_connect_args(self) -> dict[str, Any]:
        """Return additional connection arguments for the engine.

        Returns:
            A dictionary of connection arguments (default is empty).
        """
        return {}

    def _get_pool_kwargs(self, configs: ConfigT) -> dict[str, Any]:
        """Return connection pool arguments for the engine.

        Override when the dialect picks a pool class that rejects queue-pool arguments.

        Args:
            configs: SQLAlchemy configuration.

        Returns:
            A dictionary of pool-related keyword arguments for engine creation.
        """
        return {
            "pool_pre_ping": configs.POOL_PRE_PING,
            "pool_size": configs.POOL_SIZE,
            "pool_recycle": configs.POOL_RECYCLE_SECONDS,
            "pool_reset_on_return": configs.POOL_RESET_ON_RETURN,
            "pool_timeout": configs.POOL_TIMEOUT,
            "pool_use_lifo": configs.POOL_USE_LIFO,
            "max_overflow": configs.POOL_MAX_OVERFLOW,
        }

    def _get_session_generator(self) -> scoped_session:
        """Create a scoped session factory for synchronous sessions.

        Returns:
            A scoped_session instance used by `get_session` to provide thread-safe sessions.

        Raises:
            DatabaseConfigurationError: If there's an error in the database configuration.
        """
        try:
            session_maker = sessionmaker(self.engine)
            return scoped_session(session_maker)
        except SQLAlchemyError as e:
            if "configuration" in str(e).lower():
                raise DatabaseConfigurationError(
                    database=self._get_database_name(),
                ) from e
            raise DatabaseError(
                database=self._get_database_name(),
            ) from e

    @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

    @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.base.sqlalchemy.session_managers.BaseSQLAlchemySessionManager.engine instance-attribute

engine = self._create_engine(orm_config)

archipy.adapters.base.sqlalchemy.session_managers.BaseSQLAlchemySessionManager.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.base.sqlalchemy.session_managers.BaseSQLAlchemySessionManager.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.base.sqlalchemy.session_managers.AsyncBaseSQLAlchemySessionManager

Bases: AsyncSessionManagerPort

Base asynchronous SQLAlchemy session manager.

Implements the AsyncSessionManagerPort interface to provide session management for asynchronous database operations. Database-specific session managers should inherit from this class and implement database-specific async engine creation.

Parameters:

Name	Type	Description	Default
orm_config	ConfigT	SQLAlchemy configuration. Must match the expected config type for the database.	required

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

class AsyncBaseSQLAlchemySessionManager[ConfigT: SQLAlchemyConfig](AsyncSessionManagerPort):
    """Base asynchronous SQLAlchemy session manager.

    Implements the AsyncSessionManagerPort interface to provide session management for
    asynchronous database operations. Database-specific session managers should inherit
    from this class and implement database-specific async engine creation.

    Args:
        orm_config: SQLAlchemy configuration. Must match the expected config type for the database.
    """

    @classmethod
    def _default_config(cls) -> ConfigT:
        """Return the configuration used when none is passed, normally the global config's section.

        Returns:
            The default configuration for this database.

        Raises:
            NotImplementedError: If the subclass does not define it.
        """
        raise NotImplementedError("Subclasses must implement _default_config")

    @classmethod
    def _singleton_key(cls, orm_config: ConfigT | None = None) -> str:
        """Key the ``Singleton`` metaclass uses: one manager per distinct configuration.

        Args:
            orm_config: The configuration passed to the constructor, or None for the default.

        Returns:
            The configuration serialised to JSON.
        """
        return (orm_config or cls._default_config()).model_dump_json()

    def __init__(self, orm_config: ConfigT) -> None:
        """Initialize the base async session manager.

        Args:
            orm_config: SQLAlchemy configuration.

        Raises:
            InvalidArgumentError: If the configuration type is invalid.
            DatabaseConnectionError: If there's an error creating the database connection.
            DatabaseConfigurationError: If there's an error in the database configuration.
        """
        if not isinstance(orm_config, self._expected_config_type()):
            raise InvalidArgumentError(
                argument_name="orm_config",
                additional_data={
                    "expected": self._expected_config_type().__name__,
                    "got": type(orm_config).__name__,
                },
            )
        try:
            self.engine = self._create_async_engine(orm_config)
            self._session_generator = self._get_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

    @abstractmethod
    def _expected_config_type(self) -> type[SQLAlchemyConfig]:
        """Return the expected configuration type for the database.

        Returns:
            The SQLAlchemy configuration class expected by this session manager.
        """

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

        Returns:
            str: The name of the database (e.g., 'postgresql', 'sqlite', 'starrocks').
        """

    @abstractmethod
    def _create_url(self, configs: ConfigT) -> URL:
        """Create a database connection URL.

        Args:
            configs: Database-specific configuration.

        Returns:
            A SQLAlchemy URL object for the database.

        Raises:
            DatabaseConnectionError: If there's an error creating the URL.
        """

    def _create_async_engine(self, configs: ConfigT) -> AsyncEngine:
        """Create an async SQLAlchemy engine with common configuration.

        Args:
            configs: SQLAlchemy configuration.

        Returns:
            A configured async SQLAlchemy engine.

        Raises:
            DatabaseConnectionError: If there's an error creating the engine.
            DatabaseConfigurationError: If there's an error in the database configuration.
        """
        from sqlalchemy.ext.asyncio import create_async_engine

        try:
            url = self._create_url(configs)
            return create_async_engine(
                url,
                isolation_level=configs.ISOLATION_LEVEL,
                echo=configs.ECHO,
                echo_pool=configs.ECHO_POOL,
                enable_from_linting=configs.ENABLE_FROM_LINTING,
                hide_parameters=configs.HIDE_PARAMETERS,
                query_cache_size=configs.QUERY_CACHE_SIZE,
                connect_args=self._get_connect_args(),
                **self._get_pool_kwargs(configs),
            )
        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

    def _get_connect_args(self) -> dict[str, Any]:
        """Return additional connection arguments for the engine.

        Returns:
            A dictionary of connection arguments (default is empty).
        """
        return {}

    def _get_pool_kwargs(self, configs: ConfigT) -> dict[str, Any]:
        """Return connection pool arguments for the engine.

        Override when the dialect picks a pool class that rejects queue-pool arguments.

        Args:
            configs: SQLAlchemy configuration.

        Returns:
            A dictionary of pool-related keyword arguments for engine creation.
        """
        return {
            "pool_pre_ping": configs.POOL_PRE_PING,
            "pool_size": configs.POOL_SIZE,
            "pool_recycle": configs.POOL_RECYCLE_SECONDS,
            "pool_reset_on_return": configs.POOL_RESET_ON_RETURN,
            "pool_timeout": configs.POOL_TIMEOUT,
            "pool_use_lifo": configs.POOL_USE_LIFO,
            "max_overflow": configs.POOL_MAX_OVERFLOW,
        }

    def _get_session_generator(self) -> async_scoped_session:
        """Create an async scoped session factory.

        Returns:
            An async_scoped_session instance used by `get_session` to provide task-safe sessions.

        Raises:
            DatabaseConfigurationError: If there's an error in the database configuration.
        """
        from sqlalchemy.ext.asyncio import async_scoped_session, async_sessionmaker

        try:
            session_maker = async_sessionmaker(self.engine)
            return async_scoped_session(session_maker, scopefunc=current_task)
        except SQLAlchemyError as e:
            if "configuration" in str(e).lower():
                raise DatabaseConfigurationError(
                    database=self._get_database_name(),
                ) from e
            raise DatabaseError(
                database=self._get_database_name(),
            ) from e

    @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

    @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

archipy.adapters.base.sqlalchemy.session_managers.AsyncBaseSQLAlchemySessionManager.engine instance-attribute

engine = self._create_async_engine(orm_config)

archipy.adapters.base.sqlalchemy.session_managers.AsyncBaseSQLAlchemySessionManager.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.base.sqlalchemy.session_managers.AsyncBaseSQLAlchemySessionManager.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

Adapters

The base SQLAlchemy adapter implements generic CRUD operations that concrete database adapters (PostgreSQL, SQLite, StarRocks) inherit from.

archipy.adapters.base.sqlalchemy.adapters re-exports everything below, so existing imports keep working.

Base synchronous SQLAlchemy adapter.

archipy.adapters.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter

Bases: SQLAlchemyPort, SQLAlchemyPaginationMixin, SQLAlchemySortMixin, SQLAlchemyFilterMixin, SQLAlchemyExceptionHandlerMixin

Base synchronous SQLAlchemy adapter for ORM operations.

Provides a standardized interface for CRUD operations, pagination, sorting, and filtering. Specific database adapters should inherit from this class and provide their own session manager.

Parameters:

Name	Type	Description	Default
orm_config	ConfigT	Configuration for SQLAlchemy. If None, uses global config.	required

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

class BaseSQLAlchemyAdapter[ConfigT: SQLAlchemyConfig](
    SQLAlchemyPort,
    SQLAlchemyPaginationMixin,
    SQLAlchemySortMixin,
    SQLAlchemyFilterMixin,
    SQLAlchemyExceptionHandlerMixin,
):
    """Base synchronous SQLAlchemy adapter for ORM operations.

    Provides a standardized interface for CRUD operations, pagination, sorting, and filtering.
    Specific database adapters should inherit from this class and provide their own session manager.

    Args:
        orm_config: Configuration for SQLAlchemy. If None, uses global config.
    """

    def __init__(self, orm_config: ConfigT) -> None:
        """Initialize the base adapter with a session manager.

        Args:
            orm_config: Configuration for SQLAlchemy. Subclasses resolve the
                appropriate default from global config when not provided.
        """
        self.session_manager: BaseSQLAlchemySessionManager[ConfigT] = self._create_session_manager(
            orm_config,
        )
        self._cached_count = _build_count_cache(orm_config)
        self._count_threshold = orm_config.COUNT_THRESHOLD

    def _create_session_manager(self, configs: ConfigT) -> BaseSQLAlchemySessionManager[ConfigT]:
        """Create a session manager for the specific database.

        Args:
            configs: SQLAlchemy configuration.

        Returns:
            A session manager instance.
        """
        raise NotImplementedError("Subclasses must implement _create_session_manager")

    def _count_total(
        self,
        session: Session,
        entity: type[T],
        query: Select,
        count_threshold: int | None,
    ) -> int:
        """Return the total of a search query, from the count cache when it is enabled and still valid."""
        if self._cached_count is None:
            return self._count_rows(session, entity, query, count_threshold)
        key = self._count_cache_key(self._build_count_source(entity, query), count_threshold)
        return self._cached_count(_CountRequest(key, lambda: self._count_rows(session, entity, query, count_threshold)))

    def _count_rows(
        self,
        session: Session,
        entity: type[T],
        query: Select,
        count_threshold: int | None,
    ) -> int:
        """Count the rows of a search query, bounded and estimated above ``count_threshold`` when it is set."""
        if count_threshold is None:
            return session.execute(self._build_count_query(self._build_count_source(entity, query))).scalar_one()
        bounded = session.execute(
            self._build_count_query(self._build_count_source(entity, query, limit=count_threshold + 1)),
        ).scalar_one()
        if bounded <= count_threshold:
            return bounded
        statement = self._build_estimate_statement(self._build_count_source(entity, query), session.get_bind().dialect)
        if statement is not None:
            estimate = self._parse_estimate(session.execute(statement).scalar())
            if estimate is not None:
                return max(estimate, bounded)
        return session.execute(self._build_count_query(self._build_count_source(entity, query))).scalar_one()

    @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,
        *,
        count_threshold: int | None = None,
    ) -> 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: Set when the query selects several entities; only instances of ``entity``
                from the first column of each row are returned.
            count_threshold: When set, the total is exact up to this many rows. Above it the total is a database
                estimate (PostgreSQL) so large tables are never counted in full. Other databases count exactly.
                Defaults to the ``COUNT_THRESHOLD`` config value.

        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)
            results = _extract_entities(result_set, entity, has_multiple_entities)
            total_count = self._total_from_page(results, pagination, has_multiple_entities)
            if total_count is None:
                total_count = self._count_total(
                    session,
                    entity,
                    query,
                    count_threshold if count_threshold is not None else self._count_threshold,
                )
        except SQLAlchemyError 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

    @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()

    @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.
        """
        _ensure_entity(entity)

        try:
            session = self.get_session()
            session.add(entity)
            session.flush()
        except SQLAlchemyError 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

    @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.
        """
        _ensure_entities(entities)

        try:
            session = self.get_session()
            session.add_all(entities)
            session.flush()
        except SQLAlchemyError 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

    @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.
        """
        _ensure_entity_type(entity_type)

        try:
            session = self.get_session()
            result = session.get(entity_type, entity_uuid)
        except SQLAlchemyError 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

    @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.
        """
        _ensure_entity(entity)

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

    @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.
        """
        _ensure_entities(entities)

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

    @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 SQLAlchemyError 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

    @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 SQLAlchemyError 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.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.session_manager instance-attribute

session_manager: BaseSQLAlchemySessionManager[ConfigT] = (
    self._create_session_manager(orm_config)
)

archipy.adapters.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.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,
    *,
    count_threshold: int | None = None,
) -> 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	Set when the query selects several entities; only instances of entity from the first column of each row are returned.	False
count_threshold	int | None	When set, the total is exact up to this many rows. Above it the total is a database estimate (PostgreSQL) so large tables are never counted in full. Other databases count exactly. Defaults to the COUNT_THRESHOLD config value.	None

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/sync_adapter.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,
    *,
    count_threshold: int | None = None,
) -> 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: Set when the query selects several entities; only instances of ``entity``
            from the first column of each row are returned.
        count_threshold: When set, the total is exact up to this many rows. Above it the total is a database
            estimate (PostgreSQL) so large tables are never counted in full. Other databases count exactly.
            Defaults to the ``COUNT_THRESHOLD`` config value.

    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)
        results = _extract_entities(result_set, entity, has_multiple_entities)
        total_count = self._total_from_page(results, pagination, has_multiple_entities)
        if total_count is None:
            total_count = self._count_total(
                session,
                entity,
                query,
                count_threshold if count_threshold is not None else self._count_threshold,
            )
    except SQLAlchemyError 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.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.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/sync_adapter.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.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.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/sync_adapter.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.
    """
    _ensure_entity(entity)

    try:
        session = self.get_session()
        session.add(entity)
        session.flush()
    except SQLAlchemyError 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.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.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/sync_adapter.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.
    """
    _ensure_entities(entities)

    try:
        session = self.get_session()
        session.add_all(entities)
        session.flush()
    except SQLAlchemyError 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.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.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/sync_adapter.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.
    """
    _ensure_entity_type(entity_type)

    try:
        session = self.get_session()
        result = session.get(entity_type, entity_uuid)
    except SQLAlchemyError 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.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.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/sync_adapter.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.
    """
    _ensure_entity(entity)

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

archipy.adapters.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.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/sync_adapter.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.
    """
    _ensure_entities(entities)

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

archipy.adapters.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.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/sync_adapter.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 SQLAlchemyError 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.base.sqlalchemy.sync_adapter.BaseSQLAlchemyAdapter.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/sync_adapter.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 SQLAlchemyError 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

Base asynchronous SQLAlchemy adapter.

archipy.adapters.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter

Bases: AsyncSQLAlchemyPort, SQLAlchemyPaginationMixin, SQLAlchemySortMixin, SQLAlchemyFilterMixin, SQLAlchemyExceptionHandlerMixin

Base asynchronous SQLAlchemy adapter for ORM operations.

Provides a standardized interface for CRUD operations, pagination, sorting, and filtering. Specific database adapters should inherit from this class and provide their own session manager.

Parameters:

Name	Type	Description	Default
orm_config	ConfigT	Configuration for SQLAlchemy. If None, uses global config.	required

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

class AsyncBaseSQLAlchemyAdapter[ConfigT: SQLAlchemyConfig](
    AsyncSQLAlchemyPort,
    SQLAlchemyPaginationMixin,
    SQLAlchemySortMixin,
    SQLAlchemyFilterMixin,
    SQLAlchemyExceptionHandlerMixin,
):
    """Base asynchronous SQLAlchemy adapter for ORM operations.

    Provides a standardized interface for CRUD operations, pagination, sorting, and filtering.
    Specific database adapters should inherit from this class and provide their own session manager.

    Args:
        orm_config: Configuration for SQLAlchemy. If None, uses global config.
    """

    def __init__(self, orm_config: ConfigT) -> None:
        """Initialize the base async adapter with a session manager.

        Args:
            orm_config: Configuration for SQLAlchemy. Subclasses resolve the
                appropriate default from global config when not provided.
        """
        self.session_manager: AsyncBaseSQLAlchemySessionManager[ConfigT] = self._create_async_session_manager(
            orm_config,
        )
        self._cached_count = _build_async_count_cache(orm_config)
        self._count_threshold = orm_config.COUNT_THRESHOLD

    def _create_async_session_manager(self, configs: ConfigT) -> AsyncBaseSQLAlchemySessionManager[ConfigT]:
        """Create an async session manager for the specific database.

        Args:
            configs: SQLAlchemy configuration.

        Returns:
            An async session manager instance.
        """
        raise NotImplementedError("Subclasses must implement _create_async_session_manager")

    async def _count_total(
        self,
        session: AsyncSession,
        entity: type[T],
        query: Select,
        count_threshold: int | None,
    ) -> int:
        """Return the total of a search query, from the count cache when it is enabled and still valid."""
        if self._cached_count is None:
            return await self._count_rows(session, entity, query, count_threshold)
        key = self._count_cache_key(self._build_count_source(entity, query), count_threshold)
        return await self._cached_count(
            _CountRequest(key, lambda: self._count_rows(session, entity, query, count_threshold)),
        )

    async def _count_rows(
        self,
        session: AsyncSession,
        entity: type[T],
        query: Select,
        count_threshold: int | None,
    ) -> int:
        """Count the rows of a search query, bounded and estimated above ``count_threshold`` when it is set."""
        if count_threshold is None:
            return (
                await session.execute(self._build_count_query(self._build_count_source(entity, query)))
            ).scalar_one()
        bounded = (
            await session.execute(
                self._build_count_query(self._build_count_source(entity, query, limit=count_threshold + 1)),
            )
        ).scalar_one()
        if bounded <= count_threshold:
            return bounded
        statement = self._build_estimate_statement(self._build_count_source(entity, query), session.get_bind().dialect)
        if statement is not None:
            estimate = self._parse_estimate((await session.execute(statement)).scalar())
            if estimate is not None:
                return max(estimate, bounded)
        return (await session.execute(self._build_count_query(self._build_count_source(entity, query)))).scalar_one()

    @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,
        *,
        count_threshold: int | None = None,
    ) -> 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: Set when the query selects several entities; only instances of ``entity``
                from the first column of each row are returned.
            count_threshold: When set, the total is exact up to this many rows. Above it the total is a database
                estimate (PostgreSQL) so large tables are never counted in full. Other databases count exactly.
                Defaults to the ``COUNT_THRESHOLD`` config value.

        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)
            results = _extract_entities(result_set, entity, has_multiple_entities)
            total_count = self._total_from_page(results, pagination, has_multiple_entities)
            if total_count is None:
                total_count = await self._count_total(
                    session,
                    entity,
                    query,
                    count_threshold if count_threshold is not None else self._count_threshold,
                )
        except SQLAlchemyError 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

    @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()

    @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.
        """
        _ensure_entity(entity)

        try:
            session = self.get_session()
            session.add(entity)
            await session.flush()
        except SQLAlchemyError 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

    @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.
        """
        _ensure_entities(entities)

        try:
            session = self.get_session()
            session.add_all(entities)
            await session.flush()
        except SQLAlchemyError 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

    @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.
        """
        _ensure_entity_type(entity_type)

        try:
            session = self.get_session()
            result = await session.get(entity_type, entity_uuid)
        except SQLAlchemyError 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

    @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.
        """
        _ensure_entity(entity)

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

    @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.
        """
        _ensure_entities(entities)

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

    @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 SQLAlchemyError 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

    @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 SQLAlchemyError 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.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.session_manager instance-attribute

session_manager: AsyncBaseSQLAlchemySessionManager[
    ConfigT
] = self._create_async_session_manager(orm_config)

archipy.adapters.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.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,
    *,
    count_threshold: int | None = None,
) -> 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	Set when the query selects several entities; only instances of entity from the first column of each row are returned.	False
count_threshold	int | None	When set, the total is exact up to this many rows. Above it the total is a database estimate (PostgreSQL) so large tables are never counted in full. Other databases count exactly. Defaults to the COUNT_THRESHOLD config value.	None

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/async_adapter.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,
    *,
    count_threshold: int | None = None,
) -> 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: Set when the query selects several entities; only instances of ``entity``
            from the first column of each row are returned.
        count_threshold: When set, the total is exact up to this many rows. Above it the total is a database
            estimate (PostgreSQL) so large tables are never counted in full. Other databases count exactly.
            Defaults to the ``COUNT_THRESHOLD`` config value.

    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)
        results = _extract_entities(result_set, entity, has_multiple_entities)
        total_count = self._total_from_page(results, pagination, has_multiple_entities)
        if total_count is None:
            total_count = await self._count_total(
                session,
                entity,
                query,
                count_threshold if count_threshold is not None else self._count_threshold,
            )
    except SQLAlchemyError 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.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.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/async_adapter.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.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.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/async_adapter.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.
    """
    _ensure_entity(entity)

    try:
        session = self.get_session()
        session.add(entity)
        await session.flush()
    except SQLAlchemyError 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.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.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/async_adapter.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.
    """
    _ensure_entities(entities)

    try:
        session = self.get_session()
        session.add_all(entities)
        await session.flush()
    except SQLAlchemyError 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.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.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/async_adapter.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.
    """
    _ensure_entity_type(entity_type)

    try:
        session = self.get_session()
        result = await session.get(entity_type, entity_uuid)
    except SQLAlchemyError 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.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.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/async_adapter.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.
    """
    _ensure_entity(entity)

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

archipy.adapters.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.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/async_adapter.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.
    """
    _ensure_entities(entities)

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

archipy.adapters.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.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/async_adapter.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 SQLAlchemyError 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.base.sqlalchemy.async_adapter.AsyncBaseSQLAlchemyAdapter.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/async_adapter.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 SQLAlchemyError 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

Query mixins shared by the sync and async base SQLAlchemy adapters: errors, filtering, pagination, sorting.

archipy.adapters.base.sqlalchemy.mixins.T module-attribute

T = TypeVar('T', bound=BaseEntity)

archipy.adapters.base.sqlalchemy.mixins.ConfigT module-attribute

ConfigT = TypeVar('ConfigT', bound=SQLAlchemyConfig)

archipy.adapters.base.sqlalchemy.mixins.SQLAlchemyExceptionHandlerMixin

Mixin providing centralized exception handling for SQLAlchemy operations.

This mixin provides a standard method for handling database exceptions and converting them to appropriate application-specific exceptions.

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

class SQLAlchemyExceptionHandlerMixin:
    """Mixin providing centralized exception handling for SQLAlchemy operations.

    This mixin provides a standard method for handling database exceptions and
    converting them to appropriate application-specific exceptions.
    """

    @classmethod
    def _handle_db_exception(cls, exception: Exception, db_name: str | None = None) -> NoReturn:
        """Handle database exceptions and raise appropriate errors.

        Args:
            exception: The exception to handle.
            db_name: Optional database name for error context.

        Raises:
            DatabaseSerializationError: If a serialization failure (SQLSTATE 40001) is detected.
            DatabaseDeadlockError: If a deadlock (SQLSTATE 40P01, MySQL 1213, SQLITE_BUSY/LOCKED) is detected.
            DatabaseResourceExhaustedError: If the server ran out of memory or connections.
            DatabaseTimeoutError: If a timeout is detected.
            DatabaseConnectionError: If a connection error is detected.
            DatabaseTransactionError: If a transaction error is detected.
            DatabaseIntegrityError: If an integrity violation is detected.
            DatabaseConstraintError: If a constraint violation is detected.
            DatabaseQueryError: For other database errors.
        """
        # Classify by driver error code first (SQLSTATE / errno); message matching is only the fallback.
        error_class = classify_driver_error(exception) or cls._classify_by_message(str(exception).lower())
        log_driver_error(exception, error_class, db_name)
        raise error_class(database=db_name) from exception

    @staticmethod
    def _classify_by_message(message: str) -> type[DatabaseError]:
        """Fallback classification for drivers that expose no usable error code."""
        if "timeout" in message:
            return DatabaseTimeoutError
        if "integrity" in message:
            return DatabaseIntegrityError
        if "constraint" in message:
            return DatabaseConstraintError
        if "connection" in message:
            return DatabaseConnectionError
        if "transaction" in message:
            return DatabaseTransactionError
        return DatabaseQueryError

archipy.adapters.base.sqlalchemy.mixins.SQLAlchemyFilterMixin

Mixin providing filtering capabilities for SQLAlchemy queries.

Supports equality, inequality, string operations, list operations, and NULL checks.

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

class SQLAlchemyFilterMixin:
    """Mixin providing filtering capabilities for SQLAlchemy queries.

    Supports equality, inequality, string operations, list operations, and NULL checks.
    """

    @staticmethod
    def _validate_list_operation(
        value: str | float | bool | Decimal | list[Any] | UUID | datetime | date | Enum | None,
        operation: FilterOperationType,
    ) -> list[Any]:
        """Validate that value is a list for list operations."""
        if not isinstance(value, list):
            raise InvalidArgumentError(
                argument_name="value",
                additional_data={
                    "operation": operation.value,
                    "expected_type": "list",
                    "got": type(value).__name__,
                },
            )
        return value

    @staticmethod
    def _apply_filter(
        query: Select | Update | Delete,
        field: InstrumentedAttribute,
        value: str | float | bool | Decimal | list[Any] | UUID | datetime | date | Enum | None,
        operation: FilterOperationType,
    ) -> Select | Update | Delete:
        """Apply a filter to a SQLAlchemy query based on the specified operation.

        Args:
            query: The SQLAlchemy query to apply the filter to.
            field: The model attribute/column to filter on.
            value: The value to compare against.
            operation: The type of filter operation to apply.

        Returns:
            The updated query with the filter applied.
        """
        # Skip filter if value is None (except for IS_NULL/IS_NOT_NULL operations)
        if value is None and operation not in (FilterOperationType.IS_NULL, FilterOperationType.IS_NOT_NULL):
            return query

        if operation in (FilterOperationType.IN_LIST, FilterOperationType.NOT_IN_LIST):
            expression = field.in_(SQLAlchemyFilterMixin._validate_list_operation(value, operation))
            return query.where(~expression if operation == FilterOperationType.NOT_IN_LIST else expression)

        build_expression = _FILTER_EXPRESSIONS.get(operation)
        if build_expression is None:
            return query
        return query.where(build_expression(field, value))

archipy.adapters.base.sqlalchemy.mixins.SQLAlchemyPaginationMixin

Mixin providing pagination capabilities for SQLAlchemy queries.

Supports limiting results and applying offsets for paginated queries.

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

class SQLAlchemyPaginationMixin:
    """Mixin providing pagination capabilities for SQLAlchemy queries.

    Supports limiting results and applying offsets for paginated queries.
    """

    @staticmethod
    def _apply_pagination(query: Select, pagination: PaginationDTO | None) -> Select:
        """Apply pagination to a SQLAlchemy query.

        Args:
            query: The SQLAlchemy query to paginate.
            pagination: Pagination settings (page size and offset).

        Returns:
            The paginated query.
        """
        if pagination is None:
            return query
        return query.limit(pagination.page_size).offset(pagination.offset)

    @staticmethod
    def _total_from_page(
        results: list[Any],
        pagination: PaginationDTO | None,
        has_multiple_entities: bool,
    ) -> int | None:
        """Return the total count when the fetched page already proves it, otherwise None.

        The page proves the total when nothing was paginated, or when the first page is not full.
        Skipping the COUNT query then saves a full scan of the filtered rows.

        Args:
            results: The fetched page.
            pagination: Pagination settings used for the query.
            has_multiple_entities: True when rows may have been dropped while extracting entities.

        Returns:
            The total count, or None when a COUNT query is needed.
        """
        if has_multiple_entities:
            return None
        if pagination is None or (pagination.offset == 0 and len(results) < pagination.page_size):
            return len(results)
        return None

    @staticmethod
    def _build_count_source(entity: type[BaseEntity], query: Select, limit: int | None = None) -> Select:
        """Build the inner SELECT that a COUNT runs over.

        Drops the ordering, and selects only the primary key of ``entity`` so the database can answer from the
        primary key index instead of reading every selected column. The columns are kept as they are when they
        change the row count (DISTINCT, GROUP BY, HAVING, LIMIT, OFFSET).

        Args:
            entity: The entity class the query selects.
            query: The unpaginated search query.
            limit: Optional cap on the rows counted. Ignored when the query has its own LIMIT or OFFSET.

        Returns:
            The inner SELECT.
        """
        source = query.order_by(None)
        shapes_rows = (
            source._distinct
            or source._group_by_clauses
            or source._having_criteria
            or source._limit_clause is not None
            or source._offset_clause is not None
        )
        if not shapes_rows:
            source = source.with_only_columns(*inspect(entity).primary_key, maintain_column_froms=True)
            if limit is not None:
                source = source.limit(limit)
        return source

    @staticmethod
    def _count_cache_key(source: Select, count_threshold: int | None) -> tuple[str, str, int | None]:
        """Build the cache key of a count query from its SQL text and bound values."""
        compiled = source.compile()
        return str(compiled), repr(sorted(compiled.params.items())), count_threshold

    @staticmethod
    def _build_count_query(source: Select) -> Select:
        """Wrap an inner SELECT in a COUNT.

        Args:
            source: The inner SELECT from ``_build_count_source``.

        Returns:
            The COUNT query.
        """
        return select(func.count()).select_from(source.subquery())

    @staticmethod
    def _build_estimate_statement(source: Select, dialect: Dialect) -> Executable | None:
        """Build the statement that estimates the rows of an inner SELECT.

        Args:
            source: The inner SELECT from ``_build_count_source``, without a LIMIT.
            dialect: The dialect of the connected database.

        Returns:
            An EXPLAIN statement, or None when the dialect has no supported estimate.
        """
        if dialect.name != "postgresql":
            return None
        return _ExplainJson(source)

    @staticmethod
    def _parse_estimate(plan: object) -> int | None:
        """Read the estimated row count from a PostgreSQL JSON plan.

        Args:
            plan: The first column of the EXPLAIN result, a JSON string or an already parsed list.

        Returns:
            The estimated rows, or None when the plan has an unexpected shape.
        """
        try:
            parsed: Any = json.loads(plan) if isinstance(plan, str | bytes) else plan
            return int(parsed[0]["Plan"]["Plan Rows"])
        except KeyError, IndexError, TypeError, ValueError:
            return None

archipy.adapters.base.sqlalchemy.mixins.SQLAlchemySortMixin

Mixin providing sorting capabilities for SQLAlchemy queries.

Supports dynamic column selection and ascending/descending order.

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

class SQLAlchemySortMixin:
    """Mixin providing sorting capabilities for SQLAlchemy queries.

    Supports dynamic column selection and ascending/descending order.
    """

    @staticmethod
    def _apply_sorting(entity: type[T], query: Select, sort_info: SortDTO | None) -> Select:
        """Apply sorting to a SQLAlchemy query.

        Args:
            entity: The entity class to query.
            query: The SQLAlchemy query to sort.
            sort_info: Sorting information (column and direction).

        Returns:
            The sorted query.

        Raises:
            InvalidArgumentError: If the sort order is invalid.
        """
        if sort_info is None:
            return query
        if isinstance(sort_info.column, str):
            sort_column = getattr(entity, sort_info.column)
        elif isinstance(sort_info.column, Enum):
            sort_column = getattr(entity, sort_info.column.name.lower())
        else:
            sort_column = sort_info.column

        order_value = sort_info.order.value if isinstance(sort_info.order, Enum) else sort_info.order
        match order_value:
            case SortOrderType.ASCENDING.value:
                return query.order_by(sort_column.asc())
            case SortOrderType.DESCENDING.value:
                return query.order_by(sort_column.desc())
            case _:
                raise InvalidArgumentError(argument_name="sort_info.order")

options: show_root_toc_entry: false heading_level: 3

Atomic Transaction Decorators

Decorators that wrap a function in a SQLAlchemy transaction, rolling back on failure.

SQLAlchemy atomic transaction decorators.

The unit of work for the SQLAlchemy adapters: it wraps a function in a transaction on the database's session manager. This module provides decorators for managing SQLAlchemy transactions with automatic commit/rollback and support for different database types (PostgreSQL, SQLite, StarRocks, MySQL).

archipy.adapters.base.sqlalchemy.atomic.logger module-attribute

logger = logging.getLogger(__name__)

archipy.adapters.base.sqlalchemy.atomic.ATOMIC_BLOCK_CONFIGS module-attribute

ATOMIC_BLOCK_CONFIGS = {
    "postgres": {
        "flag": "in_postgres_sqlalchemy_atomic_block",
        "registry": "archipy.adapters.postgres.sqlalchemy.session_manager_registry.PostgresSessionManagerRegistry",
    },
    "sqlite": {
        "flag": "in_sqlite_sqlalchemy_atomic_block",
        "registry": "archipy.adapters.sqlite.sqlalchemy.session_manager_registry.SQLiteSessionManagerRegistry",
    },
    "starrocks": {
        "flag": "in_starrocks_sqlalchemy_atomic_block",
        "registry": "archipy.adapters.starrocks.sqlalchemy.session_manager_registry.StarRocksSessionManagerRegistry",
    },
    "mysql": {
        "flag": "in_mysql_sqlalchemy_atomic_block",
        "registry": "archipy.adapters.mysql.sqlalchemy.session_manager_registry.MySQLSessionManagerRegistry",
    },
}

archipy.adapters.base.sqlalchemy.atomic.R module-attribute

R = TypeVar('R')

archipy.adapters.base.sqlalchemy.atomic.postgres_sqlalchemy_atomic_decorator module-attribute

postgres_sqlalchemy_atomic_decorator = deprecated_alias(
    "postgres_sqlalchemy_atomic_decorator",
    postgres_sqlalchemy_atomic,
)

archipy.adapters.base.sqlalchemy.atomic.async_postgres_sqlalchemy_atomic_decorator module-attribute

async_postgres_sqlalchemy_atomic_decorator = (
    deprecated_alias(
        "async_postgres_sqlalchemy_atomic_decorator",
        postgres_sqlalchemy_atomic,
    )
)

archipy.adapters.base.sqlalchemy.atomic.sqlite_sqlalchemy_atomic_decorator module-attribute

sqlite_sqlalchemy_atomic_decorator = deprecated_alias(
    "sqlite_sqlalchemy_atomic_decorator",
    sqlite_sqlalchemy_atomic,
)

archipy.adapters.base.sqlalchemy.atomic.async_sqlite_sqlalchemy_atomic_decorator module-attribute

async_sqlite_sqlalchemy_atomic_decorator = deprecated_alias(
    "async_sqlite_sqlalchemy_atomic_decorator",
    sqlite_sqlalchemy_atomic,
)

archipy.adapters.base.sqlalchemy.atomic.starrocks_sqlalchemy_atomic_decorator module-attribute

starrocks_sqlalchemy_atomic_decorator = deprecated_alias(
    "starrocks_sqlalchemy_atomic_decorator",
    starrocks_sqlalchemy_atomic,
)

archipy.adapters.base.sqlalchemy.atomic.async_starrocks_sqlalchemy_atomic_decorator module-attribute

async_starrocks_sqlalchemy_atomic_decorator = (
    deprecated_alias(
        "async_starrocks_sqlalchemy_atomic_decorator",
        starrocks_sqlalchemy_atomic,
    )
)

archipy.adapters.base.sqlalchemy.atomic.mysql_sqlalchemy_atomic_decorator module-attribute

mysql_sqlalchemy_atomic_decorator = deprecated_alias(
    "mysql_sqlalchemy_atomic_decorator",
    mysql_sqlalchemy_atomic,
)

archipy.adapters.base.sqlalchemy.atomic.async_mysql_sqlalchemy_atomic_decorator module-attribute

async_mysql_sqlalchemy_atomic_decorator = deprecated_alias(
    "async_mysql_sqlalchemy_atomic_decorator",
    mysql_sqlalchemy_atomic,
)

archipy.adapters.base.sqlalchemy.atomic.sqlalchemy_atomic

sqlalchemy_atomic(
    db_type: str,
    function: Callable[P, R],
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R]

sqlalchemy_atomic(
    db_type: str,
    function: None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

sqlalchemy_atomic(
    db_type: str,
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> (
    Callable[P, R]
    | Callable[[Callable[P, R]], Callable[P, R]]
)

Run a function inside a SQLAlchemy transaction for the given database type.

If the function succeeds, the transaction is committed; otherwise it is rolled back and the error is mapped to a domain database error. Synchronous and async def functions are both supported; the kind is detected automatically.

Parameters:

Name	Type	Description	Default
db_type	str	The database type ("postgres", "sqlite", "starrocks", or "mysql").	required
function	Callable | None	The function to wrap. If None, returns a decorator to apply later.	None
registry	type[SessionManagerRegistry] | None	Registry that supplies the session manager. Defaults to the registry of db_type, which reads the global config; pass a registry subclass to run on another database of the same type (see the multiple databases section of the docs).	None

Returns:

Name	Type	Description
Callable	Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]	The wrapped function, or a decorator to apply later when function is None.

Raises:

Type	Description
InvalidArgumentError	If an invalid db_type is provided.
DatabaseSerializationError	If a serialization failure is detected.
DatabaseDeadlockError	If an operational error occurs due to a deadlock.
DatabaseTransactionError	If a transaction-related error occurs.
DatabaseQueryError	If a query-related error occurs.
DatabaseConnectionError	If a connection-related error occurs.
DatabaseConstraintError	If a constraint violation is detected.
DatabaseIntegrityError	If an integrity violation is detected.
DatabaseTimeoutError	If a database operation times out.
DatabaseConfigurationError	If there's an error in the database configuration.

Example

Synchronous PostgreSQL example

@sqlalchemy_atomic("postgres") def update_user(id: int, name: str) -> None: # Database operations pass

Asynchronous SQLite example

@sqlalchemy_atomic("sqlite") async def update_record(id: int, data: str) -> None: # Async database operations pass

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

def sqlalchemy_atomic[**P, R](
    db_type: str,
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
    """Run a function inside a SQLAlchemy transaction for the given database type.

    If the function succeeds, the transaction is committed; otherwise it is rolled back and the
    error is mapped to a domain database error. Synchronous and ``async def`` functions are both
    supported; the kind is detected automatically.

    Args:
        db_type (str): The database type ("postgres", "sqlite", "starrocks", or "mysql").
        function (Callable | None): The function to wrap. If None, returns a decorator to apply later.
        registry (type[SessionManagerRegistry] | None): Registry that supplies the session manager. Defaults to the
            registry of ``db_type``, which reads the global config; pass a registry subclass to run on another
            database of the same type (see the multiple databases section of the docs).

    Returns:
        Callable: The wrapped function, or a decorator to apply later when ``function`` is None.

    Raises:
        InvalidArgumentError: If an invalid db_type is provided.
        DatabaseSerializationError: If a serialization failure is detected.
        DatabaseDeadlockError: If an operational error occurs due to a deadlock.
        DatabaseTransactionError: If a transaction-related error occurs.
        DatabaseQueryError: If a query-related error occurs.
        DatabaseConnectionError: If a connection-related error occurs.
        DatabaseConstraintError: If a constraint violation is detected.
        DatabaseIntegrityError: If an integrity violation is detected.
        DatabaseTimeoutError: If a database operation times out.
        DatabaseConfigurationError: If there's an error in the database configuration.

    Example:
        # Synchronous PostgreSQL example
        @sqlalchemy_atomic("postgres")
        def update_user(id: int, name: str) -> None:
            # Database operations
            pass

        # Asynchronous SQLite example
        @sqlalchemy_atomic("sqlite")
        async def update_record(id: int, data: str) -> None:
            # Async database operations
            pass
    """
    if db_type not in ATOMIC_BLOCK_CONFIGS:
        raise InvalidArgumentError(
            argument_name="db_type",
            additional_data={"valid_values": list(ATOMIC_BLOCK_CONFIGS.keys())},
        )

    if function is None:
        return cast("Callable[[Callable[P, R]], Callable[P, R]]", _bind_db_type(db_type, registry))

    atomic_flag = ATOMIC_BLOCK_CONFIGS[db_type]["flag"]
    if inspect.iscoroutinefunction(function):
        return cast(
            "Callable[P, R]",
            _create_async_atomic_wrapper(
                function,
                db_type=db_type,
                atomic_flag=atomic_flag,
                registry_override=registry,
            ),
        )
    return _create_sync_atomic_wrapper(function, db_type=db_type, atomic_flag=atomic_flag, registry_override=registry)

archipy.adapters.base.sqlalchemy.atomic.postgres_sqlalchemy_atomic

postgres_sqlalchemy_atomic(
    function: Callable[P, R],
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R]

postgres_sqlalchemy_atomic(
    function: None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

postgres_sqlalchemy_atomic(
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> (
    Callable[P, R]
    | Callable[[Callable[P, R]], Callable[P, R]]
)

Decorator for PostgreSQL atomic transactions; works for sync and async def functions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a decorator to apply later.	None
registry	type[SessionManagerRegistry] | None	Registry for another database of the same type; defaults to the registry that reads the global config.	None

Returns:

Name	Type	Description
Callable	Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]	The wrapped function, or a decorator to apply later when function is None.

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

def postgres_sqlalchemy_atomic[**P, R](
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator for PostgreSQL atomic transactions; works for sync and ``async def`` functions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a decorator to apply later.
        registry (type[SessionManagerRegistry] | None): Registry for another database of the same type; defaults to
            the registry that reads the global config.

    Returns:
        Callable: The wrapped function, or a decorator to apply later when ``function`` is None.
    """
    return _atomic_for("postgres", function, registry)

archipy.adapters.base.sqlalchemy.atomic.sqlite_sqlalchemy_atomic

sqlite_sqlalchemy_atomic(
    function: Callable[P, R],
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R]

sqlite_sqlalchemy_atomic(
    function: None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

sqlite_sqlalchemy_atomic(
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> (
    Callable[P, R]
    | Callable[[Callable[P, R]], Callable[P, R]]
)

Decorator for SQLite atomic transactions; works for sync and async def functions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a decorator to apply later.	None
registry	type[SessionManagerRegistry] | None	Registry for another database of the same type; defaults to the registry that reads the global config.	None

Returns:

Name	Type	Description
Callable	Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]	The wrapped function, or a decorator to apply later when function is None.

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

def sqlite_sqlalchemy_atomic[**P, R](
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator for SQLite atomic transactions; works for sync and ``async def`` functions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a decorator to apply later.
        registry (type[SessionManagerRegistry] | None): Registry for another database of the same type; defaults to
            the registry that reads the global config.

    Returns:
        Callable: The wrapped function, or a decorator to apply later when ``function`` is None.
    """
    return _atomic_for("sqlite", function, registry)

archipy.adapters.base.sqlalchemy.atomic.starrocks_sqlalchemy_atomic

starrocks_sqlalchemy_atomic(
    function: Callable[P, R],
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R]

starrocks_sqlalchemy_atomic(
    function: None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

starrocks_sqlalchemy_atomic(
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> (
    Callable[P, R]
    | Callable[[Callable[P, R]], Callable[P, R]]
)

Decorator for StarRocks atomic transactions; works for sync and async def functions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a decorator to apply later.	None
registry	type[SessionManagerRegistry] | None	Registry for another database of the same type; defaults to the registry that reads the global config.	None

Returns:

Name	Type	Description
Callable	Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]	The wrapped function, or a decorator to apply later when function is None.

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

def starrocks_sqlalchemy_atomic[**P, R](
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator for StarRocks atomic transactions; works for sync and ``async def`` functions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a decorator to apply later.
        registry (type[SessionManagerRegistry] | None): Registry for another database of the same type; defaults to
            the registry that reads the global config.

    Returns:
        Callable: The wrapped function, or a decorator to apply later when ``function`` is None.
    """
    return _atomic_for("starrocks", function, registry)

archipy.adapters.base.sqlalchemy.atomic.mysql_sqlalchemy_atomic

mysql_sqlalchemy_atomic(
    function: Callable[P, R],
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R]

mysql_sqlalchemy_atomic(
    function: None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

mysql_sqlalchemy_atomic(
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> (
    Callable[P, R]
    | Callable[[Callable[P, R]], Callable[P, R]]
)

Decorator for MySQL atomic transactions; works for sync and async def functions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a decorator to apply later.	None
registry	type[SessionManagerRegistry] | None	Registry for another database of the same type; defaults to the registry that reads the global config.	None

Returns:

Name	Type	Description
Callable	Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]	The wrapped function, or a decorator to apply later when function is None.

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

def mysql_sqlalchemy_atomic[**P, R](
    function: Callable[P, R] | None = None,
    *,
    registry: type[SessionManagerRegistry] | None = None,
) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator for MySQL atomic transactions; works for sync and ``async def`` functions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a decorator to apply later.
        registry (type[SessionManagerRegistry] | None): Registry for another database of the same type; defaults to
            the registry that reads the global config.

    Returns:
        Callable: The wrapped function, or a decorator to apply later when ``function`` is None.
    """
    return _atomic_for("mysql", function, registry)

archipy.adapters.base.sqlalchemy.atomic.sqlalchemy_atomic_decorator

sqlalchemy_atomic_decorator(
    db_type: str,
    is_async: bool = False,
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any]

Deprecated: use :func:sqlalchemy_atomic, which detects async functions itself.

Parameters:

Name	Type	Description	Default
db_type	str	The database type ("postgres", "sqlite", "starrocks", or "mysql").	required
is_async	bool	Ignored; the function kind is detected automatically.	False
function	Callable | None	The function to wrap. If None, returns a decorator to apply later.	None

Returns:

Name	Type	Description
Callable	Callable[..., Any]	The wrapped function, or a decorator to apply later when function is None.

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

def sqlalchemy_atomic_decorator(
    db_type: str,
    is_async: bool = False,
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any]:
    """Deprecated: use :func:`sqlalchemy_atomic`, which detects async functions itself.

    Args:
        db_type (str): The database type ("postgres", "sqlite", "starrocks", or "mysql").
        is_async (bool): Ignored; the function kind is detected automatically.
        function (Callable | None): The function to wrap. If None, returns a decorator to apply later.

    Returns:
        Callable: The wrapped function, or a decorator to apply later when ``function`` is None.
    """
    del is_async
    warn_deprecated_name("sqlalchemy_atomic_decorator", "sqlalchemy_atomic")
    return sqlalchemy_atomic(db_type, function)

options: show_root_toc_entry: false heading_level: 3