Consumindo APIs REST com requests e httpx

APIs REST são o vocabulário comum da internet moderna. Serviços de pagamento, mapas, clima, autenticação, inteligência artificial — quase tudo que um sistema profissional consome ou expõe segue esse padrão. Python tem duas bibliotecas excelentes para isso: requests, a mais usada da história do ecossistema, e httpx, sua sucessora moderna com suporte a requisições assíncronas.

O Protocolo HTTP em Resumo

Antes do código, os conceitos fundamentais:

Método   Significado          Uso típico
─────────────────────────────────────────
GET      Buscar dados         Listar, buscar
POST     Criar recurso        Cadastrar, enviar
PUT      Substituir recurso   Atualizar completo
PATCH    Atualizar parcial    Atualizar campos
DELETE   Remover recurso      Deletar

Código   Significado
──────────────────────────
200      OK
201      Created
204      No Content
400      Bad Request
401      Unauthorized
403      Forbidden
404      Not Found
422      Unprocessable Entity
429      Too Many Requests
500      Internal Server Error

requests: Instalação e Primeiros Passos

pip install requests

import requests

# GET simples
resposta = requests.get("https://api.github.com")

print(resposta.status_code)     # 200
print(resposta.headers["Content-Type"])
print(type(resposta.json()))    # <class 'dict'>

Um aviso que vale para todos os exemplos com requests: ele não tem timeout padrão. Sem o argumento timeout, uma requisição a um servidor que aceitou a conexão e parou de responder espera para sempre, e o programa trava sem erro. Os primeiros exemplos omitem o argumento para ficarem curtos; em código de verdade, passe-o sempre — de preferência como tupla, timeout=(5, 30), com o limite para conectar e o limite entre dois pedaços de resposta.

GET com Parâmetros

import requests

# Query parameters — ?q=python&per_page=5
resposta = requests.get(
    "https://api.github.com/search/repositories",
    params={
        "q":        "python web framework",
        "sort":     "stars",
        "order":    "desc",
        "per_page": 5
    }
)

if resposta.status_code == 200:
    dados = resposta.json()
    print(f"Total encontrado: {dados['total_count']}")
    for repo in dados["items"]:
        print(f"  ★ {repo['stargazers_count']:>6} — {repo['full_name']}")
else:
    print(f"Erro: {resposta.status_code}")

POST, PUT, PATCH e DELETE

import requests

BASE = "https://jsonplaceholder.typicode.com"

# POST — criando recurso
novo_post = {
    "title":  "Dominando Python",
    "body":   "Python é a linguagem mais versátil do mundo.",
    "userId": 1
}

resposta = requests.post(f"{BASE}/posts", json=novo_post)
print(f"POST {resposta.status_code}")
print(resposta.json())   # {"id": 101, "title": ...}

# PUT — substituição completa
resposta = requests.put(
    f"{BASE}/posts/1",
    json={"id": 1, "title": "Título atualizado", "body": "Novo conteúdo", "userId": 1}
)
print(f"PUT {resposta.status_code}")

# PATCH — atualização parcial
resposta = requests.patch(
    f"{BASE}/posts/1",
    json={"title": "Apenas o título muda"}
)
print(f"PATCH {resposta.status_code}")
print(resposta.json())

# DELETE
resposta = requests.delete(f"{BASE}/posts/1")
print(f"DELETE {resposta.status_code}")  # 200

Headers e Autenticação

import requests
import os

TOKEN = os.getenv("GITHUB_TOKEN")

# Headers personalizados
headers = {
    "Authorization": f"Bearer {TOKEN}",
    "Accept":        "application/vnd.github.v3+json",
    "User-Agent":    "MeuApp/1.0"
}

resposta = requests.get(
    "https://api.github.com/user",
    headers=headers
)

usuario = resposta.json()
print(f"Usuário: {usuario.get('login')}")
print(f"Nome:    {usuario.get('name')}")
print(f"Repos:   {usuario.get('public_repos')}")

# Basic Auth
resposta = requests.get(
    "https://api.exemplo.com/dados",
    auth=("usuario", "senha")
)

# Bearer Token com Session
session = requests.Session()
session.headers.update({"Authorization": f"Bearer {TOKEN}"})

r1 = session.get("https://api.github.com/user")
r2 = session.get("https://api.github.com/user/repos")

Session e Reutilização de Conexão

Session reutiliza conexões TCP — muito mais eficiente para múltiplas requisições ao mesmo host:

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def criar_session(retries=3, backoff=0.5):
    """Session com retry automático em falhas temporárias."""
    session = requests.Session()

    retry_strategy = Retry(
        total=retries,
        backoff_factor=backoff,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"]   # POST fica de fora — veja abaixo
    )

    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://",  adapter)

    return session


