¿Cómo probar y filtrar una API RESTful de manera eficiente?

En el desarrollo de una API RESTful, uno de los aspectos clave es asegurarse de que todos sus puntos finales estén correctamente probados y filtrados según sea necesario para facilitar la búsqueda y manipulación de datos. A continuación, exploraremos cómo estructurar y probar una API de manera efectiva utilizando herramientas como pytest y FastAPI, además de cómo implementar un filtrado y búsqueda avanzado para mejorar la funcionalidad de la API.

Para comenzar, es fundamental definir el entorno de prueba adecuado. Usamos un módulo conftest.py, que es estándar en proyectos que emplean pytest, para definir una base de datos de prueba común a todo el proyecto. Esto se logra mediante la configuración de una lista de tareas de prueba y la creación de un archivo CSV que actúa como la base de datos para las pruebas. De esta forma, garantizamos que todos los tests interactúan con un entorno controlado.

Un aspecto clave en este proceso es la fixture de pytest, que asegura que cada test se ejecute sobre una base de datos limpia y preparada antes de cada prueba. La fixture se configura para ejecutarse automáticamente en cada prueba, evitando así la necesidad de configurar manualmente el entorno antes de cada ejecución. Esto no solo simplifica el proceso, sino que también mejora la fiabilidad y consistencia de las pruebas.

Por ejemplo, podemos definir tareas de prueba de la siguiente manera:

python
Copy code
TEST_DATABASE_FILE = "test_tasks.csv"

TEST_TASKS_CSV = [
    { "id": "1", "title": "Test Task One", "description": "Test Description One", "status": "Incomplete" },
    { "id": "2", "title": "Test Task Two", "description": "Test Description Two", "status": "Ongoing" },
]

En este caso, TEST_TASKS_CSV es un conjunto de datos de ejemplo que se usará para las pruebas. Para cada prueba, el archivo CSV se crea automáticamente, y después de que la prueba se complete, el archivo se elimina para evitar que las pruebas posteriores se vean afectadas.

Realizando las Pruebas de los Endpoints

Por ejemplo, para probar el endpoint GET /tasks, que lista todas las tareas en la base de datos, se puede realizar el siguiente test:

python
Copy code
def test_endpoint_read_all_tasks():
    response = client.get("/tasks")
    assert response.status_code == 200
    assert response.json() == TEST_TASKS

Este test valida que el endpoint responde correctamente (código de estado 200) y que los datos devueltos coinciden con las tareas predefinidas en TEST_TASKS.

Un ejemplo de test para el endpoint POST /task sería:

python
Copy code
def test_endpoint_create_task():
    task = { "title": "To Define", "description": "Will be done", "status": "Ready" }
    response = client.post("/task", json=task)
    assert response.status_code == 200
    assert response.json() == {**task, "id": 3}

Este test verifica que, al enviar una tarea nueva, la API asigna un nuevo id y devuelve los datos correctos.

Implementación de Filtrado y Búsqueda Avanzada

Una vez que los endpoints básicos están probados, se puede mejorar la funcionalidad del API al implementar capacidades de filtrado y búsqueda avanzada, lo cual es fundamental para sistemas que manejan grandes volúmenes de datos.

Aquí hay un ejemplo de cómo hacerlo:

python
Copy code
@app.get("/tasks", response_model=list[TaskWithID])
def get_tasks(status: Optional[str] = None, title: Optional[str] = None):
    tasks = read_all_tasks()
    if status:
        tasks = [task for task in tasks if task.status == status]
    if title:
        tasks = [task for task in tasks if task.title == title]
    return tasks

Con esta implementación, los usuarios pueden filtrar las tareas por su status o title directamente desde la URL de la solicitud. Si ambos parámetros son proporcionados, la API devolverá solo aquellas tareas que coincidan con ambos criterios.

Un ejemplo de cómo se puede implementar un endpoint de búsqueda sería:

python
Copy code
@app.get("/tasks/search", response_model=list[TaskWithID])
def search_tasks(keyword: str):
    tasks = read_all_tasks()

    filtered_tasks = [task for task in tasks if keyword.lower() in (task.title + task.description).lower()]

    return filtered_tasks

