Skip to content

    Chapter 34, Projects

    Blog API: assembling, testing, and running

    The application set up, the test suite that runs on SQLite and on PostgreSQL for real, and the actual execution against the database, with the responses and the schema that PostgreSQL created.

    The application

    The criar_app is a factory: receives the configurations and returns a new application, with its own engine. This is what allows the test to create an application with an in-memory database, and production, another pointed to PostgreSQL. The lifespan creates the tables on startup and closes the connections on shutdown. CORS is only enabled if there are configured origins, and with methods and headers listed (chapter 22):

    projeto_blog/blog_api/app/main.py
    from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import Configuracoes, obter_configuracoes
from app.database import Base, criar_engine, criar_fabrica
from app.routers import auth, posts


def criar_app(config: Configuracoes | None = None) -> FastAPI:
    config = config or obter_configuracoes()
    engine = criar_engine(config.banco_url)

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        # Para aprender e para os testes. Em produção, as tabelas vêm de migrações (Alembic).
        Base.metadata.create_all(engine)
        yield
        engine.dispose()

    app = FastAPI(title="Blog API", version="1.0.0", lifespan=lifespan)
    app.state.config = config
    app.state.engine = engine
    app.state.fabrica_sessao = criar_fabrica(engine)

    if config.origens_permitidas:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=config.origens_permitidas,
            allow_credentials=True,
            allow_methods=["GET", "POST", "PUT", "DELETE"],
            allow_headers=["authorization", "content-type"],
        )

    app.include_router(auth.router)
    app.include_router(posts.router)

    @app.get("/saude", tags=["Infra"])
    def saude() -> dict[str, str]:
        return {"status": "ok"}

    return app


app = criar_app()

    create_all and migrations

    The create_all creates the tables that do not exist, and nothing more. In production, the schema evolves through migrations (Alembic, chapter 57 of the Python course): a new column in a table with real data cannot be made by create_all. I left it in the lifespan to run the project with a single command, and the comment in the code notes that this is for learning and testing.

    The tests

    The conftest.py delivers, with each test, an application with the database clean (and it deletes it at the end, also closing the connection pool: without the engine.dispose(), each test would leave an open connection in PostgreSQL), and functions to create already logged-in users. The variable TEST_DATABASE_URL chooses the database: without it, SQLite in memory (fast, without installing anything), and with it, the real PostgreSQL:

    projeto_blog/blog_api/tests/conftest.py
    import os

# A chave precisa existir antes de importar a aplicação (que lê as configurações na partida).
os.environ.setdefault("BLOG_CHAVE_SECRETA", "chave-de-teste-com-no-minimo-32-caracteres-0123")

from collections.abc import Callable, Iterator  # noqa: E402

import pytest  # noqa: E402
from fastapi import FastAPI  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402

from app.config import Configuracoes  # noqa: E402
from app.database import Base  # noqa: E402
from app.main import criar_app  # noqa: E402

# Por padrão, SQLite em memória. Para testar no PostgreSQL de verdade:
#   TEST_DATABASE_URL=postgresql+psycopg://blog:blog_senha@localhost:5432/blog_teste uv run pytest
URL_TESTE = os.environ.get("TEST_DATABASE_URL", "sqlite://")


@pytest.fixture
def app() -> Iterator[FastAPI]:
    aplicacao = criar_app(Configuracoes(banco_url=URL_TESTE, _env_file=None))
    yield aplicacao
    # Cada teste começa com o banco vazio: nenhum teste depende do que outro deixou.
    engine = aplicacao.state.engine
    Base.metadata.drop_all(engine)
    engine.dispose()  # fecha o pool: sem isso, cada teste deixaria uma conexão aberta


@pytest.fixture
def cliente(app: FastAPI) -> Iterator[TestClient]:
    with TestClient(app) as c:
        yield c


Entrar = Callable[[str, str], dict[str, str]]


@pytest.fixture
def novo_usuario(cliente: TestClient) -> Entrar:
    """Cria um usuário, faz login e devolve o cabeçalho Authorization."""

    def criar(nome: str, email: str) -> dict[str, str]:
        senha = "senha-segura-123"
        cliente.post("/usuarios", json={"nome": nome, "email": email, "senha": senha})
        resposta = cliente.post("/auth/token", data={"username": email, "password": senha})
        return {"Authorization": f"Bearer {resposta.json()['access_token']}"}

    return criar


