4. Arquitectura de la API de backend

Estado: v1. Estructura de backend completa para apps/skillnet-api/. Alineado con data-model.md, screens.md y architecture.md.

4.1 Estructura del proyecto

apps/skillnet-api/
├── pyproject.toml                  # proyecto uv, dependencias
├── alembic.ini                     # configuración de migraciones de BD
├── alembic/
│   ├── env.py
│   └── versions/                   # Ficheros de migración
│
├── src/
│   ├── __init__.py
│   ├── main.py                     # factoría de la app FastAPI, lifespan, middleware
│   ├── config.py                   # Pydantic Settings (variables de entorno)
│   │
│   ├── auth/                       # Autenticación (fastapi-users)
│   │   ├── __init__.py
│   │   ├── backend.py              # CookieTransport + estrategia de sesión
│   │   ├── manager.py              # UserManager (crear, verificar, etc.)
│   │   ├── schemas.py              # UserRead, UserCreate, UserUpdate
│   │   └── router.py               # /auth/login, /auth/logout, /auth/me
│   │
│   ├── models/                     # Modelos ORM de SQLAlchemy (1 fichero por tabla)
│   │   ├── __init__.py             # Reexporta todos los modelos
│   │   ├── base.py                 # DeclarativeBase, mixins comunes (TimestampMixin, UUIDMixin)
│   │   ├── organization.py
│   │   ├── user.py                 # Extiende el modelo SQLAlchemy de fastapi-users
│   │   ├── document.py
│   │   ├── document_chunk.py
│   │   ├── skill.py
│   │   ├── skill_category.py
│   │   ├── course.py
│   │   ├── module.py
│   │   ├── lesson.py
│   │   ├── exercise.py
│   │   ├── skill_checkpoint.py
│   │   ├── manual.py
│   │   ├── enrollment.py
│   │   ├── exercise_attempt.py
│   │   ├── user_skill.py
│   │   ├── spaced_repetition.py
│   │   ├── generation_job.py
│   │   ├── course_feedback.py
│   │   ├── chat_session.py
│   │   ├── chat_message.py
│   │   ├── background_job.py
│   │   ├── user_session.py
│   │   ├── audit_log.py
│   │   ├── api_key.py
│   │   ├── webhook.py
│   │   └── webhook_delivery.py
│   │
│   ├── repositories/               # Capa de acceso a datos (consultas async de SQLAlchemy)
│   │   ├── __init__.py
│   │   ├── base.py                 # BaseRepository[T] — CRUD genérico
│   │   ├── user_repo.py
│   │   ├── document_repo.py
│   │   ├── course_repo.py          # Incluye la carga anidada de módulo/lección/ejercicio
│   │   ├── enrollment_repo.py
│   │   ├── exercise_attempt_repo.py
│   │   ├── skill_repo.py           # Skills + categorías + user_skills + consulta de la matriz
│   │   ├── manual_repo.py
│   │   ├── spaced_repetition_repo.py
│   │   ├── generation_job_repo.py
│   │   ├── course_feedback_repo.py
│   │   ├── document_chunk_repo.py  # Búsqueda de similitud vectorial
│   │   ├── chat_session_repo.py
│   │   ├── chat_message_repo.py
│   │   ├── background_job_repo.py
│   │   ├── user_session_repo.py
│   │   ├── audit_log_repo.py
│   │   ├── api_key_repo.py
│   │   ├── webhook_repo.py
│   │   └── webhook_delivery_repo.py
│   │
│   ├── services/                   # Lógica de negocio (sin conocimiento de BD ni de HTTP)
│   │   ├── __init__.py
│   │   ├── user_service.py         # Invitar, invitación masiva, desactivar
│   │   ├── document_service.py     # Subida a disco, dispara el procesamiento
│   │   ├── course_service.py       # CRUD, ciclo de vida publicar/archivar
│   │   ├── enrollment_service.py   # Asignar, cálculo de progreso, completar
│   │   ├── exercise_service.py     # Calificar intento, comprobación automática, calificado por IA
│   │   ├── skill_service.py        # Matriz, sugerencias de mentoría, verify_skill
│   │   ├── manual_service.py       # CRUD, vinculado al curso
│   │   ├── spaced_repetition_service.py  # Algoritmo HLR, repasos pendientes, enviar repaso
│   │   ├── generation_service.py   # Orquesta el pipeline de generación
│   │   ├── chat_service.py         # Chat de tutor + admin (RAG + streaming)
│   │   ├── alert_service.py        # Calcula alertas a partir de los datos de progreso
│   │   ├── stats_service.py        # Agregaciones del dashboard
│   │   └── feedback_service.py     # Envío + generación del informe de revisión
│   │
│   ├── routes/                     # Routers de FastAPI (finos — validan, llaman al service, devuelven)
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── users.py
│   │   ├── documents.py
│   │   ├── courses.py              # Incluye las acciones /generate, /publish, /archive
│   │   ├── modules.py              # Anidado bajo courses
│   │   ├── lessons.py              # Anidado bajo modules
│   │   ├── exercises.py            # Incluye la acción /attempt
│   │   ├── enrollments.py
│   │   ├── skills.py               # Categorías, skills, matriz, mentoría
│   │   ├── manuals.py
│   │   ├── chat.py                 # Endpoints SSE (tutor + admin)
│   │   ├── spaced_repetition.py
│   │   ├── generation_jobs.py
│   │   ├── stats.py                # /stats, /alerts
│   │   ├── settings.py             # Ajustes de la organización, config de LLM
│   │   └── feedback.py
│   │
│   ├── schemas/                    # Modelos Pydantic para petición/respuesta
│   │   ├── __init__.py
│   │   ├── common.py               # Paginación, ErrorResponse, SuccessMessage
│   │   ├── user.py
│   │   ├── document.py
│   │   ├── course.py               # Incluye los esquemas anidados de módulo/lección/ejercicio
│   │   ├── enrollment.py
│   │   ├── exercise.py             # AttemptRequest, AttemptResponse, por tipo de ejercicio
│   │   ├── skill.py                # SkillMatrix, MentorshipSuggestion
│   │   ├── manual.py
│   │   ├── chat.py                 # ChatMessage, ChatEvent (SSE)
│   │   ├── spaced_repetition.py
│   │   ├── generation_job.py
│   │   ├── stats.py                # DashboardStats, Alert
│   │   ├── settings.py
│   │   └── feedback.py
│   │
│   ├── deps/                       # Inyección de dependencias de FastAPI
│   │   ├── __init__.py
│   │   ├── db.py                   # get_async_session
│   │   ├── auth.py                 # current_user, current_active_user, require_admin
│   │   ├── llm.py                  # get_llm_client
│   │   └── embedding.py            # get_embedding_service
│   │
│   ├── llm/                        # Integración de LLM (agnóstica de proveedor)
│   │   ├── __init__.py
│   │   ├── client.py               # wrapper de AsyncOpenAI, lee variables de entorno
│   │   ├── prompts/                # Plantillas de prompt como .py o .txt
│   │   │   ├── tutor_system.py
│   │   │   ├── admin_system.py
│   │   │   ├── grading.py          # Prompts de evaluación de ejercicios
│   │   │   └── generation.py       # Prompts de generación de curso/manual
│   │   └── embedding.py            # Servicio de embeddings (API o modelo local)
│   │
│   ├── agents/                     # Definiciones de agentes LangGraph (diferido — Fase 2+)
│   │   ├── __init__.py
│   │   ├── content_agent.py        # Pipeline de generación multipaso
│   │   └── tutor_agent.py          # Tutor conversacional RAG
│   │
│   └── core/                       # Utilidades compartidas
│       ├── __init__.py
│       ├── exceptions.py           # Clases de excepción propias de la app
│       ├── security.py             # Hash de contraseñas, config de cookies
│       └── pagination.py           # Ayudante de paginación por offset
│
├── tests/
│   ├── conftest.py                 # Fixtures: cliente async, BD de test, usuario de test
│   ├── factories.py                # Factorías de Factory Boy para datos de test
│   ├── test_auth.py
│   ├── test_users.py
│   ├── test_courses.py
│   ├── test_enrollments.py
│   ├── test_exercises.py
│   ├── test_skills.py
│   ├── test_chat.py
│   └── test_spaced_repetition.py
│
├── uploads/                        # Documentos subidos (montaje de volumen Docker)
└── Dockerfile