Este endpoint permitirá a los usuarios realizar búsquedas de texto completas, donde la palabra clave proporcionada se buscará tanto en el título como en la descripción de las tareas.

Al integrar estas funcionalidades de búsqueda y filtrado, mejoramos la usabilidad de la API, permitiendo a los usuarios encontrar rápidamente las tareas relevantes sin tener que navegar por grandes cantidades de datos.

Es importante destacar que el manejo eficiente de filtros y búsquedas en una API puede mejorar significativamente la experiencia del usuario, especialmente cuando se trabaja con grandes cantidades de datos. Las buenas prácticas incluyen el uso de índices en la base de datos para acelerar las búsquedas y el filtrado, además de limitar la cantidad de resultados devueltos para evitar sobrecargar el servidor.

¿Cómo Optimizar Consultas SQL en una Aplicación?

Uno de los problemas más comunes al trabajar con bases de datos es la ineficiencia en las consultas SQL. Esto no solo afecta el rendimiento de la aplicación, sino que también puede incrementar el consumo de recursos del servidor y provocar tiempos de respuesta lentos. Para evitar estos problemas, existen varias técnicas de optimización que se pueden aplicar en diferentes contextos. A continuación, exploramos tres de estas técnicas utilizando ejemplos prácticos.

Uno de los problemas que frecuentemente se presenta al realizar consultas es el conocido como "N+1 queries". Este problema ocurre cuando se realiza una consulta para obtener una lista de elementos y luego, por cada uno de esos elementos, se realiza una nueva consulta para obtener datos relacionados, lo que da lugar a un gran número de consultas adicionales.

Supongamos que necesitamos mostrar una lista de eventos junto con los patrocinadores asociados a cada uno. Un enfoque inicial podría ser hacer una consulta para obtener los eventos y luego recorrerlos para obtener los patrocinadores de cada evento mediante consultas adicionales. Este enfoque sería altamente ineficiente, ya que se generaría una consulta inicial para obtener los eventos y luego N consultas adicionales (una para cada evento) para obtener los patrocinadores.

python
Copy code
async def get_events_with_sponsors(db_session: AsyncSession) -> list[Event]:
    query = (
        select(Event)
        .options(joinedload(Event.sponsors))
    )
    async with db_session as session:
        result = await session.execute(query)
        events = result.scalars().all()
    return events

Otro concepto importante a la hora de optimizar consultas es el uso adecuado de la declaración JOIN. Aunque los JOIN pueden hacer que las consultas sean más legibles, es fundamental utilizarlos de forma moderada y solo cuando sea necesario. A veces, hacer un JOIN innecesario puede afectar el rendimiento de la consulta, ya que está uniendo tablas que no se utilizan realmente en el resultado final.

Imaginemos que necesitamos obtener una lista de patrocinadores junto con la cantidad que han aportado a un evento específico, ordenada de mayor a menor cantidad. Una forma de hacerlo sería utilizando múltiples JOIN para obtener los datos de tres tablas diferentes (patrocinadores, aportaciones y eventos). El código sería el siguiente:

python
Copy code
async def get_event_sponsorships_with_amount(db_session: AsyncSession, event_id: int):

    query = (
        select(Sponsor.name, Sponsorship.amount)
        .join(Sponsorship, Sponsorship.sponsor_id == Sponsor.id)
        .join(Event, Sponsorship.event_id == Event.id)
        .order_by(Sponsorship.amount.desc())
    )
    async with db_session as session:
        result = await session.execute(query)
        sponsor_contributions = result.fetchall()
    return sponsor_contributions

Sin embargo, en este caso, estamos realizando un JOIN con la tabla de eventos, aunque realmente no necesitamos esos datos en el resultado. La forma más eficiente de escribir esta consulta sería eliminar el JOIN innecesario con la tabla de eventos, como se muestra a continuación:

