Arquitecturas Monolíticas

Ingeniería del Software para Inteligencia Artificial

ATENCIÓN

El código de esta presentación funciona con Python 3.12

Versiones anteriores pueden requerir cambios en el código

Arquitectura de software

• Es el conjunto de decisiones estructurales sobre el diseño y organización de un sistema de software.
• Define componentes, relaciones, flujos de datos y patrones de comunicación.
• Actúa como una guía para el desarrollo de software, asegurando escalabilidad y mantenimiento.

Importancia en sistemas IA

• Gestión del ciclo de vida del modelo: Facilita la automatización de pruebas, despliegue y monitoreo (MLOps).
• Escalabilidad y rendimiento: Permite entrenar y ejecutar modelos en paralelo, gestionando los recursos de forma óptima.
• Modularidad y reutilización: Facilita la integración de nuevos componentes o modelos sin rediseñar todo el sistema.

Ejemplo

• Mala arquitectura: El modelo de IA está embebido en el código de la aplicación y necesita actualización manual.
• Buena arquitectura: El modelo es un servicio independiente en la nube, que puede reentrenarse y desplegarse sin afectar la aplicación principal.

Arquitectura monolítica

Una arquitectura monolítica es un modelo de desarrollo de software donde todos los componentes de una aplicación están integrados en una sola unidad.

La interfaz de usuario, la lógica de negocio y el acceso a datos están dentro del mismo código base y se despliegan juntos.

Ventajas

• Fácil de desarrollar y probar: No requiere integración entre múltiples servicios.
• Menor latencia interna: No hay comunicación externa entre servicios.
• Menos sobrecarga técnica: No requiere orquestación de servicios.

Desventajas

• Difícil de escalar: Si crece, es complicado dividirlo en partes más pequeñas.
• Despliegues lentos: Cualquier cambio requiere volver a desplegar toda la aplicación.
• Menos flexible: Si un módulo falla, puede afectar toda la aplicación.
• Difícil de mantener en equipos grandes, el código se vuelve complejo.

Cuándo usar

• Para aplicaciones pequeñas o medianas donde no se requiere escalabilidad extrema.
• Cuando el equipo es pequeño y no es necesario dividir responsabilidades en microservicios.
• Para desarrollar un MVP (Producto Mínimo Viable) rápidamente.
• Cuando los requerimientos no cambian frecuentemente.

Casos prácticos

Vamos a desarrollar dos casos prácticos para entender cómo implementar una arquitectura monolítica:

• API de inferencia: Recibe datos y devuelve predicciones de un modelo IA.
• Backend de aplicación: Gestiona los datos y la lógica de negocio.

API de inferencia

API de inferencia

• Vamos a implementar una aplicación que recibe un texto en español y devuelve su traducción al inglés usando un modelo de traducción preentrenado de Hugging Face.
• Es un caso de uso común en sistemas de IA, donde se necesita exponer un modelo de inferencia a través de una API.
• La implementación es sencilla y no necesita estructurar demasiado el código.

API: Application Programming Interface

Comunicación con el exterior

• El software como servicio necesita comunicarse con otros servicios y aplicaciones.
• Existen distintos métodos de comunicación:
  • API REST
  • gRPC
  • WebSockets
  • Colas de mensajes (Kafka, RabbitMQ)
  • Bases de datos

Vamos a realizar la implementación de una API REST.

APIs REST

Recordad que una API REST debe seguir los principios RESTful:

• Recursos: Cada recurso tiene una URL única
• Verbos HTTP: GET, POST, PUT, DELETE
• Códigos de estado: 200, 404, 500, …
• Formato de datos: JSON

Implementación de APIs

Frameworks comunes para crear APIs REST en Python:

• Flask: Framework ligero para crear APIs REST.
• FastAPI: Framework moderno y rápido para APIs REST.
• Django REST Framework: Extensión de Django para APIs REST.

Vamos a utilizar FastAPI para implementar nuestra API de inferencia.

FastAPI: Hello World

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def hello_world():
    return {"message": "Hello World"}

FastAPI: Servicio de traducción

from fastapi import FastAPI
from transformers import pipeline

app = FastAPI()

@app.post("/translate")
def translate(prompt: str):
    # Carga el modelo de traducción
    pipe = pipeline("translation", model="Helsinki-NLP/opus-mt-es-en")
    translated_text = pipe(prompt)[0]['translation_text']
    return {"translated_text": translated_text}

