Patrones y prácticas para usar SQLAlchemy 2.0 con FastAPI

Si bien Django y Flask siguen siendo las primeras opciones para muchos ingenieros de Python, FastAPI ya ha sido reconocida como una elección innegablemente confiable. Es un marco estructurado altamente flexible y bien optimizado que brinda al desarrollador infinitas posibilidades para crear aplicaciones de back-end.

Trabajar con bases de datos es un aspecto esencial de la mayoría de las aplicaciones de back-end. Como resultado, el ORM juega un papel fundamental en el código de back-end. Sin embargo, a diferencia de Django, FastAPI no tiene un ORM incorporado. Es enteramente responsabilidad del desarrollador seleccionar una biblioteca adecuada e integrarla en el código base.

Los ingenieros de Python consideran ampliamente que SQLAlchemy es el ORM más popular disponible. Es una biblioteca legendaria que ha estado en uso desde 2006 y ha sido adoptada por miles de proyectos. En 2023, recibió una importante actualización a la versión 2.0. Al igual que FastAPI, SQLAlchemy proporciona a los desarrolladores potentes funciones y utilidades sin obligarlos a usarlas de una manera específica. Esencialmente, es un conjunto de herramientas versátil que permite a los desarrolladores usarlo como mejor les parezca.

FastAPI y SQLAlchemy son una combinación perfecta. Ambas son tecnologías confiables, de alto rendimiento y modernas, que permiten la creación de aplicaciones poderosas y únicas. Este artículo explora la creación de una aplicación de back-end FastAPI que utiliza SQLAlchemy 2.0 como ORM. El contenido cubre:

• construyendo modelos usando Mapped y mapped_column
• definiendo un modelo abstracto
• manejo de sesión de base de datos
• utilizando el ORM
• creando una clase de repositorio común para todos los modelos
• preparación de una configuración de prueba y adición de pruebas

Luego, podrá combinar la aplicación FastAPI con SQLAlchemy ORM fácilmente. Además, se familiarizará con las mejores prácticas y patrones para crear aplicaciones bien estructuradas, sólidas y de alto rendimiento.

requisitos previos

Los ejemplos de código incluidos en el artículo provienen del proyecto alchemist , que es una API básica para crear y leer objetos de ingredientes y pociones. El enfoque principal del artículo es explorar la combinación de FastAPI y SQLAlchemy. No cubre otros temas, tales como:

• configurar la configuración de Docker
• iniciando el servidor uvicorn
• configurando pelusa

Si está interesado en estos temas, puede explorarlos por su cuenta examinando el código base. Para acceder al repositorio de código del proyecto alquimista, siga este enlace aquí . Además, puede encontrar la estructura de archivos del proyecto a continuación:

 alchemist ├─ alchemist │ ├─ api │ │ ├─ v1 │ │ │ ├─ __init__.py │ │ │ └─ routes.py │ │ ├─ v2 │ │ │ ├─ __init__.py │ │ │ ├─ dependencies.py │ │ │ └─ routes.py │ │ ├─ __init__.py │ │ └─ models.py │ ├─ database │ │ ├─ __init__.py │ │ ├─ models.py │ │ ├─ repository.py │ │ └─ session.py │ ├─ __init__.py │ ├─ app.py │ └─ config.py ├─ requirements │ ├─ base.txt │ └─ dev.txt ├─ scripts │ ├─ create_test_db.sh │ ├─ migrate.py │ └─ run.sh ├─ tests │ ├─ conftest.py │ └─ test_api.py ├─ .env ├─ .gitignore ├─ .pre-commit-config.yaml ├─ Dockerfile ├─ Makefile ├─ README.md ├─ docker-compose.yaml ├─ example.env └─ pyproject.toml

Aunque el árbol puede parecer grande, algunos de los contenidos no son relevantes para el punto principal de este artículo. Además, el código puede parecer más simple de lo necesario en ciertas áreas. Por ejemplo, el proyecto carece de:

• etapa de producción en el Dockerfile
• instalación de alambique para migraciones
• subdirectorios para pruebas

Esto se hizo intencionalmente para reducir la complejidad y evitar gastos generales innecesarios. Sin embargo, es importante tener en cuenta estos factores si se trata de un proyecto más listo para la producción.

Requisitos de la API