Responsabilidades por capa:

Flujo de datos: petición HTTP -> route -> service -> repository -> base de datos y de vuelta. Cada capa solo habla con su vecino inmediato.

4.2 Endpoints de la API

Todos los endpoints con el prefijo /api/v1. Autenticación vía cookie de sesión en cada petición (excepto el login).

Auth

La respuesta del login redirige según el rol: empleado -> /dashboard, admin -> /admin. El frontend lee el rol desde /auth/me al cargar la página.

Modo de espacio de trabajo. /auth/me (y /settings) también llevan workspace_mode (organization | individual; ver audience-modes.md). En un despliegue individual los endpoints colectivos, exclusivos de organización — empleados (listar/crear/reiniciar), talento, /stats, asignación de cursos (POST/DELETE /enrollments, asignación por carpeta) y el catálogo de skills — devuelven 404 vía la dependencia require_organization_workspace: esos conceptos no existen en un espacio de trabajo personal. Es un cumplimiento del lado del servidor; la SPA además oculta las secciones como UX.

Usuarios

Perfil del aprendiz (personalización)

Router src/routes/learner_profile.py, prefijo /users/me/learner-profile. Ver personalization.md.

Artefactos multimedia

Router src/routes/media.py, prefijo /media. Ver media-artifacts.md.

