Introducción a los Microservicios en FastAPI: Guía básica

Vamos a comenzar, igual que hicimos con el de Laravel, con dos simples servicios: uno de tareas y otro para usuarios. El chiste se cuenta solo: un CRUD de tareas —ya lo vamos a ver en unos momentos, por aquí ya lo tenemos— y, para hacer cada operación, obviamente el usuario tiene que estar autenticado. Y aquí la autenticación no es local a esta aplicación de tasks —es simplemente una aplicación en FastAPI—: depende de otro servicio. Voy a emplear «microservicio», «servicio» y «aplicación» como sinónimos. En este caso, el otro servicio sería el de user-service, y este se conecta a él mediante peticiones HTTP. Sí que existen formas más eficientes —a lo largo de esta guía iremos viendo otras cosas y cómo configurarlas correctamente—, pero la idea es simplemente dar una introducción a los microservicios con FastAPI: la esencia de FastAPI, principalmente, e ir desde la mínima expresión hasta llegar a algo un poco más completo. Así que, otra vez, es una guía introductoria.

Primer microservicio

En esta primera parte, vamos a usar dos microervicios:

Lo siguiente: tal cual puedes ver, la implementación ya la tengo; no la voy a hacer desde cero. Tú tienes el código fuente, y yo te voy a dar las aplicaciones —los archivos— más importantes. Esto lo hago así para no alargar esta sección y que se haga diez veces más larga.

Y como siempre, ya doy por hecho que tienes conocimientos al menos básicos de FastAPI. Si no los tienes: en mi blog hay varias publicaciones, tengo un libro comunitario que puedes iniciar completamente gratis o, si me quieres apoyar, también puedes ir por la solución completa; al igual que el curso, que lo tengo tanto en mi plataforma como en Udemy, y puedes comenzar gratis. Los conocimientos necesarios son, más que nada, que sepas cómo hacer un CRUD básico y lo que ello implica. Así que, ahora sí, vamos a comenzar.

Sin UI de servidor: la documentación automática

Lo comentado antes: tenemos dos servicios. A diferencia de la aplicación en Laravel, que estábamos renderizando con Blade, aquí no quise hacerlo para no meter vistas y complicar un poco más el asunto, ya que creo que lo podemos aprovechar más adelante conectando una aplicación —por ejemplo, en Vue o en React— y consumiendo los mismos servicios. Por lo tanto, de momento vamos a emplear la documentación automática, cosa que no tenemos en Laravel. Y aquí un poco lo mismo: dependiendo del ecosistema en el que te encuentres cuando estamos comenzando a desarrollar, podemos utilizar un esquema u otro. En este caso, como aquí tenemos la documentación automática, para hacer estas pruebas y conexiones iniciales nos viene de perlas, y luego vemos que levantamos una UI dependiendo de los requisitos del proyecto, del cliente o qué sé yo. Como te digo, en Laravel usualmente —como es un framework un poco más tradicional— ya le metí Blade para no complicar el asunto con Vue y demás.

Los proyectos y los puertos (8000 / 8001)

Los proyectos están justamente por aquí. Aquí levantamos el del usuario —importante que está en el puerto 8000, porque obviamente tenemos un hardcode en la aplicación de tareas que va a hacer la petición al de usuarios, por lo comentado inicialmente, en la forma de autenticarnos—; y el de tareas sería el 8001. Aunque aquí no importa cuál es el que quieras colocar. Y es esto básicamente, en pocas palabras: la autenticación va a ser, de momento, mediante una petición HTTP al user-service.

task-service/app/config.py

import os

def _int(name: str, default: int) -> int:
    return int(os.environ.get(name, str(default)))

# Dónde vive el IdP. En producción, variable de entorno; en dev, puerto 8000.
USER_SERVICE_URL = os.environ.get("USER_SERVICE_URL", "http://127.0.0.1:8000")
HTTP_TIMEOUT_SECONDS = _int("HTTP_TIMEOUT_SECONDS", 2)