@pytest.fixture
def ana(novo_usuario: Entrar) -> dict[str, str]:
    return novo_usuario("Ana", "ana@exemplo.com")


@pytest.fixture
def bia(novo_usuario: Entrar) -> dict[str, str]:
    return novo_usuario("Bia", "bia@exemplo.com")

    The authentication tests cover what is most dangerous to get wrong: the response never does not bring the password or the hash, the repeated email, the validations, the same response for non-existent email and wrong password, and the expired token:

    projeto_blog/blog_api/tests/test_auth.py
    import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.security import criar_token

CADASTRO = {"nome": "Ana", "email": "Ana@Exemplo.com", "senha": "senha-segura-123"}


def test_cadastro_nao_devolve_a_senha_nem_o_hash(cliente: TestClient) -> None:
    resposta = cliente.post("/usuarios", json=CADASTRO)
    assert resposta.status_code == 201
    corpo = resposta.json()
    assert corpo["email"] == "ana@exemplo.com"
    assert set(corpo) == {"id", "email", "nome"}


def test_email_repetido_devolve_409(cliente: TestClient) -> None:
    cliente.post("/usuarios", json=CADASTRO)
    repetido = cliente.post("/usuarios", json={**CADASTRO, "email": "ana@exemplo.com"})
    assert repetido.status_code == 409


@pytest.mark.parametrize(
    "mudanca",
    [{"senha": "curta"}, {"email": "nao-e-email"}, {"nome": ""}, {"senha": "x" * 129}],
)
def test_cadastro_invalido_devolve_422(cliente: TestClient, mudanca: dict[str, str]) -> None:
    assert cliente.post("/usuarios", json={**CADASTRO, **mudanca}).status_code == 422


def test_login_com_sucesso_e_acesso_aos_proprios_dados(cliente: TestClient) -> None:
    cliente.post("/usuarios", json=CADASTRO)
    login = cliente.post(
        "/auth/token", data={"username": "ana@exemplo.com", "password": "senha-segura-123"}
    )
    assert login.status_code == 200 and login.json()["token_type"] == "bearer"
    cabecalho = {"Authorization": f"Bearer {login.json()['access_token']}"}
    assert cliente.get("/usuarios/eu", headers=cabecalho).json()["email"] == "ana@exemplo.com"


def test_senha_errada_e_email_inexistente_dao_a_mesma_resposta(cliente: TestClient) -> None:
    cliente.post("/usuarios", json=CADASTRO)
    errada = cliente.post("/auth/token", data={"username": "ana@exemplo.com", "password": "errada"})
    inexistente = cliente.post("/auth/token", data={"username": "zeca@x.com", "password": "errada"})
    assert errada.status_code == inexistente.status_code == 401
    assert errada.json() == inexistente.json()


def test_sem_token_ou_com_token_ruim_devolve_401(cliente: TestClient) -> None:
    assert cliente.get("/usuarios/eu").status_code == 401
    ruim = cliente.get("/usuarios/eu", headers={"Authorization": "Bearer lixo"})
    assert ruim.status_code == 401
    assert ruim.headers["www-authenticate"] == "Bearer"


def test_token_vencido_e_recusado(app: FastAPI, cliente: TestClient) -> None:
    cadastro = cliente.post("/usuarios", json=CADASTRO).json()
    vencido = criar_token(app.state.config, cadastro["id"], minutos=-1)
    resposta = cliente.get("/usuarios/eu", headers={"Authorization": f"Bearer {vencido}"})
    assert resposta.status_code == 401

    The post tests cover the business rules: require login, only the author can edit and delete (with the check that the post has not changed after the 403), the entire pagination (ids without repetition between pages), the search without case sensitivity, the % treated as text and the filter by author:

    projeto_blog/blog_api/tests/test_posts.py
    from fastapi.testclient import TestClient

NOVO = {"titulo": "Meu primeiro post", "conteudo": "Olá, mundo do FastAPI"}


def criar_post(
    cliente: TestClient, cabecalho: dict[str, str], **mudancas: str
) -> dict[str, object]:
    resposta = cliente.post("/posts", json={**NOVO, **mudancas}, headers=cabecalho)
    assert resposta.status_code == 201, resposta.text
    corpo: dict[str, object] = resposta.json()
    return corpo


def test_criar_exige_login(cliente: TestClient) -> None:
    assert cliente.post("/posts", json=NOVO).status_code == 401