Al comenzar a desarrollar una aplicación, es fundamental considerar los modelos que usará su aplicación. Estos modelos representarán los objetos y entidades con los que trabajará su aplicación y estarán expuestos en la API. En el caso de la aplicación alquimista, hay dos entidades: ingredientes y pociones. La API debería permitir la creación y recuperación de estas entidades. El archivo alchemist/api/models.py contiene los modelos que se utilizarán en la API:

 import uuid from pydantic import BaseModel, Field class Ingredient(BaseModel): """Ingredient model.""" pk: uuid.UUID name: str class Config: orm_mode = True class IngredientPayload(BaseModel): """Ingredient payload model.""" name: str = Field(min_length=1, max_length=127) class Potion(BaseModel): """Potion model.""" pk: uuid.UUID name: str ingredients: list[Ingredient] class Config: orm_mode = True class PotionPayload(BaseModel): """Potion payload model.""" name: str = Field(min_length=1, max_length=127) ingredients: list[uuid.UUID] = Field(min_items=1)

La API devolverá modelos Ingredient y Potion . Establecer orm_mode en True en la configuración facilitará el trabajo con los objetos SQLAlchemy en el futuro. Los modelos Payload se utilizarán para crear nuevos objetos.

El uso de pydantic hace que las clases sean más detalladas y claras en sus roles y funciones. Ahora es el momento de crear los modelos de base de datos.

Declaración de modelos

Un modelo es esencialmente una representación de algo. En el contexto de las API, los modelos representan lo que espera el backend en el cuerpo de la solicitud y lo que devolverá en los datos de respuesta. Los modelos de bases de datos, por otro lado, son más complejos y representan las estructuras de datos almacenadas en la base de datos y los tipos de relaciones entre ellos.

El archivo alchemist/database/models.py contiene modelos para objetos de ingredientes y pociones:

 import uuid from sqlalchemy import Column, ForeignKey, Table, orm from sqlalchemy.dialects.postgresql import UUID class Base(orm.DeclarativeBase): """Base database model.""" pk: orm.Mapped[uuid.UUID] = orm.mapped_column( primary_key=True, default=uuid.uuid4, ) potion_ingredient_association = Table( "potion_ingredient", Base.metadata, Column("potion_id", UUID(as_uuid=True), ForeignKey("potion.pk")), Column("ingredient_id", UUID(as_uuid=True), ForeignKey("ingredient.pk")), ) class Ingredient(Base): """Ingredient database model.""" __tablename__ = "ingredient" name: orm.Mapped[str] class Potion(Base): """Potion database model.""" __tablename__ = "potion" name: orm.Mapped[str] ingredients: orm.Mapped[list["Ingredient"]] = orm.relationship( secondary=potion_ingredient_association, backref="potions", lazy="selectin", )

Cada modelo en SQLAlchemy comienza con la clase DeclarativeBase . Heredar de él permite construir modelos de base de datos que sean compatibles con los verificadores de tipo Python.

También es una buena práctica crear un modelo abstracto (clase Base en este caso) que incluya campos obligatorios en todos los modelos. Estos campos incluyen la clave principal, que es un identificador único de cada objeto. El modelo abstracto a menudo también almacena las fechas de creación y actualización de un objeto, que se configuran automáticamente cuando se crea o actualiza un objeto. Sin embargo, el modelo Base se mantendrá simple.

Pasando al modelo Ingredient , el atributo __tablename__ especifica el nombre de la tabla de la base de datos, mientras que el campo name usa la nueva sintaxis de SQLAlchemy, lo que permite que los campos del modelo se declaren con anotaciones de tipo. Este enfoque conciso y moderno es poderoso y ventajoso para los verificadores de tipo y los IDE, ya que reconoce el campo name como una cadena.

Las cosas se vuelven más complejas en el modelo Potion . También incluye __tablename__ y atributos name , pero además de eso, almacena la relación con los ingredientes. El uso de Mapped[list["Ingredient"]] indica que la poción puede contener varios ingredientes y, en este caso, la relación es de muchos a muchos (M2M). Esto significa que un solo ingrediente se puede asignar a varias pociones.