# sqlite de este servicio; en tests, DATABASE_URL=sqlite:///:memory:
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///database.sqlite")

Configuración del cliente HTTP: el timeout

Fíjate que tenemos lo comentado: el servicio de usuarios con el puerto 8000, y algunas configuraciones adicionales —como es totalmente normal—: el timeout. Si se pasa de dos segundos, la desechamos, por si acaso hay algunos problemas: una protección muy básica que tenemos que ir mejorando poco a poco.

Ese timeout es el que usa el cliente HTTP cuando hace la petición (lo vemos en el bloque 14); y la URL del IdP se puede cambiar con la variable de entorno USER_SERVICE_URL, así el hardcode del 8000 solo es el valor de desarrollo.

La base de datos (SQLite, SessionLocal, get_db)

Estamos empleando SQLite. Aquí tenemos la conexión a la base de datos, un poco lo normal: aquí la declaramos —y esto sería para las pruebas, con sqlite:///:memory:—. Aquí tenemos lo de siempre: el engine, el SessionLocal, y el get_db —el yield— para devolver la conexión. Por lo tanto, este es el método que vamos a utilizar mediante una dependencia.

task-service/app/db.py

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app import config

class Base(DeclarativeBase):
    pass

def _make_engine(url: str) -> "create_engine":  # type: ignore[valid-type]
    kwargs = {"connect_args": {"check_same_thread": False}}
    if url.endswith("//:memory:"):
        from sqlalchemy.pool import StaticPool

        kwargs["poolclass"] = StaticPool
    return create_engine(url, **kwargs)


engine = _make_engine(config.DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

 Arrancar la aplicación (main.py)

Para arrancar la aplicación. Muy normal: simplemente es el lifespan —con él se crean las tablas—, el FastAPI, levantamos la aplicación, y el include_router, con el router que había antes.

task-service/app/main.py

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.db import Base, engine
from app.routes_tasks import router as tasks_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(engine)
    yield

app = FastAPI(
    title="task-service",
    description="Tareas por usuario, autenticadas vía el IdP (capítulo 1).",
    version="0.1.0",
    lifespan=lifespan,
)
app.include_router(tasks_router)

El modelo de tareas: el usuario es un entero, no una FK

El modelo de tareas muy sencillo: simplemente es un modelo base y lo que quieras colocar: el id, el usuario —fíjate que es simplemente un entero—, porque aquí, otra vez, él no sabe qué demonios es un usuario. Por lo tanto, aquí no podemos tener una relación de tipo FK, porque, otra vez, por convención, el usuario vive en otra aplicación, también su base de datos vive en esa aplicación, y aquí no se sabe absolutamente nada. Por lo demás, muy normal: título, status, y lo que tú quieras colocar por ahí.

task-service/app/models.py

from datetime import UTC, datetime

from sqlalchemy import DateTime, String, Text
from sqlalchemy.orm import Mapped, mapped_column

from app.db import Base

STATUS_PENDING = "pending"
STATUS_COMPLETED = "completed"
STATUSES = (STATUS_PENDING, STATUS_COMPLETED)

def utcnow() -> datetime:
    # Naive en UTC: sqlite no conserva tzinfo.
    return datetime.now(UTC).replace(tzinfo=None)

class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(index=True)
    title: Mapped[str] = mapped_column(Text)
    status: Mapped[str] = mapped_column(String(20), default=STATUS_PENDING)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow, onupdate=utcnow)

Los esquemas (Pydantic)

El esquema: nada raro. Simplemente, bueno, los de Pydantic te lo de siempre; y bueno, sería TaskUpdate y TaskCreate, los que vamos a manejar; el de salidas —para pintarlo por pantalla—, muy normal; y el del cliente. El normalize_status es la validación del status: fuera de pending|completed, un 422.

task-service/app/schemas.py

from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field

from app.models import STATUSES

