¿Cómo configurar una base de datos SQL con FastAPI y SQLAlchemy para operaciones asíncronas?

La integración de FastAPI con bases de datos SQL a través de SQLAlchemy permite crear aplicaciones web eficientes y escalables, aprovechando la potencia de las bases de datos relacionales junto con la capacidad de ejecución asíncrona de Python. En este proceso, se configura un sistema que permite interactuar con bases de datos como SQLite, PostgreSQL o MySQL, de manera asíncrona, optimizando la eficiencia y la escalabilidad de las aplicaciones web.

Para comenzar, es necesario instalar los paquetes correspondientes. Esto se puede hacer fácilmente con el siguiente comando en la terminal:

bash
Copy code
$ pip install fastapi[all] "sqlalchemy>=2.0.0" aiosqlite

Una vez que los paquetes estén correctamente instalados, se puede proceder a configurar la conexión a la base de datos SQL. Los pasos generales incluyen la creación de clases de mapeo que coincidan con las tablas de la base de datos, la creación de capas de abstracción, un motor de conexión y una sesión para interactuar con la base de datos, y finalmente la inicialización de la conexión en el inicio del servidor.

Creación de las clases de mapeo de la base de datos

python
Copy code
from sqlalchemy import Column, Float, ForeignKey, Table
from sqlalchemy.orm import (DeclarativeBase, Mapped, mapped_column)
class Base(DeclarativeBase):
    pass
class Ticket(Base):
    __tablename__ = "tickets"
    id: Mapped[int] = mapped_column(primary_key=True)
    price: Mapped[float] = mapped_column(nullable=True)
    show: Mapped[str | None]
    user: Mapped[str | None]

Este modelo de clase Ticket servirá para reflejar la estructura de la tabla tickets en la base de datos.

Creación de las capas de abstracción

Primero, creamos una función que devuelva el motor de conexión. En un archivo nuevo llamado db_connection.py, dentro de la carpeta de la aplicación, se debe escribir lo siguiente:

python
Copy code
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite:///.database.db"
def get_engine():
    return create_async_engine(SQLALCHEMY_DATABASE_URL, echo=True)

En este caso, se está utilizando SQLite como base de datos, a través de aiosqlite, una librería asincrónica que permite ejecutar operaciones de forma asíncrona utilizando asyncio.

Luego, creamos un objeto AsyncSessionLocal que configurará la sesión de la base de datos para ser asíncrona:

python
Copy code
from sqlalchemy.ext.asyncio import AsyncSession

AsyncSessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=get_engine(),
    class_=AsyncSession,
)
async def get_db_session():
    async with AsyncSessionLocal() as session:
        yield session

La función get_db_session servirá como una dependencia para cada uno de los puntos finales (endpoints) que interactúan con la base de datos.

Inicialización de la conexión a la base de datos

Finalmente, necesitamos configurar el servidor FastAPI para que inicie la conexión a la base de datos al iniciar el servidor. En el archivo main.py, que se encuentra en la carpeta de la aplicación, se puede hacer lo siguiente:

python
Copy code
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.database import Base
from app.db_connection import AsyncSessionLocal, get_db_session
@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = get_engine()
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()
app = FastAPI(lifespan=lifespan)

Aquí, hemos utilizado el parámetro lifespan para especificar acciones del servidor al evento de inicio, asegurando que las tablas de la base de datos sean creadas si aún no existen.

Implementación de las operaciones CRUD

Por ejemplo, para agregar un nuevo ticket:

python
Copy code
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from app.database import Ticket
async def create_ticket(
    db_session: AsyncSession, 
    show_name: str, 
    user: str = None, 
    price: float = None
) -> int:
    ticket = Ticket(show=show_name, user=user, price=price)
    async with db_session.begin():
        db_session.add(ticket)
        await db_session.flush()
        ticket_id = ticket.id
        await db_session.commit()
    return ticket_id

Para obtener un ticket:

python
Copy code
async def get_ticket(db_session: AsyncSession, ticket_id: int) -> Ticket | None:

    query = select(Ticket).where(Ticket.id == ticket_id)
    async with db_session as session:
        tickets = await session.execute(query)
        return tickets.scalars().first()

