Construir APIs CRUD asíncronas

Aprende a construir APIs CRUD de alto rendimiento aprovechando las capacidades asíncronas de FastAPI. En este tutorial implementaremos E/S de archivos asíncrona y procesamiento eficiente de datos con la plantilla fastapi-async-crud.

Lo que aprenderás en este tutorial

• Entender las aplicaciones FastAPI asíncronas
• Operaciones CRUD asíncronas con la sintaxis async/await
• Procesamiento asíncrono de archivos con aiofiles
• Escribir y ejecutar pruebas asíncronas
• Técnicas de optimización de rendimiento

Requisitos previos

• Haber completado el tutorial del servidor API básico
• Conceptos básicos de async/await en Python
• FastAPI-fastkit instalado

Por qué hace falta procesamiento asíncrono

Entendamos la diferencia entre procesamiento síncrono y asíncrono:

Procesamiento síncrono

def process_items():
    item1 = read_file("item1.json")      # Espera 2 segundos
    item2 = read_file("item2.json")      # Espera 2 segundos
    item3 = read_file("item3.json")      # Espera 2 segundos
    return [item1, item2, item3]         # Total: 6 segundos

Procesamiento asíncrono

async def process_items():
    item1_task = read_file_async("item1.json")  # Arranca en paralelo
    item2_task = read_file_async("item2.json")  # Arranca en paralelo
    item3_task = read_file_async("item3.json")  # Arranca en paralelo

    items = await asyncio.gather(item1_task, item2_task, item3_task)
    return items                                # Total: 2 segundos

Paso 1: Crear un proyecto CRUD asíncrono

Crea un proyecto con la plantilla fastapi-async-crud:

$ fastkit startdemo fastapi-async-crud
Enter the project name: async-todo-api
Enter the author name: Developer Kim
Enter the author email: developer@example.com
Enter the project description: Asynchronous todo management API
Deploying FastAPI project using 'fastapi-async-crud' template

           Project Information
┌──────────────┬─────────────────────────────────────────┐
│ Project Name │ async-todo-api                          │
│ Author       │ Developer Kim                           │
│ Author Email │ developer@example.com                   │
│ Description  │ Asynchronous todo management API        │
└──────────────┴─────────────────────────────────────────┘

       Template Dependencies
┌──────────────┬───────────────────┐
│ Dependency 1 │ fastapi           │
│ Dependency 2 │ uvicorn           │
│ Dependency 3 │ pydantic          │
│ Dependency 4 │ pydantic-settings │
│ Dependency 5 │ aiofiles          │
│ Dependency 6 │ pytest-asyncio    │
└──────────────┴───────────────────┘

Select package manager (pip, uv, pdm, poetry) [uv]: uv
Do you want to proceed with project creation? [y/N]: y

✨ FastAPI project 'async-todo-api' from 'fastapi-async-crud' has been created successfully!

Paso 2: Analizar la estructura del proyecto

Examinemos las diferencias clave del proyecto generado:

async-todo-api/
├── src/
│   ├── main.py                    # Aplicación FastAPI asíncrona
│   ├── api/
│   │   └── routes/
│   │       └── items.py          # Endpoints CRUD asíncronos
│   ├── crud/
│   │   └── items.py              # Lógica asíncrona de procesamiento de datos
│   ├── schemas/
│   │   └── items.py              # Modelos de datos (los mismos)
│   ├── mocks/
│   │   └── mock_items.json       # Base de datos en archivo JSON
│   └── core/
│       └── config.py             # Archivo de configuración
└── tests/
    ├── conftest.py               # Configuración de pruebas asíncronas
    └── test_items.py             # Casos de prueba asíncronos

Diferencias clave

1. aiofiles: procesamiento asíncrono de E/S de archivos
2. pytest-asyncio: soporte de pruebas asíncronas
3. Patrón async/await: todas las operaciones CRUD están implementadas de forma asíncrona

Paso 3: Entender la lógica CRUD asíncrona

Procesamiento asíncrono de datos (src/crud/items.py)

import json
import asyncio
from typing import List, Optional
from aiofiles import open as aio_open
from pathlib import Path

from src.schemas.items import Item, ItemCreate, ItemUpdate