class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    status: str | None = None

class TaskUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    status: str | None = None

class TaskOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    user_id: int
    title: str
    status: str
    created_at: datetime
    updated_at: datetime

def normalize_status(value: str | None, default: str) -> str:
    if value is None:
        return default
    if value not in STATUSES:
        from fastapi import HTTPException

        raise HTTPException(
            status_code=422,
            detail=f"status debe ser uno de: {', '.join(STATUSES)}",
        )
    return value

Las rutas: dueño de la tarea, 404 y 403

Vamos a dejar aquí el de rutas, donde se conecta todo. Obviamente, porque tenemos, por ejemplo, el task_id, para encontrar una tarea, y que obviamente yo sea el dueño, para no ver las tareas de otros usuarios; todo esto se realiza por aquí. Con el task_id buscamos; si existe la tarea, damos el 200; si es nula, entonces, a freír gatos: un 404; y si no eres dueño, pues no es tu tarea: un 403.

Y por aquí, como cosa interesante, tenemos el de current_user, que vamos a ver qué es. Por lo demás, es el mismo flujo para el resto de los endpoints: para crear, simplemente, otra vez, el usuario identificado —siempre tenemos esta dependencia, ya la vamos a ver—; creamos la tarea, insertamos en la base de datos. Como dependencia aquí también tenemos el get_db —por aquí lo puedes ver, ahí lo tienes como dependencia—. Al igual que el de obtener todas las tareas: simplemente es una petición para nuestra base de datos, con el query correspondiente, de obtener todas las tareas del usuario, obviamente con el current_user, aquí lo puedes ver. Al igual que obtener una tarea en particular, o para actualizarla, o eliminarla: es exactamente lo mismo que vemos en el curso básico o en el libro, así que no lo voy a indicar mucho más.

task-service/app/routes_tasks.py

def find_owned(task_id: int, current_user: dict[str, Any], db: Session) -> Task:
    task = db.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    if task.user_id != current_user["id"]:
        # El 403 va antes que cualquier error interno: no existe para ti.
        raise HTTPException(status_code=403, detail="No es tu tarea")
    return task

@router.post("/tasks", response_model=TaskOut, status_code=201)
def create_task(
    payload: TaskCreate,
    current_user: dict[str, Any] = Depends(get_current_user),
    db: Session = Depends(get_db),
) -> Task:
    task = Task(
        title=payload.title,
        status=normalize_status(payload.status, STATUS_PENDING),
        # El dueño lo pone el sistema: el cliente no manda user_id.
        user_id=current_user["id"],
    )
    db.add(task)
    db.commit()
    db.refresh(task)
    return task

@router.get("/tasks", response_model=list[TaskOut])
def list_tasks(
    current_user: dict[str, Any] = Depends(get_current_user),
    db: Session = Depends(get_db),
) -> list[Task]:
    return list(
        db.query(Task)
        .filter(Task.user_id == current_user["id"])
        .order_by(Task.id.desc())
        .all()
    )

(show_task, update_task y delete_task tienen la misma forma: siempre Depends(get_current_user) + Depends(get_db), y find_owned al frente.)

@router.get("/tasks/{task_id}", response_model=TaskOut)
def show_task(
    task_id: int,
    current_user: dict[str, Any] = Depends(get_current_user),
    db: Session = Depends(get_db),
) -> Task:
    return find_owned(task_id, current_user, db)

deps.py: la credencial en la cabecera

Vamos con la parte del current_user; fíjate que aquí tenemos una parte más interesante. Aquí tenemos esto para la cabecera; recuerda que, por aquí —por ejemplo, esta sería la de usuarios—, cuando vamos a hacer una petición, tenemos que configurar lo que es la cabecera. Para tener aquí la parte de las credenciales: hay como cuatro formas —en la documentación automática—, pero esta es la que estoy utilizando, porque sí, porque patata. También, aquí la autenticación que estamos teniendo es muy básica; más adelante la vamos a mejorar con un JWT, pero de a poco.