Para actualizar el precio de un ticket:

python
Copy code
async def update_ticket_price(
    db_session: AsyncSession, 
    ticket_id: int, 
    new_price: float
) -> bool:
    query = update(Ticket).where(Ticket.id == ticket_id).values(price=new_price)
    async with db_session as session:
        ticket_updated = await session.execute(query)
        await session.commit()
        return ticket_updated.rowcount > 0

Consejos adicionales

Además, al integrar con bases de datos más grandes, como PostgreSQL o MySQL, se debe verificar que los controladores necesarios (como aiomysql o asyncpg) estén instalados y configurados correctamente para garantizar la compatibilidad con la funcionalidad asíncrona.

¿Cómo configurar el entorno de desarrollo para trabajar con FastAPI?

Para comenzar a trabajar con FastAPI, es esencial tener el entorno adecuado configurado en tu sistema. FastAPI, siendo un marco de desarrollo web basado en Python, requiere de ciertas herramientas y componentes para que su instalación y uso sean lo más eficientes y sin problemas posible. A continuación, se describen los pasos clave para configurar correctamente este entorno de desarrollo, cubriendo desde la instalación de Python hasta las herramientas necesarias para un flujo de trabajo ágil.

Una vez que Python está instalado, es necesario verificar que también tienes pip, el gestor de paquetes de Python, correctamente instalado. Para ello, puedes ejecutar el comando pip --version en la terminal. Si ves la versión de pip, entonces puedes continuar con la instalación de FastAPI y sus dependencias. FastAPI se instala a través de pip con el comando pip install fastapi[all], que no solo instala FastAPI sino también su servidor ASGI, Uvicorn, que es esencial para ejecutar las aplicaciones FastAPI de forma rápida y eficiente. Uvicorn es un servidor asincrónico que proporciona un alto rendimiento para manejar solicitudes HTTP.

Además de FastAPI y Uvicorn, y el IDE, es importante integrar herramientas de control de versiones como Git. Git es esencial para gestionar los cambios en tu código, especialmente si trabajas en equipo o planeas hacer mejoras continuas en tu proyecto. GitHub, además, te permitirá alojar tus repositorios y colaborar con otros desarrolladores. Si aún no tienes una cuenta de GitHub, crear una es rápido y sencillo. Con Git y GitHub correctamente configurados, podrás realizar un seguimiento preciso de los cambios, revertir a versiones anteriores y colaborar de manera eficiente.

En cuanto a la prueba de tus aplicaciones, FastAPI facilita este proceso gracias a la generación automática de una interfaz de usuario de Swagger para que puedas interactuar con tus endpoints API directamente desde tu navegador. Swagger UI es una herramienta que FastAPI genera por defecto para que puedas probar y explorar los puntos de entrada de tu aplicación de forma visual y sin necesidad de escribir código adicional para las pruebas.

Es importante entender que, más allá de instalar las herramientas necesarias, es crucial tener un buen conocimiento sobre cómo usar Git para control de versiones, cómo organizar tu código de manera eficiente y cómo aprovechar las características avanzadas de tu IDE elegido, como el autocompletado de código, la depuración y la integración con servicios de control de versiones.

Otro punto clave que deberías dominar es la comprensión de cómo manejar correctamente las excepciones y errores en FastAPI. Aunque este aspecto no se detalla en este capítulo, es fundamental para construir aplicaciones robustas que respondan adecuadamente ante fallos. FastAPI permite definir y personalizar modelos de solicitud y respuesta que no solo validan los datos entrantes, sino que también te permiten manejar errores y excepciones de manera eficiente, ofreciendo respuestas claras y útiles al cliente.

Es esencial también familiarizarse con las mejores prácticas en desarrollo web, como la correcta estructuración del proyecto, la modularidad del código, el uso de entornos virtuales para aislar las dependencias y cómo gestionar las bases de datos de forma efectiva. Además, no debes subestimar la importancia de realizar pruebas unitarias y funcionales para asegurar la estabilidad y calidad de tu código.