class AsyncItemCRUD:
    def __init__(self, data_file: str = "src/mocks/mock_items.json"):
        self.data_file = Path(data_file)

    async def _read_data(self) -> List[dict]:
        """Lee datos del archivo JSON de forma asíncrona"""
        try:
            async with aio_open(self.data_file, 'r', encoding='utf-8') as f:
                content = await f.read()
                return json.loads(content)
        except FileNotFoundError:
            return []

    async def _write_data(self, data: List[dict]) -> None:
        """Escribe datos al archivo JSON de forma asíncrona"""
        async with aio_open(self.data_file, 'w', encoding='utf-8') as f:
            await f.write(json.dumps(data, indent=2, ensure_ascii=False))

    async def get_items(self) -> List[Item]:
        """Recupera todos los items (asíncrono)"""
        data = await self._read_data()
        return [Item(**item) for item in data]

    async def get_item(self, item_id: int) -> Optional[Item]:
        """Recupera un item concreto (asíncrono)"""
        data = await self._read_data()
        item_data = next((item for item in data if item["id"] == item_id), None)
        return Item(**item_data) if item_data else None

    async def create_item(self, item: ItemCreate) -> Item:
        """Crea un item nuevo (asíncrono)"""
        data = await self._read_data()
        new_id = max([item["id"] for item in data], default=0) + 1

        new_item = Item(id=new_id, **item.dict())
        data.append(new_item.dict())

        await self._write_data(data)
        return new_item

    async def update_item(self, item_id: int, item_update: ItemUpdate) -> Optional[Item]:
        """Actualiza un item (asíncrono)"""
        data = await self._read_data()

        for i, item in enumerate(data):
            if item["id"] == item_id:
                update_data = item_update.dict(exclude_unset=True)
                data[i].update(update_data)
                await self._write_data(data)
                return Item(**data[i])

        return None

    async def delete_item(self, item_id: int) -> bool:
        """Elimina un item (asíncrono)"""
        data = await self._read_data()
        original_length = len(data)

        data = [item for item in data if item["id"] != item_id]

        if len(data) < original_length:
            await self._write_data(data)
            return True

        return False

Endpoints asíncronos de la API (src/api/routes/items.py)

from typing import List
from fastapi import APIRouter, HTTPException, status

from src.schemas.items import Item, ItemCreate, ItemUpdate
from src.crud.items import AsyncItemCRUD

router = APIRouter()
crud = AsyncItemCRUD()

@router.get("/", response_model=List[Item])
async def read_items():
    """Recupera todos los items (asíncrono)"""
    return await crud.get_items()

@router.get("/{item_id}", response_model=Item)
async def read_item(item_id: int):
    """Recupera un item concreto (asíncrono)"""
    item = await crud.get_item(item_id)
    if item is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item with id {item_id} not found"
        )
    return item

@router.post("/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemCreate):
    """Crea un item nuevo (asíncrono)"""
    return await crud.create_item(item)

@router.put("/{item_id}", response_model=Item)
async def update_item(item_id: int, item_update: ItemUpdate):
    """Actualiza un item (asíncrono)"""
    updated_item = await crud.update_item(item_id, item_update)
    if updated_item is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item with id {item_id} not found"
        )
    return updated_item

@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    """Elimina un item (asíncrono)"""
    deleted = await crud.delete_item(item_id)
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item with id {item_id} not found"
        )

Paso 4: Ejecutar el servidor y probarlo

Entra en el directorio del proyecto y arranca el servidor:

$ cd async-todo-api
$ fastkit runserver
Starting FastAPI server at 127.0.0.1:8000...

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [12345] using WatchFiles
INFO:     Started server process [12346]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Pruebas de rendimiento

Verifiquemos el rendimiento del procesamiento asíncrono enviando varias peticiones a la vez:

Test de peticiones concurrentes (script Python)

import asyncio
import aiohttp
import time

async def create_item(session, item_data):
    async with session.post("http://127.0.0.1:8000/items/", json=item_data) as response:
        return await response.json()

