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.

Classes:

Name	Description
SQLiteSQLAlchemySessionManager	Synchronous SQLAlchemy session manager for SQLite.
AsyncSQLiteSQLAlchemySessionManager	Asynchronous SQLAlchemy session manager for SQLite.

SQLiteSQLAlchemySessionManager

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

AsyncSQLiteSQLAlchemySessionManager

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

options: show_root_toc_entry: false heading_level: 3

Session Manager Registry

Registry for SQLite session manager instances.

Classes:

Name	Description
SQLiteSessionManagerRegistry	Registry for SQLite SQLAlchemy session managers.

SQLiteSessionManagerRegistry

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)

Methods:

Name	Description
get_sync_manager	Get the synchronous SQLite session manager instance.
set_sync_manager	Register a synchronous session manager.
get_async_manager	Get the asynchronous SQLite session manager instance.
set_async_manager	Register an asynchronous session manager.
reset	Reset the registry to its initial state.

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

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

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

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

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

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.

Classes:

Name	Description
SQLiteSQLAlchemyAdapter	Synchronous SQLAlchemy adapter for SQLite.
AsyncSQLiteSQLAlchemyAdapter	Asynchronous SQLAlchemy adapter for SQLite.

SQLiteSQLAlchemyAdapter

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)

AsyncSQLiteSQLAlchemyAdapter

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)

options: show_root_toc_entry: false heading_level: 3