Entonces, ¿qué es lo que tenemos aquí? Recuerda que estamos todavía en la aplicación de tareas. Aquí obtenemos el token: las credenciales, pues otra vez, el token que colocamos aquí —de igual manera, después lo puedes ver en el curl, para saber exactamente cómo lo está pasando, o cómo lo tienes que pasar, desde, por ejemplo, una aplicación en Vue—. Aquí lo estoy imprimiendo para que lo veas.

task-service/app/deps.py

from typing import Any

from fastapi import Header, HTTPException

from app.user_service_client import verify_token

def get_current_user(
    authorization: str | None = Header(default=None),
) -> dict[str, Any]:
    parts = authorization.split(" ", 1) if authorization else []
    if len(parts) != 2 or parts[0] != "Bearer" or not parts[1].strip():
        # Header ausente o malformado: problema del cliente, se rechaza aquí
        # sin llamar a la red.
        raise HTTPException(status_code=401, detail="Token no presentado o malformado")
    return verify_token(parts[1].strip())

verify_token: la petición HTTP al user-service

Aquí tenemos uno llamado verify_token. Otra vez: seguimos aquí en tareas. Y fíjate aquí lo que estamos haciendo: una petición a la aplicación de usuarios. De momento no nos importa qué es lo que está haciendo, pero obviamente, por contexto, ya te lo puedes esperar: aquí tenemos la respuesta, y la respuesta —si todo va OK— es el usuario, que es lo que finalmente manejamos desde nuestra aplicación. Entonces, aquí puedes ver un poquito el sentido de que, por ejemplo, cuando estamos preguntando por el id y demás, es porque lo estamos obteniendo mediante una petición HTTP —a secas, el sistema más simple posible; lo «bueno», como HTTPS o firmas, viene después—. Por lo demás, es simplemente una petición HTTP que estamos haciendo. Como te digo, hay formas más eficientes —por ejemplo, al final haremos gRPC de la manera correcta—, pero de momento, así, sencillito, para que se entienda el ejercicio. Y aquí lo que yo quiero que entiendas es básicamente eso: un microservicio es eso, comunicarse, de cierta forma y con ciertos estándares, con otro servicio. Y aquí el problema que vamos a tener, sobre todo, es la latencia.

Bueno, aquí son validaciones típicas: si se devuelve 200, entonces tenemos el usuario; y si no, bueno: token inválido o expirado. Y eso es todo lo que tenemos.

task-service/app/user_service_client.py

def verify_token(token: str) -> dict[str, Any]:
    """¿Quién es este token? Toda la frontera de red vive en este archivo:
    los demás módulos te mandan el dict del usuario como si fuera local."""
    try:
        resp = _http_get(
            f"{config.USER_SERVICE_URL}/api/user",
            headers={"Authorization": f"Bearer {token}"},
        )
    except (httpx2.ConnectError, httpx2.ConnectTimeout):
        # Es un fallo de infraestructura, no de identidad: el cliente merece
        # saber qué servicio falta y cómo levantarlo (503, no `401` genérico).
        raise HTTPException(
            status_code=503,
            detail=(
                "user-service no está en pie (se esperaba "
                f"{config.USER_SERVICE_URL}); levántalo con "
                "`uv run uvicorn app.main:app --port 8000` en user-service/"
            ),
        )

    if resp.status_code == 200:
        user = resp.json()
        # El cliente del IdP devuelve {id, name, email}; nada más entra aquí.
        return {"id": int(user["id"]), "name": str(user["name"]), "email": str(user["email"])}

    if resp.status_code == 401:
        # La distinción que importa: `401` NO es un fallo del servicio, es
        # identidad inválida del token presentado.
        raise HTTPException(status_code=401, detail="Token inválido o expirado")

    raise HTTPException(
        status_code=502,
        detail=f"Respuesta inesperada del user-service (HTTP {resp.status_code})",
    )

