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, 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. FastAPI 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 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. que SQLAlchemy 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 y Mapped 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 , 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: alchemist 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 . Además, puede encontrar la estructura de archivos del proyecto a continuación: aquí 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 y 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 contiene los modelos que se utilizarán en la API: objetos entidades alchemist/api/models.py 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 y . Establecer en en la configuración facilitará el trabajo con los objetos SQLAlchemy en el futuro. Los modelos se utilizarán para crear nuevos objetos. Ingredient Potion orm_mode True Payload 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 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 almacenadas en la base de datos y los tipos entre ellos. representación estructuras de datos de relaciones El archivo contiene modelos para objetos de ingredientes y pociones: alchemist/database/models.py 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 . Heredar de él permite construir modelos de base de datos que sean compatibles con los verificadores de tipo Python. DeclarativeBase También es una buena práctica crear un modelo (clase 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 se mantendrá simple. abstracto Base Base Pasando al modelo , el atributo especifica el nombre de la tabla de la base de datos, mientras que el campo 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 como una cadena. Ingredient __tablename__ name name Las cosas se vuelven más complejas en el modelo . También incluye y atributos , pero además de eso, almacena la relación con los ingredientes. El uso de 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. Potion __tablename__ name Mapped[list["Ingredient"]] 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 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. potion_ingredient_association La función configura la relación entre la poción y sus ingredientes. El argumento 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 significa que los ingredientes se cargarán con la poción, eliminando la necesidad de consultas adicionales en el código. relationship lazy selectin 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 . De acuerdo con la documentación de SQLAlchemy, el objeto del motor es responsable de conectar el y 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 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. motor Pool Dialect conexión Una 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. sesión 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. Dialect El signo de interrogación final es el . 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. conjunto de conexiones Con ese conocimiento cubierto, ahora puede echar un vistazo al archivo , que contiene una función que se utilizará como dependencia para conectarse a la base de datos: alchemist/database/session.py 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 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. get_db_session Las dos primeras líneas de la función 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. get_db_session 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 , la sesión se confirma si todo va bien y se revierte si se genera una excepción. get_db_session 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 . Esto generalmente se hace usando una biblioteca de migración como alambique, que rastrea cada modelo y actualiza la base de datos en consecuencia. aplicarlos 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 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 . migrate_tables python scripts/migrate.py Los modelos ahora están presentes tanto en el código como en la base de datos y puede facilitar las interacciones con la base de datos. Ahora puede comenzar a trabajar en la lógica de la API. get_db_session 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 , 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 inicializa la sesión y habilita las interacciones con la base de datos. /ingredients get_db_session En el cuerpo de la función real, se llevan a cabo cinco pasos: Se crea un objeto de ingrediente a partir de la carga útil entrante. El método 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. add La sesión está comprometida. El objeto del ingrediente se actualiza para garantizar que sus atributos coincidan con el estado de la base de datos. 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 dentro de la función . Al consultar la base de datos usando SQLAlchemy, el método se usa a menudo con una consulta como argumento. Mientras que el método devuelve tuplas en forma de fila, devuelven entidades ORM directamente, lo que hace que el punto final sea más limpio. scalars get_ingredients execute execute scalars 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 , 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. scalars 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 especificado en la relación. lazy="selectin" 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 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. Model.objects El próximo capítulo presenta el patrón de repositorio: una solución elegante para extraer código ORM. Repositorio El patrón 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. de repositorio 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 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 , lo que permite reutilizar el código para todos los modelos en lugar de duplicarlo en cada punto final. DatabaseRepository __init__ Además, utiliza genéricos de Python, con el tipo genérico 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. DatabaseRepository Model Debido a que el repositorio necesita usar la sesión de la base de datos, debe inicializarse junto con la dependencia . Considere la nueva dependencia en el archivo : get_db_session 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 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 . 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: get_repository alchemist/api/v2/routes.py 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 , una nueva forma de trabajar con las dependencias de FastAPI. Al especificar el tipo de retorno de la dependencia como y declarar su uso con puede terminar usando anotaciones de tipo simple en el punto final: . Annotated DatabaseRepository[db_models.Ingredient] Depends(get_repository(db_models.Ingredient)) 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 del archivo . volumes 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 logra ambos objetivos. Su cuerpo sigue los siguientes pasos: db_session Cree un motor con una URL de base de datos modificada. Elimine todas las tablas existentes para asegurarse de que la prueba tenga una base de datos limpia. Cree todas las tablas dentro de la base de datos (el mismo código que en el script de migración). Crear y generar un objeto de sesión. 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: es la instancia de FastAPI del archivo , con la dependencia reemplazada por el accesorio test_app alchemist/app.py get_db_session db_session es el httpx que realizará solicitudes de API contra client AsyncClient test_app Con todo esto configurado, finalmente se pueden escribir las pruebas reales. Para ser conciso, el siguiente ejemplo del archivo muestra solo una prueba para crear un ingrediente: tests/test_api.py 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 También publicado . aquí

Patrones y prácticas para usar SQLAlchemy 2.0 con FastAPI

About Author

COMENTARIOS

ETIQUETAS

ESTE ARTÍCULO FUE PRESENTADO EN

Related Stories

De los foros a los feeds: cómo los algoritmos de las redes sociales influyen en la interacción digital

La guía completa para una migración exitosa a la nube: estrategias y mejores prácticas

Las capas invisibles: por qué las entrevistas con los usuarios son un activo irremplazable

Toque para ganar: Telegram puede incorporar a los próximos 10 mil millones de usuarios de criptomonedas antes de Solana

De los foros a los feeds: cómo los algoritmos de las redes sociales influyen en la interacción digital

La guía completa para una migración exitosa a la nube: estrategias y mejores prácticas

Las capas invisibles: por qué las entrevistas con los usuarios son un activo irremplazable

Toque para ganar: Telegram puede incorporar a los próximos 10 mil millones de usuarios de criptomonedas antes de Solana

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps