template-web-backend-python/code/app/db.py

"""
Capa de conexión SQLAlchemy.
- engine único por proceso, configurado al startup
- sesiones por request (dependency injection FastAPI)
- ping ligero para /health
"""

from collections.abc import Iterator

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, sessionmaker

from app.settings import settings

engine = create_engine(
    settings.DATABASE_URL,
    pool_pre_ping=True,
    pool_recycle=3600,
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def get_session() -> Iterator[Session]:
    """Dependency FastAPI: una sesión por request, cierre garantizado."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def db_alive() -> bool:
    """Ping de conexión usado por /health. Retorna False ante cualquier excepción."""
    try:
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        return True
    except Exception:
        return False