Esquema de curso (flujo de creación)

Router src/routes/course_schema.py. Ciclo de vida del esquema de curso para el admin: proponer -> PUT (genera packs) -> revisar -> validar -> precalentar. Ver create-course-flow y learning-experience-architecture.md.

Documentos

Cursos

Módulos

Lecciones

Ejercicios

Cuerpo de la petición de intento según el tipo de ejercicio:

test:           { "selected": 1 }
true_false:     { "answer": true }
fill_blank:     { "answers": ["unused", "tags"] }
order_steps:    { "order": [0, 2, 1, 3] }
practical_case: { "response": "I would explain the 30-day policy..." }
dialogue:       { "messages": [{"role": "user", "content": "..."}] }

Para test, true_false, fill_blank, order_steps: el servidor califica de forma determinista (compara contra la respuesta correcta en el contenido del ejercicio).

Para practical_case: el servidor envía la respuesta + la rúbrica al LLM para su evaluación. Devuelve una puntuación + feedback por criterio.

Para dialogue: el servidor ejecuta una conversación multiturno vía LLM con el system_prompt del contenido del ejercicio. Evaluación al alcanzar max_turns.

Inscripciones

Cálculo del progreso: módulos_completados / módulos_totales. Un módulo está completo cuando todos los ejercicios de sus lecciones tienen al menos un intento aprobado.

Skills

Checkpoints de skill

Cuando se completa un módulo, el sistema comprueba los checkpoints y actualiza user_skills en consecuencia. El nivel nunca baja por los checkpoints (solo una anulación manual del admin puede bajarlo).

Chat (SSE)

Protocolo SSE:

1. El cliente envía POST /chat con el cuerpo del mensaje
2. El servidor devuelve Content-Type: text/event-stream
3. El servidor emite eventos:

event: token
data: {"content": "The"}

event: token
data: {"content": " return"}

event: citations
data: {"citations": [{"document": "Manual_Devoluciones.pdf", "section": "Plazos", "page": 3}]}

event: suggestions
data: {"prompts": ["What is the return window?", "How do I process a refund?"]}

event: done
data: {"message_id": "uuid"}

