Alberto Méndez

API REST con FastAPI, PostgreSQL y Docker: Guía definitiva con código fuente

sept. 15, 2025

En el ecosistema actual de desarrollo backend, FastAPI se ha posicionado como una de las opciones más populares para crear APIs REST modernas y eficientes. Combina la simplicidad de Python con un rendimiento excepcional, documentación automática y tipado estático que mejora significativamente la experiencia de desarrollo.

A diferencia de frameworks tradicionales como Flask o Django REST, FastAPI aprovecha las características modernas de Python como type hints y programación asíncrona para ofrecer mejor rendimiento, detección de errores en tiempo de desarrollo y documentación automática con Swagger UI. Esto se traduce en menos bugs, desarrollo más rápido y APIs más robustas.

En este tutorial construiremos una aplicación completa de gestión de tareas (Task Manager) que incluye registro de usuarios, autenticación JWT, operaciones CRUD y deployment en producción. El proyecto será completamente funcional y servirá como base sólida para APIs más complejas.

Utilizaremos un stack moderno y probado: FastAPI como framework principal, PostgreSQL como base de datos, SQLAlchemy como ORM, Docker para containerización y pytest para testing.

¿Por qué FastAPI para APIs modernas?

FastAPI ha ganado popularidad rápidamente en la comunidad de Python por varias razones fundamentales que lo distinguen de otros frameworks:

Rendimiento excepcional

FastAPI está construido sobre Starlette para el manejo de requests y Pydantic para validación de datos, lo que le permite alcanzar rendimientos comparables a frameworks de Go o Node.js. En benchmarks independientes, FastAPI supera consistentemente a Django REST Framework y se acerca al rendimiento de FastHTTP en Go.

Documentación automática

Una de las características más apreciadas es la generación automática de documentación interactiva. FastAPI crea automáticamente documentación Swagger UI y ReDoc basándose en los type hints de Python, eliminando la necesidad de mantener documentación por separado.

Desarrollo moderno con type hints

El uso extensivo de tipado estático permite que IDEs como PyCharm o VS Code ofrezcan autocompletado inteligente, detección de errores en tiempo real y refactoring seguro. Esto reduce significativamente los bugs y mejora la productividad del equipo.

Compatibilidad asíncrona nativa

FastAPI soporta tanto funciones síncronas como asíncronas de forma transparente, permitiendo aprovechar al máximo las ventajas de la programación asíncrona cuando es necesario sin complicar el código.

El rendimiento superior de FastAPI se debe a su arquitectura basada en ASGI y el uso de librerías optimizadas como Starlette y Pydantic, lo que resulta en una experiencia de desarrollo más ágil y aplicaciones más rápidas.

Preparando el entorno de desarrollo

Antes de comenzar con el código, necesitamos configurar un entorno de desarrollo consistente y reproducible. Utilizaremos Docker para garantizar que el entorno funcione igual en todos los sistemas.

Estructura del proyecto

La organización del código es fundamental para mantener un proyecto escalable y fácil de mantener. FastAPI no impone una estructura específica, pero seguiremos las mejores prácticas de la comunidad que separan responsabilidades de forma clara:

Crearemos una estructura organizada que siga las mejores prácticas para proyectos FastAPI:


    mkdir task-manager-api
    cd task-manager-api

    # Estructura de directorios
    mkdir -p {app/{api,core,db,models,schemas,services},tests,scripts}

    # Archivos de configuración
    touch {requirements.txt,Dockerfile,docker-compose.yml,.env.example,.gitignore}

La estructura final quedará así:

task-manager-api/
├── app/
│   ├── __init__.py
│   ├── main.py              # Punto de entrada de la aplicación
│   ├── api/                 # Endpoints y rutas
│   │   ├── __init__.py
│   │   ├── deps.py          # Dependencias compartidas
│   │   └── v1/              # Versión 1 de la API
│   │       ├── __init__.py
│   │       ├── auth.py      # Endpoints de autenticación
│   │       └── tasks.py     # Endpoints de tareas
│   ├── core/                # Configuración y utilidades
│   │   ├── __init__.py
│   │   ├── config.py        # Variables de entorno
│   │   └── security.py      # JWT y hashing
│   ├── db/                  # Base de datos
│   │   ├── __init__.py
│   │   ├── database.py      # Conexión a BD
│   │   └── base.py          # Base para modelos
│   ├── models/              # Modelos SQLAlchemy
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── task.py
│   ├── schemas/             # Modelos Pydantic
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── task.py
│   └── services/            # Lógica de negocio
│       ├── __init__.py
│       ├── auth_service.py
│       └── task_service.py
├── tests/                   # Tests automatizados
├── scripts/                 # Scripts de utilidad
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
└── .env.example

Dependencias del proyecto

Es importante entender qué hace cada dependencia en nuestro proyecto. Cada librería tiene un propósito específico:

  • FastAPI: El framework principal para crear la API
  • Uvicorn: Servidor ASGI para ejecutar FastAPI en producción
  • Pydantic: Validación de datos y serialización con type hints
  • SQLAlchemy: ORM para interactuar con PostgreSQL de forma pythónica
  • psycopg2-binary: Driver de PostgreSQL para Python
  • python-jose: Librería para manejar tokens JWT
  • passlib: Hashing seguro de contraseñas con bcrypt
  • pytest: Framework de testing con excelente soporte para FastAPI

Definiremos las dependencias necesarias en requirements.txt:


    # requirements.txt
    fastapi==0.104.1
    uvicorn[standard]==0.24.0
    pydantic==2.5.0
    pydantic-settings==2.1.0
    sqlalchemy==2.0.23
    psycopg2-binary==2.9.9
    alembic==1.13.1
    python-jose[cryptography]==3.3.0
    python-multipart==0.0.6
    passlib[bcrypt]==1.7.4
    pytest==7.4.3
    pytest-asyncio==0.21.1
    httpx==0.25.2

