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, o FastAPI 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.

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 SQLAlchemy 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.

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 Mapped e 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 alchemist , 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:

• 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 aqui . Além disso, você pode encontrar a estrutura de arquivos do projeto abaixo:

 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 objetos e entidades 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 alchemist/api/models.py contém os modelos que serão utilizados na 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)

A API retornará modelos Ingredient e Potion . Definir orm_mode como True na configuração facilitará o trabalho com os objetos SQLAlchemy no futuro. Modelos Payload serão utilizados para criar novos objetos.

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 representação 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 estruturas de dados armazenadas no banco de dados e os tipos de relacionamento entre elas.

O arquivo alchemist/database/models.py contém modelos para objetos de ingredientes e poções:

 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 DeclarativeBase . Herdar dele permite construir modelos de banco de dados compatíveis com os verificadores de tipo Python.

Também é uma boa prática criar um modelo abstrato — classe Base 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 Base será mantido simples.

Passando para o modelo Ingredient , o atributo __tablename__ especifica o nome da tabela do banco de dados, enquanto o campo name 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 name como uma string.

As coisas ficam mais complexas no modelo Potion . Ele também inclui os atributos __tablename__ e name , mas, além disso, armazena o relacionamento com os ingredientes. O uso de Mapped[list["Ingredient"]] 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.

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 potion_ingredient_association 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.

A função relationship configura a relação entre a poção e seus ingredientes. O argumento lazy 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 selectin significa que os ingredientes serão carregados com a poção, eliminando a necessidade de consultas adicionais no código.

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 motor . De acordo com a documentação do SQLAlchemy, o objeto do mecanismo é responsável por conectar o Pool e Dialect 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 conexão fornece funcionalidades de alto nível, como executar instruções SQL, gerenciar transações e recuperar resultados do banco de dados.

Uma sessão é 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.

Dialect é 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.

O ponto de interrogação final é o pool de conexões . 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.

Com esse conhecimento coberto, agora você pode dar uma olhada no arquivo alchemist/database/session.py , que contém uma função que será usada como dependência para conexão com o banco de dados:

 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 get_db_session é 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.

As duas primeiras linhas da função get_db_session 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.

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 get_db_session , a sessão é confirmada se tudo correr bem e revertida se uma exceção for levantada.

É 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 aplicá- 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.

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 migrate_tables lê a estrutura dos modelos e a recria no banco de dados usando o mecanismo SQLAlchemy. Para executar este script, use o comando python scripts/migrate.py .

Os modelos agora estão presentes tanto no código quanto no banco de dados e get_db_session pode facilitar as interações com o banco de dados. Agora você pode começar a trabalhar na lógica da API.

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 /ingredients , 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 get_db_session inicializa a sessão e permite interações com o banco de dados.

No corpo da função real, há cinco etapas ocorrendo:

1. Um objeto de ingrediente é criado a partir da carga recebida.
2. O método add 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.
3. A sessão é confirmada.
4. O objeto de ingrediente é atualizado para garantir que seus atributos correspondam ao estado do banco de dados.
5. 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 scalars dentro da função get_ingredients . Ao consultar o banco de dados usando SQLAlchemy, o método execute geralmente é usado com uma consulta como argumento. Enquanto o método execute retorna tuplas semelhantes a linhas, scalars retorna entidades ORM diretamente, tornando o endpoint mais limpo.

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 scalars é 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.

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 lazy="selectin" especificado no relacionamento.

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 Model.objects 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.

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 de repositório 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.

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 DatabaseRepository 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 __init__ , possibilitando o reaproveitamento do código para todos os models ao invés de duplicá-lo em cada endpoint.

Além disso, o DatabaseRepository usa genéricos do Python, com o tipo genérico Model 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.

Como o repositório precisa usar a sessão do banco de dados, ele deve ser inicializado junto com a dependência get_db_session . Considere a nova dependência no arquivo 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 get_repository é 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 alchemist/api/v2/routes.py . Ele mostra apenas os endpoints do POST, mas deve ser o suficiente para dar uma ideia mais clara de como o código é aprimorado:

 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 Annotated , uma nova maneira de trabalhar com dependências FastAPI. Especificando o tipo de retorno da dependência como DatabaseRepository[db_models.Ingredient] e declarando seu uso com Depends(get_repository(db_models.Ingredient)) você pode acabar usando anotações de tipo simples no terminal: 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 volumes do arquivo 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 db_session cumpre esses dois objetivos. Seu corpo segue as seguintes etapas:

1. Crie um mecanismo com um URL de banco de dados modificado.
2. Exclua todas as tabelas existentes para garantir que o teste tenha um banco de dados limpo.
3. Crie todas as tabelas dentro do banco de dados (mesmo código do script de migração).
4. Crie e produza um objeto de sessão.
5. 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:

• test_app é a instância FastAPI do arquivo alchemist/app.py , com a dependência get_db_session substituída pelo acessório db_session
• client é o httpx AsyncClient que fará solicitações de API contra o test_app

Com tudo isso configurado, finalmente os testes reais podem ser escritos. Para concisão, o exemplo abaixo do arquivo tests/test_api.py mostra apenas um teste para criar um 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

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 .