• El valor de prompt se recibe en el Query String: http://dominio/translate?prompt=texto
• El texto se traduce usando un modelo preentrenado.

Demo

Petición POST con Query Params

Buenas prácticas

• Carga eficiente del modelo en el arranque.

@app.post("/translate")
def translate(prompt: str):
    # ❌ Carga el modelo en cada petición
    pipe = pipeline("translation", model="Helsinki-NLP/opus-mt-es-en")
    translated_text = pipe(prompt)[0]['translation_text']
    return {"translated_text": translated_text}

Carga eficiente del modelo

from fastapi import FastAPI
from transformers import pipeline
app = FastAPI()

# ✅ Carga única del modelo al arrancar la aplicación
# ❌ Todavía no se gestiona la liberación de recursos
pipe = pipeline("translation", model="Helsinki-NLP/opus-mt-es-en")

# Añadimos este método para simplificar el endpoint
def translate_text(text: str) -> str:
    if pipe is None:
        raise RuntimeError("Translation service not initialized")
    return pipe(text)[0]['translation_text']

@app.post("/translate")
def translate(prompt: str):
    translated_text = translate_text(prompt)
    return {"translated_text": translated_text}

Buenas prácticas

• Carga eficiente del modelo en el arranque.
• Gestión de recursos en memoria.

Gestión de recursos con lifespan

from contextlib import asynccontextmanager
from fastapi import FastAPI
from transformers import pipeline

pipe = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global pipe  # Declare global to modify the outer scope
    # Initialize the translation pipeline
    pipe = pipeline("translation", model="Helsinki-NLP/opus-mt-es-en")
    yield
    # Cleanup
    del pipe

app = FastAPI(lifespan=lifespan)  # Attach lifespan handler

yield

• yield pausa la ejecución de la función. Cuando se vuelve a llamar, la función continúa desde donde se quedó.
• FastAPI llama a la función lifespan al arrancar la aplicación y la vuelve a llamar al cerrarla.

Inversión de control

Recuerda, muchas librerías y frameworks utilizan el patrón de Inversión de Control para gestionar el ciclo de vida de los recursos.

async

async permite que una función se ejecute de forma asíncrona, lo que es útil para tareas que pueden bloquear la ejecución principal.

Asynchronous Server Gateway Interface (ASGI)

FastAPI soporta funciones asíncronas para gestionar tareas que requieren tiempo.

WSGI vs. ASGI

Buenas prácticas

• Carga eficiente del modelo en el arranque.
• Gestión de recursos en memoria.
• Validación de datos de entrada y salida.

Validación de datos de entrada

• Pydantic permite validar los datos de forma sencilla.
• Mapea la entrada recibida como JSON en el cuerpo (body) de la petición HTTP.

from pydantic import BaseModel

class TranslationRequest(BaseModel):
    text: str

@app.post("/translate")
async def translate(request: TranslationRequest):
    prompt = request.text
    translated_text = translate_text(prompt)
    return {"translated_text": translated_text}

Demo

Petición POST con JSON

Validación de datos de salida

• También podemos definir la estructura de la respuesta.
• Los type hints documentan el código y facilitan su mantenimiento.

class TranslationResponse(BaseModel):
    translated_text: str

@app.post("/translate", response_model=TranslationResponse)
async def translate(request: TranslationRequest) -> dict[str, str]:
    prompt = request.text
    translated_text = translate_text(prompt)
    response = TranslationResponse(translated_text=translated_text)
    return response
    # return {"translation_text": translated_text}  # Alternativa

Type hints

• Desde Python 3.9+ podemos usar list, dict, tuple y set como tipos genéricos.
• En versiones anteriores hay que usar la librería typing.

from typing import Dict

async def translate(request: TranslationRequest) -> Dict[str, str]:

Buenas prácticas

• Carga eficiente del modelo en el arranque.
• Gestión de recursos en memoria.
• Validación de datos de entrada y salida.
• Logging para registrar eventos y errores.

Logging

• En un servidor, es importante registrar eventos y errores para facilitar el mantenimiento y la depuración.
• Los logs se pueden guardar en un archivo o enviar a un servicio de monitoreo, ya que normalmente no tenemos acceso a la salida por consola.

Logging

import logging

# Create logs folder if it doesn't exist
os.makedirs('logs', exist_ok=True)