python
Copy code
async def get_event_sponsorships_with_amount(db_session: AsyncSession, event_id: int):

    query = (
        select(Sponsor.name, Sponsorship.amount)
        .join(Sponsorship, Sponsorship.sponsor_id == Sponsor.id)
        .where(Sponsorship.event_id == event_id)
        .order_by(Sponsorship.amount.desc())
    )
    async with db_session as session:
        result = await session.execute(query)
        sponsor_contributions = result.fetchall()
    return sponsor_contributions

Al eliminar el JOIN con la tabla de eventos, hemos simplificado la consulta y reducido el trabajo innecesario que la base de datos debe hacer, lo que mejora la eficiencia general.

La tercera técnica que se debe considerar es la minimización de los datos que se traen de la base de datos. Si no necesitamos toda la información de una tabla, es innecesario cargarla completamente, ya que esto solo aumentará el tiempo de ejecución de la consulta y el consumo de memoria. SQLAlchemy ofrece una función llamada load_only, que permite especificar solo las columnas que necesitamos, lo cual es útil especialmente cuando trabajamos con grandes volúmenes de datos.

Por ejemplo, supongamos que necesitamos obtener una lista de boletos de un evento, pero solo nos interesan tres columnas: el ID del boleto, el usuario y el precio. En lugar de traer todos los datos de la tabla de boletos, podemos utilizar load_only para limitar los datos que se cargan:

python
Copy code
async def get_events_tickets_with_user_price(db_session: AsyncSession, event_id: int) -> list[Ticket]:

    query = (
        select(Ticket)
        .where(Ticket.event_id == event_id)
        .options(load_only(Ticket.id, Ticket.user, Ticket.price))
    )
    async with db_session as session:
        result = await session.execute(query)
        tickets = result.scalars().all()
    return tickets

En este caso, solo se traerán las columnas necesarias (ID, usuario y precio), lo que reduce la cantidad de datos transferidos y mejora la velocidad de la consulta, especialmente en aplicaciones con bases de datos grandes.

Es importante mencionar que la optimización de consultas SQL no se limita a las técnicas mencionadas anteriormente. Hay otras consideraciones que pueden influir en el rendimiento, como el uso adecuado de índices, la partición de tablas, el uso de cachés de consultas, entre otras. Además, las bases de datos SQL pueden tener características propias que optimizan consultas en diferentes contextos, como el uso de motores de almacenamiento especializados o la implementación de modelos de transacción que afectan el rendimiento de las consultas.

Por lo tanto, al trabajar con bases de datos, es crucial entender las necesidades específicas de las consultas y elegir las técnicas de optimización que mejor se adapten a los requisitos del proyecto. Siempre es recomendable realizar pruebas de rendimiento para asegurarse de que las optimizaciones realmente mejoren el tiempo de respuesta y la eficiencia general de la aplicación.

¿Cómo mejorar el rendimiento de las consultas en MongoDB mediante índices?

Cuando trabajamos con bases de datos que almacenan colecciones extensas, como las que contienen información sobre canciones y álbumes, las consultas simples pueden volverse lentas si no están optimizadas. Consideremos, por ejemplo, una consulta que busque canciones lanzadas en un año específico. Si no se ha creado un índice adecuado sobre el campo release_year del álbum, cada consulta tendrá que recorrer toda la colección para encontrar los documentos correspondientes. Esto puede ser muy costoso en términos de tiempo y recursos, especialmente si la colección es extensa.

Para mejorar este rendimiento, MongoDB permite la creación de índices. Un índice es básicamente una estructura de datos que organiza de manera eficiente los valores de un campo determinado, de modo que las consultas puedan localizar rápidamente los documentos relevantes sin necesidad de escanear toda la colección. En nuestro ejemplo, si creamos un índice sobre el campo album.release_year, las consultas que filtren por el año de lanzamiento de las canciones se ejecutarán mucho más rápido, ya que MongoDB podrá buscar directamente los documentos relacionados con ese valor sin tener que revisar cada uno de ellos.

Imaginemos que necesitamos crear un endpoint que devuelva todas las canciones de un álbum lanzado en un año específico. Para realizar esta búsqueda de manera eficiente, debemos primero crear el índice en la base de datos. Esto se puede hacer al inicio del servidor, utilizando el método create_index de MongoDB. La siguiente línea de código ilustra cómo crear un índice en el campo album.release_year, ordenado de forma descendente:

python
Copy code
await db.songs.create_index({"album.release_year": -1})

Este índice optimiza las consultas que buscan canciones basadas en el año de lanzamiento, haciendo que las búsquedas sean mucho más ágiles. Es importante señalar que el índice no solo mejora el tiempo de respuesta de las consultas, sino que también garantiza la integridad y la consistencia en la ejecución de las operaciones de búsqueda. Para verificar que la consulta realmente utiliza el índice, podemos agregar un log que nos muestre el índice que está siendo utilizado:

python
Copy code
explained_query = await query.explain()

logger.info("Index used: %s", explained_query.get("queryPlanner", {}).get("winningPlan", {}).get("inputStage", {}).get("indexName", "No index used"))

A medida que las consultas se vuelven más complejas, es probable que necesitemos otros tipos de índices, como índices de texto. Los índices de texto permiten realizar búsquedas eficientes en campos que contienen texto, como el nombre del artista. En este caso, para buscar canciones de un artista específico, creamos un índice de texto en el campo artist de la colección songs:

python
Copy code
await db.songs.create_index({"artist": "text"})

Una vez creado este índice, podemos realizar búsquedas de texto completas sobre el campo artist, lo que nos permite encontrar todas las canciones de un artista, incluso si solo conocemos una parte de su nombre. La consulta se realizaría de la siguiente manera:

python
Copy code
query = db.songs.find({"$text": {"$search": artist}})

Es importante destacar que la creación de índices no debe ser tomada a la ligera. Aunque los índices mejoran significativamente el rendimiento de las consultas, también requieren recursos adicionales para su mantenimiento. Cada vez que se inserta, actualiza o elimina un documento, MongoDB debe mantener actualizado el índice, lo que puede afectar el rendimiento en operaciones de escritura. Por lo tanto, es recomendable crear índices solo en los campos que realmente se utilizan en las consultas de búsqueda más frecuentes. Además, en colecciones con datos extremadamente grandes, la elección adecuada de índices puede ser un factor decisivo en el rendimiento general del sistema.

MongoDB ofrece una variedad de tipos de índices más allá de los de texto y los simples, como los índices compuestos, que combinan varios campos, o los índices 2D, que son útiles para almacenar y consultar datos geoespaciales. Para sacar el máximo provecho de las capacidades de MongoDB, es recomendable explorar todas las opciones de índices que la base de datos ofrece y elegir los más adecuados según las necesidades específicas de la aplicación.

¿Cómo desarrollar middleware personalizado en ASGI para modificar solicitudes y respuestas?

En el desarrollo de aplicaciones web, uno de los elementos más poderosos para la personalización y el control es el middleware. Este sirve para interceptar y modificar tanto las solicitudes como las respuestas antes de que lleguen al cliente o servidor, brindando un sinfín de oportunidades para ajustar el comportamiento de nuestra aplicación según necesidades específicas. Este capítulo se centra en cómo desarrollar middleware personalizado para modificar solicitudes y respuestas utilizando ASGI en el contexto de FastAPI, una de las bibliotecas más utilizadas para la creación de aplicaciones web rápidas y modernas en Python.

El middleware en ASGI actúa como una capa intermedia que maneja las solicitudes HTTP entre el cliente y el servidor. A través de esta capa, es posible ejecutar tareas de procesamiento adicionales, como la autenticación, la validación de datos o la modificación de contenido antes de que llegue a su destino final. Desarrollar middleware adecuado para la modificación de solicitudes y respuestas puede mejorar la flexibilidad y seguridad de nuestra aplicación.

Desarrollo de middleware para la modificación de solicitudes

El primer paso en la creación de middleware personalizado es interceptar la solicitud antes de que llegue al endpoint correspondiente. En este caso, se desarrollará un middleware para modificar el cuerpo de la solicitud mediante un hash, lo que podría ser útil para asegurar la integridad de los datos.

Supongamos que tenemos un endpoint de tipo POST que recibe un mensaje en el cuerpo de la solicitud. Lo primero será configurar el endpoint en el archivo principal main.py para que acepte contenido del cuerpo. Un ejemplo de cómo se hace esto es:

python
Copy code
import logging

logger = logging.getLogger("uvicorn")
@app.post("/send")
async def send(message: str = Body()):
    logger.info(f"Mensaje: {message}")
    return message

Una vez que tenemos este endpoint, podemos proceder a crear un middleware que modifique el cuerpo de la solicitud antes de que llegue al endpoint. Este middleware podría, por ejemplo, aplicar un hash al contenido de la solicitud para asegurarse de que no ha sido alterado.

A continuación, se presentan los pasos para crear dicho middleware en un archivo request_middleware.py:

1. Importar los tipos necesarios:
   El primer paso es importar los tipos desde Starlette, que es la biblioteca base sobre la que FastAPI construye su funcionalidad de ASGI.

   python
   Copy code
   from starlette.types import ASGIApp, Scope, Receive, Send
   
   from hashlib import sha1
   

2. Crear la clase del middleware:
   Definimos una clase llamada HashBodyContentMiddleware, que tomará la aplicación ASGI y una lista de rutas permitidas como parámetros. Este middleware se encargará de procesar solo las rutas que se especifiquen.

   python
   Copy code
   class HashBodyContentMiddleware:
   
       def __init__(self, app: ASGIApp, allowed_paths: list[str]):
   
           self.app = app
           self.allowed_paths = allowed_paths
   

3. Modificar el cuerpo de la solicitud:
   En el método __call__ de la clase, verificamos si la solicitud es del tipo HTTP y si el path está en la lista de rutas permitidas. Si ambas condiciones se cumplen, modificamos el cuerpo de la solicitud aplicando un hash SHA-1.

   python
   Copy code
   async def __call__(self, scope: Scope, receive: Receive, send: Send):
       if scope["type"] != "http" or scope["path"] not in self.allowed_paths:
           await self.app(scope, receive, send)
           return
       async def receive_with_new_body() -> Message:
           message = await receive()
           assert message["type"] == "http.request"
           body = message.get("body", b"")
           message["body"] = f'"{sha1(body).hexdigest()}"'.encode()
           return message
       await self.app(scope, receive_with_new_body, send)
   

4. Agregar el middleware a la aplicación:
   Finalmente, agregamos el middleware a nuestra instancia de FastAPI utilizando el método add_middleware.

   python
   Copy code
   app.add_middleware(HashBodyContentMiddleware, allowed_paths=["/send"])
   
   

Este es un ejemplo básico de cómo puedes crear un middleware para modificar las solicitudes antes de que lleguen a su destino. Cuando se realiza una solicitud POST a /send, el cuerpo de la solicitud se modificará y se enviará un hash en lugar del contenido original.

Desarrollo de middleware para la modificación de respuestas

En un módulo denominado response_middleware.py, comenzamos importando los tipos necesarios y creando una clase similar a la del middleware de solicitudes. La diferencia principal es que, en lugar de trabajar con el cuerpo de la solicitud, estaremos manipulando las respuestas.

python
Copy code
from typing import Sequence

from starlette.types import ASGIApp, Scope, Receive, Send, Message
class ResponseMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app
    async def __call__(self, scope: Scope, receive: Receive, send: Send):

        async def send_with_custom_header(message: Message):

            if "headers" in message:
                message["headers"].append((b"X-Custom-Header", b"Value"))
            await send(message)
        await self.app(scope, receive, send_with_custom_header)

En este caso, la clase ResponseMiddleware intercepta las respuestas antes de que se envíen al cliente y agrega una cabecera personalizada X-Custom-Header. De manera similar al middleware de solicitudes, este middleware también se agrega a la aplicación FastAPI mediante el método add_middleware.

python
Copy code
app.add_middleware(ResponseMiddleware)

Este middleware asegurará que todas las respuestas que se envíen desde la aplicación incluyan la cabecera personalizada especificada.

Reflexión final