¿Cómo manejar errores y excepciones en conexiones WebSocket con FastAPI?

El manejo adecuado de conexiones WebSocket en aplicaciones FastAPI no se ограничивается лишь установлением соединения и разрывом связи. Важно обеспечить также правильное управление ошибками, что способствует созданию надежных и отзывчивых приложений. WebSocket-соединения могут быть подвержены множеству ошибок и исключений, которые возникают в процессе их жизненного цикла. Эти ошибки могут включать сбои при подключении, ошибки разбора сообщений и неожиданные разрывы соединения. Важно не только правильно обработать такие ситуации, но и грамотно сообщить об ошибках клиенту.

Примером может служить улучшение обработки WebSocket-соединения с использованием метода websocket.close(), где можно передать код статуса и причину закрытия соединения. Рассмотрим следующий пример:

python
Copy code
from fastapi import status
@app.websocket("/ws")

async def chatroom(websocket: WebSocket):

    if not websocket.headers.get("Authorization"):
        return await websocket.close()
    await websocket.accept()
    await websocket.send_text("Welcome to the chat room!")
    try:
        while True:
            data = await websocket.receive_text()
            logger.info(f"Message received: {data}")
            if data == "disconnect":
                logger.warn("Disconnecting...")
                return await websocket.close(
                    code=status.WS_1000_NORMAL_CLOSURE,
                    reason="Disconnecting..."
                )
    except WebSocketDisconnect:
        logger.warn("Connection closed by the client")

В данном примере при получении строки "disconnect" сервер корректно завершает соединение с кодом состояния 1000 и сопровождает разрыв поясняющим сообщением. Такая практика помогает избежать неоправданных ошибок в процессе работы приложения, и клиент получает ясную информацию о причине завершения сессии.

python
Copy code
@app.websocket("/ws")

async def ws_endpoint(websocket: WebSocket):

    await websocket.accept()
    await websocket.send_text("Welcome to the chat room!")
    try:
        while True:
            data = await websocket.receive_text()
            logger.info(f"Message received: {data}")
            if data == "disconnect":
                logger.warn("Disconnecting...")
                return await websocket.close(
                    code=status.WS_1000_NORMAL_CLOSURE,
                    reason="Disconnecting..."
                )
            if "bad message" in data:
                raise WebSocketException(
                    code=status.WS_1008_POLICY_VIOLATION,
                    reason="Inappropriate message"
                )
    except WebSocketDisconnect:
        logger.warn("Connection closed by the client")

В этом примере если клиент отправляет сообщение, содержащее строку "bad message", сервер немедленно разрывает соединение, предоставляя соответствующий код и сообщение об ошибке. Этот подход полезен для защиты от нежелательных действий со стороны клиента и соблюдения строгих правил безопасности.

Для более глубокого понимания и использования WebSocket в FastAPI стоит обратить внимание на следующие моменты:

• Стандарты и спецификации WebSocket. Поскольку WebSocket — это относительно новый протокол по сравнению с HTTP, важно ознакомиться с его стандартами и кодами завершения соединений. Например, статус-коды для WebSocket можно найти в WebSocket Close Code Number Registry.

• Совместимость с браузерами. Для обеспечения совместимости с различными браузерами стоит ознакомиться с документацией о событиях WebSocket CloseEvent.

• Документация FastAPI. В FastAPI есть отдельная документация для работы с исключениями WebSocket, в том числе для использования WebSocketException, что делает обработку ошибок еще более гибкой и персонализируемой. Более подробную информацию можно найти в документации FastAPI.

Кроме того, следует помнить, что ошибки, такие как неправильные сообщения, неправильный формат данных или нарушение условий безопасности, должны обрабатываться на стороне сервера с должной информативностью для клиента. Важно сообщать о причине разрыва соединения, чтобы клиент мог правильно интерпретировать ошибку и предпринять необходимые действия. Также стоит помнить о возможности логирования этих ошибок для дальнейшего анализа и устранения потенциальных проблем.