event: error
data: {"message": "Model unavailable"}

El chat del tutor usa RAG: la consulta se convierte a embedding, se recuperan los top-k chunks de document_chunks, y se incluyen en el contexto del LLM junto con el contexto del curso/lección actual del empleado.

El chat de admin tiene acceso a los datos de toda la organización (empleados, skills, inscripciones) para consultas operativas.

Manuales

Repetición espaciada

El algoritmo HLR se ejecuta al enviar:

• Correcto: half_life *= 2, review_count += 1
• Incorrecto: half_life /= 2, review_count += 1
• next_review_at = ahora + tiempo hasta que P(olvido) > 0.3

Trabajos de generación

Pasos del pipeline de generación: pending -> extracting -> structuring -> generating -> reviewing -> published o failed.

Cada paso actualiza el estado del trabajo. El frontend hace polling a GET /generation-jobs/{id} para mostrar el progreso.

Feedback de curso

Estadísticas y alertas

Cada alerta incluye:

{
  "type": "deadline_risk",
  "severity": "high",
  "message": "Carlos has 0% progress on 'Returns' — deadline in 3 days",
  "action_url": "/admin/users/{user_id}",
  "related_ids": {"user_id": "...", "enrollment_id": "..."}
}

Las alertas se calculan por petición (no se almacenan). El servicio consulta las inscripciones, los intentos y la repetición espaciada para detectar las condiciones.

Ajustes de la organización

Usuarios — Invitaciones

Health

Admin — Trabajos en segundo plano

Los ajustes de LLM se almacenan en el jsonb organizations.settings. Las variables de entorno (LLM_BASE_URL, LLM_API_KEY, LLM_MODEL) son los valores por defecto; los ajustes a nivel de organización las sobrescriben.

4.3 Inyección de dependencias

El sistema Depends() de FastAPI proporciona recursos compartidos a los manejadores de ruta. Todas las dependencias se definen en src/deps/.

Sesión de base de datos

# src/deps/db.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from src.config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=False)
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)

async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        yield session

# Alias de tipo para las firmas de las rutas
DBSession = Annotated[AsyncSession, Depends(get_async_session)]

Usuario actual (a partir de la cookie de sesión)

fastapi-users proporciona el backend de cookie de sesión. La cadena de dependencias extrae el usuario de la cookie automáticamente.

# src/deps/auth.py
from fastapi_users import FastAPIUsers
from src.auth.backend import auth_backend
from src.auth.manager import get_user_manager
from src.models.user import User

fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])

# Dependencias base (proporcionadas por fastapi-users)
current_user = fastapi_users.current_user(active=True)
current_optional_user = fastapi_users.current_user(active=True, optional=True)

# Alias de tipo
CurrentUser = Annotated[User, Depends(current_user)]

Acceso basado en roles

# src/deps/auth.py (continuación)
from fastapi import HTTPException, status

def require_admin(user: CurrentUser) -> User:
    if user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required",
        )
    return user

def require_employee(user: CurrentUser) -> User:
    if user.role != "employee":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Employee access required",
        )
    return user

# Alias de tipo para las firmas de las rutas
AdminUser = Annotated[User, Depends(require_admin)]
EmployeeUser = Annotated[User, Depends(require_employee)]

Uso en las rutas:

# src/routes/courses.py
@router.get("/courses")
async def list_courses(user: AdminUser, db: DBSession):
    ...

@router.post("/exercises/{id}/attempt")
async def attempt_exercise(user: EmployeeUser, db: DBSession, ...):
    ...

@router.get("/courses/{id}")
async def get_course(user: CurrentUser, db: DBSession, ...):
    # Ambos roles pueden acceder, pero el empleado solo lo ve si está inscrito
    ...

Cliente LLM

# src/deps/llm.py
from openai import AsyncOpenAI
from src.config import settings

