Pular para o conteúdo
desenvolvimento

Python Backend com FastAPI: Arquitetura para Aplicações de Produção

Como estruturar um backend Python com FastAPI para produção: estrutura de projeto, autenticação, async, banco de dados com SQLAlchemy, testes e deploy.

Douglas M. Pereira4 min de leitura
pythonfastapibackendsqlalchemyarquiteturaapiasync

Por que FastAPI virou referência em backend Python

FastAPI combina alta performance (comparável a Node.js e Go), type hints nativos do Python, validação automática via Pydantic, documentação OpenAPI gerada automaticamente e suporte a async I/O de primeira classe.

Para APIs REST e microsserviços em Python, FastAPI substituiu Flask e Django REST Framework como primeira escolha na maioria dos projetos novos.

Estrutura de projeto para produção

backend/
├── app/
│   ├── __init__.py
│   ├── main.py              # entry point
│   ├── config.py            # configurações via pydantic-settings
│   ├── dependencies.py      # deps compartilhadas (db session, user auth)
│   ├── models/              # SQLAlchemy ORM models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── order.py
│   ├── schemas/             # Pydantic schemas (request/response)
│   │   ├── user.py
│   │   └── order.py
│   ├── routers/             # routes por domínio
│   │   ├── auth.py
│   │   ├── users.py
│   │   └── orders.py
│   ├── services/            # lógica de negócio
│   │   ├── auth_service.py
│   │   └── order_service.py
│   └── db/
│       ├── session.py       # engine e sessão
│       └── migrations/      # alembic
├── tests/
├── Dockerfile
├── docker-compose.yml
└── pyproject.toml

Configuração com Pydantic Settings

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    app_name: str = "Minha API"
    debug: bool = False
    
    # Banco de dados
    database_url: str
    
    # JWT
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    
    # Rate limiting
    rate_limit_requests: int = 100
    rate_limit_window: int = 60

    class Config:
        env_file = ".env"

@lru_cache()
def get_settings() -> Settings:
    return Settings()

Banco de dados com SQLAlchemy async

# app/db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from app.config import get_settings

settings = get_settings()

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,  # verifica conexão antes de usar
)

AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Base = declarative_base()

# app/dependencies.py  
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Router e Schemas

# app/schemas/user.py
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from uuid import UUID

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=100)
    email: EmailStr
    password: str = Field(..., min_length=8)

class UserResponse(BaseModel):
    id: UUID
    name: str
    email: str
    created_at: datetime
    
    model_config = {"from_attributes": True}

# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_db, get_current_user
from app.schemas.user import UserCreate, UserResponse
from app.services.user_service import UserService

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    payload: UserCreate,
    db: AsyncSession = Depends(get_db),
):
    service = UserService(db)
    user = await service.create(payload)
    return user

@router.get("/me", response_model=UserResponse)
async def get_current_user_info(
    current_user = Depends(get_current_user),
):
    return current_user

Autenticação JWT

# app/services/auth_service.py
from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import HTTPException, status

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(subject: str, secret: str, expires_minutes: int) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=expires_minutes)
    payload = {"sub": subject, "exp": expire, "iat": datetime.now(timezone.utc)}
    return jwt.encode(payload, secret, algorithm="HS256")

def decode_token(token: str, secret: str) -> dict:
    try:
        return jwt.decode(token, secret, algorithms=["HS256"])
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token inválido ou expirado",
        )

Middleware e tratamento de erros

# app/main.py
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import logging
import time

app = FastAPI(title="Minha API", version="1.0.0")

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.minha-empresa.com"],
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

# Logging de requisições
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    logging.info(f"{request.method} {request.url.path} {response.status_code} {duration:.3f}s")
    return response

# Handler de erros global
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(status_code=422, content={"detail": str(exc)})

Testes com pytest-asyncio

# tests/test_users.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users/", json={
        "name": "Douglas",
        "email": "douglas@teste.com",
        "password": "senha123!"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "douglas@teste.com"
    assert "password" not in data  # senha nunca deve vazar na resposta

Deploy com Docker

FROM python:3.12-slim

WORKDIR /app

RUN pip install uv

COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev

COPY app/ ./app/

CMD ["uv", "run", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

FastAPI com 4 workers Uvicorn processa facilmente 5.000–10.000+ req/s em endpoints simples com I/O assíncrono.