M2M requiere una configuración adicional, que generalmente involucra la creación de una tabla de asociación que almacena las conexiones entre las dos entidades. En este caso, el objeto potion_ingredient_association almacena solo los identificadores del ingrediente y la poción, pero también podría incluir atributos adicionales, como la cantidad de un ingrediente específico necesario para la poción.

La función relationship configura la relación entre la poción y sus ingredientes. El argumento lazy especifica cómo deben cargarse los elementos relacionados. En otras palabras: ¿qué debe hacer SQLAlchemy con los ingredientes relacionados cuando busca una poción? Establecerlo en selectin significa que los ingredientes se cargarán con la poción, eliminando la necesidad de consultas adicionales en el código.

La construcción de modelos bien diseñados es crucial cuando se trabaja con un ORM. Una vez hecho esto, el siguiente paso es establecer la conexión con la base de datos.

Controlador de sesión

Cuando se trabaja con una base de datos, particularmente cuando se usa SQLAlchemy, es esencial comprender los siguientes conceptos:

• dialecto
• motor
• conexión
• grupo de conexiones
• sesión

De todos estos términos, el más importante es el motor . De acuerdo con la documentación de SQLAlchemy, el objeto del motor es responsable de conectar el Pool y Dialect para facilitar la conectividad y el comportamiento de la base de datos. En términos más simples, el objeto del motor es el origen de la conexión a la base de datos, mientras que la conexión proporciona funcionalidades de alto nivel como la ejecución de instrucciones SQL, la gestión de transacciones y la recuperación de resultados de la base de datos.

Una sesión es una unidad de trabajo que agrupa operaciones relacionadas dentro de una sola transacción. Es una abstracción sobre las conexiones de base de datos subyacentes y administra de manera eficiente las conexiones y el comportamiento transaccional.

Dialect es un componente que brinda soporte para un backend de base de datos específico. Actúa como intermediario entre SQLAlchemy y la base de datos, manejando los detalles de la comunicación. El proyecto alchemist utiliza Postgres como base de datos, por lo que el dialecto debe ser compatible con este tipo de base de datos específico.

El signo de interrogación final es el conjunto de conexiones . En el contexto de SQLAlchemy, un grupo de conexiones es un mecanismo que administra una colección de conexiones de base de datos. Está diseñado para mejorar el rendimiento y la eficiencia de las operaciones de la base de datos al reutilizar las conexiones existentes en lugar de crear otras nuevas para cada solicitud. Al reutilizar las conexiones, el conjunto de conexiones reduce la sobrecarga de establecer nuevas conexiones y eliminarlas, lo que da como resultado un mejor rendimiento.

Con ese conocimiento cubierto, ahora puede echar un vistazo al archivo alchemist/database/session.py , que contiene una función que se utilizará como dependencia para conectarse a la base de datos:

 from collections.abc import AsyncGenerator from sqlalchemy import exc from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.config import settings async def get_db_session() -> AsyncGenerator[AsyncSession, None]: engine = create_async_engine(settings.DATABASE_URL) factory = async_sessionmaker(engine) async with factory() as session: try: yield session await session.commit() except exc.SQLAlchemyError as error: await session.rollback() raise

El primer detalle importante a notar es que la función get_db_session es una función generadora. Esto se debe a que el sistema de dependencia de FastAPI admite generadores. Como resultado, esta función puede manejar tanto escenarios exitosos como fallidos.

Las dos primeras líneas de la función get_db_session crean un motor de base de datos y una sesión. Sin embargo, el objeto de sesión también se puede utilizar como administrador de contexto. Esto le da más control sobre posibles excepciones y resultados exitosos.

Aunque SQLAlchemy maneja el cierre de las conexiones, es una buena práctica declarar explícitamente cómo manejar la conexión una vez que se haya realizado. En la función get_db_session , la sesión se confirma si todo va bien y se revierte si se genera una excepción.

Es importante tener en cuenta que este código se basa en la extensión asyncio. Esta función de SQLAlchemy permite que la aplicación interactúe con la base de datos de forma asíncrona. Significa que las solicitudes a la base de datos no bloquearán otras solicitudes de API, lo que hace que la aplicación sea mucho más eficiente.

Una vez que se configuran los modelos y la conexión, el siguiente paso es asegurarse de que los modelos se agreguen a la base de datos.

Migraciones rápidas