session = criar_session()

# Todas as requisições desta session usam retry automático
for user_id in range(1, 6):
    r = session.get(
        f"https://jsonplaceholder.typicode.com/users/{user_id}",
        timeout=10
    )
    user = r.json()
    print(f"  {user['name']:25} — {user['email']}")

O allowed_methods merece atenção, porque a versão que inclui "POST" é comum e perigosa. Repetir uma requisição só é seguro quando ela é idempotente — quando executá-la duas vezes dá o mesmo resultado que uma. GET, PUT e DELETE são; POST não. Um 503 não garante que o servidor deixou de processar o pedido: ele pode ter gravado o pagamento e falhado ao responder. Medido contra um servidor local que devolve 503 duas vezes antes de aceitar, a sessão com POST na lista fez o servidor receber três pagamentos para uma única chamada. Sem o allowed_methods, o padrão do urllib3 já exclui o POST. Quando a API oferece uma chave de idempotência — um cabeçalho como Idempotency-Key, que faz o servidor reconhecer a repetição —, aí sim o POST pode ser repetido.

Tratamento de Erros

import requests
from requests.exceptions import (
    Timeout,
    ConnectionError,
    HTTPError,
    RequestException
)


def requisicao_segura(url, **kwargs):
    try:
        resposta = requests.get(url, timeout=10, **kwargs)
        resposta.raise_for_status()   # lança HTTPError para 4xx e 5xx
        return resposta.json()

    except Timeout:
        print(f"Timeout: {url} demorou mais de 10s.")
    except ConnectionError:
        print(f"Falha de conexão: verifique a URL ou rede.")
    except HTTPError as e:
        codigo = e.response.status_code
        if codigo == 404:
            print(f"Recurso não encontrado: {url}")
        elif codigo == 401:
            print("Não autorizado — verifique as credenciais.")
        elif codigo == 429:
            print("Rate limit atingido — aguarde antes de tentar novamente.")
        else:
            print(f"Erro HTTP {codigo}: {e}")
    except RequestException as e:
        print(f"Erro inesperado: {e}")

    return None


dados = requisicao_segura("https://api.github.com/users/python")
if dados:
    print(f"Python org: {dados['public_repos']} repositórios públicos")

Upload de Arquivos e Formulários

import requests

# Enviando formulário (application/x-www-form-urlencoded)
resposta = requests.post(
    "https://httpbin.org/post",
    data={"campo1": "valor1", "campo2": "valor2"}
)

# Upload de arquivo (multipart/form-data)
with open("relatorio.pdf", "rb") as arquivo:
    resposta = requests.post(
        "https://api.exemplo.com/upload",
        files={"arquivo": ("relatorio.pdf", arquivo, "application/pdf")},
        headers={"Authorization": "Bearer TOKEN"}
    )

# Download de arquivo grande — streaming
def baixar_arquivo(url, destino):
    with requests.get(url, stream=True, timeout=(5, 30)) as r:
        r.raise_for_status()
        total = int(r.headers.get("content-length", 0))
        baixado = 0

        with open(destino, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)
                baixado = r.raw.tell()   # bytes recebidos da rede, não descomprimidos
                if total:
                    pct = baixado / total * 100
                    print(f"\rDownload: {pct:.1f}%", end="")
        print(f"\nSalvo em: {destino}")

O cálculo do progresso usa r.raw.tell(), e não a soma de len(chunk), por causa da compressão. Quando o servidor responde com Content-Encoding: gzip, o Content-Length é o tamanho comprimido, e o iter_content entrega os dados já descomprimidos. Medido com uma resposta de 100.000 bytes comprimida para 133, a soma dos pedaços levou a barra a 75.188%. O tell() conta o que chegou pela rede e termina em 100%.

httpx: A Alternativa Moderna

httpx tem API quase idêntica ao requests, mas adiciona suporte nativo a HTTP/2 e requisições assíncronas:

pip install httpx

import httpx

# Uso síncrono — igual ao requests
with httpx.Client(timeout=10.0) as client:
    resposta = client.get(
        "https://api.github.com/users/python",
        headers={"Accept": "application/vnd.github.v3+json"}
    )
    dados = resposta.json()
    print(f"Repositórios: {dados['public_repos']}")