async def get_llm_client(db: DBSession) -> AsyncOpenAI:
    """Devuelve un cliente AsyncOpenAI configurado a partir de los ajustes de la org o de las variables de entorno."""
    # Primero comprueba si hay una anulación a nivel de organización
    org = await db.execute(select(Organization).limit(1))
    org_settings = org.scalar_one().settings

    base_url = org_settings.get("llm_base_url") or settings.LLM_BASE_URL
    api_key = org_settings.get("llm_api_key") or settings.LLM_API_KEY

    return AsyncOpenAI(base_url=base_url, api_key=api_key)

LLMClient = Annotated[AsyncOpenAI, Depends(get_llm_client)]

El cliente AsyncOpenAI funciona con cualquier API compatible con OpenAI. El nombre del modelo se resuelve de forma similar (ajuste de la org > variable de entorno) y se pasa por llamada, no al crear el cliente.

Servicio de embeddings

# src/deps/embedding.py
from src.llm.embedding import EmbeddingService

async def get_embedding_service(db: DBSession) -> EmbeddingService:
    """Devuelve el servicio de embeddings (basado en API o modelo local)."""
    org = await db.execute(select(Organization).limit(1))
    org_settings = org.scalar_one().settings

    return EmbeddingService(
        base_url=org_settings.get("embedding_base_url") or settings.EMBEDDING_BASE_URL,
        api_key=org_settings.get("embedding_api_key") or settings.EMBEDDING_API_KEY,
        model=org_settings.get("embedding_model") or settings.EMBEDDING_MODEL,
    )

EmbeddingSvc = Annotated[EmbeddingService, Depends(get_embedding_service)]

Ejemplo de composición de dependencias

Un manejador de ruta compone lo que necesita:

@router.post("/chat")
async def tutor_chat(
    user: EmployeeUser,
    db: DBSession,
    llm: LLMClient,
    embeddings: EmbeddingSvc,
    body: ChatMessageRequest,
):
    service = ChatService(db, llm, embeddings)
    return StreamingResponse(
        service.tutor_stream(user, body.message, body.context),
        media_type="text/event-stream",
    )

Los services reciben sus dependencias por inyección de constructor (desde el manejador de ruta), no a través de estado global.

4.4 Manejo de errores

Formato de respuesta de error

Cada error devuelve la misma forma JSON:

{
  "detail": "Course not found",
  "code": "NOT_FOUND",
  "field": null
}

Para errores de validación (422):

{
  "detail": "Validation failed",
  "code": "VALIDATION_ERROR",
  "errors": [
    {"field": "email", "message": "Invalid email format"},
    {"field": "full_name", "message": "Required field"}
  ]
}

Excepciones de la aplicación

# src/core/exceptions.py
class AppError(Exception):
    """Error base de la aplicación."""
    def __init__(self, message: str, code: str, status_code: int = 400):
        self.message = message
        self.code = code
        self.status_code = status_code

class NotFoundError(AppError):
    def __init__(self, resource: str, id: str):
        super().__init__(f"{resource} not found", "NOT_FOUND", 404)

class ForbiddenError(AppError):
    def __init__(self, message: str = "Access denied"):
        super().__init__(message, "FORBIDDEN", 403)

class ConflictError(AppError):
    def __init__(self, message: str):
        super().__init__(message, "CONFLICT", 409)

class LLMError(AppError):
    def __init__(self, message: str = "LLM service unavailable"):
        super().__init__(message, "LLM_ERROR", 502)

Manejador global de excepciones

# src/main.py
@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message, "code": exc.code, "field": None},
    )

@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        field = ".".join(str(loc) for loc in error["loc"] if loc != "body")
        errors.append({"field": field, "message": error["msg"]})
    return JSONResponse(
        status_code=422,
        content={"detail": "Validation failed", "code": "VALIDATION_ERROR", "errors": errors},
    )

Uso de códigos de estado HTTP

Enfoque de validación

1. Los esquemas Pydantic validan la forma y los tipos de la petición en la capa de ruta
2. La capa de service valida las reglas de negocio (p. ej., “el curso debe tener al menos 1 módulo para publicarse”)
3. Las restricciones de la base de datos son la última línea de defensa (unique, FK, check constraints)

