Python

Consumindo APIs REST com requests e httpx Já leu

9 min de leitura

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 du

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'>

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"]
    )

    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']}")

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) 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 += len(chunk)
                if total:
                    pct = baixado / total * 100
                    print(f"\rDownload: {pct:.1f}%", end="")
        print(f"\nSalvo em: {destino}")

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']}")

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:
        tarefas    = {cep: self.buscar(cep) for cep in ceps}
        resultados = {}
        for cep, tarefa in tarefas.items():
            resultados[cep] = await tarefa
        return 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())

Resumo

  • requests é a biblioteca síncrona mais usada — simples, expressiva e robusta
  • Use Session para múltiplas requisições ao mesmo host — reutiliza conexões TCP
  • raise_for_status() lança exceção automaticamente para códigos 4xx e 5xx
  • Sempre defina timeout — requisições sem timeout podem travar o programa indefinidamente
  • HTTPAdapter com Retry adiciona resiliência automática contra falhas temporárias
  • httpx oferece API compatível com requests e adiciona suporte nativo a async e HTTP/2
  • asyncio.gather() executa múltiplas requisições em paralelo — ideal para APIs com muitos endpoints
  • Nunca coloque tokens e senhas no código — use variáveis de ambiente

Referências e Leituras Complementares

  • 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.
Comentários

Mais em Python

Trabalhando com Datas e Horas
Trabalhando com Datas e Horas

Datas e horas estão em todo lugar no desenvolvimento de software — registros...

Artigo 47 — Design Patterns em Python
Artigo 47 — Design Patterns em Python

Design Patterns — padrões de projeto — são soluções reutilizáveis para proble...

Artigo 38 — Processamento de Linguagem Natural (NLP)
Artigo 38 — Processamento de Linguagem Natural (NLP)

Processamento de Linguagem Natural é o campo da IA que lida com texto e fala...