async def test_concurrent_requests():
    start_time = time.time()

    items_to_create = [
        {"name": f"Item {i}", "description": f"Description {i}", "price": i * 10, "tax": i}
        for i in range(1, 11)  # Crear 10 items en paralelo
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [create_item(session, item) for item in items_to_create]
        results = await asyncio.gather(*tasks)

    end_time = time.time()
    print(f"Created 10 items in: {end_time - start_time:.2f} seconds")
    print(f"Number of items created: {len(results)}")

# Ejecutar test
# asyncio.run(test_concurrent_requests())

Paso 5: Escribir pruebas asíncronas

Configuración de pruebas (tests/conftest.py)

import pytest
import asyncio
from httpx import AsyncClient
from src.main import app

@pytest.fixture(scope="session")
def event_loop():
    """Configuración del event loop"""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture
async def async_client():
    """Cliente de pruebas asíncrono"""
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

Casos de prueba asíncronos (tests/test_items.py)

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_create_item_async(async_client: AsyncClient):
    """Test asíncrono de creación de item"""
    item_data = {
        "name": "Test Item",
        "description": "Item for asynchronous testing",
        "price": 100.0,
        "tax": 10.0
    }

    response = await async_client.post("/items/", json=item_data)

    assert response.status_code == 201
    data = response.json()
    assert data["name"] == item_data["name"]
    assert data["price"] == item_data["price"]
    assert "id" in data

@pytest.mark.asyncio
async def test_read_items_async(async_client: AsyncClient):
    """Test asíncrono de listado de items"""
    response = await async_client.get("/items/")

    assert response.status_code == 200
    items = response.json()
    assert isinstance(items, list)

@pytest.mark.asyncio
async def test_concurrent_operations(async_client: AsyncClient):
    """Test de operaciones concurrentes"""
    import asyncio

    # Crear varios items en paralelo
    tasks = []
    for i in range(5):
        item_data = {
            "name": f"ConcurrentItem{i}",
            "description": f"Description{i}",
            "price": i * 10,
            "tax": i
        }
        task = async_client.post("/items/", json=item_data)
        tasks.append(task)

    responses = await asyncio.gather(*tasks)

    # Verificar que todas las peticiones tuvieron éxito
    for response in responses:
        assert response.status_code == 201

    # Verificar los items creados
    response = await async_client.get("/items/")
    items = response.json()
    assert len(items) >= 5

Ejecutar las pruebas

$ pytest tests/ -v --asyncio-mode=auto
======================== test session starts ========================
collected 8 items

tests/test_items.py::test_create_item_async PASSED            [ 12%]
tests/test_items.py::test_read_items_async PASSED             [ 25%]
tests/test_items.py::test_read_item_async PASSED              [ 37%]
tests/test_items.py::test_update_item_async PASSED            [ 50%]
tests/test_items.py::test_delete_item_async PASSED            [ 62%]
tests/test_items.py::test_concurrent_operations PASSED        [ 75%]
tests/test_items.py::test_item_not_found_async PASSED         [ 87%]
tests/test_items.py::test_invalid_item_data_async PASSED      [100%]

======================== 8 passed in 0.24s ========================

Paso 6: Monitorización y optimización del rendimiento

Añadir middleware de medición de tiempo de respuesta

Vamos a añadir monitorización del rendimiento en src/main.py:

import time
from fastapi import FastAPI, Request
from src.api.api import api_router
from src.core.config import settings

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.VERSION,
    description=settings.DESCRIPTION,
)

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """Añadir el tiempo de procesamiento de la petición a las cabeceras"""
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

app.include_router(api_router)

@app.get("/")
async def read_root():
    return {"message": "Welcome to the Asynchronous Todo API!"}

Implementar procesamiento por lotes asíncrono

Vamos a añadir endpoints en lote para procesar varios items a la vez:

# Añadir a src/api/routes/items.py

@router.post("/batch", response_model=List[Item])
async def create_items_batch(items: List[ItemCreate]):
    """Crear varios items en paralelo (procesamiento por lotes)"""
    import asyncio

    # Ejecutar todas las tareas de creación en paralelo
    tasks = [crud.create_item(item) for item in items]
    created_items = await asyncio.gather(*tasks)

    return created_items

@router.get("/batch/{item_ids}")
async def read_items_batch(item_ids: str):
    """Recuperar varios items en paralelo (procesamiento por lotes)"""
    import asyncio

    # Parsear los IDs separados por comas
    ids = [int(id.strip()) for id in item_ids.split(",")]

    # Ejecutar todas las tareas de recuperación en paralelo
    tasks = [crud.get_item(item_id) for item_id in ids]
    items = await asyncio.gather(*tasks)

    # Devolver solo los items no None
    return [item for item in items if item is not None]

Probar el procesamiento por lotes

# Test de creación por lotes
$ curl -X POST "http://127.0.0.1:8000/items/batch" \
  -H "Content-Type: application/json" \
  -d '[
    {"name": "Item1", "description": "Description1", "price": 10.0, "tax": 1.0},
    {"name": "Item2", "description": "Description2", "price": 20.0, "tax": 2.0},
    {"name": "Item3", "description": "Description3", "price": 30.0, "tax": 3.0}
  ]'