def test_criar_e_ler(cliente: TestClient, ana: dict[str, str]) -> None:
    criado = criar_post(cliente, ana)
    assert criado["titulo"] == NOVO["titulo"] and criado["atualizado_em"] is None
    lido = cliente.get(f"/posts/{criado['id']}")
    assert lido.status_code == 200 and lido.json()["conteudo"] == NOVO["conteudo"]


def test_leitura_e_publica(cliente: TestClient, ana: dict[str, str]) -> None:
    criar_post(cliente, ana)
    assert cliente.get("/posts").status_code == 200


def test_post_inexistente_devolve_404(cliente: TestClient) -> None:
    assert cliente.get("/posts/999").status_code == 404


def test_validacao_do_corpo(cliente: TestClient, ana: dict[str, str]) -> None:
    assert cliente.post("/posts", json={"titulo": ""}, headers=ana).status_code == 422
    assert (
        cliente.post("/posts", json={"titulo": "x" * 201, "conteudo": "c"}, headers=ana).status_code
        == 422
    )


def test_so_o_autor_altera_e_remove(
    cliente: TestClient, ana: dict[str, str], bia: dict[str, str]
) -> None:
    post_id = criar_post(cliente, ana)["id"]
    novo = {"titulo": "Título novo", "conteudo": "Conteúdo novo"}

    assert cliente.put(f"/posts/{post_id}", json=novo, headers=bia).status_code == 403
    assert cliente.delete(f"/posts/{post_id}", headers=bia).status_code == 403
    assert cliente.get(f"/posts/{post_id}").json()["titulo"] == NOVO["titulo"]

    alterado = cliente.put(f"/posts/{post_id}", json=novo, headers=ana)
    assert alterado.status_code == 200 and alterado.json()["titulo"] == "Título novo"
    assert alterado.json()["atualizado_em"] is not None

    assert cliente.delete(f"/posts/{post_id}", headers=ana).status_code == 204
    assert cliente.get(f"/posts/{post_id}").status_code == 404
    assert cliente.delete(f"/posts/{post_id}", headers=ana).status_code == 404


def test_paginacao(cliente: TestClient, ana: dict[str, str]) -> None:
    for i in range(25):
        criar_post(cliente, ana, titulo=f"Post {i:02d}")

    primeira = cliente.get("/posts?limite=10").json()
    assert (primeira["total"], primeira["paginas"], len(primeira["dados"])) == (25, 3, 10)
    assert primeira["dados"][0]["titulo"] == "Post 24"  # os mais novos primeiro

    ultima = cliente.get("/posts?limite=10&pagina=3").json()
    assert len(ultima["dados"]) == 5
    assert cliente.get("/posts?pagina=99").json()["dados"] == []

    ids = [
        p["id"]
        for n in (1, 2, 3)
        for p in cliente.get(f"/posts?limite=10&pagina={n}").json()["dados"]
    ]
    assert len(ids) == len(set(ids)) == 25
    assert cliente.get("/posts?limite=51").status_code == 422
    assert cliente.get("/posts?pagina=0").status_code == 422


def test_busca_no_titulo_e_no_conteudo(cliente: TestClient, ana: dict[str, str]) -> None:
    criar_post(cliente, ana, titulo="Introdução ao FastAPI", conteudo="rotas e modelos")
    criar_post(cliente, ana, titulo="Banco de dados", conteudo="usando SQLAlchemy")
    criar_post(cliente, ana, titulo="Outro assunto", conteudo="sobre fastapi também")

    por_titulo = cliente.get("/posts?busca=introdução").json()
    assert por_titulo["total"] == 1
    sem_caixa = cliente.get("/posts?busca=FASTAPI").json()
    assert sem_caixa["total"] == 2
    assert cliente.get("/posts?busca=nada-disso").json()["total"] == 0


def test_busca_trata_o_percentual_como_texto(cliente: TestClient, ana: dict[str, str]) -> None:
    criar_post(cliente, ana, titulo="Promoção de 50% hoje")
    criar_post(cliente, ana, titulo="Sem símbolo")
    assert cliente.get("/posts?busca=%25").json()["total"] == 1


def test_filtrar_por_autor(cliente: TestClient, ana: dict[str, str], bia: dict[str, str]) -> None:
    criar_post(cliente, ana)
    criar_post(cliente, bia)
    criar_post(cliente, bia)
    autor_bia = cliente.get("/usuarios/eu", headers=bia).json()["id"]
    assert cliente.get(f"/posts?autor_id={autor_bia}").json()["total"] == 2