logging.basicConfig(level=logging.INFO, filename='../logs/translate.log', filemode='a', format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

@app.post("/translate")
async def translate(request: TranslationRequest) -> dict[str, str]:
    prompt = request.text
    logger.info(f"Translating: {prompt}")
    translated_text = translate_text(request.text)
    response = TranslationResponse(translated_text=translated_text)
    return response

Deberíamos configurar el nivel de log según el entorno (desarrollo: INFO/DEBUG, producción: WARNING/ERROR).

Buenas prácticas

• Carga eficiente del modelo en el arranque.
• Gestión de recursos en memoria.
• Validación de datos de entrada y salida.
• Logging para registrar eventos y errores.
• Documentación de la API con OpenAPI.

Documentación de los endpoints

FastAPI genera automáticamente la documentación de la API con OpenAPI.

Se puede completar con descripciones, ejemplos y validaciones.

• Accede a http://localhost:8000/docs para ver la documentación interactiva.
• Accede a http://localhost:8000/redoc para ver la documentación en formato ReDoc.

Esta documentación es muy útil para los desarrolladores que consumen la API.

Documentación de los endpoints

@app.post(
        "/translate",
        response_model=TranslationResponse,
        summary="Translate text from Spanish to English",
        description="""
        Translate text using a pre-trained model from Hugging Face:
        https://huggingface.co/Helsinki-NLP/opus-mt-en-es
        """
    )
async def translate(request: TranslationRequest) -> dict[str, str]:
    translated_text = translate_text(request.text)
    response = TranslationResponse(translated_text=translated_text)
    return response

Buenas prácticas

• Carga eficiente del modelo en el arranque.
• Gestión de recursos en memoria.
• Validación de datos de entrada y salida.
• Logging para registrar eventos y errores.
• Documentación de la API con OpenAPI.
• Seguridad.

Seguridad

• Las APIs deben protegerse contra ataques y accesos no autorizados.
  • Si son públicos, se deben proteger con autenticación y autorización, igual que cualquier otro servicio.
  • Si son privados, se pueden proteger con mediante restricciones de red o VPN (sólo accesibles para otros servicios).

Protección mediante API KEY

db.py: base de datos (FAKE) con usuarios y claves API.

users = [
    {
        "name": "Alice",
        "api_key": "5f0c7127-3be9-4488-b801-c7b6415b45e9"
    },
    {
        "name": "Bob",
        "api_key": "e54d4431-5dab-474e-b71a-0db1fcb9e659"
    }
]

# For demonstration purposes only, YOU MUST USE A DATABASE INSTEAD
def get_user_from_api_key(api_key: str):
    for user in users:
        if user["api_key"] == api_key:
            return user
    return None

Protección mediante API KEY

auth.py: middleware para autenticar usuarios.

from fastapi import Security, HTTPException, status
from fastapi.security import APIKeyHeader
from db import get_user_from_api_key

api_key_header = APIKeyHeader(name="X-API-Key")

def get_user(api_key_header: str = Security(api_key_header)):
    user = get_user_from_api_key(api_key_header)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing or invalid API key"
        )
    return user

Protección mediante API KEY

from fastapi import Depends
from auth import get_user

@app.post("/secure/translate", response_model=TranslationResponse)
async def translate(
        request: TranslationRequest,
        user: dict = Depends(get_user)
    ) -> dict[str, str]:
    
    logger.info(f"Translation request by user {user['name']}")
    translated_text = translate_text(request.text)
    response = TranslationResponse(translated_text=translated_text)
    return response

FastAPI permite definir dependencias para inyectar datos en los endpoints.

Inyección de dependencias

• Las dependencias son funciones que se ejecutan antes de un endpoint.
• Sirven para validar datos, autenticar usuarios, etc.
• Se pueden inyectar en los endpoints para facilitar el desarrollo y la reutilización de código.
• Pueden tener a su vez otras dependencias.

async def translate(user: dict = Depends(get_user)):

def get_user(api_key_header: str = Security(api_key_header)):

• translate.py
• auth.py
• db.py

import os
import logging

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from pydantic import BaseModel

from transformers import pipeline

from auth import get_user


# Create logs folder if it doesn't exist
os.makedirs('logs', exist_ok=True)

