SQLite

The sqlite/sqlalchemy adapter provides a SQLite-specific SQLAlchemy integration suitable for development, testing, and lightweight production workloads.

Session Managers

SQLite-specific session manager handling file-based and in-memory database connections.

archipy.adapters.sqlite.sqlalchemy.session_managers.SQLiteSQLAlchemySessionManager

Bases: BaseSQLAlchemySessionManager[SQLiteSQLAlchemyConfig]

Synchronous SQLAlchemy session manager for SQLite.

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

Parameters:

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

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

class SQLiteSQLAlchemySessionManager(BaseSQLAlchemySessionManager[SQLiteSQLAlchemyConfig], metaclass=Singleton):
    """Synchronous SQLAlchemy session manager for SQLite.

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

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

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

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

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

        Returns:
            The SQLiteSQLAlchemyConfig class.
        """
        return SQLiteSQLAlchemyConfig

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

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

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

        Args:
            configs: SQLite configuration.

        Returns:
            A SQLAlchemy URL object for SQLite.

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

archipy.adapters.sqlite.sqlalchemy.session_managers.SQLiteSQLAlchemySessionManager.engine instance-attribute

engine = _create_engine(orm_config)

archipy.adapters.sqlite.sqlalchemy.session_managers.SQLiteSQLAlchemySessionManager.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.sqlite.sqlalchemy.session_managers.SQLiteSQLAlchemySessionManager.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.sqlite.sqlalchemy.session_managers.AsyncSQLiteSQLAlchemySessionManager

Bases: AsyncBaseSQLAlchemySessionManager[SQLiteSQLAlchemyConfig]

Asynchronous SQLAlchemy session manager for SQLite.

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

Parameters:

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

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

class AsyncSQLiteSQLAlchemySessionManager(
    AsyncBaseSQLAlchemySessionManager[SQLiteSQLAlchemyConfig],
    metaclass=Singleton,
):
    """Asynchronous SQLAlchemy session manager for SQLite.

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

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

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

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

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

        Returns:
            The SQLiteSQLAlchemyConfig class.
        """
        return SQLiteSQLAlchemyConfig

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

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

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

        Args:
            configs: SQLite configuration.

        Returns:
            A SQLAlchemy URL object for SQLite.

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

archipy.adapters.sqlite.sqlalchemy.session_managers.AsyncSQLiteSQLAlchemySessionManager.engine instance-attribute

engine = _create_async_engine(orm_config)

archipy.adapters.sqlite.sqlalchemy.session_managers.AsyncSQLiteSQLAlchemySessionManager.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.sqlite.sqlalchemy.session_managers.AsyncSQLiteSQLAlchemySessionManager.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 SQLite session manager instances.

archipy.adapters.sqlite.sqlalchemy.session_manager_registry.SQLiteSessionManagerRegistry

Bases: SessionManagerRegistry

Registry for SQLite SQLAlchemy session managers.

This registry provides a centralized access point for both synchronous and asynchronous SQLite 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 (SQLiteSQLAlchemySessionManager) - An asynchronous session manager (AsyncSQLiteSQLAlchemySessionManager)

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

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

    This registry provides a centralized access point for both synchronous and
    asynchronous SQLite 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 (SQLiteSQLAlchemySessionManager)
    - An asynchronous session manager (AsyncSQLiteSQLAlchemySessionManager)
    """

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

        Lazily initializes a default SQLiteSQLAlchemySessionManager if none has been set.

        Returns:
            SessionManagerPort: The registered synchronous session manager

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

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

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

        Args:
            manager: The session manager to register
        """
        cls._sync_instance = manager

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

        Lazily initializes a default AsyncSQLiteSQLAlchemySessionManager if none has been set.

        Returns:
            AsyncSessionManagerPort: The registered asynchronous session manager

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

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

    @classmethod
    def set_async_manager(cls, manager: AsyncSessionManagerPort) -> None:
        """Register an asynchronous session manager.

        Args:
            manager: The async session manager to register
        """
        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.sqlite.sqlalchemy.session_manager_registry.SQLiteSessionManagerRegistry.get_sync_manager classmethod

get_sync_manager() -> SessionManagerPort

Get the synchronous SQLite session manager instance.

Lazily initializes a default SQLiteSQLAlchemySessionManager 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/sqlite/sqlalchemy/session_manager_registry.py

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

    Lazily initializes a default SQLiteSQLAlchemySessionManager if none has been set.

    Returns:
        SessionManagerPort: The registered synchronous session manager

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

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

archipy.adapters.sqlite.sqlalchemy.session_manager_registry.SQLiteSessionManagerRegistry.set_sync_manager classmethod

set_sync_manager(manager: SessionManagerPort) -> None

Register a synchronous session manager.

Parameters:

