¿Cómo crear operaciones CRUD en una API RESTful utilizando FastAPI?

Al desarrollar una API RESTful, uno de los componentes fundamentales es la implementación de operaciones CRUD (Crear, Leer, Actualizar, Eliminar). En este caso, exploramos cómo gestionar tareas en una base de datos representada por un archivo CSV, y cómo exponer estas operaciones a través de una API utilizando FastAPI. Aquí veremos cómo desglosar cada función necesaria y cómo exponerlas mediante endpoints para interactuar con ellas de forma efectiva.

Para comenzar, se debe dividir la tarea de creación de una operación en tres funciones principales. Primero, es necesario implementar una función para obtener el siguiente ID disponible en la base de datos, lo que permite garantizar la unicidad del identificador de cada tarea. La implementación de esta función en Python sería la siguiente:

python
Copy code
def get_next_id():
    try:
        with open(DATABASE_FILENAME, "r") as csvfile:
            reader = csv.DictReader(csvfile)
            max_id = max(
                int(row["id"]) for row in reader
            )
        return max_id + 1
    except (FileNotFoundError, ValueError):
        return 1

Con esta función, se obtiene el ID más alto en el archivo CSV y se incrementa en 1 para asignar un nuevo ID a la tarea a crear. Si el archivo no existe o está vacío, la función devuelve 1, comenzando la numeración desde allí.

Una vez obtenido el ID, el siguiente paso es definir una función que escriba una tarea en el archivo CSV:

python
Copy code
def write_task_into_csv(task: TaskWithID):
    with open(DATABASE_FILENAME, mode="a", newline="") as file:
        writer = csv.DictWriter(
            file,
            fieldnames=column_fields,
        )
        writer.writerow(task.model_dump())

Esta función recibe una tarea con su ID asignado y la escribe en el archivo CSV correspondiente.

Luego, se puede crear la función que combina las dos operaciones anteriores para crear la tarea:

python
Copy code
def create_task(task: Task) -> TaskWithID:
    id = get_next_id()
    task_with_id = TaskWithID(
        id=id, **task.model_dump()
    )
    write_task_into_csv(task_with_id)
    return task_with_id

Aquí, primero se obtiene el ID disponible, luego se agrega a la tarea y se guarda en el archivo CSV.

El siguiente paso es implementar la función que permite modificar una tarea existente:

python
Copy code
def modify_task(id: int, task: dict) -> Optional[TaskWithID]:
    updated_task: Optional[TaskWithID] = None
    tasks = read_all_tasks()
    for number, task_ in enumerate(tasks):
        if task_.id == id:
            tasks[number] = updated_task = task_.model_copy(update=task)
    
    with open(DATABASE_FILENAME, mode="w", newline="") as csvfile:
        writer = csv.DictWriter(
            csvfile, fieldnames=column_fields,
        )
        writer.writeheader()
        for task in tasks:
            writer.writerow(task.model_dump())
    
    if updated_task:
        return updated_task

Esta función busca la tarea por su ID, la actualiza con los nuevos valores y luego reescribe el archivo CSV con los datos modificados.

Finalmente, para eliminar una tarea por su ID, se debe crear una función que elimine el registro correspondiente:

python
Copy code
def remove_task(id: int) -> bool:
    deleted_task: Optional[Task] = None
    tasks = read_all_tasks()
    
    with open(DATABASE_FILENAME, mode="w", newline="") as csvfile:
        writer = csv.DictWriter(
            csvfile, fieldnames=column_fields,
        )
        writer.writeheader()
        for task in tasks:
            if task.id == id:
                deleted_task = task
                continue
            writer.writerow(task.model_dump())
    
    if deleted_task:
        dict_task_without_id = deleted_task.model_dump()
        del dict_task_without_id["id"]
        return Task(**dict_task_without_id)

En este caso, la tarea a eliminar se busca por ID, y una vez eliminada, se reescribe el archivo CSV sin esa tarea.

Con estas operaciones CRUD básicas implementadas, el siguiente paso es exponerlas a través de los endpoints de la API. En FastAPI, esto se hace de manera sencilla utilizando los decoradores adecuados para cada operación HTTP (GET, POST, PUT, DELETE). A continuación, se describen los endpoints necesarios para interactuar con las tareas:

1. Para listar todas las tareas, se utiliza el siguiente endpoint:

python
Copy code
@app.get("/tasks", response_model=list[TaskWithID])
def get_tasks():
    tasks = read_all_tasks()
    return tasks

2. Para obtener una tarea específica por su ID:

python
Copy code
@app.get("/task/{task_id}")
def get_task(task_id: int):
    task = read_task(task_id)
    if not task:
        raise HTTPException(
            status_code=404, detail="task not found"
        )
    return task

3. Para agregar una nueva tarea:

python
Copy code
@app.post("/task", response_model=TaskWithID)
def add_task(task: Task):
    return create_task(task)

4. Para actualizar una tarea existente, se usa el siguiente endpoint, que permite modificar campos específicos de la tarea:

python
Copy code
@app.put("/task/{task_id}", response_model=TaskWithID)
def update_task(task_id: int, task_update: UpdateTask):
    modified = modify_task(
        task_id, task_update.model_dump(exclude_unset=True),
    )
    if not modified:
        raise HTTPException(
            status_code=404, detail="task not found"
        )
    return modified

5. Finalmente, para eliminar una tarea por su ID:

python
Copy code
@app.delete("/task/{task_id}", response_model=Task)
def delete_task(task_id: int):
    removed_task = remove_task(task_id)
    if not removed_task:
        raise HTTPException(
            status_code=404, detail="task not found"
        )
    return removed_task

Una vez definidos estos endpoints, FastAPI genera automáticamente documentación interactiva, lo que facilita la prueba y el uso de la API. Se puede acceder a esta documentación a través de http://localhost:8000/docs, donde el usuario podrá experimentar creando, actualizando, listando y eliminando tareas.

A la hora de probar la API, es recomendable utilizar herramientas como pytest para escribir pruebas unitarias que aseguren el correcto funcionamiento de cada operación. Además, es una buena práctica utilizar una base de datos de pruebas separada para evitar interferir con la base de datos de producción.

En resumen, FastAPI proporciona una forma rápida y eficiente de construir y exponer operaciones CRUD en una API RESTful. La simplicidad de la implementación, combinada con la poderosa capacidad de FastAPI para gestionar rutas y validaciones de datos, permite desarrollar aplicaciones robustas con una mínima cantidad de código.

¿Cómo manejar transacciones y concurrencia en bases de datos?

Uno de los retos fundamentales en los sistemas de bases de datos es manejar transacciones concurrentes provenientes de múltiples usuarios, mientras se preserva la consistencia e integridad de los datos. Las transacciones pueden tener diferentes requisitos en cuanto a cómo acceden y modifican los datos, así como la forma en que se enfrentan a otras transacciones que podrían entrar en conflicto con ellas. Un enfoque común para gestionar la concurrencia es el uso de bloqueos, que son mecanismos diseñados para prevenir operaciones no autorizadas o incompatibles sobre los datos. Sin embargo, el uso de bloqueos puede introducir compensaciones entre el rendimiento, la disponibilidad y la corrección de los datos.

Dependiendo de las necesidades del negocio, algunas transacciones pueden requerir bloqueos durante períodos más largos o en niveles de granularidad diferentes, como a nivel de tabla o de fila. Por ejemplo, SQLite solo permite bloqueos a nivel de base de datos, mientras que PostgreSQL permite bloqueos a nivel de tabla o de fila. Este enfoque permite manejar transacciones concurrentes de manera más eficiente y precisa.

Otro concepto clave en la gestión de transacciones concurrentes es el nivel de aislamiento, que define el grado en el que una transacción debe estar aislada de los efectos de otras transacciones concurrentes. Los niveles de aislamiento son cruciales para garantizar que las transacciones mantengan la consistencia de los datos a pesar de los accesos y modificaciones simultáneos por parte de múltiples usuarios.

El estándar SQL define cuatro niveles de aislamiento, cada uno de los cuales ofrece distintos compromisos entre la concurrencia y la consistencia de los datos:

1. READ UNCOMMITTED: Las transacciones a este nivel permiten leer datos "sucios", es decir, cambios no confirmados realizados por otras transacciones concurrentes. Pueden ocurrir lecturas no repetibles y lecturas fantasma. Este nivel proporciona la mayor concurrencia pero la menor consistencia de los datos.