logging.basicConfig(level=logging.INFO, filename='../logs/translate.log', filemode='a', format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Global variable to store the translation pipeline
pipe = None

# Initialize the translation pipeline at startup
@asynccontextmanager
async def lifespan(app: FastAPI):
    global pipe  # Declare global to modify the outer scope
    logger.info("Initializing translation pipeline")
    pipe = pipeline("translation", model="Helsinki-NLP/opus-mt-es-en")
    yield
    logger.info("Cleaning up translation pipeline")
    del pipe

# Function to translate text using the pipeline
def translate_text(text: str) -> str:
    if pipe is None:
        raise RuntimeError("Translation service not initialized")
    return pipe(text)[0]['translation_text']

# Create the FastAPI app
app = FastAPI(lifespan=lifespan)

# Define the request and response models
class TranslationRequest(BaseModel):
    text: str

class TranslationResponse(BaseModel):
    translated_text: str

# Define the translation endpoint
@app.post(
        "/translate",
        response_model=TranslationResponse,
        summary="Translate text from Spanish to English",
        description="Translate text using a pre-trained model."
    )
async def translate(request: TranslationRequest) -> dict[str, str]:
    translated_text = translate_text(request.text)
    response = TranslationResponse(translated_text=translated_text)
    return response

# Define the translation endpoint
@app.post(
        "/secure/translate",
        response_model=TranslationResponse,
        summary="Translate text from Spanish to English",
        description="Translate text using a pre-trained model."
    )
async def secure_translate(
        request: TranslationRequest,
        user: dict = Depends(get_user)
    ) -> dict[str, str]:

    logger.info(f"Translation request by user {user['name']}")
    translated_text = translate_text(request.text)
    response = TranslationResponse(translated_text=translated_text)
    return response

from fastapi import Security, HTTPException, status
from fastapi.security import APIKeyHeader
from db import get_user_from_api_key

api_key_header = APIKeyHeader(name="X-API-Key")

def get_user(api_key_header: str = Security(api_key_header)):
    user = get_user_from_api_key(api_key_header)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing or invalid API key"
        )
    return user

users = [
    {
        "name": "Alice",
        "email": "alice@example.com",
        "api_key": "5f0c7127-3be9-4488-b801-c7b6415b45e9"
    },
    {
        "name": "Bob",
        "email": "bob@example.com",
        "api_key": "e54d4431-5dab-474e-b71a-0db1fcb9e659"
    }
]

# For demonstration purposes only, YOU MUST USE A DATABASE INSTEAD
def get_user_from_api_key(api_key: str):
    for user in users:
        if user["api_key"] == api_key:
            return user
    return None

• requirements.txt
• ejecución

pydantic==2.10.6
fastapi[standard]==0.115.10
torch==2.6.0
transformers==4.49.0
sentencepiece==0.2.0
sacremoses==0.1.1

fastapi dev translate.py

async/await vs. sync (AVANZADO)

FastAPI trata internamente de forma diferente las funciones asíncronas y síncronas.

• async def se ejecuta en el bucle de eventos de FastAPI, usando asyncio.
• def se ejecuta en un hilo separado.

async/await vs. sync (AVANZADO)

Si llamamos a una función síncrona desde una función asíncrona, bloqueamos el bucle de eventos.

from fastapi import FastAPI
import time

app = FastAPI()

# This endpoint blocks the event loop
# It should be sync -> def block():
@app.get("/block")
async def block():
    time.sleep(10)
    return {"msg": "Sorry, I blocked the event loop"}

@app.get("/blocked")
async def blocked():
    return {"msg": "I had to wait for the other endpoint to finish"}

async/await vs. sync (AVANZADO)

• async/await es más eficiente para operaciones de I/O (red, disco, etc.).
• sync es más eficiente para operaciones CPU-bound (cálculos intensivos).

async/await vs. sync (AVANZADO)

Cuando llamamos a otras funciones asíncronas se ejecutan en paralelo.

from fastapi import FastAPI
import time

app = FastAPI()

async def sum(a: int, b: int) -> int:
    return a + b

@app.get("/sum")
async def endpoint_sum():
    value = sum(1, 2)
    return {"value": str(value)}

{"value":"<coroutine object sum at 0x791dff090c40>"}

async/await vs. sync (AVANZADO)

Si necesitamos su valor de retorno, usamos await.

from fastapi import FastAPI
import time

app = FastAPI()

async def sum(a: int, b: int) -> int:
    return a + b

@app.get("/sum")
async def endpoint_sum():
    value = await sum(1, 2)
    return {"value": str(value)}

Backend de aplicación

Backend de aplicación

