Tutorial completo para crear una API REST de producción con FastAPI, PostgreSQL y Docker
Meta Title: Crear API REST con FastAPI, PostgreSQL y Docker: Guía completa para desarrollo backend
Meta Description: Aprende paso a paso a construir una API REST profesional con FastAPI, autenticación OAuth2, migraciones Alembic, tareas asíncronas con Celery, pruebas con pytest y despliegue con Docker y Kubernetes.
URL Slug: tutorial-fastapi-postgresql-docker-api-rest
1. Requisitos previos e instalación
1.1 Resumen
Antes de comenzar, asegúrate de tener instalado Python 3.11+, Docker y Docker Compose, Git, PostgreSQL, Redis y las herramientas necesarias para el manejo de migraciones y pruebas.
1.2 Pasos concretos
- Verifica Python:
|
1 2 3 |
python --version # Debe indicar Python 3.11.x |
- Verifica pip o Poetry:
|
1 2 3 4 |
pip --version # o si usas Poetry: poetry --version |
- Verifica Docker y Docker Compose:
|
1 2 3 |
docker --version docker compose version |
- Verifica Git:
|
1 2 |
git --version |
-
Instalación local recomendada para PostgreSQL y Redis (alternativamente usar Docker, que cubriremos más adelante).
-
Instalación de herramientas de migración y pruebas:
|
1 2 |
pip install alembic pytest pytest-asyncio |
1.3 Manejo de entorno y secretos
- Crea un archivo
.env.examplecon variables de entorno básicas:
|
1 2 3 4 5 |
DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/dbname REDIS_URL=redis://localhost:6379/0 SECRET_KEY=changeme ACCESS_TOKEN_EXPIRE_MINUTES=30 |
-
Para desarrollo usa python-dotenv o direnv para cargar
.envlocal. -
En producción, gestiona secretos con servicios especializados (AWS Secrets Manager, Vault) y no almacenes claves sensibles en el código.
Checklist 1
- [ ] Python 3.11+ instalado
- [ ] Docker y Docker Compose instalados
- [ ] PostgreSQL y Redis configurados (local o Docker)
- [ ] Alembic y pytest instalados
- [ ] Archivo
.env.examplepreparado
2. Inicializar proyecto y estructura de carpetas
2.1 Resumen
Organiza tu proyecto con una estructura modular, idealmente usando Poetry para manejo de dependencias y virtualenvs. Incluye separaciones para API, modelos, configuración, pruebas y scripts.
2.2 Pasos concretos
- Crear proyecto y activar entorno:
|
1 2 3 4 |
mkdir fastapi_project && cd fastapi_project poetry init --name fastapi_project --dependency fastapi --dependency uvicorn --dependency asyncpg --dependency sqlmodel --dependency alembic --dependency passlib[bcrypt] --dependency python-jose --dependency celery --dependency redis --dependency pytest --dependency pytest-asyncio poetry shell |
- Crear estructura estándar:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
fastapi_project/ ├── docker/ ├── alembic/ ├── deploy/ ├── scripts/ ├── src/ │ └── app/ │ ├── main.py │ ├── api/ │ │ └── v1/ │ │ ├── routes_product.py │ │ ├── routes_auth.py │ │ ├── schemas.py │ │ └── dependencies.py │ ├── core/ │ │ ├── config.py │ │ └── security.py │ ├── models/ │ ├── db/ │ ├── services/ │ └── tasks/ ├── tests/ │ ├── test_auth.py │ └── test_products.py └── pyproject.toml |
- Ejemplo básico de
pyproject.tomlgenerado por Poetry (extracto):
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
[tool.poetry] name = "fastapi_project" version = "0.1.0" description = "API REST con FastAPI y PostgreSQL" [tool.poetry.dependencies] python = "^3.11" fastapi = "^0.95" uvicorn = {extras = ["standard"], version = "^0.21.1"} sqlmodel = "^0.0.8" asyncpg = "^0.27" alembic = "^1.11" passlib = {extras = ["bcrypt"], version = "^1.7"} python-jose = "^3.3" celery = "^5.2" redis = "^4.5" pytest = "^7.3" pytest-asyncio = "^0.20" [tool.poetry.scripts] start = "src.app.main:main" |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
Este `pyproject.toml` configura las dependencias y el script de arranque. ### Checklist 2 - [ ] Proyecto creado con poetry - [ ] Estructura de carpetas creada - [ ] `pyproject.toml` con dependencias clave presente --- ## 3. ORM y migraciones con SQLModel y Alembic ### 3.1 Resumen SQLModel combina SQLAlchemy con Pydantic, ideal para modelos y validación. Alembic permite manejar las migraciones de forma segura. ### 3.2 Pasos concretos - Define modelos en `src/app/models/models.py`: |
python
from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List
from datetime import datetime
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primarykey=True)
username: str = Field(index=True, unique=True)
hashedpassword: str
isactive: bool = Field(default=True)
orders: List[“Order”] = Relationship(backpopulates=”user”)
class Product(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
description: Optional[str]
price: float
class Order(SQLModel, table=True):
id: Optional[int] = Field(default=None, primarykey=True)
userid: int = Field(foreignkey=”user.id”)
createdat: datetime = Field(defaultfactory=datetime.utcnow)
user: Optional[User] = Relationship(backpopulates=”orders”)
items: List[“OrderItem”] = Relationship(back_populates=”order”)
class OrderItem(SQLModel, table=True):
id: Optional[int] = Field(default=None, primarykey=True)
orderid: int = Field(foreignkey=”order.id”)
productid: int = Field(foreignkey=”product.id”)
quantity: int
order: Optional[Order] = Relationship(backpopulates=”items”)
|
1 2 3 4 5 6 7 |
``` Define las entidades con relaciones. - Configura Alembic: Archivo `alembic/env.py` (extracto para async SQLModel): |
python
from logging.config import fileConfig
from sqlalchemy import enginefromconfig
from sqlalchemy import pool
from alembic import context
from sqlmodel import SQLModel
import asyncio
from src.app.models.models import User, Product, Order, OrderItem
config = context.config
fileConfig(config.configfilename)
target_metadata = SQLModel.metadata
def runmigrationsonline():
connectable = enginefromconfig(
config.getsection(config.configini_section),
prefix=’sqlalchemy.’,
poolclass=pool.NullPool,
future=True,
)
|
1 2 3 4 5 6 7 8 9 |
with connectable.connect() as connection: context.configure( connection=connection, target_metadata=target_metadata ) with context.begin_transaction(): context.run_migrations() |
if context.isofflinemode():
# runmigrationsoffline() if needed
pass
else:
runmigrationsonline()
|
1 2 3 4 |
``` - Crear la migración e inicializar base: |
bash
alembic revision –autogenerate -m “Initial migration”
alembic upgrade head
|
1 2 3 4 |
Esto creará las tablas definidas. - Configura conexión asíncrona en `src/app/db/session.py`: |
python
from sqlmodel import SQLModel
from sqlalchemy.ext.asyncio import AsyncSession, createasyncengine
from sqlalchemy.orm import sessionmaker
from src.app.core.config import settings
engine = createasyncengine(
settings.databaseurl,
echo=True,
poolpre_ping=True,
)
asyncsession = sessionmaker(
engine, class=AsyncSession, expireoncommit=False
)
async def initdb():
async with engine.begin() as conn:
await conn.runsync(SQLModel.metadata.create_all)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
``` ### Checklist 3 - [ ] Modelos definidos con SQLModel - [ ] Alembic configurado y migración inicial creada - [ ] AsyncEngine configurado para conexiones --- ## 4. Implementación de FastAPI y rutas ### 4.1 Resumen Configura una aplicación FastAPI con middlewares, rutas versionadas y ejemplos CRUD con validación y paginación. ### 4.2 Pasos concretos - Archivo `src/app/main.py`: |
python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.gzip import GZipMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
import logging
import uvicorn
from src.app.api.v1 import routesauth, routesproduct
from src.app.core.config import settings
app = FastAPI(
title=”FastAPI API REST”,
description=”API REST con FastAPI, PostgreSQL, Celery y más”,
version=”1.0.0″,
openapi_url=”/api/v1/openapi.json”
)
CORS (configurar dominios seguros)
app.addmiddleware(
CORSMiddleware,
alloworigins=settings.allowedhosts,
allowmethods=[““],
allow_headers=[““],
)
Compresión GZip
app.addmiddleware(GZipMiddleware, minimumsize=1000)
Middleware logging simple
@app.middleware(“http”)
async def logrequests(request: Request, callnext):
logger = logging.getLogger(“uvicorn.access”)
logger.info(f”Inicio request: {request.method} {request.url}”)
response = await callnext(request)
logger.info(f”Fin request: statuscode={response.status_code}”)
return response
Routers
app.includerouter(routesauth.router, prefix=”/api/v1/auth”, tags=[“auth”])
app.includerouter(routesproduct.router, prefix=”/api/v1/products”, tags=[“products”])
if name == “main“:
uvicorn.run(“src.app.main:app”, host=”0.0.0.0″, port=8000, reload=True)
|
1 2 3 4 |
``` - Ejemplo mínimo CRUD para productos en `routes_product.py`: |
python
from fastapi import APIRouter, Depends, HTTPException, Query
from typing import List
from sqlmodel import select
from src.app.models.models import Product
from src.app.db.session import async_session
from src.app.api.v1.schemas import ProductCreate, ProductRead
router = APIRouter()
@router.get(“/”, responsemodel=List[ProductRead])
async def listproducts(
limit: int = Query(10, le=100),
offset: int = Query(0),
):
async with async_session() as session:
query = select(Product).limit(limit).offset(offset)
result = await session.execute(query)
products = result.scalars().all()
return products
@router.post(“/”, responsemodel=ProductRead)
async def createproduct(productin: ProductCreate):
product = Product.fromorm(productin)
async with asyncsession() as session:
session.add(product)
await session.commit()
await session.refresh(product)
return product
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
``` ### Checklist 4 - [ ] FastAPI configurado en `main.py` con middlewares - [ ] Routers versionados y organizados - [ ] CRUD básico para Product implementado --- ## 5. Autenticación y autorización: OAuth2 con JWT ### 5.1 Resumen Implementa OAuth2 con Password Flow, usando JWT para manejo de access y refresh tokens, roles y scopes, junto a almacenamiento seguro de contraseñas. ### 5.2 Pasos concretos - Archivo `src/app/core/security.py`: |
python
from passlib.context import CryptContext
from datetime import datetime, timedelta
from jose import jwt, JWTError
pwd_context = CryptContext(schemes=[“bcrypt”], deprecated=”auto”)
SECRETKEY = “SUPERSECRETKEYCHANGE”
ALGORITHM = “HS256”
ACCESSTOKENEXPIREMINUTES = 30
def verifypassword(plainpassword, hashedpassword):
return pwdcontext.verify(plainpassword, hashedpassword)
def getpasswordhash(password):
return pwd_context.hash(password)
def createaccesstoken(data: dict, expiresdelta: timedelta | None = None):
toencode = data.copy()
expire = datetime.utcnow() + (expiresdelta or timedelta(minutes=ACCESSTOKENEXPIREMINUTES))
toencode.update({“exp”: expire})
encodedjwt = jwt.encode(toencode, SECRETKEY, algorithm=ALGORITHM)
return encoded_jwt
def decodetoken(token: str):
try:
payload = jwt.decode(token, SECRETKEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
|
1 2 3 4 |
``` - Implementa endpoints en `routes_auth.py` para registro, login y refresh: |
python
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel
from datetime import timedelta
from src.app.models.models import User
from src.app.db.session import asyncsession
from src.app.core.security import getpasswordhash, verifypassword, createaccesstoken
from sqlmodel import select
router = APIRouter()
class UserRegister(BaseModel):
username: str
password: str
class Token(BaseModel):
accesstoken: str
tokentype: str = “bearer”
@router.post(“/register”, responsemodel=Token)
async def register(userin: UserRegister):
async with asyncsession() as session:
query = select(User).where(User.username == userin.username)
result = await session.execute(query)
user = result.scalaroneornone()
if user:
raise HTTPException(statuscode=400, detail=”Usuario ya existe”)
hashedpassword = getpasswordhash(userin.password)
user = User(username=userin.username, hashedpassword=hashedpassword)
session.add(user)
await session.commit()
accesstoken = createaccesstoken({“sub”: user.username})
return {“accesstoken”: accesstoken}
@router.post(“/login”, responsemodel=Token)
async def login(formdata: UserRegister):
async with asyncsession() as session:
query = select(User).where(User.username == formdata.username)
result = await session.execute(query)
user = result.scalaroneornone()
if not user or not verifypassword(formdata.password, user.hashedpassword):
raise HTTPException(statuscode=401, detail=”Credenciales inválidas”)
accesstoken = createaccesstoken({“sub”: user.username})
return {“accesstoken”: accesstoken}
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
``` - Añade una dependencia para extraer el usuario actual de un token en `dependencies.py`. ### Checklist 5 - [ ] Seguridad de contraseñas con passlib - [ ] Creación y validación de JWT - [ ] Endpoints para registro y login - [ ] Dependencia para `current_user` --- ## 6. Background tasks y workers con Celery y Redis ### 6.1 Resumen Celery y Redis facilitan la ejecución de tareas asíncronas pesadas como envío de emails o generación de reportes, despejando la API. ### 6.2 Pasos concretos - Configura Celery en `src/app/tasks/celery_app.py`: |
python
from celery import Celery
from src.app.core.config import settings
celeryapp = Celery(
“worker”,
broker=settings.redisurl,
backend=settings.redis_url
)
celeryapp.conf.taskroutes = {“src.app.tasks.tasks.*”: “main-queue”}
|
1 2 3 4 |
``` - Define tareas en `src/app/tasks/tasks.py`: |
python
from src.app.tasks.celeryapp import celeryapp
@celeryapp.task
def sendemail_task(email: str, subject: str, body: str):
print(f”Enviando email a {email} con asunto {subject}”)
# Aquí implementaríamos el envío real
|
1 2 |
- Para usar desde un endpoint: |
python
from fastapi import APIRouter
from src.app.tasks.tasks import sendemailtask
router = APIRouter()
@router.post(“/send-email”)
async def sendemail(email: str):
sendemail_task.delay(email, “Bienvenido”, “Gracias por registrarte”)
return {“message”: “Email encolado”}
|
1 2 |
- Para ejecutar worker y beat: |
bash
celery -A src.app.tasks.celeryapp.celeryapp worker –loglevel=info
celery -A src.app.tasks.celeryapp.celeryapp beat –loglevel=info
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
### Checklist 6 - [ ] Celery configurado con Redis - [ ] Tareas creadas en tasks.py - [ ] Worker y Beat funcionando - [ ] Ejecución de tareas desde FastAPI --- ## 7. Comunicación y eventos: Webhooks idempotentes ### 7.1 Resumen Implementa webhooks con verificación y tolerancia a duplicados para integraciones seguras. ### Pasos clave - Crear endpoint que valide firma del webhook (usando HMAC). - Guardar eventos procesados para evitar repeticiones. - Implementar reintentos basados en HTTP codes. ### Diagrama ASCII |
Cliente webhook –> API (verifica firma, valida idempotencia)–> DB
|
Respuesta
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
### Checklist 7 - [ ] Verificación HMAC implementada - [ ] Almacenamiento de eventos procesados - [ ] Lógica de reintentos --- ## 8. Manejo de errores y respuestas estandarizadas ### 8.1 Resumen Maneja errores globalmente para mantener respuestas consistentes y útiles. ### Ejemplo en `src/app/main.py`: |
python
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
@app.exceptionhandler(StarletteHTTPException)
async def httpexceptionhandler(request: Request, exc: StarletteHTTPException):
return JSONResponse({“detail”: exc.detail, “statuscode”: exc.statuscode}, statuscode=exc.status_code)
@app.exceptionhandler(RequestValidationError)
async def validationexceptionhandler(request: Request, exc: RequestValidationError):
return JSONResponse({“detail”: exc.errors(), “statuscode”: 422}, status_code=422)
|
1 2 |
Estándar basado en RFC 7807 para errores: |
json
{
“type”: “https://example.com/probs/out-of-credit”,
“title”: “You do not have enough credit.”,
“status”: 403,
“detail”: “Your current balance is 30, but that costs 50.”,
“instance”: “/account/12345/msgs/abc”
}
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
### Checklist 8 - [ ] Handlers globales para HTTPException y ValidationError - [ ] Respuesta de error con formato estándar --- ## 9. Documentación y OpenAPI ### 9.1 Resumen FastAPI genera automáticamente OpenAPI con Swagger UI y ReDoc, que puede personalizarse. ### Pasos concretos - Metadata personalizada en `FastAPI()` (ya visto en main.py). - Agrega `tags` y `description` a routers y endpoints. - Proteger documentación en producción (ejemplo con autenticación básica para Swagger). Ejemplo para proteger docs: |
python
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi import Depends
security = HTTPBasic()
@app.get(“/docs”)
async def getswagger(credentials: HTTPBasicCredentials = Depends(security)):
correctusername = secrets.comparedigest(credentials.username, “admin”)
correctpassword = secrets.comparedigest(credentials.password, “secret”)
if not (correctusername and correctpassword):
raise HTTPException(statuscode=401, detail=”No autorizado”)
return getswaggeruihtml(openapiurl=”/api/v1/openapi.json”, title=”Docs”)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
### Checklist 9 - [ ] Metadata completa - [ ] Ejemplos y esquemas claros - [ ] Documentación protegida en producción --- ## 10. Observabilidad y métricas ### 10.1 Resumen Configura logging estructurado, métricas Prometheus y trazabilidad con OpenTelemetry. ### Pasos concretos - Logging con `structlog` o `loguru` (ejemplo básico con logging estándar ya mostrado). - Métricas Prometheus: expose `/metrics` con `prometheus_fastapi_instrumentator`: |
bash
pip install prometheus-fastapi-instrumentator
|
1 2 |
En `main.py`: |
python
from prometheusfastapiinstrumentator import Instrumentator
@app.on_event(“startup”)
async def startup():
Instrumentator().instrument(app).expose(app)
|
1 2 |
- Health Checks: |
python
@app.get(“/healthz”)
async def healthcheck():
return {“status”: “ok”}
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
- OpenTelemetry puede integrarse para tracing distribuidos, fuera de este tutorial por brevedad. ### Checklist 10 - [ ] Logging estructurado - [ ] Métricas prometheus configuradas y expuestas - [ ] Endpoints de healthz implementados --- ## 11. Tests y calidad ### 11.1 Resumen Escribe pruebas unitarias y de integración usando pytest y pytest-asyncio, incluyendo arranque de dependencias en CI. ### Pasos concretos - Ejemplo de test async en `tests/test_auth.py`: |
python
import pytest
from httpx import AsyncClient
from src.app.main import app
@pytest.mark.asyncio
async def testregisterandlogin():
async with AsyncClient(app=app, baseurl=”http://test”) as ac:
response = await ac.post(“/api/v1/auth/register”, json={“username”:”testuser”,”password”:”testpass”})
assert response.statuscode == 200
data = response.json()
assert “accesstoken” in data
|
1 2 3 4 5 |
login_resp = await ac.post("/api/v1/auth/login", json={"username":"testuser","password":"testpass"}) assert login_resp.status_code == 200 login_data = login_resp.json() assert "access_token" in login_data |
|
1 2 3 4 |
``` - Testcontainers para postgres y redis en CI: |
python
Con testcontainers
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
@pytest.fixture(scope=”session”)
def postgrescontainer():
with PostgresContainer(“postgres:15”) as postgres:
yield postgres.getconnection_url()
@pytest.fixture(scope=”session”)
def rediscontainer():
with RedisContainer() as redis:
yield redis.getconnection_url()
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
- Ejecutar migraciones en setup de tests para asegurar esquema correcto. ### Checklist 11 - [ ] Tests async con pytest-asyncio - [ ] Uso de Testcontainers o docker-compose en CI - [ ] Cobertura para autenticación y productos --- ## 12. Contenerización y orquestación local ### 12.1 Resumen Crea un Dockerfile multi-stage para optimizar la imagen y docker-compose para levantar servicio con postgres y redis. ### Pasos concretos - Dockerfile (guardado en raíz): |
dockerfile
Builder
FROM python:3.11-slim AS builder
WORKDIR /app
RUN pip install –upgrade pip
COPY pyproject.toml poetry.lock* /app/
RUN pip install poetry
RUN poetry config virtualenvs.create false
RUN poetry install –no-dev –no-interaction –no-ansi
COPY . /app
Runtime
FROM python:3.11-slim
WORKDIR /app
COPY –from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY –from=builder /app /app
ENV PYTHONUNBUFFERED=1
CMD [“uvicorn”, “src.app.main:app”, “–host”, “0.0.0.0”, “–port”, “8000”]
|
1 2 3 4 |
``` - `.dockerignore` |
pycache
*.pyc
.env
.git
|
1 2 |
- `docker-compose.yml`: |
yaml
version: “3.9”
services:
api:
build: .
ports:
– 8000:8000
environment:
– DATABASEURL=postgresql+asyncpg://postgres:postgres@db:5432/fastapi
– REDISURL=redis://redis:6379/0
– SECRETKEY=changeme
dependson:
– db
– redis
db:
image: postgres:15
restart: always
environment:
POSTGRESUSER: postgres
POSTGRESPASSWORD: postgres
POSTGRES_DB: fastapi
volumes:
– pgdata:/var/lib/postgresql/data
redis:
image: redis:7-alpine
restart: always
volumes:
pgdata:
|
1 2 3 4 |
``` - Levantar todo: |
bash
docker compose up –build
|
1 2 |
- Ejecutar migraciones dentro del contenedor: |
bash
docker compose run api alembic upgrade head
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
### Checklist 12 - [ ] Dockerfile multi-stage - [ ] docker-compose configurado con api, postgres y redis - [ ] Migraciones ejecutadas en contenedor --- ## 13. Kubernetes (opcional pero recomendado) ### 13.1 Resumen Manifiestos para despliegue básico con readiness y liveness probes, configuración de secrets y autoscaling. ### Ejemplo `deploy/k8s/api-deployment.yaml`: |
yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: fastapi-api
spec:
replicas: 2
selector:
matchLabels:
app: fastapi-api
template:
metadata:
labels:
app: fastapi-api
spec:
containers:
– name: api
image: yourrepo/fastapi_api:latest
ports:
– containerPort: 8000
readinessProbe:
httpGet:
path: /healthz
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
livenessProbe:
httpGet:
path: /healthz
port: 8000
initialDelaySeconds: 15
periodSeconds: 20
resources:
requests:
cpu: “250m”
memory: “256Mi”
limits:
cpu: “500m”
memory: “512Mi”
envFrom:
– secretRef:
name: fastapi-secrets
|
1 2 |
- Service básico: |
yaml
apiVersion: v1
kind: Service
metadata:
name: fastapi-api-service
spec:
selector:
app: fastapi-api
ports:
- protocol: TCP
port: 80
targetPort: 8000
type: ClusterIP
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
### Checklist 13 - [ ] Deployment con probes y recursos claros - [ ] Uso de Secrets y ConfigMaps para configuración - [ ] Servicios para exponer pods --- ## 14. CI/CD con GitHub Actions ### 14.1 Resumen Configura pipeline para linting, tests, construcción y despliegue seguros. ### Ejemplo workflow `.github/workflows/ci.yml`: |
yaml
name: CI
on: [push, pullrequest]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:15
env:
POSTGRESDB: fastapi
POSTGRESUSER: postgres
POSTGRESPASSWORD: postgres
ports:
– 5432:5432
redis:
image: redis:7-alpine
ports:
– 6379:6379
steps:
– uses: actions/checkout@v3
– name: Setup Python
uses: actions/setup-python@v4
with:
python-version: 3.11
– name: Install dependencies
run: |
python -m pip install –upgrade pip
pip install poetry
poetry install
– name: Run migrations
run: alembic upgrade head
– name: Run tests
run: pytest -v
– name: Lint
run: |
pip install ruff
ruff check src tests
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
``` ### Checklist 14 - [ ] Entorno levantado con postgres y redis - [ ] Tests y migraciones ejecutadas - [ ] Calidad de código validada --- ## 15. Rendimiento y buenas prácticas ### Puntos clave: - Usa pool de conexiones configurado correctamente (asyncpg). - Indexar columnas usadas en filtros. - Paginación con límites y offsets razonables. - Cache en Redis para recursos muy consultados. - HTTP cache headers para respuestas estáticas. - Compresión GZip habilitada. - Worker uvicorn con múltiples workers (e.g. `--workers 4`). - Prefiere async en I/O, pero evalúa tareas CPU-bound si es necesario. --- ## 16. Seguridad en producción - TLS/HTTPS obligatorio, con terminación en proxy o Ingress. - Configura headers de seguridad: - X-Content-Type-Options - X-Frame-Options - Content-Security-Policy - Rate limiting en login para evitar brute-force. - Rotación periódica de claves JWT. - Protege secrets con vault o proveedores cloud. - Considera WAF para ataques comunes. --- ## 17. Resolución de problemas comunes - **Puerto ocupado:** cambia puerto uvicorn o verifica procesos activos. - **Error conexión DB en Docker:** verifica variables de entorno y que contenedores inicien. - **Migraciones Alembic fallan:** revisar revision y conexión DB. - **Tokens JWT inválidos:** sincronizar hora, validar claves. - **Celery no encola:** revisar broker Redis, logs worker y confirmación delay. - **Tests CI fallan por permisos:** usa usuarios correctos, vars env y dependencias instaladas. - **Errores CORS en producción:** configura dominios permitidos exactos. - **Webhook firma mismatch:** confirmar secreto compartido y HMAC exacto. --- ## 18. Archivos clave - Ejemplos completos ### 18.1 pyproject.toml (extracto) |
toml
[tool.poetry]
name = “fastapi_project”
version = “0.1.0”
[tool.poetry.dependencies]
python = “^3.11”
fastapi = “^0.95”
uvicorn = {extras = [“standard”], version = “^0.21.1”}
sqlmodel = “^0.0.8”
asyncpg = “^0.27”
alembic = “^1.11”
passlib = {extras = [“bcrypt”], version = “^1.7”}
python-jose = “^3.3”
celery = “^5.2”
redis = “^4.5”
pytest = “^7.3”
pytest-asyncio = “^0.20”
[tool.poetry.scripts]
start = “src.app.main:main”
|
1 2 3 4 5 6 |
### 18.2 src/app/main.py (Ver sección 4.2, contiene FastAPI instanciado, middleware y routers) ### 18.3 src/app/core/config.py |
python
from pydantic import BaseSettings
class Settings(BaseSettings):
databaseurl: str
redisurl: str
secretkey: str
accesstokenexpireminutes: int = 30
allowed_hosts: list[str] = [“*”]
|
1 2 3 |
class Config: env_file = ".env" |
settings = Settings()
|
1 2 3 4 5 6 7 8 |
Explica carga confiable de variables desde `.env`. ### 18.4 src/app/models/models.py (Ver sección 3.2, define User, Product, Order, OrderItem) ### 18.5 src/app/schemas.py |
python
from pydantic import BaseModel
class ProductCreate(BaseModel):
name: str
description: str | None = None
price: float
class ProductRead(BaseModel):
id: int
name: str
description: str | None = None
price: float
|
1 2 3 |
class Config: orm_mode = True |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
### 18.6 src/app/api/v1/routes_product.py (Ver sección 4.2 ejemplo CRUD) ### 18.7 src/app/api/v1/routes_auth.py (Ver sección 5.2, incluye register y login) ### 18.8 alembic/env.py (Ver sección 3.2, maneja migraciones con SQLModel, AsyncEngine) ### 18.9 src/app/tasks/celery_app.py y tasks.py (Ver sección 6.2, configuración de Celery y tarea ejemplo) ### 18.10 Dockerfile y docker-compose.yml (Ver sección 12) ### 18.11 deploy/k8s/api-deployment.yaml y api-service.yaml (Ver sección 13) ### 18.12 tests/test_auth.py y test_products.py (Ver sección 11, incluye fixtures y tests async) ### 18.13 .env.example |
env
DATABASEURL=postgresql+asyncpg://postgres:postgres@localhost:5432/fastapi
REDISURL=redis://localhost:6379/0
SECRETKEY=yoursecretkey
ACCESSTOKENEXPIREMINUTES=30
ALLOWED_HOSTS=*
|
1 2 |
### 18.14 README (extracto comandos usados) |
bash
Instalar dependencias y activar entorno
poetry install
poetry shell
Ejecutar migraciones
alembic upgrade head
Levantar entorno local con Docker Compose
docker compose up –build
Ejecutar tests
pytest -q
`
FAQ
¿Qué versión de Python necesito? 3.11 o superior para aprovechar async y mejoras.
¿Por qué usar SQLModel y no solo SQLAlchemy? SQLModel integra validación Pydantic y ORM, facilitando validación y modelos.
¿Puedo usar otro broker diferente a Redis para Celery? Sí, RabbitMQ es otra opción popular.
¿Cómo protejo mis tokens JWT? Usa un secret fuerte y guarda refresh tokens de forma segura.
¿Por qué usar Testcontainers? Para pruebas fiables levantando instancias reales de DB y servicios.
¿Cómo escalo mi API en producción? Usa múltiples workers Uvicorn, balanceo de carga y Kubernetes/HPA.
¿Qué hacer si Celery tasks no se ejecutan? Verificar conexión a Redis, logs de worker y uso correcto de .delay().
¿Cómo debuggeo errores con Alembic? Revisa la configuración de URL, revisa env.py y el script de migración.
CHECKLIST GLOBAL PARA PRODUCCIÓN
- [ ] Entorno configurado con variables y secretos seguros
- [ ] Base de datos y migraciones desplegadas y actualizadas
- [ ] API con endpoints organizados y testing cubierto
- [ ] Seguridad implementada: autenticación, autorización, HTTPS
- [ ] Tareas asíncronas con Celery funcionando
- [ ] Observabilidad activa con logging, métricas y health checks
- [ ] Contenedores construidos y orquestados localmente
- [ ] Pipeline CI/CD configurado y probado
- [ ] Manifiestos Kubernetes o equivalente preparados
- [ ] Tests automatizados y cobertura aceptable
- [ ] Políticas de seguridad y escalado listas
Esta guía le da a un desarrollador principiante-intermedio una ruta detallada, práctica y profesional para crear APIs REST robustas con FastAPI y entorno moderno, integrando pruebas, despliegue y buenas prácticas.
¡Comienza hoy mismo y construye APIs escalables y seguras con FastAPI!
