MySQL

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

Session Managers

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

MySQL SQLAlchemy session manager implementations.

archipy.adapters.mysql.sqlalchemy.session_managers.MySQLSQlAlchemySessionManager

Bases: BaseSQLAlchemySessionManager[MySQLSQLAlchemyConfig]

Synchronous SQLAlchemy session manager for MySQL.

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

Parameters:

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

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

class MySQLSQlAlchemySessionManager(BaseSQLAlchemySessionManager[MySQLSQLAlchemyConfig], metaclass=Singleton):
    """Synchronous SQLAlchemy session manager for MySQL.

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

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

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

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

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

        Returns:
            The MySQLSQLAlchemyConfig class.
        """
        return MySQLSQLAlchemyConfig

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

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

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

        Args:
            configs: MySQL configuration.

        Returns:
            A SQLAlchemy URL object for MySQL.

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

archipy.adapters.mysql.sqlalchemy.session_managers.MySQLSQlAlchemySessionManager.engine instance-attribute

engine = self._create_engine(orm_config)

archipy.adapters.mysql.sqlalchemy.session_managers.MySQLSQlAlchemySessionManager.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.mysql.sqlalchemy.session_managers.MySQLSQlAlchemySessionManager.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.mysql.sqlalchemy.session_managers.AsyncMySQLSQlAlchemySessionManager

Bases: AsyncBaseSQLAlchemySessionManager[MySQLSQLAlchemyConfig]

Asynchronous SQLAlchemy session manager for MySQL.

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

Parameters:

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

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

class AsyncMySQLSQlAlchemySessionManager(
    AsyncBaseSQLAlchemySessionManager[MySQLSQLAlchemyConfig],
    metaclass=Singleton,
):
    """Asynchronous SQLAlchemy session manager for MySQL.

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

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

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

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

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

        Returns:
            The MySQLSQLAlchemyConfig class.
        """
        return MySQLSQLAlchemyConfig

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

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

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

        For async operations, MySQL uses the mysql+asyncmy driver which is provided
        by the asyncmy2 package.

        Args:
            configs: MySQL configuration.

        Returns:
            A SQLAlchemy URL object for MySQL with async driver.

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

archipy.adapters.mysql.sqlalchemy.session_managers.AsyncMySQLSQlAlchemySessionManager.engine instance-attribute

engine = self._create_async_engine(orm_config)

archipy.adapters.mysql.sqlalchemy.session_managers.AsyncMySQLSQlAlchemySessionManager.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.mysql.sqlalchemy.session_managers.AsyncMySQLSQlAlchemySessionManager.remove_session async

remove_session() -> None

Remove the current session from the registry.

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

Raises:

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Session Manager Registry

Registry for MySQL session manager instances.

Registry for MySQL SQLAlchemy session managers.

archipy.adapters.mysql.sqlalchemy.session_manager_registry.MySQLSessionManagerRegistry

Bases: SessionManagerRegistry