(Este es el cliente con el timeout del bloque 7: httpx2.Client(timeout=config.HTTP_TIMEOUT_SECONDS); si la petición tarda más de dos segundos, se desecha.)

deps ↔ user_service_client: modularidad

Aquí lo que te recomiendo es: copia el nombre de la función y, con Ctrl+Shift+F, busca; puedes ver exactamente dónde se está referenciando. En este caso, el de deps es el que llama a todo lo que acabaste de ver, de la conexión con el servicio de la autenticación y demás. Otra vez, aquí puedes ver, por ejemplo, verify_token, dónde se utiliza: se está utilizando desde user_service_client, que lo tenemos por aquí. Entonces, el de deps lo que hace es llamar a este módulo; es así de simple, para tener todo un poco más modular. Así que, como te digo, estudia un poquito esto, pero a la final lo que hacemos es resolver un usuario que vive en otra aplicación.

user-service: la semilla y security (hash y token)

Hay muchas formas de hacer esto, así que no le des muchas vueltas. Son funciones utilitarias, obviamente, para manipular el token —esto sería para comparar el token: el hash_token—; son funciones de ese tipo. Así que ya creo que se entiende un poquito ahí.

user-service/app/seed.py

"""Usuarios demo de la primera parte (los mismos de la serie Laravel):

    uv run python -m app.seed
"""

from sqlalchemy.orm import Session

from app import security
from app.db import Base, engine, SessionLocal
from app.models import User

DEMO_USERS = [
    {"name": "Ana Torres", "email": "ana.torres@example.com", "password": "password"},
    {"name": "Luis Ramirez", "email": "luis.ramirez@example.com", "password": "password"},
]

def run() -> None:
    Base.metadata.create_all(engine)
    db: Session = SessionLocal()
    try:
        for spec in DEMO_USERS:
            exists = db.query(User).filter(User.email == spec["email"]).first()
            if exists:
                continue
            db.add(
                User(
                    name=spec["name"],
                    email=spec["email"],
                    password=security.hash_password(spec["password"]),
                )
            )
        db.commit()
        users = db.query(User).count()
        print(f"{users} usuario(s) en la base de datos")
    finally:
        db.close()

if __name__ == "__main__":
    run()

user-service/app/security.py

import hashlib
import hmac
import secrets
from datetime import UTC, datetime, timedelta

def hash_password(password: str) -> str:
    # PBKDF2 de la biblioteca estándar: sin dependencias, determinista.
    salt = secrets.token_hex(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 100_000).hex()
    return f"pbkdf2_sha256$100000${salt}${digest}"

def verify_password(password: str, stored: str) -> bool:
    try:
        _algo, _iters, salt, digest = stored.split("$")
    except ValueError:
        return False
    candidate = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 100_000).hex()
    # Comparación en tiempo constante: nunca `==` directo.
    return hmac.compare_digest(candidate, digest)

def issue_token() -> tuple[str, str]:
    """Devuelve (token crudo, hash para guardar). El crudo viaja una sola vez."""
    token = secrets.token_hex(32)
    return token, hashlib.sha256(token.encode()).hexdigest()

def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


def expiry(now: datetime | None = None) -> datetime:
    from app import config

    # Naive UTC para la columna; el route lo vuelve aware al responder.
    return (now or models_now_naive()) + timedelta(minutes=config.TOKEN_TTL_MINUTES)

Los esquemas de usuarios

Ahí, el esquema… volvemos aquí… por seguridad: el esquema que tenemos, muy normal, para el token y el usuario, que son los modelos de Pydantic.

user-service/app/schemas.py

from datetime import datetime

from pydantic import BaseModel, ConfigDict, EmailStr, Field

class TokenRequest(BaseModel):
    email: EmailStr
    password: str = Field(min_length=1)
    device_name: str | None = Field(default="api-token", max_length=100)

