Inventario de reactivos en SQLite

[project]
name = "inventario"
version = "0.1.0"
description = "Inventario de reactivos de laboratorio (proyectos 3–10 del cuaderno)"
requires-python = ">=3.12"
dependencies = []

[project.scripts]
inv = "inventario.cli:app"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/inventario"]

[tool.pytest.ini_options]
testpaths = ["tests"]

[tool.ruff]
line-length = 100
extend-exclude = ["migrations"]  # código generado por alembic: no lo reescribimos a mano

"""Modelo de datos del inventario: qué existe y cómo se relaciona.

Ubicacion 1---* Lote *---1 Reactivo
                 |
                 *
             Movimiento
"""

from datetime import date, datetime

from sqlmodel import Field, SQLModel


class Ubicacion(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    nombre: str = Field(unique=True, index=True)  # "Congelador -20 A"


class Reactivo(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    nombre: str = Field(index=True)  # "Taq polimerasa"
    cas: str | None = None  # número CAS, si aplica
    unidad: str  # "µL", "mL", "g", "unidades"


class Lote(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    reactivo_id: int = Field(foreign_key="reactivo.id")
    ubicacion_id: int = Field(foreign_key="ubicacion.id")
    codigo: str = Field(unique=True, index=True)  # "TAQ-2401"
    cantidad: float  # en la unidad del reactivo
    vencimiento: date


class Movimiento(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    lote_id: int = Field(foreign_key="lote.id")
    cantidad: float  # negativo = salida, positivo = entrada
    fecha: datetime  # siempre en UTC
    nota: str | None = None

"""Errores propios del inventario. Nombran el problema en el idioma del laboratorio."""


class InventarioError(Exception):
    """Base de todos los errores del inventario."""


class LoteNoExiste(InventarioError):
    pass


class StockInsuficiente(InventarioError):
    pass

"""Conexión a la base de datos. La URL viene del entorno; por defecto, SQLite local."""

import os
from collections.abc import Iterator
from contextlib import contextmanager

from sqlmodel import Session, SQLModel, create_engine

DEFAULT_URL = "sqlite:///inventario.db"


def get_engine(url: str | None = None, echo: bool = False):
    url = url or os.environ.get("INVENTARIO_DB_URL", DEFAULT_URL)
    return create_engine(url, echo=echo)


_engine = None


def engine():
    """Un solo engine por proceso (se crea la primera vez que se pide)."""
    global _engine
    if _engine is None:
        _engine = get_engine()
    return _engine


@contextmanager
def get_session(eng=None) -> Iterator[Session]:
    """Sesión con transacción: commit si todo va bien, rollback si algo falla."""
    with Session(eng or engine()) as session:
        try:
            yield session
            session.commit()
        except Exception:
            session.rollback()
            raise


def init_db(eng) -> None:
    """Crea las tablas directamente (para tests). En producción se usan las migraciones."""
    SQLModel.metadata.create_all(eng)

"""Fixtures compartidas: una base de datos en memoria, nueva para cada test."""

from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine

from inventario import services


@pytest.fixture
def session():
    # sqlite:// sin ruta = en memoria. StaticPool: una sola conexión, para que
    # todas las sesiones del test vean la misma DB. Se crea y se destruye por test.
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as s:
        yield s
    SQLModel.metadata.drop_all(engine)


@pytest.fixture
def datos(session):
    """Los datos de ejemplo del cuaderno: 2 ubicaciones, 3 reactivos, 3 lotes."""
    congelador = services.crear_ubicacion(session, "Congelador -20 A")
    estante = services.crear_ubicacion(session, "Estante 3")
    taq = services.crear_reactivo(session, "Taq polimerasa", "µL")
    etoh = services.crear_reactivo(session, "Etanol 96%", "mL", cas="64-17-5")
    aga = services.crear_reactivo(session, "Agarosa", "g", cas="9012-36-6")
    hoy = datetime.now(UTC).date()
    services.agregar_lote(session, taq.id, congelador.id, "TAQ-2401", 500, hoy + timedelta(days=10))
    services.agregar_lote(
        session, etoh.id, estante.id, "ETOH-2312", 2500, hoy + timedelta(days=400)
    )
    services.agregar_lote(session, aga.id, estante.id, "AGA-2405", 250, hoy + timedelta(days=200))
    session.commit()
    return {"taq": taq, "etoh": etoh, "aga": aga, "congelador": congelador, "estante": estante}

import pytest
from sqlmodel import select

from inventario import services
from inventario.errors import LoteNoExiste, StockInsuficiente
from inventario.models import Lote, Movimiento


def test_crear_reactivo_asigna_id(session):
    r = services.crear_reactivo(session, "Taq polimerasa", "µL")
    assert r.id is not None
    assert r.unidad == "µL"


def test_agregar_lote_registra_movimiento_de_alta(session, datos):
    movimientos = list(session.exec(select(Movimiento)))
    assert len(movimientos) == 3
    assert all(m.cantidad > 0 for m in movimientos)


def test_registrar_salida_descuenta_y_anota(session, datos):
    mov = services.registrar_salida(session, "TAQ-2401", 50, nota="PCR placa 12")
    lote = session.exec(select(Lote).where(Lote.codigo == "TAQ-2401")).one()
    assert lote.cantidad == 450
    assert mov.cantidad == -50
    assert mov.nota == "PCR placa 12"


def test_salida_de_lote_inexistente_falla(session, datos):
    with pytest.raises(LoteNoExiste):
        services.registrar_salida(session, "NO-EXISTE", 1)


def test_salida_mayor_al_stock_falla_y_no_deja_la_db_a_medias(session, datos):
    with pytest.raises(StockInsuficiente):
        services.registrar_salida(session, "TAQ-2401", 9999)
    session.rollback()  # lo que haría get_session() al ver la excepción
    lote = session.exec(select(Lote).where(Lote.codigo == "TAQ-2401")).one()
    assert lote.cantidad == 500  # intacto
    salidas = [m for m in session.exec(select(Movimiento)) if m.cantidad < 0]
    assert salidas == []  # ningún movimiento de salida quedó grabado


def test_salida_con_cantidad_negativa_falla(session, datos):
    with pytest.raises(ValueError):
        services.registrar_salida(session, "TAQ-2401", -5)


def test_lotes_por_vencer_solo_los_proximos(session, datos):
    codigos = [lote.codigo for lote in services.lotes_por_vencer(session, dias=30)]
    assert codigos == ["TAQ-2401"]


def test_lotes_por_vencer_ignora_lotes_agotados(session, datos):
    services.registrar_salida(session, "TAQ-2401", 500)
    assert services.lotes_por_vencer(session, dias=30) == []


def test_buscar_reactivo_ignora_mayusculas(session, datos):
    nombres = [r.nombre for r in services.buscar_reactivo(session, "taq")]
    assert nombres == ["Taq polimerasa"]


def test_buscar_reactivo_con_comilla_no_rompe_nada(session, datos):
    # Si esto se armara con f-string, la comilla rompería el SQL (o peor).
    assert services.buscar_reactivo(session, "' OR 1=1 --") == []

"""Reglas del inventario. Aquí vive la lógica; el CLI (y luego la API) solo llaman a esto.

Todas las funciones reciben la sesión como primer argumento y NO hacen commit:
quien abre la sesión decide cuándo se confirma la transacción.
"""

from datetime import UTC, date, datetime, timedelta

from sqlmodel import Session, select

from inventario.errors import LoteNoExiste, StockInsuficiente
from inventario.models import Lote, Movimiento, Reactivo, Ubicacion


def crear_ubicacion(session: Session, nombre: str) -> Ubicacion:
    ubicacion = Ubicacion(nombre=nombre)
    session.add(ubicacion)
    session.flush()  # asigna el id sin cerrar la transacción
    return ubicacion


def crear_reactivo(session: Session, nombre: str, unidad: str, cas: str | None = None) -> Reactivo:
    reactivo = Reactivo(nombre=nombre, unidad=unidad, cas=cas)
    session.add(reactivo)
    session.flush()
    return reactivo


def agregar_lote(
    session: Session,
    reactivo_id: int,
    ubicacion_id: int,
    codigo: str,
    cantidad: float,
    vencimiento: date,
) -> Lote:
    if cantidad <= 0:
        raise ValueError("la cantidad inicial debe ser positiva")
    lote = Lote(
        reactivo_id=reactivo_id,
        ubicacion_id=ubicacion_id,
        codigo=codigo,
        cantidad=cantidad,
        vencimiento=vencimiento,
    )
    session.add(lote)
    session.flush()
    session.add(
        Movimiento(lote_id=lote.id, cantidad=cantidad, fecha=datetime.now(UTC), nota="alta")
    )
    return lote


def registrar_salida(
    session: Session, codigo_lote: str, cantidad: float, nota: str | None = None
) -> Movimiento:
    """Descuenta stock de un lote. Falla si el lote no existe o no alcanza.

    Las dos escrituras (restar al lote, anotar el movimiento) van en la misma
    transacción: o pasan las dos o no pasa ninguna.
    """
    if cantidad <= 0:
        raise ValueError("la cantidad a retirar debe ser positiva")
    lote = session.exec(select(Lote).where(Lote.codigo == codigo_lote)).first()
    if lote is None:
        raise LoteNoExiste(f"no existe el lote {codigo_lote!r}")
    if lote.cantidad < cantidad:
        raise StockInsuficiente(
            f"el lote {codigo_lote} tiene {lote.cantidad:g} y pediste {cantidad:g}"
        )
    lote.cantidad -= cantidad
    movimiento = Movimiento(lote_id=lote.id, cantidad=-cantidad, fecha=datetime.now(UTC), nota=nota)
    session.add(lote)
    session.add(movimiento)
    session.flush()
    return movimiento


def lotes_por_vencer(session: Session, dias: int = 30) -> list[Lote]:
    hoy = datetime.now(UTC).date()  # explícito en UTC: ver P6, "Fechas sin zona horaria"
    limite = hoy + timedelta(days=dias)
    consulta = (
        select(Lote)
        .where(Lote.vencimiento <= limite, Lote.cantidad > 0)
        .order_by(Lote.vencimiento)
    )
    return list(session.exec(consulta))


def buscar_reactivo(session: Session, texto: str) -> list[Reactivo]:
    consulta = select(Reactivo).where(Reactivo.nombre.ilike(f"%{texto}%"))  # parametrizado
    return list(session.exec(consulta))

"""Puerta de entrada por línea de comandos. Solo traduce argumentos → servicios → texto."""

from datetime import date

import typer

from inventario import services
from inventario.db import get_session
from inventario.errors import InventarioError

app = typer.Typer(help="Inventario de reactivos del laboratorio", no_args_is_help=True)


@app.command("add-ubicacion")
def add_ubicacion(nombre: str):
    with get_session() as s:
        u = services.crear_ubicacion(s, nombre)
        typer.echo(f"Ubicación #{u.id}: {u.nombre}")


@app.command("add-reactivo")
def add_reactivo(nombre: str, unidad: str, cas: str | None = None):
    with get_session() as s:
        r = services.crear_reactivo(s, nombre, unidad, cas)
        typer.echo(f"Reactivo #{r.id}: {r.nombre} ({r.unidad})")


@app.command("add-lote")
def add_lote(reactivo_id: int, ubicacion_id: int, codigo: str, cantidad: float, vencimiento: str):
    """VENCIMIENTO en formato AAAA-MM-DD."""
    with get_session() as s:
        lote = services.agregar_lote(
            s, reactivo_id, ubicacion_id, codigo, cantidad, date.fromisoformat(vencimiento)
        )
        typer.echo(f"Lote {lote.codigo}: {lote.cantidad:g}, vence {lote.vencimiento}")


@app.command("take")
def take(codigo: str, cantidad: float, nota: str | None = None):
    """Registra una salida de stock."""
    try:
        with get_session() as s:
            services.registrar_salida(s, codigo, cantidad, nota)
            typer.echo(f"Salida de {cantidad:g} del lote {codigo}")
    except InventarioError as e:
        typer.echo(f"Error: {e}", err=True)
        raise typer.Exit(code=1)


@app.command("expiring")
def expiring(days: int = typer.Option(30, "--days")):
    """Lotes que vencen en los próximos DAYS días."""
    with get_session() as s:
        lotes = services.lotes_por_vencer(s, days)
        if not lotes:
            typer.echo(f"Nada vence en {days} días.")
        for lote in lotes:
            typer.echo(f"{lote.vencimiento}  {lote.codigo:<10} {lote.cantidad:g}")


@app.command("find")
def find(texto: str):
    with get_session() as s:
        for r in services.buscar_reactivo(s, texto):
            typer.echo(f"#{r.id}  {r.nombre} ({r.unidad})")


if __name__ == "__main__":
    app()

from typer.testing import CliRunner

from inventario.cli import app

runner = CliRunner()


def test_cli_muestra_ayuda_sin_argumentos():
    resultado = runner.invoke(app, [])
    assert "add-lote" in resultado.output
    assert "expiring" in resultado.output

import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE reactivo (nombre TEXT, confidencial TEXT)")
con.execute("INSERT INTO reactivo VALUES ('Taq polimerasa', 'no'), ('Cepa secreta X', 'sí')")

texto = "' OR 1=1 --"   # lo que "escribe el usuario"

# MAL: el texto entra al SQL y cambia la consulta
mal = f"SELECT nombre FROM reactivo WHERE nombre = '{texto}'"
print("consulta armada:", mal)
print("con f-string devuelve:", con.execute(mal).fetchall())

# BIEN: el texto viaja como parámetro; nunca es SQL
bien = con.execute("SELECT nombre FROM reactivo WHERE nombre = ?", (texto,)).fetchall()
print("parametrizado devuelve:", bien)

import timeit


def codigo(i):
    return f"L-{i:06d}"


lotes = [codigo(i) for i in range(100_000)]
objetivo = codigo(99_999)          # el peor caso para la lista: está al final
indice = set(lotes)                 # el "índice": pagar una vez para buscar rápido

t_lista = timeit.timeit(lambda: objetivo in lotes, number=1_000)
t_set = timeit.timeit(lambda: objetivo in indice, number=1_000)
print(f"en lista  (O(n)): {t_lista * 1000:8.1f} ms por 1000 búsquedas")
print(f"en set    (O(1)): {t_set * 1000:8.1f} ms por 1000 búsquedas")
print(f"factor: {t_lista / t_set:,.0f}×")

FROM python:3.12-slim
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
WORKDIR /app
COPY pyproject.toml uv.lock alembic.ini ./
COPY migrations ./migrations
COPY src ./src
COPY tests ./tests
RUN uv sync --frozen
# Al arrancar: aplica migraciones y muestra la ayuda del CLI
CMD ["sh", "-c", "uv run alembic upgrade head && uv run inv --help"]

services:
  inv:
    build: .
    environment:
      INVENTARIO_DB_URL: sqlite:////data/inventario.db
    volumes:
      - datos:/data      # la DB vive en un volumen: sobrevive al contenedor

volumes:
  datos:

class Ubicacion(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    nombre: str = Field(unique=True, index=True)  # "Congelador -20 A"
    descripcion: str | None = None  # agregada en la 2ª migración