Como consumir API de CPF em Flask com blueprints e requests

flask_cpf_app/
├── app/
│ ├── __init__.py
│ ├── config.py
│ ├── cpf/
│ │ ├── __init__.py
│ │ ├── routes.py
│ │ └── services.py
│ └── errors.py
├── .env
├── requirements.txt
└── run.py

FLASK_APP=run.py
FLASK_ENV=development
CPFHUB_API_KEY=SUA_CHAVE_DE_API
CPFHUB_BASE_URL=https://api.cpfhub.io
CPFHUB_TIMEOUT=5

# app/config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    CPFHUB_API_KEY = os.getenv("CPFHUB_API_KEY")
    CPFHUB_BASE_URL = os.getenv("CPFHUB_BASE_URL", "https://api.cpfhub.io")
    CPFHUB_TIMEOUT = int(os.getenv("CPFHUB_TIMEOUT", "5"))

# app/cpf/services.py
import re
import requests
from flask import current_app

class CpfHubError(Exception):
    """Exceção personalizada para erros da API CPFHub."""
    def __init__(self, message: str, status_code: int = 500):
    self.message = message
    self.status_code = status_code
    super().__init__(self.message)

def consultar_cpf(cpf: str) -> dict:
    """
    Consulta dados de um CPF na API da CPFHub.io.
    Retorna o dicionário com os dados do titular.
    """
    cpf_limpo = re.sub(r"\D", "", cpf)

    if len(cpf_limpo) != 11:
    raise CpfHubError("CPF deve conter exatamente 11 dígitos", status_code=400)

    config = current_app.config
    url = f"{config['CPFHUB_BASE_URL']}/cpf/{cpf_limpo}"
    headers = {
    "x-api-key": config["CPFHUB_API_KEY"],
    "Accept": "application/json",
    }

    try:
    response = requests.get(
    url,
    headers=headers,
    timeout=config["CPFHUB_TIMEOUT"],
    )
    except requests.exceptions.Timeout:
    raise CpfHubError("Timeout ao consultar a API da CPFHub", status_code=504)
    except requests.exceptions.ConnectionError:
    raise CpfHubError("Erro de conexão com a API da CPFHub", status_code=502)
    except requests.exceptions.RequestException as e:
    raise CpfHubError(f"Erro na requisição: {str(e)}", status_code=502)

    error_map = {
    400: "CPF com formato inválido",
    401: "Chave de API inválida ou ausente",
    404: "CPF não encontrado na base de dados",
    }

    if response.status_code != 200:
    message = error_map.get(
    response.status_code,
    f"Erro inesperado: HTTP {response.status_code}",
    )
    raise CpfHubError(message, status_code=response.status_code)

    data = response.json()
    if not data.get("success"):
    raise CpfHubError("Resposta inesperada da API", status_code=500)

    return data["data"]

# app/cpf/__init__.py
from flask import Blueprint

cpf_bp = Blueprint("cpf", __name__, url_prefix="/api/cpf")

from . import routes # noqa: E402, F401

# app/cpf/routes.py
from flask import jsonify
from . import cpf_bp
from .services import consultar_cpf, CpfHubError

@cpf_bp.route("/<string:cpf>", methods=["GET"])
def consulta_cpf(cpf: str):
    """
    Endpoint para consulta de CPF.
    GET /api/cpf/<cpf>
    """
    try:
    dados = consultar_cpf(cpf)
    return jsonify({"success": True, "data": dados}), 200
    except CpfHubError as e:
    return jsonify({"success": False, "error": e.message}), e.status_code
    except Exception:
    return jsonify({"success": False, "error": "Erro interno do servidor"}), 500

# app/errors.py
from flask import jsonify

def register_error_handlers(app):
    @app.errorhandler(404)
    def not_found(error):
    return jsonify({"success": False, "error": "Recurso não encontrado"}), 404

    @app.errorhandler(405)
    def method_not_allowed(error):
    return jsonify({"success": False, "error": "Método não permitido"}), 405

    @app.errorhandler(500)
    def internal_error(error):
    return jsonify({"success": False, "error": "Erro interno do servidor"}), 500

# app/__init__.py
from flask import Flask
from .config import Config
from .cpf import cpf_bp
from .errors import register_error_handlers

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)

    # Registrar blueprints
    app.register_blueprint(cpf_bp)

    # Registrar error handlers
    register_error_handlers(app)

    return app

# run.py
from app import create_app

app = create_app()

if __name__ == "__main__":
    app.run(debug=True, port=5000)

python run.py

curl -X GET http://localhost:5000/api/cpf/12345678900

{
    "success": true,
    "data": {
    "cpf": "12345678900",
    "name": "João da Silva",
    "nameUpper": "JOÃO DA SILVA",
    "gender": "M",
    "birthDate": "15/06/1990",
    "day": 15,
    "month": 6,
    "year": 1990
    }
}

# app/cpf/services.py (versão com Session)
import re
import requests
from flask import current_app

_session = None

def _get_session() -> requests.Session:
    global _session
    if _session is None:
    _session = requests.Session()
    _session.headers.update({
    "x-api-key": current_app.config["CPFHUB_API_KEY"],
    "Accept": "application/json",
    })
    return _session

def consultar_cpf_batch(cpfs: list[str]) -> list[dict]:
    """Consulta múltiplos CPFs reutilizando a mesma sessão HTTP."""
    session = _get_session()
    timeout = current_app.config["CPFHUB_TIMEOUT"]
    base_url = current_app.config["CPFHUB_BASE_URL"]
    resultados = []

    for cpf in cpfs:
    cpf_limpo = re.sub(r"\D", "", cpf)
    url = f"{base_url}/cpf/{cpf_limpo}"
    try:
    resp = session.get(url, timeout=timeout)
    if resp.status_code == 200:
    data = resp.json()
    resultados.append({"cpf": cpf_limpo, "data": data.get("data"), "error": None})
    else:
    resultados.append({"cpf": cpf_limpo, "data": None, "error": f"HTTP {resp.status_code}"})
    except requests.exceptions.RequestException as e:
    resultados.append({"cpf": cpf_limpo, "data": None, "error": str(e)})

    return resultados