"""Database connection management.""" from __future__ import annotations import logging import os import threading from typing import Any, Optional from hermes.config.settings import get_settings logger = logging.getLogger(__name__) # ── Singleton ───────────────────────────────────────────────────────── _db_manager: Optional["DatabaseManager"] = None _db_lock: threading.Lock = threading.Lock() def get_database_manager() -> Optional["DatabaseManager"]: """Get or create the global DatabaseManager singleton (thread-safe). Returns None if DATABASE_URL / POSTGRES_URL is not configured. """ global _db_manager if _db_manager is not None: return _db_manager with _db_lock: if _db_manager is not None: return _db_manager # Check if DB is configured settings = get_settings() db_url = settings.database.postgres_url if not db_url: # Also check raw env db_url = os.environ.get("DATABASE_URL", "") or os.environ.get("POSTGRES_URL", "") if not db_url: logger.debug("No DATABASE_URL configured, DB features disabled") return None _db_manager = DatabaseManager() return _db_manager def reset_database_manager() -> None: """Reset the singleton (for testing).""" global _db_manager with _db_lock: _db_manager = None class DatabaseManager: """Database connection manager.""" def __init__(self) -> None: self.settings = get_settings() self._engine: Any = None self._session_factory: Any = None @property def async_engine(self) -> Any: """Return the async SQLAlchemy engine (may be None if not initialized).""" return self._engine async def initialize(self) -> None: """Initialize database connections. Reads pool configuration from DatabaseConfig: DB_POSTGRES_URL, DB_POOL_SIZE, DB_MAX_OVERFLOW, DB_POOL_TIMEOUT, DB_POOL_RECYCLE, DB_POOL_PRE_PING, DB_ECHO_SQL. """ try: from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine db_cfg = self.settings.database self._engine = create_async_engine( db_cfg.postgres_url, echo=db_cfg.echo_sql, pool_size=db_cfg.pool_size, max_overflow=db_cfg.max_overflow, pool_timeout=db_cfg.pool_timeout, pool_recycle=db_cfg.pool_recycle, pool_pre_ping=db_cfg.pool_pre_ping, ) self._session_factory = async_sessionmaker( self._engine, class_=AsyncSession, expire_on_commit=False ) logger.info( "Database initialized (pool_size=%d, max_overflow=%d, pool_recycle=%ds)", db_cfg.pool_size, db_cfg.max_overflow, db_cfg.pool_recycle, ) except Exception as e: logger.warning("Could not initialize database: %s", e) async def get_session(self) -> Any: """Get a database session.""" if self._session_factory: return self._session_factory() return None async def close(self) -> None: """Close database connections.""" if self._engine: await self._engine.dispose() logger.info("Database connections closed") async def create_tables(self) -> None: """Create database tables.""" if self._engine: from hermes.database.models import Base async with self._engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("Database tables created")