PostgreSQL

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

Session Managers

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

Classes:

Name	Description
PostgresSQlAlchemySessionManager	Synchronous SQLAlchemy session manager for PostgreSQL.
AsyncPostgresSQlAlchemySessionManager	Asynchronous SQLAlchemy session manager for PostgreSQL.

PostgresSQlAlchemySessionManager

Synchronous SQLAlchemy session manager for PostgreSQL.

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

Parameters:

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

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

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

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

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

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

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

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

        Returns:
            The PostgresSQLAlchemyConfig class.
        """
        return PostgresSQLAlchemyConfig

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

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

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

        Args:
            configs: PostgreSQL configuration.

        Returns:
            A SQLAlchemy URL object for PostgreSQL.

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

AsyncPostgresSQlAlchemySessionManager

Asynchronous SQLAlchemy session manager for PostgreSQL.

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

Parameters:

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

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

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

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

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

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

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

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

        Returns:
            The PostgresSQLAlchemyConfig class.
        """
        return PostgresSQLAlchemyConfig

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

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

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

        Args:
            configs: PostgreSQL configuration.

        Returns:
            A SQLAlchemy URL object for PostgreSQL.

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

options: show_root_toc_entry: false heading_level: 3

Session Manager Registry

Registry for PostgreSQL session manager instances.

Classes:

Name	Description
PostgresSessionManagerRegistry	Registry for PostgreSQL SQLAlchemy session managers.

PostgresSessionManagerRegistry

Registry for PostgreSQL SQLAlchemy session managers.

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

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

Methods:

Name	Description
get_sync_manager	Get the synchronous PostgreSQL session manager instance.
set_sync_manager	Set a custom synchronous session manager.
get_async_manager	Get the asynchronous PostgreSQL session manager instance.
set_async_manager	Set a custom asynchronous session manager.
reset	Reset the registry to its initial state.

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

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

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

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

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

        Lazily initializes a default PostgresSQlAlchemySessionManager if none has been set.

        Returns:
            SessionManagerPort: The registered synchronous session manager

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

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

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

        Args:
            manager: An instance implementing SessionManagerPort

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

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

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

        Lazily initializes a default AsyncPostgresSQlAlchemySessionManager if none has been set.

        Returns:
            AsyncSessionManagerPort: The registered asynchronous session manager

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

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

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

        Args:
            manager: An instance implementing AsyncSessionManagerPort

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

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

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

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

get_sync_manager classmethod

get_sync_manager() -> SessionManagerPort

Get the synchronous PostgreSQL session manager instance.

Lazily initializes a default PostgresSQlAlchemySessionManager if none has been set.

Returns:

Name	Type	Description
SessionManagerPort	SessionManagerPort	The registered synchronous session manager

Raises:

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

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

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

    Lazily initializes a default PostgresSQlAlchemySessionManager if none has been set.

    Returns:
        SessionManagerPort: The registered synchronous session manager

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

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

set_sync_manager classmethod

set_sync_manager(manager: SessionManagerPort) -> None

Set a custom synchronous session manager.

Parameters:

Name	Type	Description	Default
manager	SessionManagerPort	An instance implementing SessionManagerPort	required

Raises:

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

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

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

    Args:
        manager: An instance implementing SessionManagerPort

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

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

get_async_manager classmethod

get_async_manager() -> AsyncSessionManagerPort

Get the asynchronous PostgreSQL session manager instance.

Lazily initializes a default AsyncPostgresSQlAlchemySessionManager if none has been set.

Returns:

Name	Type	Description
AsyncSessionManagerPort	AsyncSessionManagerPort	The registered asynchronous session manager

Raises:

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

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

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

    Lazily initializes a default AsyncPostgresSQlAlchemySessionManager if none has been set.

    Returns:
        AsyncSessionManagerPort: The registered asynchronous session manager

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

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

set_async_manager classmethod

set_async_manager(manager: AsyncSessionManagerPort) -> None

Set a custom asynchronous session manager.

Parameters:

Name	Type	Description	Default
manager	AsyncSessionManagerPort	An instance implementing AsyncSessionManagerPort	required

Raises:

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

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

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

    Args:
        manager: An instance implementing AsyncSessionManagerPort

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

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

reset classmethod

reset() -> None

Reset the registry to its initial state.

This method clears both registered managers, useful for testing.

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

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

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

options: show_root_toc_entry: false heading_level: 3

Adapters

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

Classes:

Name	Description
PostgresSQLAlchemyAdapter	Synchronous SQLAlchemy adapter for PostgreSQL.
AsyncPostgresSQLAlchemyAdapter	Asynchronous SQLAlchemy adapter for PostgreSQL.

PostgresSQLAlchemyAdapter

Synchronous SQLAlchemy adapter for PostgreSQL.

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

Parameters:

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

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

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

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

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

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

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

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

        Args:
            configs: PostgreSQL configuration.

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

AsyncPostgresSQLAlchemyAdapter

Asynchronous SQLAlchemy adapter for PostgreSQL.

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

Parameters:

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

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

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

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

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

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

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

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

        Args:
            configs: PostgreSQL configuration.

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

options: show_root_toc_entry: false heading_level: 3