A API é parecida, mas não idêntica, e três diferenças aparecem logo na migração. O httpx não segue redirecionamentos por padrão: onde o requests segue o 301 e devolve 200, o httpx devolve o próprio 301, e o raise_for_status() levanta erro nele. Para seguir, passe follow_redirects=True. Ao contrário do requests, ele tem timeout padrão, de 5 segundos — o que é bom, mas derruba com ReadTimeout a chamada a um relatório lento que antes funcionava. E o HTTP/2 não vem ligado: exige pip install "httpx[http2]" e httpx.Client(http2=True); sem o pacote, a criação do cliente levanta ImportError.

httpx Assíncrono

import httpx
import asyncio


async def buscar_usuario(client, username):
    resposta = await client.get(f"https://api.github.com/users/{username}")
    dados    = resposta.json()
    return {"login": dados["login"], "repos": dados["public_repos"]}


async def buscar_varios_usuarios(usernames: list):
    async with httpx.AsyncClient(timeout=15.0) as client:
        tarefas   = [buscar_usuario(client, u) for u in usernames]
        resultados = await asyncio.gather(*tarefas)
        return resultados


usuarios = asyncio.run(buscar_varios_usuarios([
    "python", "django", "pallets", "encode", "tiangolo"
]))

for u in usuarios:
    print(f"  {u['login']:20} — {u['repos']} repos")

Todas as requisições são feitas em paralelo — muito mais rápido que sequencial com requests.

Paginação

APIs com muitos resultados dividem a resposta em páginas:

import requests


def buscar_todos_repos(org: str, token: str = None) -> list:
    """Percorre todas as páginas de uma API com paginação."""
    headers = {}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    repos   = []
    pagina  = 1

    while True:
        resposta = requests.get(
            f"https://api.github.com/orgs/{org}/repos",
            headers=headers,
            params={"per_page": 100, "page": pagina}
        )
        resposta.raise_for_status()
        dados = resposta.json()

        if not dados:
            break

        repos.extend(dados)
        pagina += 1

        # Verifica header Link para próxima página
        link = resposta.headers.get("Link", "")
        if 'rel="next"' not in link:
            break

    return repos


repos = buscar_todos_repos("python")
print(f"Total de repositórios: {len(repos)}")
for repo in sorted(repos, key=lambda r: r["stargazers_count"], reverse=True)[:5]:
    print(f"  ★ {repo['stargazers_count']:>5} — {repo['name']}")

Exemplo Completo: Cliente para API de CEP

import httpx
import asyncio
from dataclasses import dataclass
from typing import Optional


@dataclass
class Endereco:
    cep:         str
    logradouro:  str
    bairro:      str
    cidade:      str
    estado:      str
    ibge:        str

    def __str__(self):
        return (f"{self.logradouro}, {self.bairro}\n"
                f"{self.cidade} — {self.estado}\n"
                f"CEP: {self.cep}")


class ClienteCEP:
    """Cliente para a API ViaCEP — https://viacep.com.br"""

    BASE_URL = "https://viacep.com.br/ws"

    def __init__(self):
        self._client = httpx.AsyncClient(timeout=10.0)

    async def buscar(self, cep: str) -> Optional[Endereco]:
        cep_limpo = cep.replace("-", "").replace(".", "").strip()

        if len(cep_limpo) != 8 or not cep_limpo.isdigit():
            raise ValueError(f"CEP inválido: {cep}")

        try:
            resposta = await self._client.get(
                f"{self.BASE_URL}/{cep_limpo}/json/"
            )
            resposta.raise_for_status()
            dados = resposta.json()

            if dados.get("erro"):
                return None

            return Endereco(
                cep=        dados["cep"],
                logradouro= dados.get("logradouro", ""),
                bairro=     dados.get("bairro", ""),
                cidade=     dados["localidade"],
                estado=     dados["uf"],
                ibge=       dados.get("ibge", "")
            )

        except httpx.HTTPStatusError as e:
            print(f"Erro HTTP: {e.response.status_code}")
            return None
        except httpx.RequestError as e:
            print(f"Erro de conexão: {e}")
            return None

    async def buscar_varios(self, ceps: list) -> dict:
        # gather dispara todas as consultas de uma vez; um laço com await seria sequencial
        resultados = await asyncio.gather(*(self.buscar(cep) for cep in ceps))
        return dict(zip(ceps, resultados))

    async def fechar(self):
        await self._client.aclose()


async def main():
    cliente = ClienteCEP()

    ceps = ["01310-100", "20040-020", "30112-000", "00000-000"]

    print("=== Consulta de CEPs ===\n")
    for cep in ceps:
        endereco = await cliente.buscar(cep)
        if endereco:
            print(f"CEP {cep}:")
            print(f"  {endereco}\n")
        else:
            print(f"CEP {cep}: não encontrado.\n")

    await cliente.fechar()


asyncio.run(main())

