Postgres propio con SQLModel y alembic

name: mesa-core-practicas

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: practicas
      POSTGRES_PASSWORD: localdev   # solo local; en prod la contraseña sale de Secret Manager
      POSTGRES_DB: practicas
    ports:
      - "5445:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U practicas -d practicas"]
      interval: 5s
      timeout: 3s
      retries: 10

volumes:
  pgdata:

# Copia a env/local.env (ignorado por git) y rellena.
ENV=dev
DB_URL=postgresql+asyncpg://practicas:CAMBIAME@localhost:5445/practicas

ALEMBIC := $(VENV)/bin/alembic

db-up:
	docker compose up -d postgres

db-down:
	docker compose down

db-rev:                       # ej: make db-rev m="agrega columna x"
	$(ALEMBIC) revision --autogenerate -m "$(m)"

db-up-head:
	$(ALEMBIC) upgrade head

db-down-1:
	$(ALEMBIC) downgrade -1

    # postgresql+asyncpg://user:pass@host:5445/db — nunca literal en el código
    DB_URL: str = os.getenv("DB_URL", "")

"""Conexión y sesiones async a Postgres (nuestro esquema propio).

Calca shared/db.py de m-b-core: engine perezoso (se crea en el primer uso y
solo si hay DB_URL), sessionmaker async y un context manager `db_session()`.
"""

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from .settings import settings

_engine: AsyncEngine | None = None
_SessionLocal: async_sessionmaker[AsyncSession] | None = None


def get_engine() -> AsyncEngine:
    """Engine único del proceso. Falla claro si falta DB_URL."""
    global _engine, _SessionLocal
    if _engine is None:
        if not settings.DB_URL:
            raise ValueError("DB_URL no está configurada (env/local.env).")
        _engine = create_async_engine(
            settings.DB_URL,
            pool_pre_ping=True,
            echo=settings.ENV == "staging",  # ojo: echo=True vuelca SQL con datos
        )
        _SessionLocal = async_sessionmaker(_engine, expire_on_commit=False, class_=AsyncSession)
    return _engine


@asynccontextmanager
async def db_session() -> AsyncGenerator[AsyncSession, None]:
    """Una sesión por unidad de trabajo. Quien la usa decide la transacción."""
    get_engine()
    assert _SessionLocal is not None
    async with _SessionLocal() as session:
        yield session

"""Modelos SQLModel del esquema propio: la fuente de verdad de alembic."""

from datetime import UTC, datetime

from sqlalchemy import Column, DateTime
from sqlmodel import Field, SQLModel


def _utcnow() -> datetime:
    # aware, siempre; naive está prohibido (práctica 6)
    return datetime.now(UTC)


class LocalConfig(SQLModel, table=True):
    """Configuración propia por restaurante (lo que el legacy no tiene)."""

    __tablename__ = "local_config"

    # Local.Id del legacy: la clave viene de fuera, no la genera Postgres
    local_id: int = Field(primary_key=True, sa_column_kwargs={"autoincrement": False})
    permite_reserva_online: bool = Field(default=True)
    max_dias_anticipacion: int = Field(default=30, ge=0)
    updated_at: datetime = Field(
        default_factory=_utcnow, sa_column=Column(DateTime(timezone=True), nullable=False)
    )


class Aviso(SQLModel, table=True):
    """Registro de avisos enviados (idempotencia del worker, práctica 7)."""

    __tablename__ = "aviso"

    id: int | None = Field(default=None, primary_key=True)
    local_id: int = Field(index=True)
    tipo: str = Field(max_length=40)
    enviado_en: datetime = Field(
        default_factory=_utcnow, sa_column=Column(DateTime(timezone=True), nullable=False)
    )

script_location = alembic

# el nombre del archivo es el id de revisión, y el id lo elegimos nosotros
file_template = %%(rev)s

# para que `from shared…` funcione dentro de env.py
prepend_sys_path = .

# vacío a propósito: la URL la pone env.py desde el entorno
sqlalchemy.url =

"""Entorno de alembic (async). Calca alembic/env.py de m-b-core, con dos diferencias
deliberadas: la URL viene de shared.settings (no de os.getenv suelto) y la máscara
de la URL es la de SQLAlchemy (`render_as_string(hide_password=True)`), no un
`split('@')` a mano — ver la trampa 1 de la práctica.
"""

import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection, make_url
from sqlalchemy.ext.asyncio import async_engine_from_config
from sqlmodel import SQLModel