Los services lanzan subclases de AppError. Las routes nunca capturan excepciones — lo hace el manejador global.

4.5 Capa de acceso a la base de datos

Configuración del motor y la sesión

# src/deps/db.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine(
    settings.DATABASE_URL,      # "postgresql+asyncpg://..."
    echo=settings.DEBUG,
    pool_size=10,
    max_overflow=20,
)

async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,     # Evita problemas de lazy-load tras el commit
)

Modelo base

# src/models/base.py
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import text
from datetime import datetime
import uuid

class Base(DeclarativeBase):
    pass

class UUIDMixin:
    id: Mapped[uuid.UUID] = mapped_column(
        primary_key=True,
        server_default=text("gen_random_uuid()"),
    )

class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(
        server_default=text("now()"),
    )
    updated_at: Mapped[datetime] = mapped_column(
        server_default=text("now()"),
        onupdate=text("now()"),
    )

Ejemplo de modelo

# src/models/course.py
from sqlalchemy import ForeignKey, Enum as SAEnum
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.models.base import Base, UUIDMixin, TimestampMixin
import uuid
import enum

class ContentStatus(str, enum.Enum):
    DRAFT = "draft"
    PUBLISHED = "published"
    ARCHIVED = "archived"

class Course(UUIDMixin, TimestampMixin, Base):
    __tablename__ = "courses"

    org_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("organizations.id"), nullable=False)
    created_by: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"), nullable=False)
    source_document_id: Mapped[uuid.UUID | None] = mapped_column(ForeignKey("documents.id"))
    title: Mapped[str]
    description: Mapped[str | None]
    outcome: Mapped[str | None]
    status: Mapped[ContentStatus] = mapped_column(
        SAEnum(ContentStatus, name="content_status"),
        default=ContentStatus.DRAFT,
    )

    # Relaciones
    modules: Mapped[list["Module"]] = relationship(
        back_populates="course",
        cascade="all, delete-orphan",
        order_by="Module.position",
    )
    enrollments: Mapped[list["Enrollment"]] = relationship(back_populates="course")
    manual: Mapped["Manual | None"] = relationship(back_populates="course", uselist=False)
    checkpoints: Mapped[list["SkillCheckpoint"]] = relationship(
        back_populates="course",
        cascade="all, delete-orphan",
    )

Patrón repositorio

# src/repositories/base.py
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from src.models.base import Base
from typing import TypeVar, Generic, Sequence

T = TypeVar("T", bound=Base)

class BaseRepository(Generic[T]):
    def __init__(self, session: AsyncSession, model: type[T]):
        self.session = session
        self.model = model

    async def get_by_id(self, id: uuid.UUID) -> T | None:
        return await self.session.get(self.model, id)

    async def get_or_404(self, id: uuid.UUID) -> T:
        obj = await self.get_by_id(id)
        if not obj:
            raise NotFoundError(self.model.__tablename__, str(id))
        return obj

    async def list(
        self,
        *,
        filters: list | None = None,
        order_by=None,
        offset: int = 0,
        limit: int = 50,
    ) -> tuple[Sequence[T], int]:
        query = select(self.model)
        count_query = select(func.count()).select_from(self.model)

        if filters:
            for f in filters:
                query = query.where(f)
                count_query = count_query.where(f)

        if order_by is not None:
            query = query.order_by(order_by)

        total = (await self.session.execute(count_query)).scalar_one()
        result = await self.session.execute(query.offset(offset).limit(limit))
        return result.scalars().all(), total

    async def create(self, **kwargs) -> T:
        obj = self.model(**kwargs)
        self.session.add(obj)
        await self.session.flush()
        return obj

    async def update(self, obj: T, **kwargs) -> T:
        for key, value in kwargs.items():
            setattr(obj, key, value)
        await self.session.flush()
        return obj

    async def delete(self, obj: T) -> None:
        await self.session.delete(obj)
        await self.session.flush()