O buscar_varios usa asyncio.gather porque a versão mais intuitiva, que cria as corrotinas num dicionário e as aguarda num laço, não é concorrente: uma corrotina só começa a executar quando é aguardada, e o laço aguarda uma de cada vez. Medido com quatro consultas de meio segundo, o laço levou 2,00 s e o gather, 0,50 s. Vale notar ainda que o ViaCEP sinaliza CEP inexistente com {"erro": "true"}, com o valor em texto; o dados.get("erro") funciona porque qualquer string não vazia é verdadeira, mas uma comparação como dados.get("erro") is True nunca acertaria.

Consumir uma API REST com requests é fazer a requisição com o método certo, mandar os dados em params, json, data ou files, e ler a resposta. O que separa um script de um cliente confiável são as decisões que a biblioteca deixa para quem a usa. O requests não tem timeout padrão, e uma chamada sem ele pode travar o programa para sempre. O raise_for_status() transforma 4xx e 5xx em exceção, mas não protege de uma página HTML servida com 200. A Session reaproveita conexões e aceita um Retry, que só deve repetir métodos idempotentes — um POST repetido pode ser um pagamento em dobro. E o progresso de um download se mede pelos bytes que chegam da rede, não pelos descomprimidos.

O httpx traz a mesma ideia para o código assíncrono, com diferenças de comportamento que não aparecem na sintaxe: não segue redirecionamento sem follow_redirects=True, tem timeout padrão de 5 segundos, e só fala HTTP/2 com o extra instalado. A concorrência vem do asyncio.gather, não de um laço de await, e tem limites: o cliente abre no máximo 100 conexões, e o servidor do outro lado costuma aceitar menos que isso. Um semáforo mantém o número de requisições simultâneas sob controle, e o return_exceptions=True impede que um único erro descarte todos os resultados.

Fontes e leituras recomendadas

• requests — documentação oficial — https://docs.python-requests.org/en/latest/
• httpx — documentação oficial — https://www.python-httpx.org/
• asyncio — documentação oficial — https://docs.python.org/3/library/asyncio.html
• ViaCEP — API de CEPs brasileiros — https://viacep.com.br/
• httpbin — API para testes de HTTP — https://httpbin.org/
• JSONPlaceholder — API fake para testes — https://jsonplaceholder.typicode.com/
• MASSÉ, Mark. REST API Design Rulebook. O'Reilly Media, 2011. — fundamentos de design de APIs REST.
• PERCIVAL, Harry; GREGORY, Bob. Architecture Patterns with Python. O'Reilly Media, 2020. Cap. 8 — integração com serviços externos e testes de adaptadores.

Exercícios

Exercício 1

Um serviço web em Flask, com 8 workers, consulta a API de um parceiro a cada pedido com requests.get(url_parceiro). Numa sexta-feira, o parceiro passa a aceitar conexões mas não responde. Em minutos, o serviço inteiro para de atender — inclusive as páginas que não usam o parceiro — e os logs não mostram erro nenhum. Explique a cadeia de eventos e o que deveria estar no código.

Exercício 2

Uma rotina precisa consultar 500 CEPs e, para ganhar tempo, dispara todos de uma vez com asyncio.gather num único httpx.AsyncClient(). Contra um servidor que responde em um segundo, a rotina termina com 296 resultados e 204 exceções PoolTimeout. O desenvolvedor aumenta o timeout, e o parceiro passa a responder 429 Too Many Requests. Explique os dois comportamentos e proponha a forma certa de disparar muitas requisições.

Exercício 3

No buscar_varios deste artigo, uma das 200 entradas vem de um formulário com o CEP "1234". O buscar levanta ValueError para ele, e a rotina inteira falha: nenhum dos 199 endereços válidos é salvo, embora as requisições tenham sido feitas. Explique o comportamento do gather e corrija, mantendo a informação de qual entrada falhou.

Exercício 4

Uma integração faz requests.get(url, timeout=10), chama raise_for_status() e depois resposta.json(). Durante uma janela de manutenção do parceiro, o log se enche de JSONDecodeError: Expecting value: line 1 column 1 (char 0), e a equipe perde uma hora investigando um bug de parsing que não existe. O que o servidor estava devolvendo, e como o código deveria reagir?

Exercício 5

Um parceiro muda a API de api.parceiro.com para api2.parceiro.com e deixa um redirecionamento no endereço antigo. O cliente, com requests e o token no cabeçalho Authorization, passa a receber 401 Unauthorized em todas as chamadas, embora o token seja válido e o redirecionamento funcione no navegador. Explique por que isso acontece, e por que o comportamento, embora incômodo, está certo.