from alembic import context
from shared import models  # noqa: F401 — registra las tablas en SQLModel.metadata
from shared.settings import settings

config = context.config
fileConfig(config.config_file_name)

# La fuente de verdad del esquema son los modelos SQLModel.
target_metadata = SQLModel.metadata

if not settings.DB_URL:
    raise RuntimeError("DB_URL requerido para alembic (env/local.env)")

# Máscara correcta: SQLAlchemy sabe cuál es la contraseña.
print("[alembic] DB_URL:", make_url(settings.DB_URL).render_as_string(hide_password=True))
config.set_main_option("sqlalchemy.url", settings.DB_URL)


def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata, compare_type=True)
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()


asyncio.run(run_async_migrations())   # (la plantilla trae también el modo offline)

import sqlmodel  # tipos AutoString de SQLModel

    print(
        f"[alembic] DB_URL (masked): {DB_URL.split('@')[0]}@***:***@***"
        if "@" in DB_URL
        else "[alembic] DB_URL: <not set>"
    )

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('aviso', …)
    op.create_index(op.f('ix_aviso_local_id'), 'aviso', ['local_id'], unique=False)
    op.create_table('local_config',
    sa.Column('local_id', sa.Integer(), autoincrement=False, nullable=False),
    sa.Column('permite_reserva_online', sa.Boolean(), nullable=False),
    sa.Column('max_dias_anticipacion', sa.Integer(), nullable=False),
    sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
    sa.PrimaryKeyConstraint('local_id')
    )
    # ### end Alembic commands ###

    op.create_table('local_config',
    # Revisado a mano: local_id es el Id del legacy, no un serial de Postgres
    sa.Column('local_id', sa.Integer(), autoincrement=False, nullable=False),
    …
    sa.PrimaryKeyConstraint('local_id'),
    # Revisado a mano: `Field(ge=0)` valida en Python, no en la BD. La restricción
    # que importa vive donde viven los datos.
    sa.CheckConstraint('max_dias_anticipacion >= 0', name='ck_local_config_max_dias'),
    )

    # Nullable: los locales existentes no la tienen y no queremos inventar un valor.
    zona_horaria_override: str | None = Field(default=None, max_length=64)

revision: str = '0002_zona_horaria_override'
down_revision: Union[str, None] = '0001_local_config_aviso'


def upgrade() -> None:
    op.add_column('local_config', sa.Column('zona_horaria_override',
        sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True))


def downgrade() -> None:
    op.drop_column('local_config', 'zona_horaria_override')

"""Servicio de configuración por local: la unidad de negocio es UNA transacción.

`actualizar_config` cambia la config del local Y registra un Aviso de auditoría.
O entran los dos cambios, o ninguno: `async with session.begin()` abre la
transacción y hace commit al salir; si algo lanza dentro, hace rollback.
"""

from datetime import UTC, datetime
from sqlalchemy.ext.asyncio import AsyncSession
from shared.models import Aviso, LocalConfig
from .errors import DiasAnticipacionInvalidos

MAX_DIAS_ANTICIPACION = 365


async def obtener_config(session: AsyncSession, local_id: int) -> LocalConfig | None:
    return await session.get(LocalConfig, local_id)


async def actualizar_config(
    session: AsyncSession,
    local_id: int,
    *,
    permite_reserva_online: bool,
    max_dias_anticipacion: int,
) -> LocalConfig:
    """Crea o actualiza la config y deja un Aviso, atómicamente."""
    async with session.begin():  # una transacción por unidad de trabajo
        config = await session.get(LocalConfig, local_id)
        if config is None:
            config = LocalConfig(local_id=local_id)
            session.add(config)
        config.permite_reserva_online = permite_reserva_online
        config.max_dias_anticipacion = max_dias_anticipacion
        config.updated_at = datetime.now(UTC)
        await session.flush()  # el INSERT/UPDATE ya viajó a la BD (aún sin commit)

        # Validación de negocio DESPUÉS del flush a propósito: demuestra que si
        # falla aquí, el rollback también deshace lo de arriba.
        if not 0 <= max_dias_anticipacion <= MAX_DIAS_ANTICIPACION:
            raise DiasAnticipacionInvalidos(
                f"max_dias_anticipacion={max_dias_anticipacion} fuera de 0..{MAX_DIAS_ANTICIPACION}"
            )

        session.add(Aviso(local_id=local_id, tipo="config_actualizada"))
    return config