Ejemplo de repositorio especializado

# src/repositories/skill_repo.py
class SkillRepository(BaseRepository[Skill]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, Skill)

    async def get_matrix(self, org_id: uuid.UUID, category_id: uuid.UUID | None = None):
        """Devuelve la matriz completa de skills: [{user, skill, level}, ...]"""
        query = (
            select(
                User.id.label("user_id"),
                User.full_name,
                Skill.id.label("skill_id"),
                Skill.name.label("skill_name"),
                UserSkill.level,
            )
            .select_from(User)
            .outerjoin(UserSkill, User.id == UserSkill.user_id)
            .outerjoin(Skill, UserSkill.skill_id == Skill.id)
            .where(User.org_id == org_id, User.is_active == True)
        )
        if category_id:
            query = query.where(Skill.category_id == category_id)

        result = await self.session.execute(query.order_by(User.full_name, Skill.name))
        return result.all()

    async def get_mentorship_suggestions(self, org_id: uuid.UUID):
        """Encuentra pares de skill alto-bajo para mentoría."""
        query = (
            select(
                User.full_name.label("mentor_name"),
                User.id.label("mentor_id"),
                func.array_agg(
                    func.json_build_object(
                        "mentee_name", mentee.full_name,
                        "mentee_id", mentee.id,
                        "skill", Skill.name,
                    )
                ).label("matches"),
            )
            .select_from(UserSkill)
            .join(User, User.id == UserSkill.user_id)
            .join(
                us_low := aliased(UserSkill),
                and_(
                    us_low.skill_id == UserSkill.skill_id,
                    us_low.level == "low",
                    us_low.user_id != UserSkill.user_id,
                ),
            )
            .join(mentee := aliased(User), mentee.id == us_low.user_id)
            .join(Skill, Skill.id == UserSkill.skill_id)
            .where(UserSkill.level == "high", User.org_id == org_id)
            .group_by(User.id, User.full_name)
        )
        result = await self.session.execute(query)
        return result.all()

Búsqueda vectorial (chunks de documento)

# src/repositories/document_chunk_repo.py
from pgvector.sqlalchemy import Vector

class DocumentChunkRepository(BaseRepository[DocumentChunk]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, DocumentChunk)

    async def similarity_search(
        self,
        org_id: uuid.UUID,
        query_embedding: list[float],
        top_k: int = 5,
    ) -> list[dict]:
        """Búsqueda semántica en todos los documentos de la org."""
        query = (
            select(
                DocumentChunk.content,
                DocumentChunk.metadata,
                (1 - DocumentChunk.embedding.cosine_distance(query_embedding)).label("similarity"),
            )
            .join(Document, Document.id == DocumentChunk.document_id)
            .where(Document.org_id == org_id)
            .order_by(DocumentChunk.embedding.cosine_distance(query_embedding))
            .limit(top_k)
        )
        result = await self.session.execute(query)
        return [
            {"content": row.content, "metadata": row.metadata, "similarity": row.similarity}
            for row in result.all()
        ]

Gestión de la sesión en las rutas

Las rutas obtienen una sesión de Depends, instancian repositories y services, y la sesión hace auto-commit o rollback:

# src/routes/courses.py
@router.post("/courses", status_code=201)
async def create_course(user: AdminUser, db: DBSession, body: CourseCreate):
    repo = CourseRepository(db)
    service = CourseService(repo)
    course = await service.create(
        org_id=user.org_id,
        created_by=user.id,
        **body.model_dump(),
    )
    await db.commit()
    return CourseRead.model_validate(course)

La llamada a commit() está en el manejador de ruta, no en el repository ni en el service. Esto mantiene la frontera transaccional visible y explícita. Si algo lanza una excepción antes del commit, la sesión hace rollback automáticamente al salir del context manager.

Resumen del mapeo modelo-tabla

Migraciones

Alembic con soporte async. Una migración por cambio de esquema.

# Generar migración a partir de cambios en los modelos
uv run alembic revision --autogenerate -m "add_courses_table"