Los modelos SQLAlchemy representan las estructuras de una base de datos. Sin embargo, simplemente crearlos no genera cambios inmediatos en la base de datos. Para realizar cambios, primero debe aplicarlos . Esto generalmente se hace usando una biblioteca de migración como alambique, que rastrea cada modelo y actualiza la base de datos en consecuencia.

Dado que no se planean más cambios en los modelos en este escenario, bastará con un script de migración básico. A continuación se muestra un código de ejemplo del archivo scripts/migrate.py .

 import asyncio import logging from sqlalchemy.ext.asyncio import create_async_engine from alchemist.config import settings from alchemist.database.models import Base logger = logging.getLogger() async def migrate_tables() -> None: logger.info("Starting to migrate") engine = create_async_engine(settings.DATABASE_URL) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("Done migrating") if __name__ == "__main__": asyncio.run(migrate_tables())

En pocas palabras, la función migrate_tables lee la estructura de los modelos y la recrea en la base de datos utilizando el motor SQLAlchemy. Para ejecutar este script, use el comando python scripts/migrate.py .

Los modelos ahora están presentes tanto en el código como en la base de datos y get_db_session puede facilitar las interacciones con la base de datos. Ahora puede comenzar a trabajar en la lógica de la API.

API con el ORM

Como se mencionó anteriormente, la API para ingredientes y pociones está destinada a admitir tres operaciones:

• creando objetos
• enumerar objetos
• recuperar objetos por ID

Gracias a los preparativos previos, todas estas funciones ya se pueden implementar con SQLAlchemy como ORM y FastAPI como marco web. Para comenzar, revisa la API de ingredientes ubicada en el archivo alchemist/api/v1/routes.py :

 import uuid from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.api import models from alchemist.database import models as db_models from alchemist.database.session import get_db_session router = APIRouter(prefix="/v1", tags=["v1"]) @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = db_models.Ingredient(**data.dict()) session.add(ingredient) await session.commit() await session.refresh(ingredient) return models.Ingredient.from_orm(ingredient) @router.get("/ingredients", status_code=status.HTTP_200_OK) async def get_ingredients( session: AsyncSession = Depends(get_db_session), ) -> list[models.Ingredient]: ingredients = await session.scalars(select(db_models.Ingredient)) return [models.Ingredient.from_orm(ingredient) for ingredient in ingredients] @router.get("/ingredients/{pk}", status_code=status.HTTP_200_OK) async def get_ingredient( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = await session.get(db_models.Ingredient, pk) if ingredient is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ingredient does not exist", ) return models.Ingredient.from_orm(ingredient)

En la API /ingredients , hay tres rutas disponibles. El punto final POST toma una carga útil de ingrediente como un objeto de un modelo creado previamente y una sesión de base de datos. La función del generador get_db_session inicializa la sesión y habilita las interacciones con la base de datos.

En el cuerpo de la función real, se llevan a cabo cinco pasos:

1. Se crea un objeto de ingrediente a partir de la carga útil entrante.
2. El método add del objeto de sesión agrega el objeto de ingrediente al sistema de seguimiento de sesiones y lo marca como pendiente de inserción en la base de datos.
3. La sesión está comprometida.
4. El objeto del ingrediente se actualiza para garantizar que sus atributos coincidan con el estado de la base de datos.
5. La instancia del ingrediente de la base de datos se convierte en la instancia del modelo API mediante el método from_orm .

Para una prueba rápida, se puede ejecutar un curl simple contra la aplicación en ejecución:

 curl -X 'POST' \ 'http://localhost:8000/api/v1/ingredients' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty water"}'

En la respuesta, debe haber un objeto de ingrediente que tenga una identificación proveniente de la base de datos:

 { "pk":"2eb255e9-2172-4c75-9b29-615090e3250d", "name":"Salty water" }

Aunque las múltiples capas de abstracción de SQLAlchemy pueden parecer innecesarias para una API simple, mantienen los detalles ORM separados y contribuyen a la eficiencia y escalabilidad de SQLAlchemy. Cuando se combina con asyncio, las características de ORM funcionan excepcionalmente bien en la API.

Los dos puntos finales restantes son menos complejos y comparten similitudes. Una parte que merece una explicación más profunda es el uso del método scalars dentro de la función get_ingredients . Al consultar la base de datos usando SQLAlchemy, el método execute se usa a menudo con una consulta como argumento. Mientras que el método execute devuelve tuplas en forma de fila, scalars devuelven entidades ORM directamente, lo que hace que el punto final sea más limpio.

Ahora, considere la API de pociones, en el mismo archivo:

 @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: data_dict = data.dict() ingredients = await session.scalars( select(db_models.Ingredient).where( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) ) potion = db_models.Potion(**data_dict, ingredients=list(ingredients)) session.add(potion) await session.commit() await session.refresh(potion) return models.Potion.from_orm(potion) @router.get("/potions", status_code=status.HTTP_200_OK) async def get_potions( session: AsyncSession = Depends(get_db_session), ) -> list[models.Potion]: potions = await session.scalars(select(db_models.Potion)) return [models.Potion.from_orm(potion) for potion in potions] @router.get("/potions/{pk}", status_code=status.HTTP_200_OK) async def get_potion( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: potion = await session.get(db_models.Potion, pk) if potion is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Potion does not exist", ) return models.Potion.from_orm(potion)

Los criterios de valoración GET para las pociones son idénticos a los de los ingredientes. Sin embargo, la función POST requiere código adicional. Esto se debe a que la creación de pociones implica incluir al menos un ID de ingrediente, lo que significa que los ingredientes deben buscarse y vincularse a la poción recién creada. Para lograr esto, se usa nuevamente el método scalars , pero esta vez con una consulta que especifica los ID de los ingredientes obtenidos. La parte restante del proceso de creación de la poción es idéntica a la de los ingredientes.

Para probar el punto final, nuevamente se puede ejecutar un comando curl.

 curl -X 'POST' \ 'http://localhost:8000/api/v1/potions' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty soup", "ingredients": ["0b4f1de5-e780-418d-a74d-927afe8ac954"}'

Da como resultado la siguiente respuesta:

 { "pk": "d4929197-3998-4234-a5f7-917dc4bba421", "name": "Salty soup", "ingredients": [ { "pk": "0b4f1de5-e780-418d-a74d-927afe8ac954", "name": "Salty water" } ] }

Es importante notar que cada ingrediente se representa como un objeto completo dentro de la poción, gracias al argumento lazy="selectin" especificado en la relación.

Las API son funcionales, pero hay un problema importante con el código. Si bien SQLAlchemy le brinda la libertad de interactuar con la base de datos como desee, no ofrece ninguna utilidad de "administrador" de alto nivel similar a Model.objects de Django. Como resultado, deberá crearlo usted mismo, que es esencialmente la lógica utilizada en las API de ingredientes y pociones. Sin embargo, si mantiene esta lógica directamente en los puntos finales sin extraerla en un espacio separado, terminará con una gran cantidad de código duplicado. Además, realizar cambios en las consultas o modelos será cada vez más difícil de administrar.

El próximo capítulo presenta el patrón de repositorio: una solución elegante para extraer código ORM.

Repositorio

El patrón de repositorio permite abstraer los detalles del trabajo con la base de datos. En el caso de utilizar SQLAlchemy, como en el ejemplo del alquimista, la clase repositorio sería la encargada de gestionar múltiples modelos e interactuar con la sesión de la base de datos.

Eche un vistazo al siguiente código del archivo alchemist/database/repository.py :

 import uuid from typing import Generic, TypeVar from sqlalchemy import BinaryExpression, select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models Model = TypeVar("Model", bound=models.Base) class DatabaseRepository(Generic[Model]): """Repository for performing database queries.""" def __init__(self, model: type[Model], session: AsyncSession) -> None: self.model = model self.session = session async def create(self, data: dict) -> Model: instance = self.model(**data) self.session.add(instance) await self.session.commit() await self.session.refresh(instance) return instance async def get(self, pk: uuid.UUID) -> Model | None: return await self.session.get(self.model, pk) async def filter( self, *expressions: BinaryExpression, ) -> list[Model]: query = select(self.model) if expressions: query = query.where(*expressions) return list(await self.session.scalars(query))

La clase DatabaseRepository contiene toda la lógica que se incluyó previamente en los puntos finales. La diferencia es que permite pasar la clase de modelo específica en el método __init__ , lo que permite reutilizar el código para todos los modelos en lugar de duplicarlo en cada punto final.

Además, DatabaseRepository utiliza genéricos de Python, con el tipo genérico Model limitado al modelo de base de datos abstracto. Esto permite que la clase del repositorio se beneficie más de la verificación de tipos estáticos. Cuando se usa con un modelo específico, los tipos de devolución de los métodos del repositorio reflejarán este modelo específico.

Debido a que el repositorio necesita usar la sesión de la base de datos, debe inicializarse junto con la dependencia get_db_session . Considere la nueva dependencia en el archivo alchemist/api/v2/dependencies.py :

 from collections.abc import Callable from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models, repository, session def get_repository(  model: type[models.Base], ) -> Callable[[AsyncSession], repository.DatabaseRepository]:  def func(session: AsyncSession = Depends(session.get_db_session)):    return repository.DatabaseRepository(model, session)  return func

En pocas palabras, la función get_repository es una fábrica de dependencias. Primero toma el modelo de base de datos con el que usará el repositorio. Luego, devuelve la dependencia, que se usará para recibir la sesión de la base de datos e inicializar el objeto del repositorio. Para obtener una mejor comprensión, consulte la nueva API del archivo alchemist/api/v2/routes.py . Solo muestra los puntos finales POST, pero debería ser suficiente para darle una idea más clara de cómo se mejora el código:

 from typing import Annotated from fastapi import APIRouter, Depends, status from alchemist.api import models from alchemist.api.v2.dependencies import get_repository from alchemist.database import models as db_models from alchemist.database.repository import DatabaseRepository router = APIRouter(prefix="/v2", tags=["v2"]) IngredientRepository = Annotated[  DatabaseRepository[db_models.Ingredient],  Depends(get_repository(db_models.Ingredient)), ] PotionRepository = Annotated[  DatabaseRepository[db_models.Potion],  Depends(get_repository(db_models.Potion)), ] @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, repository: IngredientRepository, ) -> models.Ingredient: ingredient = await repository.create(data.dict()) return models.Ingredient.from_orm(ingredient) @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, ingredient_repository: IngredientRepository, potion_repository: PotionRepository, ) -> models.Potion: data_dict = data.dict() ingredients = await ingredient_repository.filter( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) potion = await potion_repository.create({**data_dict, "ingredients": ingredients}) return models.Potion.from_orm(potion)