• Vamos a implementar un backend de aplicación que gestiona los datos y la lógica de negocio.
• Es un caso de uso común en sistemas de IA, donde se necesita almacenar y procesar datos antes de enviarlos a un modelo de inferencia.
• La implementación es más compleja y necesita estructurar el código en módulos.

Una forma común de estructurar el código es siguiendo el patrón de Arquitectura en Capas.

Arquitectura en capas

La arquitectura en capas organiza un sistema en módulos independientes, facilitando el mantenimiento y escalabilidad.

• Capa de Presentación (UI/API): Interfaz para usuarios o aplicaciones.
• Capa de Servicios: Lógica de negocio y comunicación con el modelo de IA.
• Capa de Datos: Almacenamiento, gestión y preprocesamiento de datos.

Hay muchas formas de implementar una arquitectura en capas, pero esta es sencilla y efectiva.

Reglas de una arquitectura en capas

Separación de responsabilidades: Cada capa tiene una función clara y se comunican entre sí de forma jerárquica.

• Controladores: Reciben las peticiones del exterior.
• Servicios: Implementan funcionalidades.
• Modelos: Reflejan la estructura de las tablas.

Cada capa debe depender sólo de las capas inferiores

Organización del código

El código se organiza en distintos paquetes lógicos.

Ejemplo:

📂 src/
│── 📄 main.py       # Inicia el servidor FastAPI
│── 📂 db/       
│   │── 📄 db.py     # Configuración de la base de datos
│   └── 📄 models.py # Modelos de datos (SQLModel)
│── 📂 services/     # Lógica de negocio
│── 📂 controllers/  # Endpoints de la API
│── 📂 middleware/   # Autenticación, logging, etc.
└── 📂 util/         # Funciones auxiliares

Capa de datos

Capa de datos

• La capa de datos se encarga de gestionar el almacenamiento, acceso y preprocesamiento de los datos.
• Puede incluir bases de datos, sistemas de archivos, APIs externas, etc.
• La capa de datos ofrece una interfaz para acceder a los datos de forma sencilla y segura.

Bases de datos

• SQL: Bases de datos relacionales (SQLite, MySQL, PostgreSQL).
• NoSQL: Bases de datos no relacionales (MongoDB, Cassandra, Redis).

Bases de datos en Python

Los sistemas ORM (Object-Relational Mapping) facilitan el acceso a bases de datos desde Python.

Permiten definir modelos de datos y operar con ellos de forma sencilla, sin necesidad de escribir SQL.

• SQLAlchemy: ORM para bases de datos SQL.
• SQLModel: ORM para bases de datos SQL con tipado estático.
• MongoEngine: ORM para MongoDB.

SQLModel

• Basado en SQLAlchemy y Pydantic
• Ofrece una forma sencilla de definir modelos de datos y trabajar con bases de datos SQL.

Conexion a la base de datos

from sqlmodel import create_engine, Session

# Cadena de conexión a la base de datos
DATABASE_URL = "mysql+pymysql://user:password@localhost/database"

# Crear el motor de conexión
engine = create_engine(DATABASE_URL, echo=True)

# Función para obtener la sesión de base de datos
def get_session():
    session = Session(engine)
    try:
        return session  # Devolvemos la sesión directamente
    except Exception as e:
        session.rollback()  # Hacemos rollback si hay un error
        raise e
    finally:
        session.close()  # Nos aseguramos de cerrar la sesión cuando se salga de la función

Modelos de datos

• Definen la estructura de las tablas en la base de datos.
• Son clases de Python que heredan de SQLModel.
• Las tablas se crean automáticamente en la base de datos.

from sqlmodel import Field, SQLModel

# Definimos un modelo de datos para la tabla 'user'
class User(SQLModel, table=True):
    id: int = Field(primary_key=True)
    name: str
    email: str

# Se creará una tabla por cada modelo de datos
SQLModel.metadata.create_all(engine)

Modelos de datos

SQLModel se encarga de mapear los atributos de la clase con las columnas de la tabla, usando el tipo de datos más adecuado en cada motor de base de datos.

class User(SQLModel, table=True):
    id: int = Field(primary_key=True)
    name: str
    email: str

Relaciones entre modelos

• Las tablas se relacionan con claves ajenas.
• SQLModel permite definir relaciones de forma sencilla, estableciendo una relación entre dos modelos.

class User(SQLModel, table=True):
    id: int = Field(primary_key=True)
    name: str = Field(index=True)
    email: str
    posts: list["Post"] = Relationship(back_populates="user")