# Aplicar migraciones
uv run alembic upgrade head

# Deshacer
uv run alembic downgrade -1

La migración inicial crea las 27 tablas + enums + índices + extensiones (pgcrypto, vector).

4.6 Configuración

# src/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # Base de datos
    DATABASE_URL: str = "postgresql+asyncpg://skillnet:skillnet@localhost:5432/skillnet"

    # Auth
    SECRET_KEY: str                       # Obligatorio, sin valor por defecto
    SESSION_LIFETIME_SECONDS: int = 604800    # 7 días
    COOKIE_NAME: str = "skillnet_session"
    COOKIE_SECURE: bool = True                # Ponlo a False para desarrollo local sin HTTPS

    # LLM (por defecto, sobrescribible por org)
    LLM_BASE_URL: str = "https://api.openai.com/v1"
    LLM_API_KEY: str = ""
    LLM_MODEL: str = "gpt-4o-mini"

    # Embeddings
    EMBEDDING_BASE_URL: str = "https://api.openai.com/v1"
    EMBEDDING_API_KEY: str = ""
    EMBEDDING_MODEL: str = "multilingual-e5-small"
    EMBEDDING_DIMENSIONS: int = 384

    # Subidas de ficheros
    UPLOAD_DIR: str = "./uploads"
    MAX_UPLOAD_SIZE_MB: int = 50

    # App
    DEBUG: bool = False
    CORS_ORIGINS: list[str] = ["http://localhost:5173"]

    model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}

settings = Settings()

4.7 Factoría de la aplicación

# src/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from src.config import settings

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Arranque: crea el directorio de subidas, verifica la conexión a la BD
    Path(settings.UPLOAD_DIR).mkdir(parents=True, exist_ok=True)
    yield
    # Apagado: libera el motor
    await engine.dispose()

def create_app() -> FastAPI:
    app = FastAPI(
        title="SkillNet API",
        version="0.1.0",
        lifespan=lifespan,
        docs_url="/api/docs" if settings.DEBUG else None,
        redoc_url=None,
    )

    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.CORS_ORIGINS,
        allow_credentials=True,        # Necesario para las cookies
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Registra los manejadores de excepciones
    app.add_exception_handler(AppError, app_error_handler)
    app.add_exception_handler(RequestValidationError, validation_error_handler)

    # Monta los routers bajo /api/v1
    from src.routes import (
        auth, users, documents, courses, modules, lessons,
        exercises, enrollments, skills, chat, manuals,
        spaced_repetition, generation_jobs, stats, settings as settings_routes,
        feedback,
    )
    prefix = "/api/v1"
    app.include_router(auth.router, prefix=prefix, tags=["Auth"])
    app.include_router(users.router, prefix=prefix, tags=["Users"])
    app.include_router(documents.router, prefix=prefix, tags=["Documents"])
    app.include_router(courses.router, prefix=prefix, tags=["Courses"])
    app.include_router(modules.router, prefix=prefix, tags=["Modules"])
    app.include_router(lessons.router, prefix=prefix, tags=["Lessons"])
    app.include_router(exercises.router, prefix=prefix, tags=["Exercises"])
    app.include_router(enrollments.router, prefix=prefix, tags=["Enrollments"])
    app.include_router(skills.router, prefix=prefix, tags=["Skills"])
    app.include_router(chat.router, prefix=prefix, tags=["Chat"])
    app.include_router(manuals.router, prefix=prefix, tags=["Manuals"])
    app.include_router(spaced_repetition.router, prefix=prefix, tags=["Spaced Repetition"])
    app.include_router(generation_jobs.router, prefix=prefix, tags=["Generation Jobs"])
    app.include_router(stats.router, prefix=prefix, tags=["Stats"])
    app.include_router(settings_routes.router, prefix=prefix, tags=["Settings"])
    app.include_router(feedback.router, prefix=prefix, tags=["Feedback"])

    return app

app = create_app()

4.8 Decisiones de diseño clave