class UserInfo(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    email: EmailStr

class TokenIssued(BaseModel):
    token: str
    token_type: str
    expires_at: datetime
    user: UserInfo

El modelo de usuarios y el token (FK, hash, caducidad)

Y el modelo, lo típico: el usuario, y un modelo completamente personalizado por nosotros, sin ningún estándar ni nada por el estilo: el usuario, el hash del token, el nombre y fechas de expiración y creación. Y el user_id, claro está: en este caso es de tipo relacional, para tener la FK en la BD. Nuevamente, aquí tenemos una base de datos completamente independiente.

user-service/app/models.py

from datetime import UTC, datetime

from sqlalchemy import DateTime, ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db import Base

def utcnow() -> datetime:
    # Naive en UTC: sqlite no conserva tzinfo; toda la BD vive en UTC-naive.
    return datetime.now(UTC).replace(tzinfo=None)

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    password: Mapped[str] = mapped_column(String(255))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)

class ApiToken(Base):
    __tablename__ = "api_tokens"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
    # Token opaco: se guarda solo el hash, el crudo sale una vez del login.
    token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(100), default="api-token")
    expires_at: Mapped[datetime] = mapped_column(DateTime)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)

    user: Mapped[User] = relationship(User)

La BD, la configuración y el main de usuarios

Nuevamente, aquí tienes la base de datos: la misma conexión que antes; de aquí no cambia nada. Las configuraciones: un poco lo mismo —esto sería la duración del token, cuánto queremos que dure, que sería una hora—. Y en el main tenemos, para arrancar la aplicación: se crean los modelos, las rutas…

user-service/app/db.py (idéntico en la forma al de task-service)

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app import config

class Base(DeclarativeBase):
    pass

def _make_engine(url: str) -> "create_engine":  # type: ignore[valid-type]
    kwargs = {"connect_args": {"check_same_thread": False}}
    if url.endswith("//:memory:"):
        from sqlalchemy.pool import StaticPool

        kwargs["poolclass"] = StaticPool
    return create_engine(url, **kwargs)