class Post(SQLModel, table=True):
    id: int = Field(primary_key=True)
    title: str
    content: str
    user_id: int = Field(foreign_key="user.id")
    user: User = Relationship(back_populates="posts")

Relaciones entre modelos

class User(SQLModel, table=True):
    id: int = Field(primary_key=True)
    name: str = Field(index=True)
    email: str
    posts: list["Post"] = Relationship(back_populates="user")

class Post(SQLModel, table=True):
    id: int = Field(primary_key=True)
    title: str
    content: str
    user_id: int = Field(foreign_key="user.id")
    user: User = Relationship(back_populates="posts")

• user_id: clave ajena a la tabla user.
• user: almacenará el objeto User relacionado.
• posts: lista de objetos Post relacionados.

Relaciones entre modelos

Distintas formas de relacionar instancias:

with get_session() as db:
    user = User(name="Alice", email="alice@example.com")
    db.add(user)
    db.commit() # Insertamos el usuario en la base de datos para obtener el id

    post1 = Post(user_id=user.id, title="First post", content="This is the first post")
    post2 = Post(user=user, title="Second post", content="This is the second post")
    post3 = Post(title="Third post", content="This is the third post")
    post3.user = user
    db.add_all([post1, post2, post3])

    post4 = Post(title="Fourth post", content="This is the fourth post")
    user.posts.append(post4)
    DATABASE_URL.add(user)

    db.commit()

Relaciones entre modelos

Las relaciones permiten obtener los objetos relacionados de forma sencilla.

from sqlmodel import select

with get_session() as db:
    user = db.exec(select(User)).first()
    print(user)
    print(user.posts)
    print(user.posts[0].title)

    post = db.exec(select(Post)).first()
    print(post)
    print(post.user)
    print(post.user.name)

Mantenimiento de la base de datos

En las primeras fases de desarrollo podemos necesitar recrear la base de datos para probar cambios en los modelos.

@asynccontextmanager
async def lifespan(app: FastAPI):
    logging.info("Recreating database tables")
    SQLModel.metadata.drop_all(engine)
    SQLModel.metadata.create_all(engine)
    logging.info("Seeding users")
    seed_users()
    logging.info("Application started")
    yield
    logging.info("Application shutdown")

app = FastAPI(lifespan=lifespan)

Mantenimiento de la base de datos

• A medida que la aplicación crece, es necesario actualizar la estructura de la base de datos.
• En producción no podemos borrar la base de datos, es importante realizar migraciones para no perder datos.
  • Cambios manuales en la estructura de la base de datos.
  • Scripts de migración para actualizar la base de datos.
  • Herramientas de migración como Alembic o Flask-Migrate.

Mantenimiento de la base de datos

• También podemos cargar datos de prueba en la base de datos para probar la aplicación.
• Hacemos uso de seeders:

@asynccontextmanager
async def lifespan(app: FastAPI):
    logging.info("Recreating database tables")
    SQLModel.metadata.drop_all(engine)
    SQLModel.metadata.create_all(engine)
    logging.info("Seeding users")
    seed_users()
    logging.info("Application started")
    yield
    logging.info("Application shutdown")

app = FastAPI(lifespan=lifespan)

Capa de servicios

Capa de servicios

Esta capa contiene la lógica de negocio y se encarga de coordinar las operaciones entre la capa de presentación y la capa de datos.

Lógica de negocio

La lógica de negocio es el conjunto de reglas y procesos que definen cómo funciona una aplicación.

Ejemplo: en una biblioteca, no se puede prestar un libro a un usuario en determinados casos:

• Si el libro está prestado a otro usuario.
• Si el usuario tiene el máximo de préstamos activos permitidos.
• Si el usuario tiene sanciones.

Servicios

• Los servicios son clases o funciones que implementan la lógica de negocio.
• Usan los modelos de datos para acceder a la base de datos.
• Ofrecen métodos para realizar operaciones de forma segura y coherente.

Servicios

• Recibimos la sesión mediante inyección de dependencias.
• Los métodos son estáticos porque no almacenan estado.

from sqlmodel import select, Session

from db import User

class UserService:
    @staticmethod
    def get_all(db: Session):
        return db.exec(select(User)).all()

    @staticmethod
    def get_by_id(user_id: int, db: Session):
        user = db.exec(select(User).where(User.id == user_id)).first()
        if not user:
            raise ValueError("User not found")
        return user

Consultas