def test_resposta_nunca_expoe_dados_do_usuario(cliente: TestClient, ana: dict[str, str]) -> None:
    post = criar_post(cliente, ana)
    assert set(post) == {"id", "titulo", "conteudo", "autor_id", "criado_em", "atualizado_em"}


def test_saude(cliente: TestClient) -> None:
    assert cliente.get("/saude").json() == {"status": "ok"}

    Run

    Terminal
    uv sync
uv run pytest
    Output
    ......................                                                   [100%]
22 passed

    The same set, now in real PostgreSQL 16, pointing to the test database:

    Terminal
    TEST_DATABASE_URL="postgresql+psycopg://blog:blog_senha@localhost:5432/blog_teste" uv run pytest
    Output
    ......................                                                   [100%]
22 passed

    The 22 tests pass in both databases, without changing a line. This is the value of testing against the real database: subtle differences (case sensitivity in searches, uniqueness constraints, time zones) appear here, and not in production. For code quality:

    Terminal
    uv run mypy
uv run ruff check .
    Output
    Success: no issues found in 11 source files
All checks passed!

    The API running for real

    The tests use an in-memory client. To see the real server, point to the database and run:

    Terminal
    export BLOG_CHAVE_SECRETA="$(openssl rand -hex 32)"
export BLOG_BANCO_URL="postgresql+psycopg://blog:blog_senha@localhost:5432/blog_db"
uv run fastapi run app/main.py

    I started the server this way, against PostgreSQL, and took a tour of the API with an HTTP client. These are the real responses:

    Passeio pela API (servidor real, PostgreSQL 16)
    POST /usuarios               201 {"id": 1, "email": "ana@exemplo.com", "nome": "Ana"}
POST /usuarios (repetido)    409 {"detail": "E-mail já cadastrado"}
POST /auth/token             200 token_type=bearer partes_do_jwt=3
POST /posts (sem token)      401 {"detail": "Not authenticated"}
POST /posts                  201 {"id": 3, "titulo": "Post 3 sobre FastAPI", "autor_id": 1}
GET /posts?limite=2          200 total=3 paginas=2 titulos=['Post 3 sobre FastAPI', 'Post 2 sobre FastAPI']
GET /posts?busca=Post 2      200 total=1
PUT /posts/1 (sem token)     401 {"detail": "Not authenticated"}
DELETE /posts/3              204 ""
GET /posts/3                 404 {"detail": "Post não encontrado"}

    The message Not authenticated from the 401 without token comes from FastAPI itself, in English. To translate it, you can use a handler for HTTPException (chapter 12) that maps the message.

    What PostgreSQL created

    When starting, the create_all created the tables. This is the real schema of the table posts, queried with the psql (\d posts):

    psql: \d posts
                                            Table "public.posts"
    Column     |           Type           | Collation | Nullable |              Default
---------------+--------------------------+-----------+----------+-----------------------------------
 id            | integer                  |           | not null | nextval('posts_id_seq'::regclass)
 titulo        | character varying(200)   |           | not null |
 conteudo      | text                     |           | not null |
 autor_id      | integer                  |           | not null |
 criado_em     | timestamp with time zone |           | not null | now()
 atualizado_em | timestamp with time zone |           |          |
Indexes:
    "posts_pkey" PRIMARY KEY, btree (id)
    "ix_posts_autor_id" btree (autor_id)
Foreign-key constraints:
    "posts_autor_id_fkey" FOREIGN KEY (autor_id) REFERENCES usuarios(id) ON DELETE CASCADE

    Everything I wrote in the models appears there: the primary key, the index in autor_id (the listing filters by it), the now() as the database default, and the foreign key with ON DELETE CASCADE. And, in the table usuarios, the column senha_hash stored a hash that starts with $argon2id$: the real password is nowhere to be found.

    What is still missing for production

    MissingWhere it is in this course
    Schema migrationsAlembic (Python course, chapter 57)
    Limit login attemptsChapter 29 (the bucket of tokens as a dependency)
    Refresh tokens and real logoutChapter 20 (the warning about stateless JWT)
    Structured logs and metricsPython course, chapter 61 (Observability)
    DeployChapter 30
    Cache in the public listingChapter 28

    Push to GitHub

    The last lesson of the original course ends by pushing the code to a repository. The flow is the same (git init, git add, git commit, git push), with one caution: make sure that the .gitignore contains .env, .venv and __pycache__ before from the first commit. A secret that entered the Git history remains there even after being deleted from the file, and the only safe solution is to change the secret.