# Test de recuperación por lotes
$ curl -X GET "http://127.0.0.1:8000/items/batch/1,2,3"

Paso 7: Patrones asíncronos avanzados

Implementar rate limiting

import asyncio
from collections import defaultdict
from fastapi import HTTPException, Request
from datetime import datetime, timedelta

class AsyncRateLimiter:
    def __init__(self, max_requests: int = 100, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = defaultdict(list)

    async def is_allowed(self, client_ip: str) -> bool:
        now = datetime.now()
        cutoff = now - timedelta(seconds=self.window_seconds)

        # eliminar registros antiguos
        self.requests[client_ip] = [
            req_time for req_time in self.requests[client_ip]
            if req_time > cutoff
        ]

        # comprobar el número actual de peticiones
        if len(self.requests[client_ip]) >= self.max_requests:
            return False

        # añadir el registro actual
        self.requests[client_ip].append(now)
        return True

# instancia global del rate limiter
rate_limiter = AsyncRateLimiter()

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host

    if not await rate_limiter.is_allowed(client_ip):
        raise HTTPException(
            status_code=429,
            detail="Too many requests"
        )

    response = await call_next(request)
    return response

Implementar caché asíncrona

import asyncio
from typing import Optional, Any
from datetime import datetime, timedelta

class AsyncCache:
    def __init__(self):
        self._cache = {}
        self._expiry = {}

    async def get(self, key: str) -> Optional[Any]:
        # eliminar elementos caducados
        if key in self._expiry and datetime.now() > self._expiry[key]:
            del self._cache[key]
            del self._expiry[key]
            return None

        return self._cache.get(key)

    async def set(self, key: str, value: Any, ttl_seconds: int = 300):
        self._cache[key] = value
        self._expiry[key] = datetime.now() + timedelta(seconds=ttl_seconds)

    async def delete(self, key: str):
        self._cache.pop(key, None)
        self._expiry.pop(key, None)

# instancia global de la caché
cache = AsyncCache()

# modificar los métodos CRUD para usar la caché
async def get_items_cached(self) -> List[Item]:
    """Recuperar items usando la caché"""
    cache_key = "all_items"
    cached_items = await cache.get(cache_key)

    if cached_items:
        return cached_items

    # si no hay caché, leer desde el archivo
    items = await self.get_items()
    await cache.set(cache_key, items, ttl_seconds=60)  # caché de 1 minuto

    return items

Paso 8: Consideraciones de producción

Gestionar los pools de conexiones

# añadir a src/core/config.py
class Settings(BaseSettings):
    # ... configuración existente ...

    # configuración relacionada con procesamiento asíncrono
    MAX_CONCURRENT_REQUESTS: int = 100
    REQUEST_TIMEOUT: int = 30
    CONNECTION_POOL_SIZE: int = 20

settings = Settings()

Mejorar el manejo de errores

import logging
from fastapi import HTTPException
from typing import Union

logger = logging.getLogger(__name__)

async def safe_async_operation(operation, *args, **kwargs) -> Union[Any, None]:
    """Ejecutar una operación asíncrona segura"""
    try:
        return await operation(*args, **kwargs)
    except asyncio.TimeoutError:
        logger.error(f"Timeout in {operation.__name__}")
        raise HTTPException(status_code=504, detail="Request timeout")
    except Exception as e:
        logger.error(f"Error in {operation.__name__}: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")

# ejemplo de uso
@router.get("/safe/{item_id}")
async def read_item_safe(item_id: int):
    return await safe_async_operation(crud.get_item, item_id)

Próximos pasos

¡Has terminado de construir una API CRUD asíncrona! Próximos pasos:

1. Integración con base de datos - Usar PostgreSQL con SQLAlchemy asíncrono
2. Contenedorización con Docker - Contenedorizar aplicaciones asíncronas
3. Manejo personalizado de respuestas - Formatos de respuesta avanzados y manejo de errores

Resumen

En este tutorial hemos usado FastAPI asíncrono para:

• ✅ Implementar operaciones CRUD asíncronas
• ✅ Optimizar E/S de archivos con aiofiles
• ✅ Manejar peticiones concurrentes y probar el rendimiento
• ✅ Escribir y ejecutar pruebas asíncronas
• ✅ Implementar procesamiento por lotes y patrones asíncronos avanzados
• ✅ Atender consideraciones de producción (caché, manejo de errores, gestión de conexiones)

¡Dominar el procesamiento asíncrono te permite construir servidores API de alto rendimiento!