class UserService:
    @staticmethod
    def queries(db: Session):
        # all() devuelver una lista
        users = db.exec(select(User)).all()
        users = db.exec(select(User).where(
            User.email.like(f"%example.com%"))
        ).all()

        # first() devuelve el primer elemento
        user = db.exec(select(User)).first()
        user = db.exec(select(User).where(User.id == 1)).first()

Inserciones

Cualquier operación que modifique la base de datos debe confirmarse con commit().

class UserService:
    @staticmethod
    def inserts(db: Session):
        user = User(name="Mary", email="mary@example.com")
        db.add(user) # Inserta un usuario
        db.commit()  # Confirma los cambios

        users = [
            User(name="Alice", email="alice@example.com"),
            User(name="Bob", email="bob@example.com", phone="123456789"),
            User(name="Charlie", email="charlie@example.com", country="US")
        ]
        db.add_all(users) # Inserta varios usuarios
        db.commit()       # Confirma los cambios

Actualizaciones y borrados

Cualquier operación que modifique la base de datos debe confirmarse con commit().

class UserService:
    @staticmethod
    def update(db: Session):
        user = db.exec(select(User).where(User.name == "Mary")).first()
        user.name = "Maria" # Cambia el nombre del usuario
        db.add(user)        # Si tiene un id, actualiza el usuario
        db.commit()         # Confirma los cambios

    @staticmethod
    def delete(db: Session):
        user = db.exec(select(User).where(User.name == "Maria")).first()
        db.delete(user) # Elimina un usuario
        db.commit()     # Confirma los cambios

Capa de presentación

Capa de Presentación

• Interfaz de usuario (UI) o API REST.
• Recibe peticiones y envía respuestas.
• Valida datos y autentica usuarios.
• Envia solicitudes a la capa de servicios.

Uso de métodos HTTP

Método HTTP	Uso
GET	Obtener datos (GET /users/{id})
POST	Crear un nuevo recurso (POST /users/)
PUT	Actualizar un recurso (PUT /users/{id})
DELETE	Eliminar un recurso (DELETE /users/{id})

Respuestas correctas

Código	Significado	Uso en API REST
200 OK	Éxito	Respuesta con éxito a una solicitud GET o PUT.
201 Created	Recurso creado	Se usa en respuestas de POST cuando se crea un nuevo recurso.
204 No Content	Sin contenido	La solicitud terminó con éxito pero no hay datos que devolver (ej. DELETE).

Códigos de error

Código	Significado	Uso en API REST
400 Bad Request	Error del cliente	Datos enviados incorrectos o inválidos.
401 Unauthorized	No autenticado	El usuario no ha iniciado sesión.
403 Forbidden	Acceso denegado	El usuario no tiene permisos para acceder al recurso.
404 Not Found	No encontrado	El recurso no existe.
409 Conflict	Conflicto	Error debido a datos duplicados o reglas de negocio.
422 Unprocessable Content	Entidad no procesable	Error de validación de datos (usado por Pydantic).

Errores de ejecución

Código	Significado	Uso en API REST
500 Internal Server Error	Error del servidor	Fallo inesperado en el servidor.

Estos errores surgen si hay errores (excepciones) en tiempo de ejecución.

Deben ser capturados y gestionados para evitar que la aplicación falle.

Controladores

• Los controladores son funciones que gestionan las peticiones HTTP.
• Reciben y validan los datos de entrada
• Llaman a los servicios.
• Devuelven una respuesta al cliente.

Controladores

• Reciben la sesión mediante inyección de dependencias.
• Esto permite realizar todas las operaciones dentro de una transacción.

app = FastAPI()

@app.get("/users", response_model=list[UserListResponseModel])
def get_users(db : Session = Depends(get_session)) -> list[UserListResponseModel]:
    users = user_service.get_all(db)
    return users

Exponer sólo lo necesario

Los endpoints no deben exponer más información de la necesaria.

@app.get("/users/{id}")
def get_user_by_id(id: int):
    user = user_service.get_by_id(id)
    return user

• Al devolver el objeto user completo estamos exponiendo más información de la necesaria (incluyendo la contraseña).
• Debemos controlar qué información se devuelve usando objetos Pydantic.

Exponer sólo lo necesario

Al devolver modelos Pydantic se filtrarán automáticamente los campos a devolver:

class UserResponseModel(BaseModel):
    id: int
    username: str

@app.get("/users/{id}", response_model=UserResponseModel)
def get_user_by_id(id: int):
    user = user_service.get_by_id(id)
    return user