2. READ COMMITTED: Las transacciones a este nivel solo pueden ver los cambios confirmados por otras transacciones. No permiten lecturas sucias. Sin embargo, es posible que ocurran lecturas no repetibles. Este nivel busca un balance entre concurrencia y consistencia.

3. REPEATABLE READ: Las transacciones en este nivel ven una instantánea consistente de los datos durante toda la transacción. Los cambios confirmados por otras transacciones después de que haya comenzado la transacción no son visibles. Se previenen las lecturas no repetibles, pero pueden ocurrir lecturas fantasma. Este nivel proporciona una consistencia más fuerte a costa de algo de concurrencia.

4. SERIALIZABLE: Las transacciones a este nivel se comportan como si se ejecutaran de forma serial, una tras otra. Proporcionan el nivel más alto de consistencia de los datos, previniendo tanto lecturas no repetibles como lecturas fantasma. Sin embargo, esto puede reducir la concurrencia debido a un mayor uso de bloqueos.

Algunas bases de datos como SQLite permiten un solo nivel de aislamiento, mientras que otras como MySQL y PostgreSQL ofrecen todos los niveles de transacción definidos. En bases de datos que soportan estos niveles, es posible configurar el nivel de aislamiento al inicializar la conexión o el motor de la base de datos. Por ejemplo, si se desea establecer el nivel de aislamiento como "REPEATABLE READ" en PostgreSQL, se puede inicializar la conexión de la siguiente forma:

python
Copy code
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

eng = create_engine(
    "postgresql+psycopg2://usuario:contraseña@localhost/base_datos", 
    isolation_level="REPEATABLE READ"
)
Session = sessionmaker(eng)

Todas estas opciones de bloqueo e aislamiento tienen un impacto significativo en la arquitectura y diseño de los sistemas de bases de datos, ya que no todas las bases de datos SQL soportan las mismas características. Es esencial comprender los principios y mejores prácticas de las estrategias de bloqueo y cómo se relacionan con el comportamiento de las transacciones y la lógica del negocio.

La gestión adecuada de la concurrencia y las transacciones también implica la capacidad de manejar correctamente los conflictos que puedan surgir, lo cual no solo depende de la base de datos y las herramientas que utilices, sino también de las decisiones de diseño que tomes en función de los requisitos específicos de tu aplicación. Por lo tanto, al diseñar un sistema que maneje transacciones concurrentes, es crucial considerar cómo equilibrar la necesidad de acceso rápido y simultáneo a los datos con la preservación de su integridad y consistencia.

Además, aunque el uso de bloqueos e índices es importante, también es fundamental considerar otros aspectos del diseño de la base de datos, como el tipo de consultas que se ejecutan con mayor frecuencia y cómo se pueden optimizar para minimizar los bloqueos innecesarios y evitar cuellos de botella. La correcta implementación de las transacciones y su aislamiento puede ser la clave para que un sistema sea tanto eficiente como confiable, evitando errores y asegurando la estabilidad del sistema ante accesos concurrentes.

¿Cómo implementar un sistema de chat en tiempo real usando WebSockets con FastAPI?

La implementación de un sistema de chat en tiempo real con WebSockets es un proceso eficiente y directo utilizando FastAPI. WebSockets permiten una comunicación bidireccional constante entre el cliente y el servidor, lo que resulta ideal para aplicaciones de chat. En esta sección, describiremos cómo implementar dicho sistema utilizando FastAPI y WebSockets.

En primer lugar, se crea un connection manager, que es responsable de gestionar las conexiones activas de los clientes y facilitar el envío de mensajes. El administrador de conexiones tendrá varias funciones clave: la conexión y desconexión de WebSockets, el envío de mensajes a un solo cliente o a todos los clientes activos, y la notificación de desconexiones.

La implementación del manager se realiza con las siguientes funciones esenciales:

1. Conexión: Cuando un cliente se conecta, se añade su WebSocket a una lista de conexiones activas. Esto se hace mediante el método connect, que acepta como parámetro el WebSocket de la conexión.

2. Desconexión: Cuando un cliente se desconecta, el WebSocket correspondiente se elimina de la lista de conexiones activas. Esta funcionalidad se implementa en el método disconnect.

3. Mensaje personal: Para enviar un mensaje a un cliente específico, se utiliza el método send_personal_message. Este método recibe el WebSocket del destinatario y el mensaje que se desea enviar.