Configuración con Docker

Docker nos permitirá crear un entorno consistente y reproducible. El Dockerfile que crearemos está optimizado para desarrollo, con recarga automática de cambios y volúmenes montados para facilitar el debugging:

Crearemos un Dockerfile optimizado para desarrollo:


    # Dockerfile
    FROM python:3.11-slim

    WORKDIR /app

    # Instalar dependencias del sistema
    RUN apt-get update && apt-get install -y \
        gcc \
        && rm -rf /var/lib/apt/lists/*

    # Copiar requirements y instalar dependencias Python
    COPY requirements.txt .
    RUN pip install --no-cache-dir -r requirements.txt

    # Copiar código de la aplicación
    COPY ./app ./app

    # Exponer puerto
    EXPOSE 8000

    # Comando por defecto
    CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]
    ```

    Y el archivo `docker-compose.yml` para desarrollo local:

    ```yaml
    # docker-compose.yml
    version: '3.8'

    services:
    api:
        build: .
        ports:
        - "8000:8000"
        volumes:
        - ./app:/app/app
        - ./tests:/app/tests
        environment:
        - DATABASE_URL=postgresql://postgres:postgres@db:5432/taskmanager
        - SECRET_KEY=dev-secret-key-change-in-production
        depends_on:
        - db
        command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

    db:
        image: postgres:15-alpine
        environment:
        - POSTGRES_USER=postgres
        - POSTGRES_PASSWORD=postgres
        - POSTGRES_DB=taskmanager
        ports:
        - "5432:5432"
        volumes:
        - postgres_data:/var/lib/postgresql/data

    volumes:
    postgres_data:

Configuración y modelos de datos

Una configuración robusta es la base de cualquier aplicación. Utilizaremos Pydantic Settings que nos da validación automática de variables de entorno, type safety y manejo de errores claro cuando falta configuración.

Sistema de configuración

Pydantic Settings nos permite:

  • Validación automática de tipos en variables de entorno
  • Valores por defecto seguros
  • Documentación automática de la configuración requerida
  • Integración perfecta con el sistema de tipos de Python

Utilizaremos Pydantic Settings para manejar la configuración de forma type-safe:


    # app/core/config.py
    from pydantic_settings import BaseSettings
    from pydantic import PostgresDsn

    class Settings(BaseSettings):
        # Configuración de la aplicación
        PROJECT_NAME: str = "Task Manager API"
        PROJECT_VERSION: str = "1.0.0"
        API_V1_STR: str = "/api/v1"
        
        # Configuración de seguridad
        SECRET_KEY: str
        ALGORITHM: str = "HS256"
        ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
        
        # Configuración de base de datos
        DATABASE_URL: PostgresDsn
        
        # Configuración de CORS
        BACKEND_CORS_ORIGINS: list[str] = ["http://localhost:3000", "http://localhost:8080"]
        
        class Config:
            env_file = ".env"
            case_sensitive = True

    settings = Settings()

Conexión a la base de datos

SQLAlchemy es el ORM más maduro y potente de Python. Aunque FastAPI soporta ORMs asíncronos como SQLModel, SQLAlchemy tradicional sigue siendo una excelente opción por su estabilidad y ecosistema.

La configuración que implementaremos utiliza:

  • Engine: Gestiona las conexiones a la base de datos
  • SessionLocal: Factory para crear sesiones de base de datos
  • Base: Clase base para todos nuestros modelos
  • get_db(): Dependencia que proporciona sesiones de BD a nuestros endpoints

Configuraremos SQLAlchemy:


    # app/db/database.py
    from sqlalchemy import create_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker
    from app.core.config import settings

    # Crear engine de SQLAlchemy
    engine = create_engine(str(settings.DATABASE_URL))

    # Crear sesión local
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    # Base para modelos
    Base = declarative_base()

    # Dependencia para obtener sesión de BD
    def get_db():
        db = SessionLocal()
        try:
            yield db
        finally:
            db.close()

Modelos de datos

Los modelos SQLAlchemy representan las tablas de nuestra base de datos. Cada modelo define:

  • Estructura de la tabla: Columnas, tipos de datos, constrains
  • Relaciones: Cómo se conectan las tablas entre sí
  • Índices: Para optimizar consultas frecuentes
  • Metadatos: Información adicional como timestamps automáticos

Nuestro modelo de User incluye campos esenciales para autenticación y auditoría, mientras que Task implementa un sistema completo de gestión con prioridades, fechas límite y estados.

Definiremos los modelos SQLAlchemy para usuarios y tareas:


    # app/models/user.py
    from sqlalchemy import Column, Integer, String, Boolean, DateTime
    from sqlalchemy.orm import relationship
    from sqlalchemy.sql import func
    from app.db.database import Base

    class User(Base):
        __tablename__ = "users"
        
        id = Column(Integer, primary_key=True, index=True)
        email = Column(String, unique=True, index=True, nullable=False)
        username = Column(String, unique=True, index=True, nullable=False)
        hashed_password = Column(String, nullable=False)
        is_active = Column(Boolean, default=True)
        created_at = Column(DateTime(timezone=True), server_default=func.now())
        updated_at = Column(DateTime(timezone=True), onupdate=func.now())
        
        # Relación con tareas
        tasks = relationship("Task", back_populates="owner", cascade="all, delete-orphan")
    


    # app/models/task.py
    from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey, Text
    from sqlalchemy.orm import relationship
    from sqlalchemy.sql import func
    from app.db.database import Base

    class Task(Base):
        __tablename__ = "tasks"
        
        id = Column(Integer, primary_key=True, index=True)
        title = Column(String, nullable=False, index=True)
        description = Column(Text, nullable=True)
        is_completed = Column(Boolean, default=False)
        priority = Column(String, default="medium")  # low, medium, high
        due_date = Column(DateTime(timezone=True), nullable=True)
        created_at = Column(DateTime(timezone=True), server_default=func.now())
        updated_at = Column(DateTime(timezone=True), onupdate=func.now())
        
        # Clave foránea al usuario
        owner_id = Column(Integer, ForeignKey("users.id"), nullable=False)
        
        # Relación con usuario
        owner = relationship("User", back_populates="tasks")

Esquemas Pydantic

Los esquemas Pydantic son el corazón de la validación en FastAPI. Separan la lógica de presentación de la lógica de persistencia:

  • Schemas de input: Validan datos que llegan del cliente
  • Schemas de output: Controlan qué datos se devuelven al cliente
  • Schemas de actualización: Permiten actualizaciones parciales con campos opcionales

Esta separación nos da:

  • Seguridad: Nunca exponemos campos sensibles como passwords
  • Flexibilidad: Diferentes vistas de los mismos datos según el contexto
  • Validación: Automática en requests y responses
  • Documentación: Swagger UI se genera automáticamente

Los esquemas Pydantic definen la estructura de datos para requests y responses:


    # app/schemas/user.py
    from pydantic import BaseModel, EmailStr
    from datetime import datetime
    from typing import Optional

    # Schema base para usuario
    class UserBase(BaseModel):
        email: EmailStr
        username: str

    # Schema para crear usuario
    class UserCreate(UserBase):
        password: str

    # Schema para actualizar usuario
    class UserUpdate(BaseModel):
        email: Optional[EmailStr] = None
        username: Optional[str] = None
        password: Optional[str] = None

    # Schema para respuestas (sin password)
    class User(UserBase):
        id: int
        is_active: bool
        created_at: datetime
        
        class Config:
            from_attributes = True

    # Schema para login
    class UserLogin(BaseModel):
        username: str
        password: str

    # Schema para token
    class Token(BaseModel):
        access_token: str
        token_type: str = "bearer"
        
    class TokenData(BaseModel):
        username: Optional[str] = None
    


    # app/schemas/task.py
    from pydantic import BaseModel
    from datetime import datetime
    from typing import Optional

    # Schema base para tarea
    class TaskBase(BaseModel):
        title: str
        description: Optional[str] = None
        priority: str = "medium"
        due_date: Optional[datetime] = None

    # Schema para crear tarea
    class TaskCreate(TaskBase):
        pass

    # Schema para actualizar tarea
    class TaskUpdate(BaseModel):
        title: Optional[str] = None
        description: Optional[str] = None
        is_completed: Optional[bool] = None
        priority: Optional[str] = None
        due_date: Optional[datetime] = None

    # Schema para respuestas
    class Task(TaskBase):
        id: int
        is_completed: bool
        created_at: datetime
        updated_at: Optional[datetime] = None
        owner_id: int
        
        class Config:
            from_attributes = True

    # Schema para tareas con información del propietario
    class TaskWithOwner(Task):
        owner: "User"

Implementando autenticación JWT

La autenticación es uno de los aspectos más críticos de cualquier API. JWT (JSON Web Tokens) es el estándar de facto para APIs modernas porque:

  • Stateless: No requiere almacenar sesiones en el servidor
  • Escalable: Perfecto para microservicios y arquitecturas distribuidas
  • Seguro: Firmado criptográficamente para prevenir manipulación
  • Flexible: Puede incluir claims personalizados según necesidades

Nuestro sistema implementará:

  • Hashing seguro de contraseñas con bcrypt
  • Generación y verificación de tokens JWT
  • Middleware de autenticación para proteger endpoints
  • Manejo de expiración y renovación de tokens

Implementaremos un sistema robusto usando JWT tokens:

Sistema de seguridad

Este módulo centraliza toda la lógica criptográfica. bcrypt es considerado el gold standard para hashing de passwords por su resistencia a ataques de fuerza bruta y rainbow tables. Los tokens JWT incluyen un timestamp de expiración para limitar la ventana de exposición en caso de compromiso.


    # app/core/security.py
    from datetime import datetime, timedelta
    from typing import Optional
    from jose import JWTError, jwt
    from passlib.context import CryptContext
    from app.core.config import settings

    # Configuración para hashing de passwords
    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

    def verify_password(plain_password: str, hashed_password: str) -> bool:
        """Verificar password contra hash"""
        return pwd_context.verify(plain_password, hashed_password)

    def get_password_hash(password: str) -> str:
        """Generar hash de password"""
        return pwd_context.hash(password)

    def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
        """Crear JWT token"""
        to_encode = data.copy()
        
        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
        
        to_encode.update({"exp": expire})
        encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
        return encoded_jwt

    def verify_token(token: str) -> Optional[str]:
        """Verificar y extraer username del token"""
        try:
            payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
            username: str = payload.get("sub")
            if username is None:
                return None
            return username
        except JWTError:
            return None

Servicio de autenticación

El patrón Service Layer separa la lógica de negocio de los controllers (endpoints). Este servicio encapsula todas las operaciones relacionadas con usuarios:

  • Validación de credenciales
  • Creación segura de usuarios
  • Verificación de duplicados
  • Gestión del ciclo de vida de usuarios

Esto facilita testing, reutilización y mantenimiento del código.


    # app/services/auth_service.py
    from sqlalchemy.orm import Session
    from app.models.user import User
    from app.schemas.user import UserCreate
    from app.core.security import verify_password, get_password_hash
    from typing import Optional

    class AuthService:
        
        @staticmethod
        def get_user_by_username(db: Session, username: str) -> Optional[User]:
            """Obtener usuario por username"""
            return db.query(User).filter(User.username == username).first()
        
        @staticmethod
        def get_user_by_email(db: Session, email: str) -> Optional[User]:
            """Obtener usuario por email"""
            return db.query(User).filter(User.email == email).first()
        
        @staticmethod
        def authenticate_user(db: Session, username: str, password: str) -> Optional[User]:
            """Autenticar usuario con username y password"""
            user = AuthService.get_user_by_username(db, username)
            if not user:
                return None
            if not verify_password(password, user.hashed_password):
                return None
            return user
        
        @staticmethod
        def create_user(db: Session, user: UserCreate) -> User:
            """Crear nuevo usuario"""
            hashed_password = get_password_hash(user.password)
            db_user = User(
                username=user.username,
                email=user.email,
                hashed_password=hashed_password
            )
            db.add(db_user)
            db.commit()
            db.refresh(db_user)
            return db_user
        
        @staticmethod
        def is_username_taken(db: Session, username: str) -> bool:
            """Verificar si username ya existe"""
            return db.query(User).filter(User.username == username).first() is not None
        
        @staticmethod
        def is_email_taken(db: Session, email: str) -> bool:
            """Verificar si email ya existe"""
            return db.query(User).filter(User.email == email).first() is not None

Dependencias de autenticación

FastAPI usa un sistema de inyección de dependencias muy potente. Estas funciones:

  • Se ejecutan automáticamente antes de los endpoints
  • Pueden ser reutilizadas en múltiples endpoints
  • Manejan la extracción y validación de tokens
  • Proporcionan usuarios autenticados a nuestros endpoints

HTTPBearer maneja automáticamente la extracción del token del header Authorization.


    # app/api/deps.py
    from fastapi import Depends, HTTPException, status
    from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
    from sqlalchemy.orm import Session
    from app.db.database import get_db
    from app.core.security import verify_token
    from app.services.auth_service import AuthService
    from app.models.user import User

    # Configurar esquema de seguridad
    security = HTTPBearer()

    async def get_current_user(
        credentials: HTTPAuthorizationCredentials = Depends(security),
        db: Session = Depends(get_db)
    ) -> User:
        """Obtener usuario actual desde JWT token"""
        
        credentials_exception = HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
        
        # Verificar token
        username = verify_token(credentials.credentials)
        if username is None:
            raise credentials_exception
        
        # Obtener usuario de BD
        user = AuthService.get_user_by_username(db, username)
        if user is None:
            raise credentials_exception
            
        if not user.is_active:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Inactive user"
            )
        
        return user

    async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
        """Obtener usuario actual activo"""
        if not current_user.is_active:
            raise HTTPException(status_code=400, detail="Inactive user")
        return current_user

Creando endpoints de la API

Los endpoints son la interfaz pública de nuestra API. FastAPI hace que definir endpoints sea intuitivo y potente:

  • Decoradores: Definen el método HTTP y ruta
  • Type hints: Automatizan validación y documentación
  • Dependency injection: Proporciona recursos como BD y usuarios autenticados
  • Response models: Controlan exactamente qué datos se devuelven

Endpoints de autenticación

Los endpoints de autenticación manejan el ciclo completo de gestión de usuarios. OAuth2PasswordRequestForm es el estándar OAuth2 para login con username/password.


    # app/api/v1/auth.py
    from datetime import timedelta
    from fastapi import APIRouter, Depends, HTTPException, status
    from fastapi.security import OAuth2PasswordRequestForm
    from sqlalchemy.orm import Session

    from app.api.deps import get_db
    from app.core.config import settings
    from app.core.security import create_access_token
    from app.schemas.user import UserCreate, User, Token
    from app.services.auth_service import AuthService

    router = APIRouter()

    @router.post("/register", response_model=User, status_code=status.HTTP_201_CREATED)
    def register_user(user: UserCreate, db: Session = Depends(get_db)):
        """Registrar nuevo usuario"""
        
        # Verificar si username ya existe
        if AuthService.is_username_taken(db, user.username):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Username already registered"
            )
        
        # Verificar si email ya existe
        if AuthService.is_email_taken(db, user.email):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Email already registered"
            )
        
        # Crear usuario
        return AuthService.create_user(db=db, user=user)

    @router.post("/login", response_model=Token)
    def login_user(
        form_data: OAuth2PasswordRequestForm = Depends(),
        db: Session = Depends(get_db)
    ):
        """Login de usuario"""
        
        # Autenticar usuario
        user = AuthService.authenticate_user(db, form_data.username, form_data.password)
        if not user:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Incorrect username or password",
                headers={"WWW-Authenticate": "Bearer"},
            )
        
        # Crear token de acceso
        access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(
            data={"sub": user.username}, expires_delta=access_token_expires
        )
        
        return {"access_token": access_token, "token_type": "bearer"}

    @router.get("/me", response_model=User)
    def read_users_me(current_user: User = Depends(get_current_active_user)):
        """Obtener información del usuario actual"""
        return current_user

Endpoints de tareas

Este conjunto de endpoints implementa un CRUD completo con características avanzadas:

  • Paginación: Para manejar grandes volúmenes de datos
  • Filtrado: Por estado de completado y prioridad
  • Aislamiento de datos: Cada usuario solo ve sus tareas
  • Operaciones atómicas: Como toggle para cambiar estado

La validación de parámetros con Query() proporciona documentación automática y validación de tipos.


    # app/api/v1/tasks.py
    from typing import List
    from fastapi import APIRouter, Depends, HTTPException, status, Query
    from sqlalchemy.orm import Session

    from app.api.deps import get_db, get_current_active_user
    from app.models.user import User
    from app.schemas.task import Task, TaskCreate, TaskUpdate
    from app.services.task_service import TaskService

    router = APIRouter()

    @router.post("/", response_model=Task, status_code=status.HTTP_201_CREATED)
    def create_task(
        task: TaskCreate,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Crear nueva tarea"""
        return TaskService.create_task(db=db, task=task, user_id=current_user.id)

    @router.get("/", response_model=List[Task])
    def read_tasks(
        skip: int = Query(0, ge=0, description="Number of tasks to skip"),
        limit: int = Query(100, ge=1, le=100, description="Maximum number of tasks to return"),
        completed: bool = Query(None, description="Filter by completion status"),
        priority: str = Query(None, description="Filter by priority (low, medium, high)"),
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Obtener tareas del usuario actual con filtros opcionales"""
        return TaskService.get_user_tasks(
            db=db, 
            user_id=current_user.id,
            skip=skip,
            limit=limit,
            completed=completed,
            priority=priority
        )

    @router.get("/{task_id}", response_model=Task)
    def read_task(
        task_id: int,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Obtener tarea específica"""
        task = TaskService.get_task(db=db, task_id=task_id, user_id=current_user.id)
        if task is None:
            raise HTTPException(status_code=404, detail="Task not found")
        return task

    @router.put("/{task_id}", response_model=Task)
    def update_task(
        task_id: int,
        task_update: TaskUpdate,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Actualizar tarea"""
        task = TaskService.update_task(
            db=db, 
            task_id=task_id, 
            task_update=task_update, 
            user_id=current_user.id
        )
        if task is None:
            raise HTTPException(status_code=404, detail="Task not found")
        return task

    @router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
    def delete_task(
        task_id: int,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Eliminar tarea"""
        success = TaskService.delete_task(db=db, task_id=task_id, user_id=current_user.id)
        if not success:
            raise HTTPException(status_code=404, detail="Task not found")

    @router.patch("/{task_id}/toggle", response_model=Task)
    def toggle_task_completion(
        task_id: int,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Alternar estado de completado de una tarea"""
        task = TaskService.toggle_task_completion(
            db=db, 
            task_id=task_id, 
            user_id=current_user.id
        )
        if task is None:
            raise HTTPException(status_code=404, detail="Task not found")
        return task

Servicio de tareas

El TaskService implementa toda la lógica de negocio para gestión de tareas:

  • Encapsulación: Toda la lógica de BD en un lugar
  • Reutilización: Métodos que pueden usarse desde diferentes endpoints
  • Transacciones: Manejo seguro de operaciones de BD
  • Filtrado dinámico: Construcción de queries basada en parámetros

El uso de exclude_unset=True en las actualizaciones permite modificaciones parciales.


    # app/services/task_service.py
    from typing import List, Optional
    from sqlalchemy.orm import Session
    from sqlalchemy import and_

    from app.models.task import Task
    from app.schemas.task import TaskCreate, TaskUpdate

    class TaskService:
        
        @staticmethod
        def create_task(db: Session, task: TaskCreate, user_id: int) -> Task:
            """Crear nueva tarea"""
            db_task = Task(**task.dict(), owner_id=user_id)
            db.add(db_task)
            db.commit()
            db.refresh(db_task)
            return db_task
        
        @staticmethod
        def get_task(db: Session, task_id: int, user_id: int) -> Optional[Task]:
            """Obtener tarea específica del usuario"""
            return db.query(Task).filter(
                and_(Task.id == task_id, Task.owner_id == user_id)
            ).first()
        
        @staticmethod
        def get_user_tasks(
            db: Session, 
            user_id: int, 
            skip: int = 0, 
            limit: int = 100,
            completed: Optional[bool] = None,
            priority: Optional[str] = None
        ) -> List[Task]:
            """Obtener tareas del usuario con filtros"""
            query = db.query(Task).filter(Task.owner_id == user_id)
            
            # Aplicar filtros opcionales
            if completed is not None:
                query = query.filter(Task.is_completed == completed)
            
            if priority is not None:
                query = query.filter(Task.priority == priority)
            
            return query.offset(skip).limit(limit).all()
        
        @staticmethod
        def update_task(
            db: Session, 
            task_id: int, 
            task_update: TaskUpdate, 
            user_id: int
        ) -> Optional[Task]:
            """Actualizar tarea"""
            db_task = TaskService.get_task(db, task_id, user_id)
            if db_task is None:
                return None
            
            # Actualizar solo campos proporcionados
            update_data = task_update.dict(exclude_unset=True)
            for field, value in update_data.items():
                setattr(db_task, field, value)
            
            db.commit()
            db.refresh(db_task)
            return db_task
        
        @staticmethod
        def delete_task(db: Session, task_id: int, user_id: int) -> bool:
            """Eliminar tarea"""
            db_task = TaskService.get_task(db, task_id, user_id)
            if db_task is None:
                return False
            
            db.delete(db_task)
            db.commit()
            return True
        
        @staticmethod
        def toggle_task_completion(db: Session, task_id: int, user_id: int) -> Optional[Task]:
            """Alternar estado de completado"""
            db_task = TaskService.get_task(db, task_id, user_id)
            if db_task is None:
                return None
            
            db_task.is_completed = not db_task.is_completed
            db.commit()
            db.refresh(db_task)
            return db_task

Aplicación principal y configuración

Configuración de la aplicación principal

El archivo main.py es el punto de entrada de nuestra aplicación. Aquí configuramos:

  • Middleware: Para CORS, seguridad, logging, etc.
  • Routers: Organización modular de endpoints
  • Documentación: Configuración de OpenAPI/Swagger
  • Inicialización: Creación de tablas y configuración inicial

CORS es especialmente importante para permitir que frontend en diferentes dominios consuman nuestra API.


    # app/main.py
    from fastapi import FastAPI
    from fastapi.middleware.cors import CORSMiddleware
    from fastapi.middleware.trustedhost import TrustedHostMiddleware

    from app.core.config import settings
    from app.api.v1 import auth, tasks
    from app.db.database import engine
    from app.models import user, task

    # Crear tablas en la base de datos
    user.Base.metadata.create_all(bind=engine)
    task.Base.metadata.create_all(bind=engine)

    # Crear instancia de FastAPI
    app = FastAPI(
        title=settings.PROJECT_NAME,
        version=settings.PROJECT_VERSION,
        description="Una API moderna para gestión de tareas construida con FastAPI y PostgreSQL",
        openapi_url=f"{settings.API_V1_STR}/openapi.json"
    )

    # Configurar CORS
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.BACKEND_CORS_ORIGINS,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Middleware de seguridad adicional
    app.add_middleware(TrustedHostMiddleware, allowed_hosts=["*"])

    # Incluir routers
    app.include_router(auth.router, prefix=f"{settings.API_V1_STR}/auth", tags=["authentication"])
    app.include_router(tasks.router, prefix=f"{settings.API_V1_STR}/tasks", tags=["tasks"])

    # Endpoint de salud
    @app.get("/", tags=["health"])
    def health_check():
        """Endpoint de verificación de estado"""
        return {
            "message": "Task Manager API is running",
            "version": settings.PROJECT_VERSION,
            "status": "healthy"
        }

    @app.get("/health", tags=["health"])
    def health():
        """Endpoint detallado de salud"""
        return {
            "status": "healthy",
            "service": settings.PROJECT_NAME,
            "version": settings.PROJECT_VERSION
        }

Archivo de variables de entorno

Crear .env.example con la configuración necesaria:


    # .env.example
    # Configuración de la aplicación
    PROJECT_NAME="Task Manager API"
    PROJECT_VERSION="1.0.0"

    # Seguridad
    SECRET_KEY="your-super-secret-key-change-this-in-production"
    ACCESS_TOKEN_EXPIRE_MINUTES=30

    # Base de datos
    DATABASE_URL="postgresql://postgres:postgres@localhost:5432/taskmanager"

    # CORS
    BACKEND_CORS_ORIGINS=["http://localhost:3000","http://localhost:8080"]

Testing automatizado

Testing es crucial para mantener la calidad del código. FastAPI tiene excelente soporte para testing con pytest:

  • TestClient: Simula requests HTTP sin levantar servidor
  • Fixtures: Configuración reutilizable para tests
  • Dependency overrides: Permite usar BD en memoria para tests
  • Isolation: Cada test tiene una BD limpia

Configuración de tests

Usamos SQLite en memoria para tests porque es rápido y se reinicia automáticamente. El override de dependencias permite reemplazar la BD real con la de testing.


    # tests/conftest.py
    import pytest
    from fastapi.testclient import TestClient
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    from app.main import app
    from app.db.database import get_db, Base
    from app.core.config import settings

    # Base de datos en memoria para tests
    SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

    engine = create_engine(
        SQLALCHEMY_DATABASE_URL,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )

    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    def override_get_db():
        try:
            db = TestingSessionLocal()
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_get_db

    @pytest.fixture(scope="function")
    def db():
        Base.metadata.create_all(bind=engine)
        yield
        Base.metadata.drop_all(bind=engine)

    @pytest.fixture(scope="function")
    def client(db):
        with TestClient(app) as test_client:
            yield test_client

    @pytest.fixture
    def user_data():
        return {
            "username": "testuser",
            "email": "test@example.com",
            "password": "testpassword123"
        }

    @pytest.fixture
    def authenticated_client(client, user_data):
        # Registrar usuario
        client.post("/api/v1/auth/register", json=user_data)
        
        # Login
        response = client.post(
            "/api/v1/auth/login",
            data={"username": user_data["username"], "password": user_data["password"]}
        )
        token = response.json()["access_token"]
        
        # Configurar headers de autenticación
        client.headers.update({"Authorization": f"Bearer {token}"})
        return client

Tests de autenticación

Estos tests verifican el flujo completo de autenticación:

  • Registro de usuarios con validación de duplicados
  • Login con credenciales correctas e incorrectas
  • Acceso a endpoints protegidos con tokens válidos

Cada test es independiente y verifica un aspecto específico del sistema.


    # tests/test_auth.py
    import pytest
    from fastapi.testclient import TestClient

    def test_register_user(client: TestClient, user_data):
        """Test registro de usuario"""
        response = client.post("/api/v1/auth/register", json=user_data)
        
        assert response.status_code == 201
        data = response.json()
        assert data["username"] == user_data["username"]
        assert data["email"] == user_data["email"]
        assert "id" in data
        assert "hashed_password" not in data

    def test_register_duplicate_username(client: TestClient, user_data):
        """Test registro con username duplicado"""
        # Primer registro
        client.post("/api/v1/auth/register", json=user_data)
        
        # Segundo registro con mismo username
        response = client.post("/api/v1/auth/register", json=user_data)
        
        assert response.status_code == 400
        assert "Username already registered" in response.json()["detail"]

    def test_login_user(client: TestClient, user_data):
        """Test login de usuario"""
        # Registrar usuario
        client.post("/api/v1/auth/register", json=user_data)
        
        # Login
        response = client.post(
            "/api/v1/auth/login",
            data={"username": user_data["username"], "password": user_data["password"]}
        )
        
        assert response.status_code == 200
        data = response.json()
        assert "access_token" in data
        assert data["token_type"] == "bearer"

    def test_login_invalid_credentials(client: TestClient, user_data):
        """Test login con credenciales inválidas"""
        response = client.post(
            "/api/v1/auth/login",
            data={"username": "nonexistent", "password": "wrongpassword"}
        )
        
        assert response.status_code == 401
        assert "Incorrect username or password" in response.json()["detail"]

    def test_get_current_user(authenticated_client: TestClient, user_data):
        """Test obtener información del usuario actual"""
        response = authenticated_client.get("/api/v1/auth/me")
        
        assert response.status_code == 200
        data = response.json()
        assert data["username"] == user_data["username"]
        assert data["email"] == user_data["email"]

Tests de tareas

Los tests de tareas verifican el CRUD completo:

  • Creación con datos válidos
  • Listado con filtros
  • Actualización parcial
  • Eliminación y verificación
  • Operaciones especiales como toggle

Estos tests aseguran que la lógica de negocio funciona correctamente.


    # tests/test_tasks.py
    import pytest
    from fastapi.testclient import TestClient

    def test_create_task(authenticated_client: TestClient):
        """Test crear tarea"""
        task_data = {
            "title": "Test Task",
            "description": "This is a test task",
            "priority": "high"
        }
        
        response = authenticated_client.post("/api/v1/tasks/", json=task_data)
        
        assert response.status_code == 201
        data = response.json()
        assert data["title"] == task_data["title"]
        assert data["description"] == task_data["description"]
        assert data["priority"] == task_data["priority"]
        assert data["is_completed"] is False
        assert "id" in data

    def test_get_tasks(authenticated_client: TestClient):
        """Test obtener lista de tareas"""
        # Crear algunas tareas
        for i in range(3):
            authenticated_client.post("/api/v1/tasks/", json={
                "title": f"Task {i}",
                "description": f"Description {i}"
            })
        
        response = authenticated_client.get("/api/v1/tasks/")
        
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 3

    def test_get_task_by_id(authenticated_client: TestClient):
        """Test obtener tarea específica"""
        # Crear tarea
        task_response = authenticated_client.post("/api/v1/tasks/", json={
            "title": "Specific Task",
            "description": "Specific description"
        })
        task_id = task_response.json()["id"]
        
        # Obtener tarea
        response = authenticated_client.get(f"/api/v1/tasks/{task_id}")
        
        assert response.status_code == 200
        data = response.json()
        assert data["title"] == "Specific Task"

    def test_update_task(authenticated_client: TestClient):
        """Test actualizar tarea"""
        # Crear tarea
        task_response = authenticated_client.post("/api/v1/tasks/", json={
            "title": "Original Title",
            "description": "Original description"
        })
        task_id = task_response.json()["id"]
        
        # Actualizar tarea
        update_data = {"title": "Updated Title", "is_completed": True}
        response = authenticated_client.put(f"/api/v1/tasks/{task_id}", json=update_data)
        
        assert response.status_code == 200
        data = response.json()
        assert data["title"] == "Updated Title"
        assert data["is_completed"] is True

    def test_delete_task(authenticated_client: TestClient):
        """Test eliminar tarea"""
        # Crear tarea
        task_response = authenticated_client.post("/api/v1/tasks/", json={
            "title": "Task to Delete"
        })
        task_id = task_response.json()["id"]
        
        # Eliminar tarea
        response = authenticated_client.delete(f"/api/v1/tasks/{task_id}")
        
        assert response.status_code == 204
        
        # Verificar que ya no existe
        get_response = authenticated_client.get(f"/api/v1/tasks/{task_id}")
        assert get_response.status_code == 404

    def test_toggle_task_completion(authenticated_client: TestClient):
        """Test alternar estado de completado"""
        # Crear tarea
        task_response = authenticated_client.post("/api/v1/tasks/", json={
            "title": "Task to Toggle"
        })
        task_id = task_response.json()["id"]
        
        # Alternar a completado
        response = authenticated_client.patch(f"/api/v1/tasks/{task_id}/toggle")
        
        assert response.status_code == 200
        data = response.json()
        assert data["is_completed"] is True
        
        # Alternar de vuelta a no completado
        response = authenticated_client.patch(f"/api/v1/tasks/{task_id}/toggle")
        data = response.json()
        assert data["is_completed"] is False

Ejecutando los tests

Para ejecutar los tests de nuestra API, necesitamos instalar las dependencias de testing y ejecutar pytest:


    # Instalar dependencias de testing (ya incluidas en requirements.txt)
    pip install pytest pytest-asyncio httpx

    # Ejecutar todos los tests
    pytest

    # Ejecutar tests con output detallado
    pytest -v

    # Ejecutar tests de un archivo específico
    pytest tests/test_auth.py

    # Ejecutar un test específico
    pytest tests/test_auth.py::test_register_user

    # Ejecutar tests con coverage
    pip install pytest-cov
    pytest --cov=app --cov-report=html

Los tests se ejecutan rápidamente porque usan una base de datos SQLite en memoria. Cada test es independiente y la base de datos se reinicia entre tests.

Estructura de tests recomendada

Para proyectos más grandes, organiza los tests de la siguiente manera:

tests/
├── conftest.py              # Configuración compartida
├── test_auth.py            # Tests de autenticación
├── test_tasks.py           # Tests de tareas
├── test_database.py        # Tests de modelos
└── integration/            # Tests de integración
    └── test_full_flow.py   # Tests end-to-end

Consideraciones para producción

⚠️ Advertencia: Este tutorial está diseñado para desarrollo y aprendizaje. Para desplegar en producción, necesitas implementar consideraciones adicionales de seguridad, rendimiento y monitoreo.

Aspectos críticos para producción

  1. Seguridad:

    • Cambiar SECRET_KEY por uno generado criptográficamente
    • Usar HTTPS con certificados SSL válidos
    • Implementar rate limiting por IP y por usuario
    • Validar y sanitizar todas las entradas
    • Configurar CORS restrictivo para dominios específicos

  2. Base de datos:

    • Usar un servidor PostgreSQL dedicado
    • Implementar backups automáticos
    • Configurar connection pooling
    • Optimizar queries con índices apropiados

  3. Monitoreo y logging:

    • Implementar logging estructurado
    • Configurar métricas de aplicación
    • Alertas para errores y rendimiento
    • Health checks para contenedores

  4. Escalabilidad:

    • Usar múltiples workers de Uvicorn
    • Implementar load balancing
    • Cache con Redis para datos frecuentes
    • CDN para contenido estático

  5. Gestión de configuración:

    • Usar secretos de Kubernetes o Docker Swarm
    • Variables de entorno específicas por ambiente
    • Gestión centralizada de configuración

Probando la API completa

Levantando el entorno


    # Clonar variables de entorno
    cp .env.example .env

    # Editar variables según tu entorno
    nano .env

    # Levantar con Docker Compose
    docker-compose up -d

    # Verificar que esté funcionando
    curl http://localhost:8000/health

Ejemplos de uso con curl


    # 1. Registrar usuario
    curl -X POST "http://localhost:8000/api/v1/auth/register" \
        -H "Content-Type: application/json" \
        -d '{
        "username": "usuario_demo",
        "email": "demo@ejemplo.com",
        "password": "password123"
        }'

    # 2. Login y obtener token
    TOKEN=$(curl -X POST "http://localhost:8000/api/v1/auth/login" \
        -H "Content-Type: application/x-www-form-urlencoded" \
        -d "username=usuario_demo&password=password123" | jq -r '.access_token')

    # 3. Crear tarea
    curl -X POST "http://localhost:8000/api/v1/tasks/" \
        -H "Authorization: Bearer $TOKEN" \
        -H "Content-Type: application/json" \
        -d '{
        "title": "Implementar nueva funcionalidad",
        "description": "Desarrollar módulo de reportes",
        "priority": "high",
        "due_date": "2025-10-01T10:00:00"
        }'

    # 4. Listar tareas
    curl -X GET "http://localhost:8000/api/v1/tasks/" \
        -H "Authorization: Bearer $TOKEN"

    # 5. Actualizar tarea
    curl -X PUT "http://localhost:8000/api/v1/tasks/1" \
        -H "Authorization: Bearer $TOKEN" \
        -H "Content-Type: application/json" \
        -d '{
        "title": "Funcionalidad completada",
        "is_completed": true
        }'

Documentación automática

Una vez levantada la aplicación, puedes acceder a:

  • Swagger UI: http://localhost:8000/docs
  • ReDoc: http://localhost:8000/redoc
  • OpenAPI JSON: http://localhost:8000/api/v1/openapi.json

FastAPI genera automáticamente esta documentación basándose en los type hints y docstrings de tus endpoints. La interfaz es completamente interactiva y permite probar la API directamente desde el navegador.

Conclusiones y siguientes pasos

Has construido una API REST completa y moderna con las mejores prácticas actuales. El proyecto incluye:

✅ Autenticación JWT segura con registro y login de usuarios
✅ CRUD completo para gestión de tareas con filtros avanzados
✅ Validación automática de datos con Pydantic
✅ Documentación interactiva generada automáticamente
✅ Testing automatizado con alta cobertura
✅ Containerización con Docker para deployment fácil
✅ Configuración de producción con Nginx y SSL

Ventajas del stack elegido

FastAPI ha demostrado ser una elección excelente por varias razones:

  • Developer Experience: Autocompletado, detección de errores en tiempo real
  • Documentación automática: Swagger UI y ReDoc sin esfuerzo adicional
  • Rendimiento: Comparable a frameworks de Go y Node.js
  • Modernidad: Aprovecha las últimas características de Python

La integración con PostgreSQL proporciona:

  • Robustez: Base de datos ACID completa con excelente rendimiento
  • Escalabilidad: Maneja desde startups hasta enterprise
  • Ecosistema: Extensiones y herramientas maduras

Docker garantiza:

  • Consistencia: Mismo entorno en desarrollo, testing y producción
  • Simplicidad: Un comando para levantar todo el stack
  • Aislamiento: Dependencias encapsuladas sin conflictos

El tipado estático con Pydantic:

  • Previene errores: Validación automática en desarrollo
  • Mejora mantenibilidad: Refactoring seguro con IDEs
  • Documenta el código: Los tipos son documentación viva

Los tests automatizados:

  • Confianza: Refactoring sin miedo a romper funcionalidad
  • Documentación: Los tests muestran cómo usar la API
  • Calidad: Detectan regresiones antes de llegar a producción

Mejoras y extensiones posibles

  1. Cache con Redis para mejorar rendimiento
  2. Logging estructurado con herramientas como ELK Stack
  3. Rate limiting avanzado por usuario
  4. Notificaciones push y email
  5. Filtros y búsqueda más sofisticados
  6. API versioning para mantener compatibilidad
  7. Métricas y monitoring con Prometheus
  8. Backup automático de base de datos

Recursos adicionales

Esta API puede servir como base sólida para proyectos más complejos, desde MVPs hasta aplicaciones empresariales. El patrón arquitectónico y las decisiones técnicas son escalables y están alineadas con las mejores prácticas de la industria en 2025.

¿Te gustó el tutorial? Comparte tu experiencia implementando la API o cuéntanos qué funcionalidades añadirías para tu caso de uso específico.

Alberto Méndez
comments powered by Disqus

Artículos relacionados

Quizá te puedan interesar

Nuestro trabajo

De concepto a realidad

Retoque fotográfico en segundos con Inteligencia Artificial Editasteic
Aventuras sin límites generadas por IA dónde tus elecciones dan forma a la historia. LAIA
Crea historias sorprendentes con cada tirada, ¡la imaginación es tu única guía! Spark
Controla fácilmente y en profundidad las horas que dedicas a tu empleo y/ó proyectos. Sé dueño de tu tiempo. WorkIO