Devolver datos agregados

Los modelos Pydantic también nos permiten devolver información agregada de varios modelos:

class PostResponseModel(BaseModel):
    username: str
    text: str

@app.get("/posts/{id}", response_model=PostResponseModel)
def get_post_by_id(id: int):
    post = post_service.get_by_id(id)
    return PostResponseModel(username=post.user.username, text=post.text)

Prevención de ataques

• Los datos de entrada deben ser validados para evitar errores y ataques (inyección SQL, XSS, …).
• Mantener la aplicación segura es fundamental para proteger los datos y la privacidad de los usuarios.

Validación de datos de entrada

Ejemplo de validación manual:

@app.post("/users/{name}")
async def get_user_by_name(name: str):
    if not name:
        raise HTTPException(status_code=400, detail="Name cannot be empty")
    if len(name) > 1000:
        raise HTTPException(status_code=400, detail="Name is too long")
    if any(bad_word in name.lower() for bad_word in [
            "<script>",
            "drop table",
            "select *",
            "delete from",
            "update users",
            "password",
            "' or 1=1",
            "http://",
            "https://"
        ]):
        raise HTTPException(status_code=400, detail="Name contains invalid content")
    # Access the database, call the model, etc.

Validación de datos de entrada

SQLModel y Pydantic permiten realizar la validación automáticamente.

Estas librerías son muy útiles para evitar errores y proteger la aplicación.

Controladores

Para devolver errores lanzamos HTTPException.

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/users/{id}", response_model=UserResponseModel)
def get_user(id: int, db : Session = Depends(get_session)) -> UserResponseModel:
    try:
        user = user_service.get_by_id(id, db)
        return user
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))

Routers

• A medida que la aplicación crece, es útil organizar los controladores en routers.
• Los routers agrupan los controladores por funcionalidad (usuarios, posts, etc.).
• Se pueden montar en la aplicación principal.

from controllers import user_router, post_router

app.include_router(user_router, tags=["users"])
app.include_router(movie_router, tags=["movies"])

Routers

controllers/user_controller.py

from fastapi import APIRouter

router = APIRouter()

@router.get("/users", response_model=list[UserListResponseModel])
def get_users(db : Session = Depends(get_session)) -> list[UserListResponseModel]:
    users = user_service.get_all(db)
    return users

controllers/__init__.py

from .user_controller import router as user_router
from .post_controller import router as post_router

Seguridad

Seguridad

• Las aplicaciones deben protegerse contra ataques y accesos no autorizados.
• Se pueden utilizar distintas técnicas de seguridad:
  • Autenticación y autorización: Proteger los endpoints con credenciales.
  • Cifrado: Proteger los datos en tránsito y en reposo.

Autenticación y autorización

• Autenticación: Verificar la identidad de un usuario.
• Autorización: Determinar si un usuario tiene permiso para acceder a un recurso.

Autenticación y autorización

Métodos comunes de autenticación:

• API KEY: Clave de acceso para proteger los endpoints.
• Tokens JWT: Tokens de acceso con información del usuario.
• OAuth: Protocolo de autorización para aplicaciones.

Una vez el usuario se ha autenticado, la aplicación puede comprobar si tiene permisos para acceder a un recurso.

Tokens JWT

JSON Web Token (JWT)

• Cadena de texto que contiene información del usuario y una firma.
  • Payload: Datos del usuario codificados en Base64.
  • Firma: Verifica la autenticidad del token.
• Cuando el usuario introduce sus credenciales (usuario y contraseña), el servidor genera y envía un token al cliente.

Tokens JWT

Tokens JWT

Solicitud del token (autenticación)

Tokens JWT

Solicitud del token (autorización)

Tokens JWT

El cliente debe guardar el token y enviarlo en cada petición.

• Ventajas: Fácil de implementar, escalable y seguro.
• Desventajas: Si el token es robado, el atacante puede acceder a la cuenta del usuario.

Para aumentar la seguridad se pueden añadir mecánismos de refresco y revocación de tokens.

Cifrado

• Cifrado en tránsito: Proteger los datos que se envían entre el cliente y el servidor.
  • HTTPS: Protocolo seguro que cifra los datos.
• Cifrado en reposo: Proteger los datos almacenados en la base de datos.
  • Encriptación: Cifrar los datos antes de almacenarlos.
  • Hashing: Almacenar contraseñas como hash.