Neste guia completo, vamos construir uma API CRUD de gerenciamento de clientes utilizando FastAPI. Estruturaremos a aplicação seguindo o padrão arquitetural Model-View-Controller (MVC), uma abordagem que separa as responsabilidades do código e facilita a manutenção de projetos em larga escala.
A API utilizará SQLAlchemy ORM para interagir com um banco de dados MySQL e Pydantic para validação de dados. Implementaremos também um wrapper de resposta REST API para garantir consistência nas respostas da aplicação.
O que você vai aprender neste tutorial
Como organizar código usando a estrutura MVC
Implementar um wrapper de resposta REST API com campos como
status,message,dataeerrorCriar validações robustas e tratamento de erros
Usar um modelo Pydantic único para operações de criação, atualização e deleção
Gerar UUID como chave primária
Estrutura do projeto
Vamos organizar nosso projeto FastAPI da seguinte forma:
fastapi_crud_mvc/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── controllers/
│ │ ├── __init__.py
│ │ └── customer_controller.py
│ ├── models/
│ │ ├── __init__.py
│ │ └── customer.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ └── customer_schema.py
│ ├── db/
│ │ ├── __init__.py
│ │ └── database.py
│ └── utils/
│ ├── __init__.py
│ └── response_wrapper.py
└── requirements.txt
Esta estrutura segue os princípios do padrão MVC, onde:
Models representam a camada de dados e estrutura do banco
Controllers contêm a lógica de negócios
Schemas definem a validação e serialização de dados
Utils armazenam funções auxiliares reutilizáveis
Passo 1: instalando as dependências
Primeiro, crie um ambiente virtual e instale os pacotes necessários:
pip install fastapi uvicorn sqlalchemy mysqlclient pymysql pydantic
Essas bibliotecas fornecem:
FastAPI: framework web moderno e de alta performance
Uvicorn: servidor ASGI para executar a aplicação
SQLAlchemy: ORM para interação com banco de dados
Pydantic: validação de dados usando type hints do Python
Passo 2: configuração do banco de dados
Crie o arquivo de conexão com o banco de dados em db/database.py:
# db/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "mysql+pymysql://<username>:<password>@localhost:3306/<database_name>"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Importante: Substitua <username>, <password> e <database_name> pelas suas credenciais do MySQL.
A função get_db() é um gerador que fornece uma sessão do banco de dados e garante que ela seja fechada adequadamente após o uso, seguindo as melhores práticas de gerenciamento de recursos.
Passo 3: criando o modelo ORM
Defina o schema da tabela Customer usando SQLAlchemy em models/customer.py:
# models/customer.py
import uuid
from sqlalchemy import Column, String
from sqlalchemy.dialects.mysql import CHAR
from ..db.database import Base
class Customer(Base):
__tablename__ = "customers"
id = Column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()))
name = Column(String(100), nullable=False)
email = Column(String(100), unique=True, nullable=False)
address = Column(String(255), nullable=True)
Aqui estamos usando UUID como chave primária, o que oferece várias vantagens:
Identificadores únicos globalmente
Maior segurança (não são sequenciais)
Facilita a distribuição de dados entre diferentes sistemas
Passo 4: criando o schema Pydantic
Crie um CustomerSchema único para criar, atualizar e retornar dados em schemas/customer_schema.py:
# schemas/customer_schema.py
from pydantic import BaseModel, EmailStr
from typing import Optional
class CustomerSchema(BaseModel):
name: Optional[str] = None
email: Optional[EmailStr] = None
address: Optional[str] = None
class Config:
orm_mode = True
O CustomerSchema é usado tanto para requisições de criação quanto de atualização, tornando os campos opcionais. A configuração orm_mode = True permite que o Pydantic trabalhe com objetos do SQLAlchemy.
Passo 5: wrapper de resposta REST API
Para garantir respostas consistentes em toda a API, defina uma função wrapper em utils/response_wrapper.py:
# utils/response_wrapper.py
def api_response(data=None, message="Success", status=True, error=None):
return {
"status": status,
"message": message,
"data": data,
"error": error,
}
Este padrão padroniza todas as respostas da API, facilitando o consumo por aplicações frontend e melhorando a experiência do desenvolvedor.
Passo 6: implementando o controller
Defina as operações CRUD na camada de controller em controllers/customer_controller.py:
# controllers/customer_controller.py
from fastapi import APIRouter, HTTPException, Depends
from sqlalchemy.orm import Session
from ..models.customer import Customer
from ..schemas.customer_schema import CustomerSchema
from ..db.database import get_db
from ..utils.response_wrapper import api_response
router = APIRouter()
# CREATE Customer
@router.post("/customers/")
def create_customer(customer: CustomerSchema, db: Session = Depends(get_db)):
if db.query(Customer).filter(Customer.email == customer.email).first():
raise HTTPException(status_code=400, detail="Email already registered")
new_customer = Customer(**customer.dict())
db.add(new_customer)
db.commit()
db.refresh(new_customer)
return api_response(data=new_customer, message="Customer created successfully")
# READ All Customers
@router.get("/customers/")
def get_customers(db: Session = Depends(get_db)):
customers = db.query(Customer).all()
return api_response(data=customers, message="All customers retrieved")
# READ Single Customer
@router.get("/customers/{customer_id}")
def get_customer(customer_id: str, db: Session = Depends(get_db)):
customer = db.query(Customer).filter(Customer.id == customer_id).first()
if customer is None:
raise HTTPException(status_code=404, detail="Customer not found")
return api_response(data=customer, message="Customer retrieved successfully")
# UPDATE Customer
@router.put("/customers/{customer_id}")
def update_customer(customer_id: str, customer_update: CustomerSchema, db: Session = Depends(get_db)):
customer = db.query(Customer).filter(Customer.id == customer_id).first()
if not customer:
raise HTTPException(status_code=404, detail="Customer not found")
for field, value in customer_update.dict(exclude_unset=True).items():
setattr(customer, field, value)
db.commit()
db.refresh(customer)
return api_response(data=customer, message="Customer updated successfully")
# DELETE Customer
@router.delete("/customers/{customer_id}")
def delete_customer(customer_id: str, db: Session = Depends(get_db)):
customer = db.query(Customer).filter(Customer.id == customer_id).first()
if not customer:
raise HTTPException(status_code=404, detail="Customer not found")
db.delete(customer)
db.commit()
return api_response(message="Customer deleted successfully")
Passo 7: registrando o controller no main.py
Crie a aplicação principal do FastAPI e inclua o router de clientes:
# main.py
from fastapi import FastAPI
from .controllers.customer_controller import router as customer_router
app = FastAPI()
app.include_router(customer_router, prefix="/api", tags=["Customers"])
@app.get("/")
def root():
return {"message": "Welcome to the FastAPI CRUD API"}
Passo 8: executando a aplicação FastAPI
Execute a aplicação usando Uvicorn:
uvicorn app.main:app --reload
O flag --reload habilita o hot-reload, permitindo que mudanças no código sejam refletidas automaticamente sem reiniciar o servidor.
Passo 9: testando a API
Acesse o Swagger UI em:
http://127.0.0.1:8000/docs
Você pode testar todos os endpoints CRUD diretamente da documentação interativa da API, que é gerada automaticamente pelo FastAPI.
Exemplos de requisições e respostas
1. Criar um cliente (POST /api/customers/)
Requisição:
{
"name": "João Silva",
"email": "[email protected]",
"address": "Rua Principal, 123"
}
Resposta:
{
"status": true,
"message": "Customer created successfully",
"data": {
"id": "b6a25c97-5d69-4f41-82d3-1a4d5f4731d8",
"name": "João Silva",
"email": "[email protected]",
"address": "Rua Principal, 123"
},
"error": null
}
2. Buscar todos os clientes (GET /api/customers/)
Resposta:
{
"status": true,
"message": "All customers retrieved",
"data": [
{
"id": "b6a25c97-5d69-4f41-82d3-1a4d5f4731d8",
"name": "João Silva",
"email": "[email protected]",
"address": "Rua Principal, 123"
}
],
"error": null
}
Vantagens da arquitetura MVC
A implementação da arquitetura MVC traz diversos benefícios:
Vantagem | Descrição |
|---|---|
Separação de responsabilidades | Cada camada tem uma função específica, facilitando a manutenção |
Testabilidade | Componentes isolados são mais fáceis de testar |
Reutilização de código | Models e schemas podem ser usados em diferentes contextos |
Escalabilidade | Facilita a adição de novas funcionalidades sem afetar o código existente |
Colaboração em equipe | Diferentes desenvolvedores podem trabalhar em camadas diferentes simultaneamente |
Perguntas frequentes (FAQ)
Por que usar UUID ao invés de auto-incremento?
UUIDs oferecem identificadores únicos globalmente, aumentam a segurança e facilitam a integração entre sistemas distribuídos, além de evitar problemas de colisão em ambientes de alta concorrência.
É possível usar outro banco de dados além do MySQL?
Sim! O SQLAlchemy suporta diversos bancos de dados. Basta alterar a URL de conexão no arquivo database.py para PostgreSQL, SQLite, ou outros SGBDs compatíveis.
Como adicionar autenticação à API?
Você pode implementar autenticação JWT (JSON Web Tokens) usando bibliotecas como python-jose e adicionar dependências de segurança nos endpoints que precisam de proteção.
Posso usar esta estrutura em produção?
Sim, mas considere adicionar: logging estruturado, métricas de performance, cache, rate limiting, e configurações de ambiente adequadas para produção.
Próximos passos
Para aprimorar ainda mais esta API, considere implementar:
Paginação nas listagens
Filtros e ordenação customizados
Autenticação e autorização
Testes automatizados (pytest)
Documentação adicional com exemplos
Rate limiting para proteger contra abuso
Logging estruturado
Migrations com Alembic
Cache com Redis
Monitoramento e observabilidade
Conclusão
Neste tutorial, criamos uma API CRUD com Python e FastAPI usando o padrão arquitetural MVC. Utilizamos SQLAlchemy ORM para operações de banco de dados, Pydantic para validação de requisições, e MySQL como nosso banco de dados. Adicionalmente, implementamos um wrapper de resposta REST API para garantir consistência nas respostas.
Esta estrutura torna o projeto escalável, de fácil manutenção e pronto para produção. O padrão MVC, combinado com as capacidades do FastAPI, oferece uma base sólida para construir aplicações web modernas e de alta performance.
Ao seguir as melhores práticas apresentadas neste guia, você estará preparado para desenvolver APIs robustas que podem crescer junto com as necessidades do seu projeto.