Tutorial FastAPI - ES2-UFPI/TermoLOG GitHub Wiki

Tutorial: FastAPI

Neste tutorial vamos criar uma API REST com FastAPI que faz o CRUD de registros num banco SQLite usando SQLModel.

Tutorial em vídeo: https://youtu.be/zpD6kOfctxM

1. Preparando o ambiente

Usaremos o uv, um gerenciador de pacotes e projetos Python.

pip install uv                    # instala o uv
uv init nome_do_projeto --bare    # cria o projeto (só com o pyproject.toml)
cd nome_do_projeto
uv add "fastapi[standard]"        # FastAPI + servidor e CLI
uv add sqlmodel                   # ORM para o banco de dados
code main.py                      # abre o arquivo principal no VS Code

2. Primeira API

Em main.py:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello World"}

@app.get("/ufpi")
def return_ufpi():
    return {"message": "Bem vindo à UFPI!"}

Rode o servidor:

uv run fastapi dev main.py

Cada função decorada com @app.get, @app.post, @app.put ou @app.delete vira um endpoint. O dicionário retornado é convertido automaticamente em JSON.

Parâmetros de caminho

@app.post("/numero/{num}")
def post_num(num: int):
    return {"message": f"O número é {num}"}

O {num} da URL é passado para a função. Como ele está tipado como int, o FastAPI valida a entrada sozinho: /numero/abc devolve erro 422.

3. Modelo de dados

Crie registro.py:

from sqlmodel import Field, SQLModel

class Registro(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    sala: int
    placa_id: str
    ativo: str
  • table=True diz que a classe vira uma tabela no banco.
  • id é a chave primária. O valor padrão None deixa o banco gerar o número.
  • A mesma classe serve para validar o JSON que chega na API e para gravar no banco.

4. Conectando ao banco

No início de main.py:

from registro import Registro
from fastapi import FastAPI, Depends, Query
from sqlmodel import create_engine, SQLModel, Session, select
from typing import Annotated

sqlite_url = "sqlite:///database.db"
engine = create_engine(sqlite_url, connect_args={"check_same_thread": False})

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)   # cria as tabelas se não existirem

def get_session():
    with Session(engine) as session:
        yield session                      # uma sessão por requisição

SessionDep = Annotated[Session, Depends(get_session)]

app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()
  • engine: a conexão com o arquivo database.db.
  • Depends(get_session): é a injeção de dependência do FastAPI. Todo endpoint que recebe um parâmetro SessionDep ganha uma sessão aberta, que é fechada ao fim da requisição.
  • startup: cria as tabelas quando o servidor sobe.

Nas versões recentes do FastAPI, on_event está depreciado em favor de lifespan, mas continua funcionando.

5. CRUD

Create (criar)

@app.post("/create_registro")
def novo(registro: Registro, session: SessionDep):
    session.add(registro)
    session.commit()
    session.refresh(registro)   # recarrega o objeto e preenche o id gerado
    return registro

Corpo da requisição (JSON):

{ "sala": 101, "placa_id": "ABC123", "ativo": "sim" }

Read (listar, com paginação)

@app.get("/registros")
def get_registros(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
) -> list[Registro]:
    return session.exec(select(Registro).offset(offset).limit(limit)).all()

Parâmetros que não aparecem no caminho viram query params: /registros?offset=0&limit=10. O Query(le=100) limita limit a no máximo 100.

Read (buscar por id)

@app.get("/registros/{id}")
def get_registro(id: int, session: SessionDep):
    registro = session.get(Registro, id)
    if registro is None:
        return {"message": "Registro não encontrado"}
    return registro

Update (atualizar)

@app.put("/registros/{id}")
def update_registro(id: int, registro: Registro, session: SessionDep):
    db_registro = session.get(Registro, id)
    if db_registro is None:
        return {"message": "Registro não pode ser alterado pois não existe!"}
    db_registro.sala = registro.sala
    db_registro.placa_id = registro.placa_id
    db_registro.ativo = registro.ativo
    session.add(db_registro)
    session.commit()
    session.refresh(db_registro)
    return db_registro

Delete (remover)

@app.delete("/registros/{id}")
def delete_registro(id: int, session: SessionDep):
    db_registro = session.get(Registro, id)
    if db_registro is None:
        return {"message": "Registro não pode ser deletado pois não existe!"}
    session.delete(db_registro)
    session.commit()
    return {"message": "Registro deletado com sucesso!"}

Dica: retornar uma mensagem quando o registro não existe responde com status 200. O mais correto em uma API REST é devolver 404:

from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Registro não encontrado")

6. Testando

Com o servidor rodando (uv run fastapi dev main.py), abra http://127.0.0.1:8000/docs, escolha um endpoint, clique em Try it out e depois em Execute.

Método Rota Ação
GET / Hello World
GET /ufpi Mensagem de boas-vindas
POST /numero/{num} Ecoa o número
POST /create_registro Cria registro
GET /registros Lista registros
GET /registros/{id} Busca um registro
PUT /registros/{id} Atualiza um registro
DELETE /registros/{id} Remove um registro