Como consultar CNPJ em Python

export APIPJ_KEY="apipj_live_sua_chave"
pip install requests

import os
import re
import time
from concurrent.futures import ThreadPoolExecutor

import requests

API_BASE = os.environ.get("APIPJ_BASE_URL", "https://apipj.com.br").rstrip("/")
API_KEY = os.environ["APIPJ_KEY"]  # nunca deixe a chave no código-fonte

_sessao = requests.Session()  # reaproveita a conexão TLS entre consultas
_sessao.headers.update({"Authorization": f"Bearer {API_KEY}", "User-Agent": "minha-app/1.0"})


class ApipjErro(Exception):
    def __init__(self, status: int, code: str, message: str):
        super().__init__(f"{status} {code}: {message}")
        self.status = status
        self.code = code


def normalizar(cnpj: str) -> str:
    """Tira pontuação e passa para maiúsculas (o CNPJ alfanumérico tem letras)."""
    return re.sub(r"[^0-9A-Za-z]", "", cnpj).upper()


PESOS = [6, 5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2]


def cnpj_valido(cnpj: str) -> bool:
    """Formato e dígitos verificadores, numérico ou alfanumérico (valor = código ASCII - 48)."""
    c = normalizar(cnpj)
    if not re.fullmatch(r"[A-Z0-9]{12}[0-9]{2}", c):
        return False
    v = [ord(ch) - 48 for ch in c]

    def dv(n: int) -> int:  # n = 12 (primeiro DV) ou 13 (segundo DV)
        resto = sum(v[i] * PESOS[i + 13 - n] for i in range(n)) % 11
        return 0 if resto < 2 else 11 - resto

    return v[12] == dv(12) and v[13] == dv(13)


def consultar_cnpj(cnpj: str, formato: str = "completo", tentativas: int = 3) -> dict | None:
    """Documento do CNPJ, ou None se ele não existe na base da Receita."""
    c = normalizar(cnpj)
    for tentativa in range(1, tentativas + 1):
        r = _sessao.get(f"{API_BASE}/v1/cnpj/{c}", params={"formato": formato}, timeout=10)
        if r.status_code == 200:
            return r.json()
        if r.status_code == 404:
            return None
        corpo = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        code = corpo.get("code", "erro")
        # 429 por minuto: espere o Retry-After e repita. Cota do mês esgotada (quota_exceeded) não adianta repetir.
        if r.status_code == 429 and code == "rate_limited" and tentativa < tentativas:
            time.sleep(float(r.headers.get("Retry-After", "1")))
            continue
        if r.status_code >= 500 and tentativa < tentativas:
            time.sleep(2 ** tentativa)
            continue
        raise ApipjErro(r.status_code, code, corpo.get("message", r.text[:200]))
    raise ApipjErro(429, "rate_limited", "limite por minuto atingido em todas as tentativas")


def consultar_varios(cnpjs: list[str], simultaneas: int = 5) -> dict[str, dict | None]:
    """Consulta em paralelo, com poucas conexões ao mesmo tempo (o limite por minuto vale para a chave)."""
    unicos = list(dict.fromkeys(normalizar(c) for c in cnpjs if cnpj_valido(c)))
    with ThreadPoolExecutor(max_workers=simultaneas) as pool:
        return dict(zip(unicos, pool.map(consultar_cnpj, unicos)))

from apipj_cnpj import consultar_cnpj

empresa = consultar_cnpj("11.378.117/0001-20")
print(empresa["razao_social"])                              # LEADS2B S/A
print(empresa["situacao_cadastral"]["descricao"])           # ATIVA
print(empresa["endereco"]["municipio"]["codigo_ibge"])      # 4106902
print(empresa["cnae_principal"]["descricao"])

simples = consultar_cnpj("00000000000191", formato="simplificado")
print(simples["cnpj"])   # 00.000.000/0001-91

from apipj_cnpj import ApipjErro, consultar_cnpj

try:
    consultar_cnpj("11378117000121")   # dígito errado
except ApipjErro as e:
    print(e.status, e.code)            # 400 invalid_cnpj

from apipj_cnpj import consultar_varios

resultado = consultar_varios(["11378117000120", "00000000000191", "60746948000112", "invalido"])
for cnpj, empresa in resultado.items():
    print(cnpj, empresa["razao_social"] if empresa else "não encontrado")

import json, time
from pathlib import Path
from apipj_cnpj import consultar_cnpj, normalizar

CACHE = Path("cache-cnpj")
CACHE.mkdir(exist_ok=True)
VALIDADE = 30 * 24 * 3600

def consultar_com_cache(cnpj: str) -> dict | None:
    arquivo = CACHE / f"{normalizar(cnpj)}.json"
    if arquivo.exists() and time.time() - arquivo.stat().st_mtime < VALIDADE:
        return json.loads(arquivo.read_text(encoding="utf-8"))
    empresa = consultar_cnpj(cnpj)
    if empresa is not None:
        arquivo.write_text(json.dumps(empresa, ensure_ascii=False), encoding="utf-8")
    return empresa

def para_cadastro(e: dict) -> dict:
    end = e["endereco"]
    return {
        "cnpj": e["cnpj_formatado"],
        "razao_social": e["razao_social"],
        "nome_fantasia": e["nome_fantasia"] or "",
        "situacao": e["situacao_cadastral"]["descricao"],
        "cnae": e["cnae_principal"]["codigo"],
        "logradouro": " ".join(filter(None, [end["tipo_logradouro"], end["logradouro"]])),
        "numero": end["numero"] or "S/N",
        "bairro": end["bairro"] or "",
        "cep": end["cep"],
        "cidade": end["municipio"]["nome"],
        "uf": end["uf"],
        "codigo_ibge": end["municipio"]["codigo_ibge"],
        "optante_simples": bool(e["simples"] and e["simples"]["optante"]),
    }