Registry for MySQL SQLAlchemy session managers.

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

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

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

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

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

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

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

        Lazily initializes a default MySQLSQlAlchemySessionManager if none has been set.

        Returns:
            SessionManagerPort: The registered synchronous session manager

        Raises:
            DatabaseConnectionError: If there's an error initializing the session manager
        """
        if cls._sync_instance is None:
            try:
                cls._sync_instance = MySQLSQlAlchemySessionManager()
            except Exception as e:
                raise DatabaseConnectionError(
                    database="mysql",
                ) from e
        return cls._sync_instance

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

        Args:
            manager: An instance implementing SessionManagerPort

        Raises:
            InvalidArgumentError: If the manager is None
            InvalidEntityTypeError: If the manager doesn't implement SessionManagerPort
        """
        if manager is None:
            raise InvalidArgumentError(argument_name="mysql_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 MySQL session manager instance.

        Lazily initializes a default AsyncMySQLSQlAlchemySessionManager if none has been set.

        Returns:
            AsyncSessionManagerPort: The registered asynchronous session manager

        Raises:
            DatabaseConnectionError: If there's an error initializing the session manager
        """
        if cls._async_instance is None:
            try:
                cls._async_instance = AsyncMySQLSQlAlchemySessionManager()
            except Exception as e:
                raise DatabaseConnectionError(
                    database="mysql",
                ) from e
        return cls._async_instance

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

        Args:
            manager: An instance implementing AsyncSessionManagerPort

        Raises:
            InvalidArgumentError: If the manager is None
            InvalidEntityTypeError: If the manager doesn't implement AsyncSessionManagerPort
        """
        if manager is None:
            raise InvalidArgumentError(argument_name="mysql_async_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.mysql.sqlalchemy.session_manager_registry.MySQLSessionManagerRegistry.get_sync_manager classmethod

get_sync_manager() -> SessionManagerPort

Get the synchronous MySQL session manager instance.

Lazily initializes a default MySQLSQlAlchemySessionManager if none has been set.

Returns:

Name	Type	Description
SessionManagerPort	SessionManagerPort	The registered synchronous session manager

Raises:

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

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

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

    Lazily initializes a default MySQLSQlAlchemySessionManager if none has been set.

    Returns:
        SessionManagerPort: The registered synchronous session manager

    Raises:
        DatabaseConnectionError: If there's an error initializing the session manager
    """
    if cls._sync_instance is None:
        try:
            cls._sync_instance = MySQLSQlAlchemySessionManager()
        except Exception as e:
            raise DatabaseConnectionError(
                database="mysql",
            ) from e
    return cls._sync_instance

archipy.adapters.mysql.sqlalchemy.session_manager_registry.MySQLSessionManagerRegistry.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/mysql/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="mysql_session_manager")

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

archipy.adapters.mysql.sqlalchemy.session_manager_registry.MySQLSessionManagerRegistry.get_async_manager classmethod

get_async_manager() -> AsyncSessionManagerPort

Get the asynchronous MySQL session manager instance.

Lazily initializes a default AsyncMySQLSQlAlchemySessionManager if none has been set.

Returns:

Name	Type	Description
AsyncSessionManagerPort	AsyncSessionManagerPort	The registered asynchronous session manager

Raises:

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

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

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

    Lazily initializes a default AsyncMySQLSQlAlchemySessionManager if none has been set.

    Returns:
        AsyncSessionManagerPort: The registered asynchronous session manager

    Raises:
        DatabaseConnectionError: If there's an error initializing the session manager
    """
    if cls._async_instance is None:
        try:
            cls._async_instance = AsyncMySQLSQlAlchemySessionManager()
        except Exception as e:
            raise DatabaseConnectionError(
                database="mysql",
            ) from e
    return cls._async_instance

archipy.adapters.mysql.sqlalchemy.session_manager_registry.MySQLSessionManagerRegistry.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/mysql/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="mysql_async_session_manager")

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

archipy.adapters.mysql.sqlalchemy.session_manager_registry.MySQLSessionManagerRegistry.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/mysql/sqlalchemy/session_manager_registry.py

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

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

options: show_root_toc_entry: false heading_level: 3

Adapters

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

MySQL SQLAlchemy adapter implementations.

archipy.adapters.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter

Bases: BaseSQLAlchemyAdapter[MySQLSQLAlchemyConfig]

Synchronous SQLAlchemy adapter for MySQL.

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

Parameters:

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

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

class MySQLSQLAlchemyAdapter(BaseSQLAlchemyAdapter[MySQLSQLAlchemyConfig]):
    """Synchronous SQLAlchemy adapter for MySQL.

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

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

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

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

    @override
    def _create_session_manager(self, configs: MySQLSQLAlchemyConfig) -> MySQLSQlAlchemySessionManager:
        """Create a MySQL-specific session manager.

        Args:
            configs: MySQL configuration.

        Returns:
            A MySQL session manager instance.
        """
        return MySQLSQlAlchemySessionManager(configs)

archipy.adapters.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.session_manager instance-attribute

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

archipy.adapters.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.get_session

get_session() -> Session

Get a database session.

Returns:

Name	Type	Description
Session	Session	A SQLAlchemy session.

Raises:

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

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

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

    Returns:
        Session: A SQLAlchemy session.

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

archipy.adapters.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.execute_search_query

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

Execute a search query with pagination and sorting.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

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

    Raises:
        DatabaseQueryError: If the database query fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        sort_info = sort_info or SortDTO.default()
        session = self.get_session()
        sorted_query = self._apply_sorting(entity, query, sort_info)
        paginated_query = self._apply_pagination(sorted_query, pagination)
        result_set = session.execute(paginated_query)
        if has_multiple_entities:
            # For multiple entities, fetchall returns list of Row objects
            raw_results = list(result_set.fetchall())
            # Convert to list[T] - each row contains entities of type T
            # Use tuple unpacking to access the first element
            results: list[T] = []
            for row in raw_results:
                if row:
                    # Row supports indexing and tuple unpacking
                    row_tuple = tuple(row)
                    if row_tuple:
                        first_entity = row_tuple[0]
                        # first_entity is T (entity type), verify it's an instance
                        if isinstance(first_entity, entity):
                            results.append(first_entity)
        else:
            # For single entity, scalars() returns list[T] directly
            results = list(result_set.scalars().all())
        count_query = select(func.count()).select_from(query.subquery())
        total_count = session.execute(count_query).scalar_one()
    except 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.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.create

create(entity: T) -> T | None

Create a new entity in the database.

Parameters:

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

Returns:

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

Raises:

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

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

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

    Args:
        entity: The entity to create.

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

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

    try:
        session = self.get_session()
        session.add(entity)
        session.flush()
    except 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.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.bulk_create

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

Creates multiple entities in a single database operation.

Parameters:

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

Returns:

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

Raises:

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

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

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

    Args:
        entities: List of entities to create.

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

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

    try:
        session = self.get_session()
        session.add_all(entities)
        session.flush()
    except 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.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.get_by_uuid

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

Retrieve an entity by its UUID.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The entity if found, None otherwise.

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

    try:
        session = self.get_session()
        result = session.get(entity_type, entity_uuid)
    except 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.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.delete

delete(entity: T) -> None

Delete an entity from the database.

Parameters:

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

Raises:

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

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

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

    Args:
        entity: The entity to delete.

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

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

archipy.adapters.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.bulk_delete

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

Delete multiple entities from the database.

Parameters:

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

Raises:

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

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

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

    Args:
        entities: List of entities to delete.

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

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

archipy.adapters.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.execute

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

Execute a SQLAlchemy statement.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The result of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = session.execute(statement, params or {})
    except 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.mysql.sqlalchemy.adapters.MySQLSQLAlchemyAdapter.scalars

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

Execute a SQLAlchemy statement and return scalar results.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The scalar results of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = session.scalars(statement, params or {})
    except 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.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter

Bases: AsyncBaseSQLAlchemyAdapter[MySQLSQLAlchemyConfig]

Asynchronous SQLAlchemy adapter for MySQL.

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

Parameters:

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

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

class AsyncMySQLSQLAlchemyAdapter(AsyncBaseSQLAlchemyAdapter[MySQLSQLAlchemyConfig]):
    """Asynchronous SQLAlchemy adapter for MySQL.

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

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

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

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

    @override
    def _create_async_session_manager(self, configs: MySQLSQLAlchemyConfig) -> AsyncMySQLSQlAlchemySessionManager:
        """Create an async MySQL-specific session manager.

        Args:
            configs: MySQL configuration.

        Returns:
            An async MySQL session manager instance.
        """
        return AsyncMySQLSQlAlchemySessionManager(configs)

archipy.adapters.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.session_manager instance-attribute

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

archipy.adapters.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.get_session

get_session() -> AsyncSession

Get a database session.

Returns:

Name	Type	Description
AsyncSession	AsyncSession	A SQLAlchemy async session.

Raises:

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

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

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

    Returns:
        AsyncSession: A SQLAlchemy async session.

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

archipy.adapters.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.execute_search_query async

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

Execute a search query with pagination and sorting.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

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

    Raises:
        DatabaseQueryError: If the database query fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        sort_info = sort_info or SortDTO.default()
        session = self.get_session()
        sorted_query = self._apply_sorting(entity, query, sort_info)
        paginated_query = self._apply_pagination(sorted_query, pagination)
        result_set = await session.execute(paginated_query)
        if has_multiple_entities:
            # For multiple entities, fetchall returns list of Row objects
            raw_results = list(result_set.fetchall())
            # Convert to list[T] - each row contains entities of type T
            # Use tuple unpacking to access the first element
            results: list[T] = []
            for row in raw_results:
                if row:
                    # Row supports indexing and tuple unpacking
                    row_tuple = tuple(row)
                    if row_tuple:
                        first_entity = row_tuple[0]
                        # first_entity is T (entity type), verify it's an instance
                        if isinstance(first_entity, entity):
                            results.append(first_entity)
        else:
            # For single entity, scalars() returns list[T] directly
            results = list(result_set.scalars().all())
        count_query = select(func.count()).select_from(query.subquery())
        total_count_result = await session.execute(count_query)
        total_count = total_count_result.scalar_one()
    except 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.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.create async

create(entity: T) -> T | None

Create a new entity in the database.

Parameters:

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

Returns:

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

Raises:

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

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

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

    Args:
        entity: The entity to create.

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

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

    try:
        session = self.get_session()
        session.add(entity)
        await session.flush()
    except 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.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.bulk_create async

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

Creates multiple entities in a single database operation.

Parameters:

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

Returns:

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

Raises:

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

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

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

    Args:
        entities: List of entities to create.

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

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

    try:
        session = self.get_session()
        session.add_all(entities)
        await session.flush()
    except 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.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.get_by_uuid async

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

Retrieve an entity by its UUID.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The entity if found, None otherwise.

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

    try:
        session = self.get_session()
        result = await session.get(entity_type, entity_uuid)
    except 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.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.delete async

delete(entity: BaseEntity) -> None

Delete an entity from the database.

Parameters:

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

Raises:

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

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

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

    Args:
        entity: The entity to delete.

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

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

archipy.adapters.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.bulk_delete async

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

Delete multiple entities from the database.

Parameters:

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

Raises:

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

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

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

    Args:
        entities: List of entities to delete.

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

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

archipy.adapters.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.execute async

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

Execute a SQLAlchemy statement.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The result of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = await session.execute(statement, params or {})
    except 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.mysql.sqlalchemy.adapters.AsyncMySQLSQLAlchemyAdapter.scalars async

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

Execute a SQLAlchemy statement and return scalar results.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The scalar results of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = await session.scalars(statement, params or {})
    except 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