core/database.py

from contextlib import contextmanager
from typing import Any, Iterator

try:
    import psycopg2
    from psycopg2.extras import RealDictCursor
except Exception:  # pragma: no cover - optional dependency safety
    psycopg2 = None
    RealDictCursor = None

from core.config import settings
from core.logger import logger


class Database:
    def is_configured(self) -> bool:
        return bool(settings.DATABASE_URL)

    def connect(self):
        if psycopg2 is None or RealDictCursor is None:
            raise RuntimeError("psycopg2 is not installed in the active Python environment")

        if not self.is_configured():
            raise RuntimeError("DATABASE_URL is not configured")

        return psycopg2.connect(
            settings.DATABASE_URL,
            cursor_factory=RealDictCursor,
        )

    @contextmanager
    def cursor(self) -> Iterator[Any]:
        connection = None
        cursor = None

        try:
            connection = self.connect()
            cursor = connection.cursor()
            yield cursor
            connection.commit()
        except Exception:
            if connection is not None:
                connection.rollback()
            raise
        finally:
            if cursor is not None:
                cursor.close()
            if connection is not None:
                connection.close()

    def fetch_all(self, query: str, params: tuple[Any, ...] | None = None) -> list[dict[str, Any]]:
        if not self.is_configured():
            return []

        try:
            with self.cursor() as cursor:
                cursor.execute(query, params or ())
                return list(cursor.fetchall())
        except Exception as exc:
            logger.warning(f"Database fetch_all failed: {exc}")
            return []

    def fetch_one(self, query: str, params: tuple[Any, ...] | None = None) -> dict[str, Any] | None:
        if not self.is_configured():
            return None

        try:
            with self.cursor() as cursor:
                cursor.execute(query, params or ())
                return cursor.fetchone()
        except Exception as exc:
            logger.warning(f"Database fetch_one failed: {exc}")
            return None

    def execute(self, query: str, params: tuple[Any, ...] | None = None) -> bool:
        if not self.is_configured():
            return False

        try:
            with self.cursor() as cursor:
                cursor.execute(query, params or ())
            return True
        except Exception as exc:
            logger.warning(f"Database execute failed: {exc}")
            return False


db = Database()