Padrões e práticas para usar SQLAlchemy 2.0 com FastAPI

Embora o Django e o Flask continuem sendo as primeiras escolhas para muitos engenheiros Python, já foi reconhecido como uma escolha inegavelmente confiável. É uma estrutura altamente flexível, bem otimizada e estruturada que oferece ao desenvolvedor possibilidades infinitas para criar aplicativos de back-end. o FastAPI Trabalhar com bancos de dados é um aspecto essencial da maioria dos aplicativos de back-end. Como resultado, o ORM desempenha um papel crítico no código de back-end. No entanto, ao contrário do Django, o FastAPI não possui um ORM integrado. É de inteira responsabilidade do desenvolvedor selecionar uma biblioteca adequada e integrá-la à base de código. Os engenheiros do Python consideram amplamente o ORM mais popular disponível. É uma biblioteca lendária que está em uso desde 2006 e foi adotada por milhares de projetos. Em 2023, recebeu uma grande atualização para a versão 2.0. Semelhante ao FastAPI, o SQLAlchemy fornece aos desenvolvedores recursos e utilitários poderosos sem forçá-los a usá-los de uma maneira específica. Essencialmente, é um kit de ferramentas versátil que capacita os desenvolvedores a usá-lo da maneira que acharem melhor. o SQLAlchemy FastAPI e SQLAlchemy são uma combinação perfeita. São tecnologias confiáveis, de alto desempenho e modernas, que permitem a criação de aplicativos poderosos e exclusivos. Este artigo explora a criação de um aplicativo de back-end FastAPI que utiliza SQLAlchemy 2.0 como o ORM. O conteúdo abrange: construindo modelos usando e Mapped mapped_column definindo um modelo abstrato manipulação de sessão de banco de dados usando o ORM criando uma classe de repositório comum para todos os modelos preparando uma configuração de teste e adicionando testes Posteriormente, você poderá combinar facilmente o aplicativo FastAPI com o SQLAlchemy ORM. Além disso, você se familiarizará com as melhores práticas e padrões para criar aplicativos bem estruturados, robustos e de alto desempenho. Pré-requisitos Os exemplos de código incluídos no artigo vêm do projeto , que é uma API básica para criar e ler objetos de ingredientes e poções. O foco principal do artigo é explorar a combinação de FastAPI e SQLAlchemy. Não abrange outros tópicos, como: alchemist configurando a configuração do Docker iniciando o servidor uvicorn configurando linting Se você estiver interessado nesses tópicos, poderá explorá-los por conta própria examinando a base de código. Para acessar o repositório de código do projeto alquimista, siga este link . Além disso, você pode encontrar a estrutura de arquivos do projeto abaixo: aqui 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 Embora a árvore possa parecer grande, alguns conteúdos não são relevantes para o ponto principal deste artigo. Além disso, o código pode parecer mais simples do que o necessário em determinadas áreas. Por exemplo, falta ao projeto: estágio de produção no Dockerfile configuração de alambique para migrações subdiretórios para testes Isso foi feito intencionalmente para reduzir a complexidade e evitar sobrecarga desnecessária. No entanto, é importante manter esses fatores em mente ao lidar com um projeto mais pronto para produção. Requisitos da API Ao começar a desenvolver um aplicativo, é fundamental considerar os modelos que seu aplicativo usará. Esses modelos representarão os e com os quais seu aplicativo trabalhará e serão expostos na API. No caso do aplicativo alquimista, existem duas entidades: ingredientes e poções. A API deve permitir a criação e recuperação dessas entidades. O arquivo contém os modelos que serão utilizados na 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) A API retornará modelos e . Definir como na configuração facilitará o trabalho com os objetos SQLAlchemy no futuro. Modelos serão utilizados para criar novos objetos. Ingredient Potion orm_mode True Payload O uso de pydantic torna as classes mais detalhadas e claras em seus papéis e funções. Agora, é hora de criar os modelos de banco de dados. Declarando modelos Um modelo é essencialmente uma de algo. No contexto das APIs, os modelos representam o que o back-end espera no corpo da solicitação e o que retornará nos dados de resposta. Os modelos de banco de dados, por outro lado, são mais complexos e representam as armazenadas no banco de dados e os tipos entre elas. representação estruturas de dados de relacionamento O arquivo contém modelos para objetos de ingredientes e poções: 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 em SQLAlchemy começa com a classe . Herdar dele permite construir modelos de banco de dados compatíveis com os verificadores de tipo Python. DeclarativeBase Também é uma boa prática criar um modelo — classe neste caso — que inclua campos obrigatórios em todos os modelos. Esses campos incluem a chave primária, que é um identificador exclusivo de cada objeto. O modelo abstrato geralmente também armazena as datas de criação e atualização de um objeto, que são definidas automaticamente, quando um objeto é criado ou atualizado. No entanto, o modelo será mantido simples. abstrato Base Base Passando para o modelo , o atributo especifica o nome da tabela do banco de dados, enquanto o campo usa a nova sintaxe SQLAlchemy, permitindo que os campos do modelo sejam declarados com anotações de tipo. Essa abordagem concisa e moderna é poderosa e vantajosa para verificadores de tipo e IDEs, pois reconhece o campo como uma string. Ingredient __tablename__ name name As coisas ficam mais complexas no modelo . Ele também inclui os atributos e , mas, além disso, armazena o relacionamento com os ingredientes. O uso de indica que a poção pode conter vários ingredientes e, nesse caso, a relação é muitos-para-muitos (M2M). Isso significa que um único ingrediente pode ser atribuído a várias poções. Potion __tablename__ name Mapped[list["Ingredient"]] O M2M requer configuração adicional, geralmente envolvendo a criação de uma tabela de associação que armazena as conexões entre as duas entidades. Nesse caso, o objeto armazena apenas os identificadores do ingrediente e da poção, mas também pode incluir atributos extras, como a quantidade de um ingrediente específico necessário para a poção. potion_ingredient_association A função configura a relação entre a poção e seus ingredientes. O argumento especifica como os itens relacionados devem ser carregados. Em outras palavras: o que o SQLAlchemy deve fazer com os ingredientes relacionados ao buscar uma poção. Defini-lo como significa que os ingredientes serão carregados com a poção, eliminando a necessidade de consultas adicionais no código. relationship lazy selectin Construir modelos bem projetados é crucial ao trabalhar com um ORM. Feito isso, o próximo passo é estabelecer a conexão com o banco de dados. Manipulador de sessão Ao trabalhar com um banco de dados, principalmente ao usar o SQLAlchemy, é essencial entender os seguintes conceitos: dialeto motor conexão pool de conexões sessão De todos esses termos, o mais importante é o . De acordo com a documentação do SQLAlchemy, o objeto do mecanismo é responsável por conectar o e para facilitar a conectividade e o comportamento do banco de dados. Em termos mais simples, o objeto do mecanismo é a origem da conexão do banco de dados, enquanto a fornece funcionalidades de alto nível, como executar instruções SQL, gerenciar transações e recuperar resultados do banco de dados. motor Pool Dialect conexão Uma é uma unidade de trabalho que agrupa operações relacionadas em uma única transação. É uma abstração sobre as conexões de banco de dados subjacentes e gerencia com eficiência as conexões e o comportamento transacional. sessão é um componente que fornece suporte para um back-end de banco de dados específico. Atua como intermediário entre o SQLAlchemy e o banco de dados, tratando dos detalhes da comunicação. O projeto alquimista usa o Postgres como banco de dados, portanto, o dialeto deve ser compatível com esse tipo de banco de dados específico. Dialect O ponto de interrogação final é o . No contexto do SQLAlchemy, um pool de conexões é um mecanismo que gerencia uma coleção de conexões de banco de dados. Ele foi projetado para melhorar o desempenho e a eficiência das operações do banco de dados reutilizando as conexões existentes em vez de criar novas para cada solicitação. Ao reutilizar conexões, o pool de conexões reduz a sobrecarga de estabelecer novas conexões e desligá-las, resultando em melhor desempenho. pool de conexões Com esse conhecimento coberto, agora você pode dar uma olhada no arquivo , que contém uma função que será usada como dependência para conexão com o banco de dados: 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 O primeiro detalhe importante a notar é que a função é uma função geradora. Isso ocorre porque o sistema de dependência FastAPI oferece suporte a geradores. Como resultado, essa função pode lidar com cenários bem-sucedidos e com falha. get_db_session As duas primeiras linhas da função criam um mecanismo de banco de dados e uma sessão. No entanto, o objeto de sessão também pode ser usado como um gerenciador de contexto. Isso lhe dá mais controle sobre possíveis exceções e resultados bem-sucedidos. get_db_session Embora o SQLAlchemy lide com o fechamento de conexões, é uma boa prática declarar explicitamente como lidar com a conexão depois de concluída. Na função , a sessão é confirmada se tudo correr bem e revertida se uma exceção for levantada. get_db_session É importante observar que esse código é criado em torno da extensão assíncrona. Esse recurso do SQLAlchemy permite que o aplicativo interaja com o banco de dados de forma assíncrona. Isso significa que as solicitações ao banco de dados não bloquearão outras solicitações de API, tornando o aplicativo muito mais eficiente. Depois que os modelos e a conexão estiverem configurados, a próxima etapa é garantir que os modelos sejam adicionados ao banco de dados. Migrações rápidas Os modelos SQLAlchemy representam as estruturas de um banco de dados. No entanto, simplesmente criá-los não resulta em alterações imediatas no banco de dados. Para fazer alterações, você deve primeiro las. Isso normalmente é feito usando uma biblioteca de migração como o alambique, que rastreia cada modelo e atualiza o banco de dados de acordo. aplicá- Como nenhuma mudança adicional nos modelos está planejada neste cenário, um script básico de migração será suficiente. Abaixo está um código de exemplo do arquivo . 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()) Simplificando, a função lê a estrutura dos modelos e a recria no banco de dados usando o mecanismo SQLAlchemy. Para executar este script, use o comando . migrate_tables python scripts/migrate.py Os modelos agora estão presentes tanto no código quanto no banco de dados e pode facilitar as interações com o banco de dados. Agora você pode começar a trabalhar na lógica da API. get_db_session API com o ORM Conforme mencionado anteriormente, a API para ingredientes e poções destina-se a suportar três operações: criando objetos listando objetos recuperando objetos por ID Graças aos preparativos anteriores, todos esses recursos já podem ser implementados com SQLAlchemy como ORM e FastAPI como estrutura da web. Para começar, revise a API de ingredientes localizada no arquivo : 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) Na API , há três rotas disponíveis. O terminal POST usa uma carga de ingrediente como um objeto de um modelo criado anteriormente e uma sessão de banco de dados. A função geradora inicializa a sessão e permite interações com o banco de dados. /ingredients get_db_session No corpo da função real, há cinco etapas ocorrendo: Um objeto de ingrediente é criado a partir da carga recebida. O método do objeto de sessão adiciona o objeto de ingrediente ao sistema de rastreamento de sessão e o marca como pendente para inserção no banco de dados. add A sessão é confirmada. O objeto de ingrediente é atualizado para garantir que seus atributos correspondam ao estado do banco de dados. A instância do ingrediente do banco de dados é convertida na instância do modelo de API usando o método . from_orm Para um teste rápido, um simples curl pode ser executado no aplicativo em execução: curl -X 'POST' \ 'http://localhost:8000/api/v1/ingredients' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty water"}' Na resposta, deve haver um objeto ingrediente que tenha um ID vindo do banco de dados: { "pk":"2eb255e9-2172-4c75-9b29-615090e3250d", "name":"Salty water" } Embora as múltiplas camadas de abstração do SQLAlchemy possam parecer desnecessárias para uma API simples, elas mantêm os detalhes do ORM separados e contribuem para a eficiência e escalabilidade do SQLAlchemy. Quando combinados com assíncrono, os recursos ORM funcionam excepcionalmente bem na API. Os dois endpoints restantes são menos complexos e compartilham semelhanças. Uma parte que merece uma explicação mais profunda é o uso do método dentro da função . Ao consultar o banco de dados usando SQLAlchemy, o método geralmente é usado com uma consulta como argumento. Enquanto o método retorna tuplas semelhantes a linhas, retorna entidades ORM diretamente, tornando o endpoint mais limpo. scalars get_ingredients execute execute scalars Agora, considere a API de poções, no mesmo arquivo: @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) Os pontos de extremidade GET para poções são idênticos aos dos ingredientes. No entanto, a função POST requer código adicional. Isso ocorre porque a criação de poções envolve a inclusão de pelo menos um ID de ingrediente, o que significa que os ingredientes devem ser buscados e vinculados à poção recém-criada. Para conseguir isso, o método é usado novamente, mas desta vez com uma consulta que especifica os IDs dos ingredientes buscados. A parte restante do processo de criação da poção é idêntica à dos ingredientes. scalars Para testar o endpoint, novamente um comando curl pode ser executado. 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"}' Resulta na seguinte resposta: { "pk": "d4929197-3998-4234-a5f7-917dc4bba421", "name": "Salty soup", "ingredients": [ { "pk": "0b4f1de5-e780-418d-a74d-927afe8ac954", "name": "Salty water" } ] } É importante notar que cada ingrediente é representado como um objeto completo dentro da poção, graças ao argumento especificado no relacionamento. lazy="selectin" As APIs são funcionais, mas há um grande problema com o código. Embora o SQLAlchemy lhe dê a liberdade de interagir com o banco de dados como quiser, ele não oferece nenhum utilitário "gerenciador" de alto nível semelhante ao do Django. Como resultado, você mesmo precisará criá-lo, que é essencialmente a lógica usada nas APIs de ingredientes e poções. No entanto, se você mantiver essa lógica diretamente nos endpoints sem extraí-la em um espaço separado, acabará com muito código duplicado. Além disso, fazer alterações nas consultas ou modelos se tornará cada vez mais difícil de gerenciar. Model.objects O próximo capítulo apresenta o padrão de repositório: uma solução elegante para extrair código ORM. Repositório O padrão permite abstrair os detalhes do trabalho com o banco de dados. No caso de usar SQLAlchemy, como no exemplo do alquimista, a classe repositório seria responsável por gerenciar vários modelos e interagir com a sessão do banco de dados. de repositório Dê uma olhada no seguinte código do arquivo : 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)) A classe contém toda a lógica que foi incluída anteriormente nos endpoints. A diferença é que ele permite que a classe específica do model seja passada no método , possibilitando o reaproveitamento do código para todos os models ao invés de duplicá-lo em cada endpoint. DatabaseRepository __init__ Além disso, o usa genéricos do Python, com o tipo genérico vinculado ao modelo de banco de dados abstrato. Isso permite que a classe de repositório se beneficie mais da verificação de tipo estático. Quando usado com um modelo específico, os tipos de retorno dos métodos de repositório refletirão esse modelo específico. DatabaseRepository Model Como o repositório precisa usar a sessão do banco de dados, ele deve ser inicializado junto com a dependência . Considere a nova dependência no arquivo : 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 Simplificando, a função é uma fábrica de dependências. Ele primeiro pega o modelo de banco de dados com o qual você usará o repositório. Em seguida, retorna a dependência, que será utilizada para receber a sessão do banco de dados e inicializar o objeto repositório. Para entender melhor, confira a nova API no arquivo . Ele mostra apenas os endpoints do POST, mas deve ser o suficiente para dar uma ideia mais clara de como o código é aprimorado: 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) O primeiro recurso importante a ser observado é o uso de , uma nova maneira de trabalhar com dependências FastAPI. Especificando o tipo de retorno da dependência como e declarando seu uso com você pode acabar usando anotações de tipo simples no terminal: . Annotated DatabaseRepository[db_models.Ingredient] Depends(get_repository(db_models.Ingredient)) repository: IngredientRepository Graças ao repositório, os endpoints não precisam armazenar toda a carga relacionada ao ORM. Mesmo no caso de poções mais complicadas, tudo o que você precisa fazer é usar dois repositórios ao mesmo tempo. Você pode se perguntar se a inicialização de dois repositórios inicializará a sessão duas vezes. A resposta é não. O sistema de dependência FastAPI armazena em cache as mesmas chamadas de dependência em uma única solicitação. Isso significa que a inicialização da sessão é armazenada em cache e ambos os repositórios usam exatamente o mesmo objeto de sessão. Ainda outro grande recurso da combinação de SQLAlchemy e FastAPI. A API é totalmente funcional e possui uma camada de acesso a dados reutilizável e de alto desempenho. A próxima etapa é garantir que os requisitos sejam atendidos escrevendo alguns testes de ponta a ponta. teste Os testes desempenham um papel crucial no desenvolvimento de software. Os projetos podem conter testes de unidade, integração e end-to-end (E2E). Embora geralmente seja melhor ter um grande número de testes de unidade significativos, também é bom escrever pelo menos alguns testes E2E para garantir que todo o fluxo de trabalho esteja funcionando corretamente. Para criar alguns testes E2E para o aplicativo alquimista, são necessárias duas bibliotecas adicionais: pytest para realmente criar e executar os testes httpx para fazer requisições assíncronas dentro dos testes Depois de instalados, a próxima etapa é ter um banco de dados de teste separado. Você não quer que seu banco de dados padrão seja poluído ou descartado. Como o alchemist inclui uma configuração do Docker, apenas um script simples é necessário para criar um segundo banco de dados. Dê uma olhada no código do arquivo : scripts/create_test_db.sh #!/bin/bash psql -U postgres psql -c "CREATE DATABASE test" Para que o script seja executado, ele deve ser adicionado como um volume ao container Postgres. Isso pode ser obtido incluindo-o na seção do arquivo . volumes docker-compose.yaml A etapa final da preparação é criar os fixtures pytest dentro do arquivo : 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 Uma coisa que é fundamental ser alterada nos testes, é como o app interage com o banco de dados. Isso inclui não apenas alterar a URL do banco de dados, mas também garantir que cada teste seja isolado iniciando com um banco de dados vazio. O acessório cumpre esses dois objetivos. Seu corpo segue as seguintes etapas: db_session Crie um mecanismo com um URL de banco de dados modificado. Exclua todas as tabelas existentes para garantir que o teste tenha um banco de dados limpo. Crie todas as tabelas dentro do banco de dados (mesmo código do script de migração). Crie e produza um objeto de sessão. Feche manualmente a sessão quando o teste for concluído. Embora a última etapa também possa ser implementada como um gerenciador de contexto, o fechamento manual funciona muito bem nesse caso. Os dois acessórios restantes devem ser bastante auto-explicativos: é a instância FastAPI do arquivo , com a dependência substituída pelo acessório test_app alchemist/app.py get_db_session db_session é o httpx que fará solicitações de API contra o client AsyncClient test_app Com tudo isso configurado, finalmente os testes reais podem ser escritos. Para concisão, o exemplo abaixo do arquivo mostra apenas um teste para criar um 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 O teste utiliza um objeto cliente criado em um fixture, que faz requisições à instância FastAPI com dependência substituída. Como resultado, o teste é capaz de interagir com um banco de dados separado que será limpo após a conclusão do teste. A estrutura do conjunto de teste restante para ambas as APIs é praticamente a mesma. Resumo FastAPI e SQLAlchemy são tecnologias excelentes para criar aplicativos de back-end modernos e poderosos. A liberdade, simplicidade e flexibilidade que eles oferecem os tornam uma das melhores opções para projetos baseados em Python. Se os desenvolvedores seguirem as melhores práticas e padrões, eles poderão criar aplicativos de alto desempenho, robustos e bem estruturados que lidam com operações de banco de dados e lógica de API com facilidade. Este artigo teve como objetivo fornecer a você um bom entendimento de como configurar e manter essa incrível combinação. Fontes O código-fonte do projeto alquimista pode ser encontrado aqui: . link Publicado também . aqui