La primera característica importante a tener en cuenta es el uso de Annotated , una nueva forma de trabajar con las dependencias de FastAPI. Al especificar el tipo de retorno de la dependencia como DatabaseRepository[db_models.Ingredient] y declarar su uso con Depends(get_repository(db_models.Ingredient)) puede terminar usando anotaciones de tipo simple en el punto final: repository: IngredientRepository .

Gracias al repositorio, los puntos finales no tienen que almacenar toda la carga relacionada con ORM. Incluso en el caso de pociones más complicado, todo lo que necesita hacer es usar dos repositorios al mismo tiempo.

Puede preguntarse si inicializar dos repositorios inicializará la sesión dos veces. La respuesta es no. El sistema de dependencia FastAPI almacena en caché las mismas llamadas de dependencia en una sola solicitud. Esto significa que la inicialización de la sesión se almacena en caché y ambos repositorios usan exactamente el mismo objeto de sesión. Otra gran característica de la combinación de SQLAlchemy y FastAPI.

La API es completamente funcional y tiene una capa de acceso a datos reutilizable y de alto rendimiento. El siguiente paso es asegurarse de que se cumplan los requisitos escribiendo algunas pruebas de extremo a extremo.

Pruebas

Las pruebas juegan un papel crucial en el desarrollo de software. Los proyectos pueden contener pruebas unitarias, de integración y de extremo a extremo (E2E). Si bien generalmente es mejor tener una gran cantidad de pruebas unitarias significativas, también es bueno escribir al menos algunas pruebas E2E para garantizar que todo el flujo de trabajo funcione correctamente.

Para crear algunas pruebas E2E para la aplicación alchemist, se requieren dos bibliotecas adicionales:

• pytest para crear y ejecutar las pruebas
• httpx para realizar solicitudes asíncronas dentro de las pruebas