La capacidad de modificar tanto las solicitudes como las respuestas es una de las características más poderosas de FastAPI y ASGI en general. A través de la creación de middleware personalizado, los desarrolladores pueden introducir funcionalidades específicas que serían difíciles de implementar de otra manera. Este tipo de herramientas es indispensable cuando se buscan soluciones avanzadas de seguridad, personalización o procesamiento de datos en tiempo real.

Además, es importante recordar que el middleware no solo es útil para modificar cuerpos o cabeceras, sino que también puede ser fundamental para la implementación de capas de seguridad adicionales, como la validación de autenticidad de los datos, la protección contra inyecciones de código o la implementación de políticas de CORS, por ejemplo. La flexibilidad que ofrece este enfoque permite a los desarrolladores construir aplicaciones web robustas y altamente personalizables, adaptadas a sus necesidades específicas.

¿Cómo mejorar la seguridad y el rendimiento en aplicaciones FastAPI?

La integración de tecnologías modernas en aplicaciones web está en constante evolución, y FastAPI, un marco de desarrollo web para Python, no es la excepción. En este contexto, aspectos clave como la seguridad, la optimización del rendimiento y la correcta integración de herramientas son esenciales para garantizar que nuestras aplicaciones sean robustas y eficientes. Este artículo aborda cómo mejorar el rendimiento y la seguridad de las aplicaciones FastAPI, tomando en cuenta prácticas recomendadas y la integración de servicios externos.

Uno de los pilares de una aplicación segura es la encriptación de datos, tanto en reposo como en tránsito. La encriptación en tránsito, como el uso de HTTPS, asegura que la información que se intercambia entre el cliente y el servidor esté protegida contra posibles intercepciones. Para implementarla correctamente en FastAPI, es crucial configurar adecuadamente el servidor con un certificado SSL, lo cual se puede hacer fácilmente usando herramientas como Let's Encrypt. Además, la encriptación en reposo es vital para proteger los datos almacenados en bases de datos o sistemas de almacenamiento. Para ello, FastAPI soporta la encriptación simétrica con bibliotecas como Fernet, que garantiza que los datos se almacenen de manera segura, incluso en caso de que el sistema sea comprometido.

Además de la encriptación y el rendimiento, la gestión de errores y excepciones juega un papel crucial en la estabilidad de la aplicación. FastAPI, por su diseño, facilita la captura de errores de manera eficiente. Sin embargo, integrar un sistema de registro robusto, como el uso de la biblioteca Python logging, puede proporcionar una visibilidad completa de lo que ocurre dentro de la aplicación. Esto permite identificar problemas antes de que afecten al usuario final, proporcionando además un sistema de auditoría que puede ser útil en situaciones de seguridad.

Cuando se trata de integrar bases de datos, FastAPI ofrece un soporte integral para bases de datos tanto relacionales como NoSQL. MongoDB, por ejemplo, es ampliamente utilizado en aplicaciones que requieren alta escalabilidad y flexibilidad en el almacenamiento de datos. Con la integración de MongoDB, es importante considerar aspectos como la creación de índices eficientes y el manejo de relaciones entre datos. FastAPI facilita esta integración a través de ODMs (Object-Document Mappers), lo que permite interactuar de manera más intuitiva con la base de datos. Además, en aplicaciones con altos requerimientos de consulta, se debe considerar la optimización de las operaciones de lectura mediante técnicas como la minimización de consultas N+1.

Finalmente, es importante tener en cuenta que la documentación automática es una de las características más potentes de FastAPI. Utilizando herramientas como Swagger UI o ReDoc, los desarrolladores pueden generar documentación interactiva para sus APIs de forma sencilla. Esto no solo mejora la experiencia del desarrollador al interactuar con la API, sino que también facilita la integración con otros servicios, permitiendo que los usuarios o desarrolladores externos puedan probar y entender rápidamente cómo funciona la API.

Aparte de estos aspectos clave, es fundamental que los desarrolladores no pierdan de vista el concepto de "seguridad por diseño". Esto implica incorporar prácticas de seguridad desde las primeras etapas del desarrollo, en lugar de implementarlas como un parche al final. La protección contra vulnerabilidades comunes, como la inyección SQL o los ataques XSS, debe ser parte integral de la construcción de la aplicación.