Name	Type	Description	Default
manager	SessionManagerPort	The session manager to register	required

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

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

    Args:
        manager: The session manager to register
    """
    cls._sync_instance = manager

archipy.adapters.sqlite.sqlalchemy.session_manager_registry.SQLiteSessionManagerRegistry.get_async_manager classmethod

get_async_manager() -> AsyncSessionManagerPort

Get the asynchronous SQLite session manager instance.

Lazily initializes a default AsyncSQLiteSQLAlchemySessionManager 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/sqlite/sqlalchemy/session_manager_registry.py

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

    Lazily initializes a default AsyncSQLiteSQLAlchemySessionManager if none has been set.

    Returns:
        AsyncSessionManagerPort: The registered asynchronous session manager

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

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

archipy.adapters.sqlite.sqlalchemy.session_manager_registry.SQLiteSessionManagerRegistry.set_async_manager classmethod

set_async_manager(manager: AsyncSessionManagerPort) -> None

Register an asynchronous session manager.

Parameters:

Name	Type	Description	Default
manager	AsyncSessionManagerPort	The async session manager to register	required

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

@classmethod
def set_async_manager(cls, manager: AsyncSessionManagerPort) -> None:
    """Register an asynchronous session manager.

    Args:
        manager: The async session manager to register
    """
    cls._async_instance = manager

archipy.adapters.sqlite.sqlalchemy.session_manager_registry.SQLiteSessionManagerRegistry.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/sqlite/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 SQLite adapter built on the base SQLAlchemy adapter with SQLite-specific configuration.

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter

Bases: BaseSQLAlchemyAdapter[SQLiteSQLAlchemyConfig]

Synchronous SQLAlchemy adapter for SQLite.

Inherits from BaseSQLAlchemyAdapter to provide SQLite-specific session management and database operations, typically used for in-memory testing.

Parameters:

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

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

class SQLiteSQLAlchemyAdapter(BaseSQLAlchemyAdapter[SQLiteSQLAlchemyConfig]):
    """Synchronous SQLAlchemy adapter for SQLite.

    Inherits from BaseSQLAlchemyAdapter to provide SQLite-specific session management
    and database operations, typically used for in-memory testing.

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

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

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

    @override
    def _create_session_manager(self, configs: SQLiteSQLAlchemyConfig) -> SQLiteSQLAlchemySessionManager:
        """Create a SQLite-specific session manager.

        Args:
            configs: SQLite configuration.

        Returns:
            A SQLite session manager instance.
        """
        return SQLiteSQLAlchemySessionManager(configs)

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.session_manager instance-attribute

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

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.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.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.execute_search_query

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

Execute a search query with pagination and sorting.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.create

create(entity: T) -> T | None

Create a new entity in the database.

Parameters:

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

Returns:

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

Raises:

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

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

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

    Args:
        entity: The entity to create.

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

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.bulk_create

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

Creates multiple entities in a single database operation.

Parameters:

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

Returns:

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

Raises:

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

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

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

    Args:
        entities: List of entities to create.

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

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.get_by_uuid

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

Retrieve an entity by its UUID.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The entity if found, None otherwise.

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.delete

delete(entity: T) -> None

Delete an entity from the database.

Parameters:

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

Raises:

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

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

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

    Args:
        entity: The entity to delete.

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.bulk_delete

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

Delete multiple entities from the database.

Parameters:

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

Raises:

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

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

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

    Args:
        entities: List of entities to delete.

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.execute

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

Execute a SQLAlchemy statement.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The result of the execution.

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

archipy.adapters.sqlite.sqlalchemy.adapters.SQLiteSQLAlchemyAdapter.scalars

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

Execute a SQLAlchemy statement and return scalar results.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The scalar results of the execution.

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

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter

Bases: AsyncBaseSQLAlchemyAdapter[SQLiteSQLAlchemyConfig]

Asynchronous SQLAlchemy adapter for SQLite.

Inherits from AsyncBaseSQLAlchemyAdapter to provide async SQLite-specific session management and database operations, typically used for in-memory testing.

Parameters:

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

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

class AsyncSQLiteSQLAlchemyAdapter(AsyncBaseSQLAlchemyAdapter[SQLiteSQLAlchemyConfig]):
    """Asynchronous SQLAlchemy adapter for SQLite.

    Inherits from AsyncBaseSQLAlchemyAdapter to provide async SQLite-specific session
    management and database operations, typically used for in-memory testing.

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

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

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

    @override
    def _create_async_session_manager(self, configs: SQLiteSQLAlchemyConfig) -> AsyncSQLiteSQLAlchemySessionManager:
        """Create an async SQLite-specific session manager.

        Args:
            configs: SQLite configuration.

        Returns:
            An async SQLite session manager instance.
        """
        return AsyncSQLiteSQLAlchemySessionManager(configs)

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.session_manager instance-attribute

session_manager: AsyncBaseSQLAlchemySessionManager[
    ConfigT
] = _create_async_session_manager(configs)

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.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.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.execute_search_query async

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

Execute a search query with pagination and sorting.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.create async

create(entity: T) -> T | None

Create a new entity in the database.

Parameters:

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

Returns:

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

Raises:

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

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

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

    Args:
        entity: The entity to create.

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

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.bulk_create async

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

Creates multiple entities in a single database operation.

Parameters:

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

Returns:

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

Raises:

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

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

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

    Args:
        entities: List of entities to create.

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

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.get_by_uuid async

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

Retrieve an entity by its UUID.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The entity if found, None otherwise.

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.delete async

delete(entity: BaseEntity) -> None

Delete an entity from the database.

Parameters:

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

Raises:

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

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

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

    Args:
        entity: The entity to delete.

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.bulk_delete async

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

Delete multiple entities from the database.

Parameters:

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

Raises:

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

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

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

    Args:
        entities: List of entities to delete.

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

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

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.execute async

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

Execute a SQLAlchemy statement.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The result of the execution.

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

archipy.adapters.sqlite.sqlalchemy.adapters.AsyncSQLiteSQLAlchemyAdapter.scalars async

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

Execute a SQLAlchemy statement and return scalar results.

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        The scalar results of the execution.

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

options: show_root_toc_entry: false heading_level: 3