4. Difusión de mensajes: El método broadcast se encarga de enviar un mensaje a todos los WebSockets activos, excluyendo uno en particular si así se especifica. Esto permite que los mensajes sean difundidos a todos los participantes en el chat, excepto al remitente del mensaje.

Una vez que el connection manager está implementado, se puede configurar el endpoint de WebSocket en FastAPI. Este endpoint manejará las conexiones de los clientes al chat y las interacciones entre ellos.

Creación del endpoint de WebSocket

El endpoint de WebSocket se define en un archivo separado, por ejemplo chat.py, y se inicializa el connection manager en este archivo. A continuación, se establece el router para gestionar las rutas y se crea el endpoint de WebSocket. Este endpoint acepta como parámetro el nombre de usuario del cliente y gestiona las interacciones a través de los métodos del connection manager.

python
Copy code
from fastapi import WebSocket, WebSocketDisconnect
from app.ws_manager import ConnectionManager

conn_manager = ConnectionManager()

@router.websocket("/chatroom/{username}")
async def chatroom_endpoint(websocket: WebSocket, username: str):
    await conn_manager.connect(websocket)
    await conn_manager.broadcast(f"{username} joined the chat", exclude=websocket)
    
    try:
        while True:
            data = await websocket.receive_text()
            await conn_manager.broadcast({"sender": username, "message": data}, exclude=websocket)
            await conn_manager.send_personal_message({"sender": "You", "message": data}, websocket)
    except WebSocketDisconnect:
        conn_manager.disconnect(websocket)
        await conn_manager.broadcast({"sender": "system", "message": f"Client #{username} left the chat"})

En este código, cuando un cliente se conecta, el servidor le envía un mensaje a todos los participantes del chat para notificarles de la llegada del nuevo usuario. Los clientes pueden luego intercambiar mensajes en tiempo real, y el servidor se asegura de que cada mensaje enviado se difunda a todos los demás participantes, con excepción del remitente.

Configuración de la interfaz de usuario

Una vez que el endpoint de WebSocket está listo, se debe proporcionar una interfaz de usuario para que los clientes puedan conectarse y participar en el chat. Esta interfaz se genera a través de un archivo HTML, que se almacena en la carpeta de plantillas del proyecto. En este caso, la página HTML incluye un campo para ingresar el nombre de usuario y un área para enviar mensajes. Además, se embebe código JavaScript en el archivo HTML para manejar la interacción con el WebSocket del servidor.

html
Copy code
<!DOCTYPE html>
<html lang="es">
<head>
    <meta charset="UTF-8">
    <title>WebSocket Chat</title>
</head>
<body>
    <h2>Bienvenido al chat</h2>
    <p>Tu ID: <span id="username"></span></p>
    <textarea id="message" rows="4" cols="50"></textarea>
    <button id="send">Enviar</button>

    <script>
        const socket = new WebSocket("ws://localhost:8000/chatroom/username");
        socket.onmessage = function(event) {
            console.log(event.data);
        };
        document.getElementById('send').addEventListener('click', function() {
            const message = document.getElementById('message').value;
            socket.send(message);
        });
    </script>
</body>
</html>

En este archivo HTML, se define un área para ingresar el nombre del usuario, un área de texto para los mensajes y un botón para enviarlos. El WebSocket se conecta al servidor, y cuando se hace clic en el botón de envío, el mensaje se transmite al servidor.

Notas importantes

El sistema de chat implementado aquí se basa en un protocolo WebSocket, que garantiza la comunicación continua entre el cliente y el servidor. Este tipo de comunicación es más eficiente que las solicitudes HTTP tradicionales, ya que no requiere que el cliente realice nuevas solicitudes para recibir mensajes.

Es crucial asegurarse de manejar adecuadamente las desconexiones de los usuarios. El servidor debe ser capaz de detectar cuándo un cliente se desconecta de manera inesperada y notificar a los demás usuarios sobre su salida. Además, el sistema debe ser capaz de manejar múltiples conexiones simultáneas sin comprometer el rendimiento.

Finalmente, aunque el código descrito hasta ahora cubre la funcionalidad básica de un chat en tiempo real, existen otras características que podrían agregarse, como la autenticación de los usuarios, la capacidad de enviar mensajes privados o la persistencia de los mensajes a través de una base de datos.