"""Tests del servicio: la transacción no deja la BD a medias."""

from services.api.app.locales import config_service as svc
from services.api.app.locales.errors import DiasAnticipacionInvalidos
from shared.models import Aviso, LocalConfig
# … + pytest, func/select

pytestmark = pytest.mark.unit


async def _cuenta(session, modelo) -> int:
    return (await session.execute(select(func.count()).select_from(modelo))).scalar_one()


async def test_actualizar_crea_config_y_aviso_juntos(sqlite_session):
    config = await svc.actualizar_config(
        sqlite_session, 11, permite_reserva_online=True, max_dias_anticipacion=30
    )
    assert config.local_id == 11
    assert await _cuenta(sqlite_session, LocalConfig) == 1
    assert await _cuenta(sqlite_session, Aviso) == 1


# … y test_actualizar_existente_no_duplica: dos llamadas seguidas dejan
# 1 fila en local_config y 2 en aviso (un aviso por cambio).


@pytest.mark.parametrize("dias", [-1, 366, 10_000])
async def test_dias_invalidos_no_deja_la_bd_a_medias(sqlite_session, dias):
    with pytest.raises(DiasAnticipacionInvalidos):
        await svc.actualizar_config(
            sqlite_session, 873, permite_reserva_online=True, max_dias_anticipacion=dias
        )
    # el flush ya había viajado a la BD; el rollback lo deshizo: ni config ni aviso
    assert await _cuenta(sqlite_session, LocalConfig) == 0
    assert await _cuenta(sqlite_session, Aviso) == 0

    config = await session.get(LocalConfig, local_id)
    …
    await session.commit()          # <- commit #1: la config ya quedó grabada

    if not 0 <= max_dias_anticipacion <= MAX_DIAS_ANTICIPACION:
        raise DiasAnticipacionInvalidos(...)

    session.add(Aviso(local_id=local_id, tipo="config_actualizada"))
    await session.commit()          # <- commit #2: nunca se llega si lo de arriba lanza
    return config

"""Fixtures de sesión: SQLite en memoria (unit) y Postgres del compose (integration)."""

from shared import models  # noqa: F401 — registra las tablas
# … + os, pytest, AsyncSession/async_sessionmaker/create_async_engine, SQLModel


@pytest.fixture()
async def sqlite_session() -> AsyncGenerator[AsyncSession, None]:
    """Una BD nueva por test: rápida, sin Docker. Para tests unit del servicio."""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    maker = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
    async with maker() as session:
        yield session
    await engine.dispose()


@pytest.fixture()
async def pg_session() -> AsyncGenerator[AsyncSession, None]:
    """Postgres real del compose (marca integration). Limpia las tablas al terminar."""
    url = os.getenv("DB_URL")
    if not url or not url.startswith("postgresql+asyncpg://"):
        pytest.skip("DB_URL de Postgres no configurada")
    engine = create_async_engine(url)

    async def _limpiar() -> None:
        async with engine.begin() as conn:
            for table in ("aviso", "local_config"):
                await conn.exec_driver_sql(f"DELETE FROM {table}")  # tablas del código, no input

    await _limpiar()  # un test no hereda filas de nadie (ni de tus inserts a mano)
    maker = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
    async with maker() as session:
        yield session
    await _limpiar()
    await engine.dispose()

    integration: Tests against real local services (Postgres del compose).

"""Lo mismo contra Postgres real: aquí además existe el CHECK de la migración."""

from sqlalchemy.exc import IntegrityError
# … + pytest, func/select, config_service as svc, Aviso/LocalConfig

pytestmark = pytest.mark.integration


async def test_actualizar_en_postgres(pg_session):
    await svc.actualizar_config(
        pg_session, 2246, permite_reserva_online=True, max_dias_anticipacion=45
    )
    total = (await pg_session.execute(select(func.count()).select_from(LocalConfig))).scalar_one()
    assert total == 1


async def test_check_constraint_lo_garantiza_la_bd(pg_session):
    """Aunque el código no validara, Postgres rechaza -5 por el CHECK de la migración 0001."""
    pg_session.add(LocalConfig(local_id=999, max_dias_anticipacion=-5))
    with pytest.raises(IntegrityError):
        await pg_session.commit()
    await pg_session.rollback()
    total = (await pg_session.execute(select(func.count()).select_from(Aviso))).scalar_one()
    assert total == 0