engine = _make_engine(config.DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Las rutas del IdP: login, /user y revocar

Las rutas: obtener el usuario; eliminar el token —que sería otra vez la conexión a la base de datos: obtener el token de autorización, que se lo tenemos que pasar, y pues nada, lo eliminamos; bueno, en este caso lo hacemos aquí localmente—. Y el login: verificamos el usuario, muy normal; obtenemos el password en texto plano, lo verificamos con el hash en la base de datos; si son iguales, perfecto. Y el token: el crudo se devuelve una sola vez —a la base de datos llega solo su hash—. El authenticate, que es lo que usan /api/user y el DELETE, busca el hash del token en la BD y comprueba que no haya caducado.

user-service/app/routes_tokens.py

@router.post("/tokens", response_model=TokenIssued, status_code=201)
def issue_token(
    payload: TokenRequest,
    db: Session = Depends(get_db),
) -> TokenIssued:
    """Inicio de sesión, una vez: email + contraseña → token opaco que caduca."""
    user = db.query(User).filter(User.email == payload.email).first()

    # Mismo mensaje y mismo status para "no existe" y "contraseña mala": si no,
    # esta respuesta sirve para enumerar qué emails están dados de alta.
    if user is None or not security.verify_password(payload.password, user.password):
        raise HTTPException(status_code=401, detail="Credenciales inválidas")

    token, token_hash = security.issue_token()
    expires_at = security.expiry()
    row = ApiToken(
        user_id=user.id,
        token_hash=token_hash,
        name=payload.device_name or "api-token",
        expires_at=expires_at,
    )
    db.add(row)
    db.commit()

    return TokenIssued(
        token=token,
        token_type="Bearer",
        expires_at=expires_at.replace(tzinfo=UTC),
        user=UserInfo.model_validate(user),
    )

@router.get("/user", response_model=UserInfo)
def whoami(user: User = Depends(require_user)) -> UserInfo:
    return UserInfo.model_validate(user)

@router.delete("/tokens", status_code=204)
def revoke(
    authorization: str | None = Header(default=None),
    db: Session = Depends(get_db),
) -> None:
    """Revoca el token con el que se hizo el request: borrar la fila es una
    revocación real e inmediata (el trade-off del token opaco)."""
    from app.auth import authenticate

    user, row = authenticate(db, authorization)
    db.query(ApiToken).filter(ApiToken.id == row.id).delete()
    db.commit()

user-service/app/auth.py

def authenticate(db: Session, authorization: str | None) -> tuple[User, ApiToken]:
    """¿Quién es este token? La respuesta es `401` ante token inválido o expirado."""
    token = _bearer_token(authorization)
    if token is None:
        raise HTTPException(status_code=401, detail="Token no presentado o malformado")

    row = db.query(ApiToken).filter(ApiToken.token_hash == security.hash_token(token)).first()
    if row is None:
        raise HTTPException(status_code=401, detail="Token inválido o expirado")

    # Token caducado: para el cliente es idéntico a uno inexistente.
    if row.expires_at <= utcnow():
        raise HTTPException(status_code=401, detail="Token inválido o expirado")

    return row.user, row

Métodos internos y pruebas unitarias

Estos métodos son empleados internamente; ya todo ello lo puedes buscar aquí. Y puedes ver que, por ejemplo, con los fixtures —que es para generar los datos, para generar el usuario de pruebas—, aquí lo estamos utilizando, porque no tengo un método de registro, entonces lo generé por ahí. Pero son métodos muy sencillos. También tenemos pruebas unitarias, que las pasé por encima para no alargar esto de más: 11 en el task-service y 7 en el user-service (en la etiqueta 0.1).

Conclusiones y la definición de microservicio

Así que lo que quiero que te quede, en este punto, son dos cosas. Una: aquí, simplemente, en la aplicación de usuarios manejamos usuarios, y en la de tareas, tareas. Pero, por lógica de negocio, tenemos que conectarlas, y para eso utilizamos conexiones HTTP, de momento, para conectar ambas aplicaciones. Y eso es simplemente eso.

Y por la temática, o el cómo están conformadas ambas aplicaciones —ambos servicios—, ya los convierte en una arquitectura de microservicios, ya que sería lo mínimo necesario para que puedan funcionar y estén conectadas. Y así, un poquito, la definición de microservicio que no le he dado hasta este punto: es simplemente un estilo de arquitectura —de desarrollo de software— en el cual una aplicación completa, que es lo que tenemos aquí —una aplicación de tareas con usuario identificado—, se divide en pequeños servicios independientes y autónomos. Esto de «autónomos» es importante, porque cuando se cae una —por ejemplo, el usuario—, la de tareas, por más que sea, debería poder hacer algo; ahorita no va a hacer ni media papa, por lo tanto, porque la respuesta HTTP nunca la va a devolver, y por lo tanto la de tareas siempre va a rechazar las peticiones. Pero vamos a eso.

Entonces, cada uno de estos servicios cumple una función específica de negocio, lo comentado desde el inicio: por eso tenemos el de usuario —gestión de usuarios— y también la parte de tareas; en nuestro caso, tareas. Pero obviamente podemos anexar más microservicios, y a la final, eso es, como quien dice, lo bonito de la teoría —como el socialismo: muy bonita—, pero la práctica es un desastre, en la cual tenemos una aplicación que hace determinadas cosas… y lo comentado antes: ya puedes ver que la cosa se va a ir enredando bastante, porque esas conexiones son bastante necesarias. Por lo tanto, a la final, el problema que tenemos con los microservicios, conforme vamos avanzando, es más que nada la conexión: nos estamos preocupando más en mantener la arquitectura de microservicios que en implementar la aplicación en sí, la lógica de negocio y lo demás, que es lo que el usuario necesita.