Una vez que estos están instalados, el siguiente paso es tener una base de datos de prueba separada. No desea que su base de datos predeterminada se contamine o se elimine. Dado que alchemist incluye una configuración de Docker, solo se necesita un script simple para crear una segunda base de datos. Eche un vistazo al código del archivo scripts/create_test_db.sh :

 #!/bin/bash psql -U postgres psql -c "CREATE DATABASE test"

Para que se ejecute el script, debe agregarse como un volumen al contenedor de Postgres. Esto se puede lograr incluyéndolo en la sección volumes del archivo docker-compose.yaml .

El paso final de preparación es crear accesorios de pytest dentro del archivo tests/conftest.py :

 from collections.abc import AsyncGenerator import pytest import pytest_asyncio from fastapi import FastAPI from httpx import AsyncClient from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.app import app from alchemist.config import settings from alchemist.database.models import Base from alchemist.database.session import get_db_session @pytest_asyncio.fixture() async def db_session() -> AsyncGenerator[AsyncSession, None]: """Start a test database session.""" db_name = settings.DATABASE_URL.split("/")[-1] db_url = settings.DATABASE_URL.replace(f"/{db_name}", "/test") engine = create_async_engine(db_url) async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.create_all) session = async_sessionmaker(engine)() yield session await session.close() @pytest.fixture() def test_app(db_session: AsyncSession) -> FastAPI: """Create a test app with overridden dependencies.""" app.dependency_overrides[get_db_session] = lambda: db_session return app @pytest_asyncio.fixture() async def client(test_app: FastAPI) -> AsyncGenerator[AsyncClient, None]: """Create an http client.""" async with AsyncClient(app=test_app, base_url="http://test") as client: yield client

Una cosa que es esencial cambiar en las pruebas es cómo la aplicación interactúa con la base de datos. Esto incluye no solo cambiar la URL de la base de datos, sino también asegurarse de que cada prueba esté aislada comenzando con una base de datos vacía.

El accesorio db_session logra ambos objetivos. Su cuerpo sigue los siguientes pasos:

1. Cree un motor con una URL de base de datos modificada.
2. Elimine todas las tablas existentes para asegurarse de que la prueba tenga una base de datos limpia.
3. Cree todas las tablas dentro de la base de datos (el mismo código que en el script de migración).
4. Crear y generar un objeto de sesión.
5. Cierre manualmente la sesión cuando finalice la prueba.

Aunque el último paso también podría implementarse como un administrador de contexto, el cierre manual funciona bien en este caso.

Los dos accesorios restantes deberían explicarse por sí mismos:

• test_app es la instancia de FastAPI del archivo alchemist/app.py , con la dependencia get_db_session reemplazada por el accesorio db_session
• client es el httpx AsyncClient que realizará solicitudes de API contra test_app

Con todo esto configurado, finalmente se pueden escribir las pruebas reales. Para ser conciso, el siguiente ejemplo del archivo tests/test_api.py muestra solo una prueba para crear un ingrediente:

 from fastapi import status class TestIngredientsAPI: """Test cases for the ingredients API.""" async def test_create_ingredient(self, client): response = await client.post("/api/v2/ingredients", json={"name": "Carrot"}) assert response.status_code == status.HTTP_201_CREATED pk = response.json().get("pk") assert pk is not None response = await client.get("/api/v2/ingredients") assert response.status_code == status.HTTP_200_OK assert len(response.json()) == 1 assert response.json()[0]["pk"] == pk

La prueba usa un objeto de cliente creado en un accesorio, que realiza solicitudes a la instancia de FastAPI con dependencia anulada. Como resultado, la prueba puede interactuar con una base de datos separada que se borrará una vez finalizada la prueba. La estructura del conjunto de pruebas restante para ambas API es prácticamente la misma.

Resumen

FastAPI y SQLAlchemy son tecnologías excelentes para crear aplicaciones de back-end modernas y potentes. La libertad, simplicidad y flexibilidad que ofrecen los convierten en una de las mejores opciones para proyectos basados en Python. Si los desarrolladores siguen las prácticas recomendadas y los patrones, pueden crear aplicaciones eficaces, sólidas y bien estructuradas que manejen las operaciones de la base de datos y la lógica de la API con facilidad. Este artículo tiene como objetivo brindarle una buena comprensión de cómo configurar y mantener esta increíble combinación.

Fuentes

El código fuente del proyecto alquimista se puede encontrar aquí: enlace .