Forem: Pedro Parker

Give your AI agent access to 67M Brazilian companies, free MCP server

Pedro Parker — Wed, 29 Apr 2026 16:20:24 +0000

Originally posted on the CNPJ Aberto blog (PT-BR). This post is the English version, kept in sync.

If you've ever asked Claude to look up a Brazilian company by its CNPJ (the federal tax ID equivalent to a US EIN), you know how it goes:

It guesses, badly.
It says "I can't access external data."
It fires up web search and lands on a captcha-walled aggregator that returns half a page.

There's a fourth option now. You can give Claude — or any Model Context Protocol (MCP) client like Cursor, Cline, or Continue — a real tool. Plug in a free API key, and the LLM can pull a structured JSON record of any of ~67 million Brazilian companies (and ~70 million establishments) directly from the public dump from Brazil's federal revenue service.

I shipped two packages for this last week: cnpjaberto on PyPI and cnpjaberto on NPM. Both are dual-purpose: each one is a thin HTTP SDK and an MCP server bundled in the same package. So you pick whichever stack you're already in.

What it looks like in practice

A real prompt I sent to Claude Desktop after wiring it up:

"Look up CNPJ 18.236.120/0001-58. When was it founded, what's the main CNAE, and who are the partners?"

Claude calls lookup_cnpj under the hood and answers with a clean paragraph: Nu Pagamentos S.A., founded May 2013, registered as financial holding (CNAE 6435-2/01), partners include David Vélez, Edward Wible, Cristina Junqueira… No hallucination, no captcha, no stale Wikipedia snippet.

Or this one, which would be ~impossible without structured data:

"Find every active company where 'Maria Silva' appears as a partner, group by state, and tell me which industries dominate."

Claude chains companies_by_owner → cnae_stats. No SQL written by hand. The whole point of MCP is that the model decides when to call which tool — you just expose them.

Setup in 60 seconds

Two paths. Same result.

Option A — Node (zero-install via `npx`)

If you have Node 18+, you don't need to install anything. Drop this in ~/Library/Application Support/Claude/claude_desktop_config.json (macOS) or %APPDATA%\Claude\claude_desktop_config.json (Windows):

{
  "mcpServers": {
    "cnpjaberto": {
      "command": "npx",
      "args": ["-y", "cnpjaberto"],
      "env": {
        "CNPJABERTO_API_KEY": "your_key_here"
      }
    }
  }
}

npx -y cnpjaberto pulls the package on demand. Nothing global. Restart Claude and you're done.

Option B — Python

pip install cnpjaberto[mcp]

{
  "mcpServers": {
    "cnpjaberto": {
      "command": "cnpjaberto-mcp",
      "env": {
        "CNPJABERTO_API_KEY": "your_key_here"
      }
    }
  }
}

The API key is free — sign up at cnpjaberto.com.br/planos, copy the key, paste it above. The free tier is 1,000 requests per day and covers every endpoint.

What tools the LLM gets

Each tool is a thin wrapper over a documented REST endpoint. Names and contracts are the same on both Python and Node packages, so MCP clients see identical schemas regardless of which one you run.

Tool	What it returns
`lookup_cnpj`	Full registry: legal name, capital, CNAE, partners, all establishments (HQ + branches), addresses, phones
`list_filiais`	Branches of a parent company, paginated, optional state filter
`search_companies`	Search by legal name, brand name, or CNPJ digits (3+ chars)
`companies_by_owner`	Companies where a person appears as partner; partial CPF disambiguates homonyms
`companies_at_same_address`	Other companies registered at a given address (postal code, street, number)
`companies_by_contact`	Companies sharing the same email or phone (DDD + number)
`cnae_stats`	Aggregated stats for a CNAE code: total, top states, top municipalities, mortality
`panorama_overview`	National view: top states, top CNAEs, capital tiers, age tiers, 10-year history
`panorama_year`	Year-by-year cut: openings, closings, monthly series, MEI share

The whole design is "give the model boring, deterministic tools and let it compose them." Two tool calls and Claude can answer questions that previously required either a paid SaaS or a custom SQL pipeline against the 4 GB monthly RFB dump.

Use it as a plain SDK too

If you're not on the LLM hype train (or you're building a backend that uses an LLM but doesn't embed MCP), the same package is a normal HTTP client. No MCP runtime in the call path.

Python:

from cnpjaberto import Client

with Client() as cnpj:                       # reads CNPJABERTO_API_KEY from env
    company = cnpj.lookup("18.236.120/0001-58")
    print(company["razao_social"])

    snap = cnpj.panorama_year(2024)
    print(f"{snap['abertas']:,} new companies in 2024")

TypeScript / JavaScript:

import { Client } from "cnpjaberto";

const cnpj = new Client();                   // reads CNPJABERTO_API_KEY from env

const company = await cnpj.lookup("18.236.120/0001-58");
console.log(company.razao_social);

const snap = await cnpj.panoramaYear(2024);
console.log(`${snap.abertas} new companies in 2024`);

The two packages are mirror images (only casing differs: snake_case in Python, camelCase in TS). Same HTTP client the MCP server uses internally.

Typed errors, because life

from cnpjaberto import Client, NotFoundError, RateLimitError, AuthError

with Client() as cnpj:
    try:
        cnpj.lookup("00000000000000")
    except NotFoundError:
        ...
    except RateLimitError as e:
        print("Daily quota:", e.payload)
    except AuthError:
        ...

import { Client, NotFoundError, RateLimitError, AuthError } from "cnpjaberto";

try {
  await cnpj.lookup("00000000000000");
} catch (e) {
  if (e instanceof NotFoundError) { /* ... */ }
  if (e instanceof RateLimitError) console.log("daily quota:", e.payload);
  if (e instanceof AuthError) { /* ... */ }
}

Three exception types is the whole hierarchy. No custom retry logic to learn — just normal try/except (or try/catch).

Why bother building this

Two reasons.

One, MCP makes "AI agents that touch real data" actually feasible. Before it, every team rolled their own function-calling glue, and every glue handled errors and tokens differently. After it, you ship one package and any compliant client picks it up: Claude Desktop today, Cursor and Cline already, OpenAI's compat layer reportedly soon, plus a long tail of open-source agent frameworks. You pay the cost once.

Two, the Brazilian company registry is uniquely valuable as MCP fodder because the underlying queries are cheap (millisecond lookups in our index) but expensive to recreate locally (the public dump is ~4 GB of CSV, with monthly drops, and joining HQ/branches/partners across the three core tables is non-trivial). It's the perfect "I'd rather call a tool than build a pipeline" surface area.

If you're working on:

Lead generation or B2B sales tooling
Compliance / KYC / AML automation
Investigative journalism or OSINT
Tax / accounting software for Brazil
Anything that touches Brazilian invoices or vendor onboarding

…you probably want this in your agent's toolkit. Free tier covers more than enough for prototyping.

What's next

v0.1 is intentionally minimal. On the roadmap:

Async client in Python (AsyncClient) for FastAPI / async agents. Node already returns Promises everywhere.
Hosted MCP at mcp.cnpjaberto.com.br over HTTP+SSE — paste a URL, no local install.
Pro endpoints programmatically: list every company in a city, lead generator, full corporate ownership trees.

PRs welcome on both repos.

Construindo um explorador de rede societária com grafos em Python

Pedro Parker — Mon, 20 Apr 2026 15:51:46 +0000

Uma das funcionalidades mais interessantes do CNPJ Aberto é a rede societária — dado um CNPJ, o sistema mapeia todos os sócios da empresa, encontra outras empresas desses mesmos sócios, e constrói um grafo de conexões.

Isso transforma dados tabulares (CSV da Receita Federal) em inteligência empresarial. Advogados usam para due diligence, jornalistas para investigação, e analistas de crédito para avaliação de risco.

Neste post, vou mostrar como construímos esse sistema.

O modelo de dados

A base da Receita Federal tem uma relação simples:

empresas (cnpj_basico) ←──1:N──→ socios (cnpj_basico, nome_socio)

Cada empresa tem N sócios. Cada sócio pode aparecer em múltiplas empresas (identificado pelo nome). E um sócio pode ser uma pessoa jurídica (outra empresa), criando conexões indiretas.

Empresa A ← sócio "João Silva" → Empresa B
Empresa B ← sócio PJ "Empresa C" → Empresa C
Empresa C ← sócio "Maria Santos" → Empresa D

Isso forma um grafo que pode revelar estruturas corporativas complexas.

Modelagem do grafo

from dataclasses import dataclass

@dataclass
class GrupoNode:
    id: str
    tipo: str       # "empresa" | "pessoa"
    label: str      # razão social ou nome
    cnpj: str | None
    situacao_cadastral: str | None
    capital_social: float | None
    uf: str | None
    is_target: bool  # é a empresa que o usuário consultou?

@dataclass
class GrupoEdge:
    source: str     # node ID
    target: str     # node ID
    label: str      # qualificação do sócio

O grafo é uma lista de nodes (empresas e pessoas) e edges (relações societárias). Simples e serializável para JSON.

O traversal recursivo

O algoritmo começa em uma empresa e expande recursivamente:

MAX_DEPTH = 2
MAX_NETWORK_NODES = 150

async def get_grupo_empresarial(cnpj_basico: str, db):
    nodes = {}
    edges = []

    await traverse_company(
        cnpj_basico, db, nodes, edges, depth=0, is_target=True
    )

    return {
        "nodes": list(nodes.values()),
        "edges": edges,
    }

traverse_company: o coração do algoritmo

async def traverse_company(cnpj_basico, db, nodes, edges, depth, is_target=False):
    if depth > MAX_DEPTH:
        return
    if len(nodes) >= MAX_NETWORK_NODES:
        return

    company_id = f"emp:{cnpj_basico}"
    if company_id in nodes:
        return  # Já visitado — evita ciclos

    # 1. Buscar dados da empresa
    empresa = db.query(Empresa).filter(
        Empresa.cnpj_basico == cnpj_basico
    ).first()

    if not empresa:
        return

    matriz = db.query(Estabelecimento).filter(
        Estabelecimento.cnpj_basico == cnpj_basico,
        Estabelecimento.identificador_matriz_filial == "1"
    ).first()

    # Adicionar node da empresa
    nodes[company_id] = GrupoNode(
        id=company_id,
        tipo="empresa",
        label=empresa.razao_social,
        cnpj=format_cnpj(cnpj_basico),
        situacao_cadastral=matriz.situacao_cadastral if matriz else None,
        capital_social=empresa.capital_social,
        uf=matriz.uf if matriz else None,
        is_target=is_target,
    )

    # 2. Buscar sócios desta empresa
    socios = db.query(Socio).filter(
        Socio.cnpj_basico == cnpj_basico
    ).all()

    for socio in socios:
        if len(nodes) >= MAX_NETWORK_NODES:
            break

        if socio.identificador_socio == "1" and socio.cpf_cnpj_socio:
            # Sócio é PESSOA JURÍDICA → seguir como outra empresa
            socio_cnpj = socio.cpf_cnpj_socio[:8]
            socio_id = f"emp:{socio_cnpj}"

            edges.append(GrupoEdge(
                source=socio_id,
                target=company_id,
                label=socio.qualificacao or "Sócio PJ",
            ))

            # Recursão: explorar a empresa sócia
            await traverse_company(
                socio_cnpj, db, nodes, edges, depth + 1
            )
        else:
            # Sócio é PESSOA FÍSICA
            person_id = f"pf:{sanitize(socio.nome_socio)}"

            if person_id not in nodes:
                nodes[person_id] = GrupoNode(
                    id=person_id,
                    tipo="pessoa",
                    label=socio.nome_socio,
                    cnpj=None,
                    situacao_cadastral=None,
                    capital_social=None,
                    uf=None,
                    is_target=False,
                )

            edges.append(GrupoEdge(
                source=person_id,
                target=company_id,
                label=socio.qualificacao or "Sócio",
            ))

            # 3. Buscar OUTRAS empresas desta pessoa
            await expand_person(
                socio.nome_socio, cnpj_basico, db, 
                nodes, edges, depth
            )

expand_person: encontrando empresas conectadas

async def expand_person(nome_socio, exclude_cnpj, db, nodes, edges, depth):
    # Buscar outros cnpj_basico onde esta pessoa é sócia
    other_companies = db.query(Socio.cnpj_basico).filter(
        Socio.nome_socio == nome_socio,
        Socio.cnpj_basico != exclude_cnpj,
    ).distinct().limit(10).all()

    person_id = f"pf:{sanitize(nome_socio)}"

    for row in other_companies:
        if len(nodes) >= MAX_NETWORK_NODES:
            break

        company_id = f"emp:{row.cnpj_basico}"

        edges.append(GrupoEdge(
            source=person_id,
            target=company_id,
            label="Sócio",
        ))

        # Continuar traversal na empresa encontrada
        await traverse_company(
            row.cnpj_basico, db, nodes, edges, depth + 1
        )

Proteções contra explosão combinatória

Sem limitações, o grafo pode explodir. Um sócio que aparece em 200 empresas, cada empresa com 5 sócios, cada sócio em 10 empresas... rapidamente vira milhões de nodes.

As proteções:

Proteção	Valor	Por quê
`MAX_DEPTH`	2	Limita a profundidade da recursão
`MAX_NETWORK_NODES`	150	Cap total de nodes no grafo
`LIMIT 10` em `expand_person`	10	Limita empresas por pessoa
Checagem `if company_id in nodes`	—	Evita ciclos e re-processamento

Depth 2 é suficiente? Na prática, sim. A maioria das estruturas societárias interessantes fica a 1-2 hops de distância. Ir mais fundo geralmente adiciona ruído sem valor.

Detecção de Red Flags

Com o grafo pronto, podemos detectar padrões suspeitos:

def detect_red_flags(cnpj_basico, db):
    flags = []

    # 1. Sócio em muitas empresas (possível laranja)
    socios = db.query(
        Socio.nome_socio, func.count(distinct(Socio.cnpj_basico))
    ).filter(
        Socio.cnpj_basico == cnpj_basico
    ).group_by(Socio.nome_socio).all()

    for nome, count in socios:
        total = db.query(func.count(distinct(Socio.cnpj_basico))).filter(
            Socio.nome_socio == nome
        ).scalar()

        if total >= 5:
            flags.append({
                "tipo": "socio_multiplas_empresas",
                "severidade": "media",
                "titulo": f"Sócio em {total} empresas",
                "descricao": f"{nome} é sócio em {total} empresas diferentes",
            })

    # 2. Muitas empresas no mesmo endereço
    matriz = get_matriz(cnpj_basico, db)
    if matriz and matriz.cep and matriz.logradouro:
        same_address = db.query(func.count()).filter(
            Estabelecimento.cep == matriz.cep,
            Estabelecimento.logradouro == matriz.logradouro,
            Estabelecimento.numero == matriz.numero,
            Estabelecimento.cnpj_basico != cnpj_basico,
            Estabelecimento.identificador_matriz_filial == "1",
        ).scalar()

        if same_address >= 3:
            flags.append({
                "tipo": "concentracao_endereco",
                "severidade": "baixa",
                "titulo": f"{same_address} empresas no mesmo endereço",
                "descricao": "Concentração incomum de empresas",
            })

    # 3. Contato compartilhado (email ou telefone)
    if matriz and matriz.email:
        shared = db.query(func.count(distinct(
            Estabelecimento.cnpj_basico
        ))).filter(
            Estabelecimento.email == matriz.email,
            Estabelecimento.cnpj_basico != cnpj_basico,
        ).scalar()

        if shared >= 1:
            flags.append({
                "tipo": "contato_compartilhado",
                "severidade": "baixa",
                "titulo": "Email usado por outra empresa",
                "descricao": f"O email {matriz.email} aparece em {shared + 1} empresas",
            })

    # Calcular score de risco (0-100)
    score = sum(
        35 if f["severidade"] == "alta" else
        20 if f["severidade"] == "media" else 10
        for f in flags
    )

    return {"score": min(score, 100), "flags": flags}

Cache: essencial para grafos

A construção do grafo envolve múltiplas queries recursivas. Sem cache, cada visualização levaria 1-3 segundos. Com Redis:

async def get_grupo_cached(cnpj_basico, db):
    cache_key = f"grupo:{cnpj_basico}"

    cached = redis.get(cache_key)
    if cached:
        return json.loads(cached)

    result = await get_grupo_empresarial(cnpj_basico, db)

    # Cache por 24h — dados mudam mensalmente
    redis.setex(cache_key, 86400, json.dumps(result))

    return result

Frontend: visualização do grafo

O JSON {nodes, edges} é renderizado no frontend com uma biblioteca de grafos. O componente React recebe os dados e renderiza nodes como cards e edges como linhas de conexão:

// Simplificado
function CorporateGroup({ cnpj }) {
  const [data, setData] = useState(null);

  useEffect(() => {
    fetch(`/api/intelligence/grupo/${cnpj}`)
      .then(r => r.json())
      .then(setData);
  }, [cnpj]);

  if (!data || data.nodes.length === 0) return null;

  return (
    <div className="relative">
      {data.nodes.map(node => (
        <NodeCard key={node.id} node={node} />
      ))}
      {data.edges.map((edge, i) => (
        <EdgeLine key={i} edge={edge} nodes={data.nodes} />
      ))}
    </div>
  );
}

Empresas ativas são destacadas em verde, inativas em vermelho. A empresa alvo da consulta fica em destaque. Sócios PF são mostrados com ícone de pessoa.

Casos de uso reais

Esse tipo de análise de rede revela coisas que dados tabulares escondem:

Due diligence: "O sócio da empresa que estou contratando também é sócio de uma empresa com situação 'Inapta'?"
Investigação: "Quem são as pessoas por trás de um grupo de empresas com o mesmo endereço?"
Análise de crédito: "Este solicitante de empréstimo tem sócios com histórico de empresas encerradas?"
Compliance: "O fornecedor que estamos avaliando tem conexões com empresas em situação irregular?"

Conclusão

Construir um explorador de rede societária envolveu:

Traversal recursivo com proteções contra explosão (depth, max nodes, ciclos)
Dual-type nodes (empresa vs pessoa) com expansão bidirecional
Heurísticas de red flags baseadas em padrões estatísticos
Cache agressivo — grafos são caros de construir e mudam raramente
Limites pragmáticos — depth 2 e 150 nodes cobrem 95% dos casos úteis

A beleza é que tudo isso roda sobre dados públicos. Não há scraping, não há API paga, não há magia. São dados da Receita Federal, organizados e conectados de forma que se tornam inteligência de verdade.

Quer explorar a rede societária de qualquer empresa brasileira? Teste no CNPJ Aberto de forma gratuita.

Rate limiting com Redis para um SaaS freemium: a arquitetura por trás de 3 tiers

Pedro Parker — Mon, 20 Apr 2026 01:23:44 +0000

No CNPJ Aberto, oferecemos consultas gratuitas de empresas brasileiras com um modelo freemium:

Tier	Limite diário	Identificação
Anônimo	50/dia	IP
Free (conta gratuita)	200/dia	User ID
Pro (R$39/mês)	5.000/dia	User ID

Implementar isso parece simples, mas os detalhes fazem a diferença entre um sistema robusto e um cheio de edge cases. Neste post, vou mostrar a arquitetura completa usando Redis e FastAPI middleware.

Por que Redis?

Rate limiting precisa de:

Contadores atômicos — múltiplas requests simultâneas não podem criar race conditions
Expiração automática — o contador deve resetar a cada período
Baixa latência — não pode adicionar overhead perceptível à request
Memória eficiente — milhões de chaves com TTL

Redis resolve tudo isso com dois comandos: INCR e EXPIRE.

A chave do rate limit

def get_rate_limit_key(identifier: str) -> str:
    today = datetime.utcnow().strftime("%Y-%m-%d")
    return f"rl:{identifier}:{today}"

Formato: rl:{identifier}:{YYYY-MM-DD}

Exemplos:

Anônimo: rl:189.44.52.100:2026-04-16
Logado: rl:user:42:2026-04-16

A data na chave garante que o contador reseta automaticamente à meia-noite UTC. Não precisamos de cron jobs para limpar contadores — o EXPIRE cuida disso.

O check: INCR + EXPIRE atômico

async def check_rate_limit(redis, identifier: str, limit: int):
    key = get_rate_limit_key(identifier)

    current = await redis.incr(key)

    if current == 1:
        # Primeira request do dia — definir TTL de 24h
        await redis.expire(key, 86400)

    if current > limit:
        return False, current  # Rate limited

    return True, current

Por que INCR primeiro? Porque INCR no Redis é atômico — se dois requests chegarem ao mesmo tempo, um vai receber 1 e o outro 2. Nunca teremos race condition.

Por que checar current == 1? Na primeira request do dia, a chave ainda não existe. O INCR cria a chave com valor 1. Nesse momento, definimos o TTL de 86400 segundos (24h). Se não fizermos isso, a chave ficaria para sempre.

Edge case: e se o EXPIRE falhar?

Se o EXPIRE falhar (Redis reiniciou entre o INCR e o EXPIRE, por exemplo), teríamos uma chave sem TTL que nunca expira. Solução de segurança:

if current == 1:
    await redis.expire(key, 86400)
elif current == 2:
    # Safeguard: verificar se o TTL foi setado
    ttl = await redis.ttl(key)
    if ttl == -1:  # Sem TTL
        await redis.expire(key, 86400)

Identificando o usuário: IP vs JWT

O rate limit precisa saber quem está fazendo a request. A ordem de identificação:

def get_identifier(request: Request) -> tuple[str, str]:
    # 1. Tentar extrair user do JWT (cookie ou header)
    token = extract_token(request)
    if token:
        user_id = decode_jwt(token).get("sub")
        if user_id:
            return f"user:{user_id}", "authenticated"

    # 2. Fallback: IP do cliente
    ip = (
        request.headers.get("X-Forwarded-For", "").split(",")[0].strip()
        or request.client.host
    )
    return ip, "anonymous"

Cuidado com X-Forwarded-For: Sempre pegar o primeiro IP da lista (o IP real do cliente). Os subsequentes são proxies intermediários. Se você não está atrás de um reverse proxy, use request.client.host.

Determinando o plano: cache de 5 minutos

Buscar o plano do usuário no banco a cada request seria caro. Usamos Redis como cache:

async def get_effective_plan(redis, db, user_id: int) -> str:
    cache_key = f"uplan:{user_id}"

    cached = await redis.get(cache_key)
    if cached:
        return cached.decode()

    user = db.query(User).filter(User.id == user_id).first()
    plan = user.plan if user else "free"

    # Cache por 5 minutos
    await redis.setex(cache_key, 300, plan)

    return plan

Por que 5 minutos? Se o usuário fizer upgrade para Pro, o novo limite entra em vigor em no máximo 5 minutos. É um tradeoff aceitável entre performance e experiência.

Invalidação: Quando o webhook de pagamento processa um upgrade, fazemos redis.delete(f"uplan:{user_id}") para invalidar o cache imediatamente.

O Middleware FastAPI

Tudo se junta no middleware que intercepta cada request:

COUNTED_PREFIXES = [
    "/api/cnpj/",
    "/api/search",
    "/api/busca-avancada",
    "/api/leads",
    "/api/socios/busca",
    "/api/pessoa/",
]

RATE_LIMITS = {
    "anon": 50,
    "free": 200,
    "pro": 5000,
}

class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        path = request.url.path

        # Só conta requests que consomem dados
        if not any(path.startswith(p) for p in COUNTED_PREFIXES):
            return await call_next(request)

        identifier, auth_type = get_identifier(request)

        if auth_type == "authenticated":
            plan = await get_effective_plan(redis, db, user_id)
        else:
            plan = "anon"

        limit = RATE_LIMITS[plan]
        allowed, current = await check_rate_limit(redis, identifier, limit)

        if not allowed:
            return JSONResponse(
                status_code=429,
                content={"detail": "Limite de consultas atingido"},
                headers=rate_limit_headers(limit, 0, plan),
            )

        response = await call_next(request)

        # Adicionar headers informativos
        response.headers.update(
            rate_limit_headers(limit, limit - current, plan)
        )

        return response

Headers X-RateLimit

def rate_limit_headers(limit, remaining, plan):
    return {
        "X-RateLimit-Limit": str(limit),
        "X-RateLimit-Remaining": str(max(0, remaining)),
        "X-RateLimit-Plan": plan,
    }

O frontend lê esses headers para mostrar um banner quando o usuário está chegando no limite:

// Frontend: lê os headers e avisa o usuário
function trackRateLimit(res: Response) {
  const remaining = res.headers.get("x-ratelimit-remaining");
  const limit = res.headers.get("x-ratelimit-limit");

  if (remaining && parseInt(remaining) <= 5) {
    // Mostra banner: "Você tem apenas X consultas restantes hoje"
    showRateLimitBanner(parseInt(limit), parseInt(remaining));
  }
}

Rate limit separado para autenticação

async def check_auth_rate_limit(redis, ip: str):
    minute = datetime.utcnow().strftime("%Y%m%d%H%M")
    key = f"rl:auth:{ip}:{minute}"

    current = await redis.incr(key)
    if current == 1:
        await redis.expire(key, 120)  # 2 minutos de janela

    return current <= 10  # Max 10 tentativas por minuto

10 tentativas por minuto por IP. Janela de 2 minutos (não 1) para cobrir o caso de requests que chegam no segundo 59 e 00.

O que não contar

É importante não consumir quota em requests que não geram valor para o usuário:

# Não conta:
# - Assets estáticos (/_next/*, /fonts/*)
# - Health checks (/api/health)
# - Auth endpoints (/api/auth/*)  ← tem rate limit próprio
# - Municipios/CNAEs search (autocomplete de filtros)

# Conta:
# - Consulta de CNPJ (/api/cnpj/{cnpj})
# - Busca textual (/api/search)
# - Busca avançada (/api/busca-avancada)
# - Leads (/api/leads)

Se contássemos autocomplete de filtros, um usuário gastaria metade do limite apenas navegando pela busca avançada.

Redirecionamento no frontend

Quando o backend retorna 429, o frontend redireciona para uma página explicativa:

async function apiFetch(url: string, init?: RequestInit) {
  const res = await fetch(url, init);

  if (res.status === 429 && typeof window !== "undefined") {
    window.location.href = "/limite";
  }

  return res;
}

A página /limite explica o que aconteceu, mostra os planos, e incentiva o upgrade. É a principal conversão do freemium.

Monitoramento

No painel admin, mostramos:

Requests totais por tier (anon/free/pro)
Quantos 429 foram servidos hoje
Top IPs por consumo
Distribuição de uso (quantos usam <10%, 10-50%, 50-100% do limite)

Isso ajuda a calibrar os limites. Se muitos anônimos estão batendo 50/dia, talvez devêssemos subir para 75 para reduzir frustração sem impactar conversão.

Conclusão

Rate limiting parece simples, mas um sistema robusto precisa:

Redis INCR + EXPIRE para contadores atômicos sem race conditions
Identificação em cascata (JWT → IP) para cobrir todos os cenários
Cache do plano para evitar query ao banco em cada request
Scoping correto — não contar requests que não geram valor
Headers informativos para o frontend mostrar o estado
Rate limits separados para auth (brute force) vs consulta (quota)

Next.js 15 com 55 milhões de páginas dinâmicas: SSR, SEO e performance

Pedro Parker — Sun, 19 Apr 2026 02:54:02 +0000

O CNPJ Aberto tem uma página dedicada para cada empresa brasileira. São 55 milhões de páginas — cada uma com título, description, OpenGraph image e JSON-LD únicos.

Gerar tudo em build time (SSG) levaria dias e ocuparia terabytes. Usar client-side rendering mataria o SEO. A solução? Server Components com SSR on-demand no Next.js 15.

Neste post, vou mostrar as decisões de arquitetura que fazem isso funcionar.

O problema: 55M páginas únicas

Cada página /cnpj/[cnpj] precisa de:

✅ Title dinâmico — "EMPRESA XPTO LTDA — CNPJ 12.345.678/0001-00"
✅ Meta description — com situação, local, CNAE, capital social
✅ OpenGraph image — gerada dinamicamente com dados da empresa
✅ JSON-LD — schema.org Organization para rich results
✅ Canonical URL — para evitar duplicatas (CNPJ formatado vs não formatado)
✅ Conteúdo completo — renderizado no servidor para crawlers

getStaticPaths com 55M paths? Impossível. getStaticProps com ISR? O cold start para 55M paths seria brutal. Server Components com generateMetadata é a resposta.

Arquitetura: Server Component puro

frontend/src/app/cnpj/[cnpj]/page.tsx  → Server Component (SSR)
frontend/src/components/CompanyDetail.tsx → Dynamic imports para client components

A page.tsx é um Server Component, sem "use client". Isso significa:

Zero JavaScript enviado ao browser para a renderização inicial
generateMetadata roda no servidor — Google recebe os meta tags corretos
Data fetching direto no componente, sem useEffect/useState

// page.tsx — Server Component
export async function generateMetadata({ params }) {
  const { cnpj } = await params;
  const empresa = await getEmpresa(cnpj);

  if (!empresa) {
    return { title: "CNPJ não encontrado", robots: { index: false } };
  }

  const matriz = empresa.estabelecimentos.find(
    e => e.identificador_matriz_filial === "Matriz"
  );
  const cnpjFormatted = formatCnpj(matriz?.cnpj || cnpj);

  return {
    title: `${empresa.razao_social} — CNPJ ${cnpjFormatted}`,
    description: [
      `CNPJ ${cnpjFormatted}`,
      `Situação: ${matriz?.situacao_cadastral}`,
      `${matriz?.municipio}/${matriz?.uf}`,
      `Capital Social: R$ ${empresa.capital_social?.toLocaleString("pt-BR")}`,
      `${empresa.socios.length} sócio(s)`,
    ].filter(Boolean).join(" · "),
    alternates: { canonical: `/cnpj/${cnpj.replace(/\D/g, "")}` },
    openGraph: { title: empresa.razao_social, type: "website" },
    twitter: { card: "summary_large_image" },
  };
}

export default async function CnpjPage({ params }) {
  const { cnpj } = await params;
  const empresa = await getEmpresa(cnpj);
  if (!empresa) notFound();

  return (
    <main className="flex-1 w-full px-3 sm:px-6 py-3 sm:py-4">
      <JsonLd data={organizationSchema} />
      <CompanyDetail empresa={empresa} />
    </main>
  );
}

Detalhe: getEmpresa é wrapped com cache do React, se generateMetadata e o componente chamam a mesma função com o mesmo argumento, a query só roda uma vez.

import { cache } from "react";

export const getEmpresa = cache(async function getEmpresa(cnpj: string) {
  const res = await apiFetch(`${getApiBase()}/api/cnpj/${cleaned}`);
  if (res.status === 404) return null;
  return res.json();
});

OpenGraph Image dinâmica

Cada empresa tem uma OG image única, gerada sob demanda pelo Next.js:

// cnpj/[cnpj]/opengraph-image.tsx
import { ImageResponse } from "next/og";

export default async function OGImage({ params }) {
  const empresa = await getEmpresa(params.cnpj);

  return new ImageResponse(
    <div style={{ display: "flex", flexDirection: "column", /* ... */ }}>
      <h1>{empresa.razao_social}</h1>
      <p>CNPJ {formatCnpj(empresa.cnpj)}</p>
      <p>Situação: {empresa.situacao_cadastral}</p>
    </div>,
    { width: 1200, height: 630 }
  );
}

Quando alguém compartilha um link de empresa no Twitter/LinkedIn, a imagem mostra dados reais da empresa. O Next.js cacheia a imagem depois da primeira geração.

Code Splitting agressivo

A página de empresa tem muitos componentes interativos: rede societária (grafo), mapa, red flags, score de saúde, notas do usuário... Carregar tudo de uma vez seria ~200KB de JavaScript.

Solução: next/dynamic para tudo que não é above-the-fold:

// CompanyDetail.tsx
import dynamic from "next/dynamic";

const CardLayoutManager = dynamic(() => import("./CardLayoutManager"));
const AddressCompanies = dynamic(() => import("./AddressCompanies"));
const MonitorButton = dynamic(() => import("./MonitorButton"));
const ProActionBar = dynamic(() => import("./ProActionBar"));

E dentro do CardLayoutManager, mais lazy loading:

const PartnerNetwork = lazy(() => import("./PartnerNetwork"));
const CorporateGroup = lazy(() => import("./CorporateGroup"));
const RedFlags = lazy(() => import("./RedFlags"));
const HealthScore = lazy(() => import("./HealthScore"));
const CompanyMap = lazy(() => import("./CompanyMap"));

Cada componente pesado é um chunk separado que só carrega quando o card entra no viewport. Resultado: o JavaScript inicial da página caiu de ~200KB para ~45KB.

Canonical URLs via Middleware

Um CNPJ pode ser digitado de várias formas:

/cnpj/12345678000100 (limpo)
/cnpj/12.345.678/0001-00 (formatado)
/cnpj/12.345.678%2F0001-00 (URL-encoded)

Todas devem apontar para a mesma página. O middleware do Next.js faz redirect 301 automático:

// middleware.ts
export function middleware(request: NextRequest) {
  const { pathname } = request.nextUrl;

  if (pathname.startsWith("/cnpj/")) {
    const raw = pathname.slice("/cnpj/".length);
    const digits = raw.replace(/\D/g, "");

    // Se tem 14 dígitos mas a URL não é limpa → redirect 301
    if (digits.length === 14 && raw !== digits) {
      const url = request.nextUrl.clone();
      url.pathname = `/cnpj/${digits}`;
      return NextResponse.redirect(url, 301);
    }
  }

  return NextResponse.next();
}

Isso garante que o Google indexe apenas a versão canônica de cada URL.

JSON-LD para Rich Results

Structured data ajuda o Google a entender a página e exibir rich snippets:

const jsonLd = {
  "@context": "https://schema.org",
  "@type": "Organization",
  name: empresa.razao_social,
  alternateName: matriz?.nome_fantasia,
  taxID: formatCnpj(cnpj),
  address: {
    "@type": "PostalAddress",
    streetAddress: `${matriz.logradouro} ${matriz.numero}`,
    addressLocality: matriz.municipio,
    addressRegion: matriz.uf,
    postalCode: matriz.cep,
    addressCountry: "BR",
  },
  telephone: matriz?.telefone,
  email: matriz?.email,
};

Componente reutilizável para injetar no <head>:

function JsonLd({ data }) {
  return (
    <script
      type="application/ld+json"
      dangerouslySetInnerHTML={{ __html: JSON.stringify(data) }}
    />
  );
}

API Proxy via Rewrites

O frontend faz requests para /api/* que o Next.js proxeia para o backend FastAPI:

// next.config.js
async rewrites() {
  return [{
    source: "/api/:path*",
    destination: `${process.env.NEXT_PUBLIC_API_URL}/api/:path*`,
  }];
}

Vantagem: O browser nunca fala direto com o backend. Sem CORS, sem exposição de IP interno, cookies funcionam transparentemente.

Para SSR, o servidor Next.js fala direto com o backend via URL interna (API_URL), evitando um hop extra:

function getApiBase() {
  if (typeof window === "undefined") {
    return process.env.API_URL || "http://localhost:8000";
  }
  return ""; // Client → usa rewrites
}

Resultados de performance

Testado com PageSpeed Insights:

Métrica	Valor
LCP (Largest Contentful Paint)	~1.2s
FID (First Input Delay)	~50ms
CLS (Cumulative Layout Shift)	0.02
Time to First Byte	~200ms
JavaScript total (initial)	~45KB gzipped

O segredo é simples: Server Components renderizam o HTML no servidor, dynamic imports carregam JavaScript sob demanda, e o cache do React evita queries duplicadas.

Conclusão

Para sites com milhões de páginas dinâmicas, Next.js 15 com App Router é uma combinação poderosa:

Server Components = SEO perfeito sem hydration cost
generateMetadata = meta tags dinâmicos sem SSG
cache() = deduplica data fetching entre metadata e componente
next/dynamic + lazy() = code splitting granular
Middleware = canonicalização de URLs sem lógica no componente
Rewrites = proxy transparente sem CORS

Busca fuzzy em 55 milhões de registros: como pg_trgm salvou meu projeto

Pedro Parker — Fri, 17 Apr 2026 15:37:12 +0000

"Buscar por nome" parece simples até você ter 55 milhões de registros e o usuário esperar resposta em menos de 300ms.

No CNPJ Aberto, os usuários buscam empresas por razão social ("Magazine Luiza"), nome fantasia ("Magalu"), CNPJ ("33.000.167/0001-01"), ou até nome de sócio ("Luiza Helena Trajano"). A busca precisa ser fuzzy (tolerar erros de digitação), rápida, e funcionar em três tabelas diferentes simultaneamente.

Neste post, vou mostrar como fizemos isso sem Elasticsearch — usando apenas PostgreSQL com pg_trgm.

Por que não Elasticsearch?

Elasticsearch é a resposta óbvia para full-text search. Mas:

Mais um serviço para manter, monitorar e escalar
Sincronização entre PostgreSQL e ES é complexa (lag, inconsistências)
Custo — ES consome muita RAM (55M docs = 16+ GB de heap)
Complexidade — para o nosso caso de uso, é um canhão para matar formiga

PostgreSQL com pg_trgm resolve o problema com uma fração da complexidade.

O que é pg_trgm?

pg_trgm (trigram) decompõe strings em conjuntos de 3 caracteres consecutivos e calcula similaridade entre eles.

SELECT show_trgm('Magazine');
-- {"  m"," ma","aga","azi","gaz","ine","mag","ne ","zin"}

Quando você busca "Magazin" (sem o 'e'), o PostgreSQL compara os trigramas e encontra alta similaridade com "Magazine". Isso é o que torna a busca tolerante a erros.

CREATE EXTENSION pg_trgm;

A estrutura das tabelas

A busca precisa cruzar três tabelas:

empresas          → razao_social        (55M linhas)
estabelecimentos  → nome_fantasia       (70M linhas)
socios            → nome_socio          (25M linhas)

Detalhe importante: O resultado final que o usuário vê é uma lista de empresas (identificadas por cnpj_basico). Então mesmo que o match venha da tabela de sócios, precisamos retornar a empresa correspondente.

Os indexes GIN

O coração da performance são os indexes GIN com gin_trgm_ops:

CREATE INDEX CONCURRENTLY ix_empresas_razao_trgm 
ON empresas USING gin (razao_social gin_trgm_ops);

CREATE INDEX CONCURRENTLY ix_estab_fantasia_trgm 
ON estabelecimentos USING gin (nome_fantasia gin_trgm_ops);

CREATE INDEX CONCURRENTLY ix_socios_nome_trgm 
ON socios USING gin (nome_socio gin_trgm_ops);

Por que GIN e não GiST? GIN é mais rápido para leitura (que é 99% do nosso caso). GiST é melhor quando há muitas atualizações. Como nossa base atualiza uma vez por mês, GIN é a escolha certa.

Tamanho dos indexes: cada um ocupa ~3-5 GB. Total de ~12 GB só de indexes trigram. Esse é o tradeoff: espaço em disco por velocidade de busca.

A query: UNION ALL com ILIKE

A estratégia é buscar em cada tabela independentemente e depois unir os resultados:

CANDIDATE_LIMIT = 1000

def search_text(query: str, page: int, per_page: int):
    sql = """
        SELECT cnpj_basico FROM empresas 
        WHERE razao_social ILIKE :pattern
        LIMIT :lim

        UNION ALL

        SELECT cnpj_basico FROM estabelecimentos 
        WHERE nome_fantasia ILIKE :pattern
        LIMIT :lim

        UNION ALL

        SELECT cnpj_basico FROM socios 
        WHERE nome_socio ILIKE :pattern
        LIMIT :lim
    """

    candidates = db.execute(sql, {
        "pattern": f"%{query}%", 
        "lim": CANDIDATE_LIMIT
    })

    # Deduplica e pagina
    unique_cnpjs = list(dict.fromkeys(
        row.cnpj_basico for row in candidates
    ))
    page_cnpjs = unique_cnpjs[(page-1)*per_page : page*per_page]

    # Hydrata com dados de exibição
    return hydrate(page_cnpjs)

Por que ILIKE e não o operador `%` (similarity)?

O operador % do pg_trgm (WHERE razao_social % 'Magazine') calcula similaridade e retorna matches acima de um threshold. É ótimo para fuzzy search puro, mas:

ILIKE '%query%' + GIN trigram é surpreendentemente eficiente — o PostgreSQL usa o index GIN para filtrar candidatos pelo trigram e depois aplica o ILIKE como filtro final
ILIKE dá matches exatos de substring, que é o que o usuário geralmente quer
O threshold do % precisa de tuning fino e pode retornar resultados irrelevantes

O LIMIT por branch

Cada branch do UNION tem LIMIT 1000. Isso é crucial:

Sem limit, uma query genérica como "COMERCIO" varreria milhões de linhas
Com limit, mesmo o pior caso retorna em < 500ms
O cap total (MAX_TOTAL_VISIBLE = 1000) garante que nunca paginamos além de resultados relevantes

Buscas numéricas: atalho por CNPJ

Quando a query é numérica, pulamos o trigram e vamos direto:

def search_cnpj(digits: str):
    if len(digits) >= 8:
        # CNPJ basico completo — equality, instantâneo
        return db.query(Empresa).filter(
            Empresa.cnpj_basico == digits[:8]
        ).all()
    else:
        # CNPJ parcial — prefix match
        return db.query(Empresa).filter(
            Empresa.cnpj_basico.like(f"{digits}%")
        ).limit(MAX_TOTAL_VISIBLE).all()

B-tree index no cnpj_basico (PK) = resposta em < 1ms.

Cache com Redis

Buscas textuais são caras (3 scans trigram + dedup + hydrate). Cache é obrigatório:

import json, hashlib

def search_with_cache(query, page, per_page):
    cache_key = f"search:{query}:{page}:{per_page}"

    cached = redis.get(cache_key)
    if cached:
        return json.loads(cached)

    results = search_text(query, page, per_page)

    # Cache por 5 minutos — dados mudam mensalmente, 
    # mas consultas novas precisam aparecer rápido
    redis.setex(cache_key, 300, json.dumps(results))

    return results

Hit rate observado: ~40% em produção. Queries populares ("Petrobras", "Magazine Luiza") são servidas do cache.

Hydratação: evitando N+1

Depois de obter a lista de cnpj_basico da busca, precisamos carregar razão social, CNPJ completo e situação cadastral para exibição. Fazer uma query por resultado seria N+1:

def hydrate(cnpj_basicos: list[str]):
    # Uma única query com IN clause
    empresas = db.query(Empresa).filter(
        Empresa.cnpj_basico.in_(cnpj_basicos)
    ).all()
    empresa_map = {e.cnpj_basico: e for e in empresas}

    # Buscar estabelecimento matriz para cada empresa
    matrizes = db.query(Estabelecimento).filter(
        Estabelecimento.cnpj_basico.in_(cnpj_basicos),
        Estabelecimento.identificador_matriz_filial == "1"
    ).all()
    matriz_map = {m.cnpj_basico: m for m in matrizes}

    results = []
    for cnpj in cnpj_basicos:
        emp = empresa_map.get(cnpj)
        mat = matriz_map.get(cnpj)
        if emp and mat:
            results.append({
                "cnpj": format_cnpj(mat.cnpj_basico, mat.cnpj_ordem, mat.cnpj_dv),
                "razao_social": emp.razao_social,
                "situacao_cadastral": mat.situacao_cadastral,
            })

    return results

2 queries no total, independente do número de resultados.

Resultados de performance

Query	Tempo (sem cache)	Tempo (com cache)
"Petrobras"	~120ms	~2ms
"Comércio varejista"	~350ms	~2ms
"33000167000101"	~5ms	~2ms
"João Silva" (sócio)	~280ms	~2ms

Tudo isso rodando em um PostgreSQL 16 com 8 GB de RAM. Sem Elasticsearch, sem Meilisearch, sem nada além do PostgreSQL.

Conclusão

pg_trgm é subestimado. Para aplicações onde:

O dataset cabe em PostgreSQL (até centenas de milhões de registros)
A busca é por substring/similaridade, não full-text semântico
Você quer manter a stack simples

...ele é a escolha certa. Adicione cache no Redis para queries repetidas e você tem um search engine que aguenta muito tráfego com hardware modesto.

Quer ver funcionando na prática? Teste uma busca no CNPJ Aberto e digite qualquer nome de empresa ou CNPJ e veja o autocomplete em tempo real.

Como importei 55 milhões de empresas para PostgreSQL em menos de 3 horas

Pedro Parker — Fri, 17 Apr 2026 00:29:21 +0000

Quando decidi construir o CNPJ Aberto, que é uma plataforma gratuita de consulta de empresas brasileiras, o primeiro desafio foi óbvio: como colocar 55 milhões de registros dentro do PostgreSQL de forma rápida e repetível (a base atualiza todo mês)?

Os dados vêm dos Dados Abertos da Receita Federal, distribuídos em dezenas de ZIPs com CSVs em latin-1, separados por ;, com campos inconsistentes (muito inconsistentes!!).

A ideia deste post rápido é mostrar as técnicas que transformaram uma importação de 12+ horas em menos de 3 horas.

O cenário

Tabela	Registros aproximados	Peso CSV
`empresas`	~55M	~4 GB
`estabelecimentos`	~70M	~15 GB
`socios`	~25M	~3 GB
`simples`	~35M	~2 GB
Tabelas auxiliares	~15K total	<1 MB

Total: ~25 GB de CSVs descompactados, distribuídos em ~40 arquivos ZIP. Nestes ponto é visível que os maiores problemas seriam empresas e estabelecimentos, não só para querys como para joins, que são muitos para criar um sistema legal e atrativo.

Tentativa 1: INSERT com ORM (12+ horas)

A abordagem ingênua com SQLAlchemy:

for row in csv_reader:
    empresa = Empresa(**parse_row(row))
    session.add(empresa)
    if i % 1000 == 0:
        session.commit()

Resultado: ~1.200 inserts/segundo. Para 55M de registros, isso dá ~12 horas só para a tabela empresas. Inaceitável. Eu acabei ingerindo tudo pois imaginava que > 12 horas de espera seirma melhores que > X horas de implementação e pesquisa para melhoria de queries. Talvez fossem, mas pelo desafio, fui além.

O ORM adiciona overhead em cada objeto: validação de tipo, tracking de estado, construção de SQL dinâmico.

Para melhorias, a IA acaba ajudando um pouco, não muito, ela alucinada bastante na ajuda e as vezes acaba piorando querys e criando index sem sentido, usei o Opus 4.6. Muito útil, mas o double check é necessário.

Tentativa 2: `executemany` com batches (4+ horas)

Removendo o ORM e usando psycopg2 direto:

BATCH_SIZE = 5000
batch = []
for row in csv_reader:
    batch.append(parse_row(row))
    if len(batch) >= BATCH_SIZE:
        cursor.executemany(INSERT_SQL, batch)
        conn.commit()
        batch.clear()

Resultado: ~4.000 inserts/segundo. Melhor, mas ainda 4+ horas.

O problema: executemany ainda gera um INSERT por linha. O PostgreSQL parseia e planeja cada statement individualmente.

Solução final: COPY + Temp Tables (< 3 horas)

O COPY é o mecanismo de bulk load nativo do PostgreSQL. Ele bypassa o parser SQL, o planner e o executor — escrevendo direto no heap da tabela. É 10-50x mais rápido que INSERT.

Passo 1: Otimizar a sessão

cursor.execute("SET synchronous_commit = off")
cursor.execute("SET work_mem = '256MB'")

synchronous_commit = off permite que o PostgreSQL confirme transações sem esperar o flush do WAL para disco. Seguro para data loads (se o servidor crashar, você reimporta). Como nesse caso os dados não são tão importantes, é possível reimportar sem precisar fazer check de nada.

Passo 2: Dropar indexes antes, recriar depois

Indexes tornam cada INSERT mais caro porque o B-tree/GIN precisa ser atualizado. Para bulk load, é mais eficiente dropar tudo, importar, e recriar:

def drop_indexes(cursor):
    cursor.execute("""
        SELECT indexname, tablename FROM pg_indexes 
        WHERE schemaname = 'public' 
        AND indexname NOT LIKE '%_pkey'
    """)
    for idx, table in cursor.fetchall():
        cursor.execute(f"DROP INDEX IF EXISTS {idx}")

def create_indexes(cursor):
    # Recriar com CONCURRENTLY para não bloquear reads
    cursor.execute("""
        CREATE INDEX CONCURRENTLY IF NOT EXISTS 
        ix_empresas_razao_trgm ON empresas 
        USING gin (razao_social gin_trgm_ops)
    """)
    # ... mais indexes

CREATE INDEX CONCURRENTLY é crucial — permite que o site continue respondendo enquanto os indexes são construídos.

Passo 3: COPY via temp table + UPSERT

Para tabelas que precisam de upsert (atualização mensal), usamos temp tables:

BATCH_SIZE = 200_000

def import_batch(cursor, table, columns, rows):
    # 1. Criar temp table com mesma estrutura
    cursor.execute(f"CREATE TEMP TABLE tmp_{table} (LIKE {table} INCLUDING DEFAULTS)")

    # 2. COPY os dados para a temp table
    csv_buffer = io.StringIO()
    writer = csv.writer(csv_buffer)
    for row in rows:
        writer.writerow(row)
    csv_buffer.seek(0)

    cursor.copy_expert(
        f"COPY tmp_{table} ({','.join(columns)}) FROM STDIN WITH (FORMAT csv, NULL '')",
        csv_buffer
    )

    # 3. UPSERT da temp table para a tabela real
    cols = ', '.join(columns)
    update_cols = ', '.join(f"{c} = EXCLUDED.{c}" for c in columns if c != pk)

    cursor.execute(f"""
        INSERT INTO {table} ({cols})
        SELECT {cols} FROM tmp_{table}
        ON CONFLICT ({pk}) DO UPDATE SET {update_cols}
    """)

    # 4. Limpar
    cursor.execute(f"DROP TABLE tmp_{table}")

Por que temp table? Porque o COPY não suporta ON CONFLICT diretamente. A temp table recebe o bulk load ultrarrápido, e depois um único INSERT ... ON CONFLICT faz o merge.

Passo 4: Paralelismo no download e importação

Os ZIPs da Receita Federal são independentes, então podemos baixar e importar em paralelo:

with ThreadPoolExecutor(max_workers=4) as pool:
    futures = []
    for zip_url in zip_urls:
        futures.append(pool.submit(download_and_import, zip_url))

    for future in as_completed(futures):
        future.result()  # propaga exceções

4 workers = 4 ZIPs sendo importados simultaneamente. Com SSDs, o PostgreSQL lida bem com escritas paralelas em tabelas diferentes.

Resultados

Abordagem	Velocidade	Tempo total
ORM (SQLAlchemy)	~1.200/s	12+ horas
`executemany` batches	~4.000/s	4+ horas
`COPY` + temp tables + parallelismo	~80.000/s	< 3 horas

67x mais rápido que a abordagem inicial.

Lições aprendidas

use COPY para bulk load no PostgreSQL. Não existe nada mais rápido sem ir para pg_bulkload ou extensões externas.
Dropar indexes antes de um bulk load e recriar depois é quase sempre mais rápido que manter os indexes durante a carga.
Temp tables são o bridge entre COPY (que não suporta upsert) e a necessidade de ON CONFLICT.
synchronous_commit = off é uma otimização segura para data loads — o pior que acontece é perder dados que você pode reimportar.
Batch size importa: 200K linhas por batch é o sweet spot. Muito menos = overhead de transação. Muito mais = uso excessivo de memória.

O resultado

Esse pipeline roda todo mês no CNPJ Aberto para atualizar a base com os dados mais recentes da Receita Federal. Qualquer pessoa pode consultar gratuitamente dados de qualquer empresa brasileira, razão social, sócios, endereço, CNAE, situação cadastral e muito mais.

Se você trabalha com dados públicos brasileiros, dá uma olhada: cnpjaberto.com.br

Obrigado!

Forem: Pedro Parker

Give your AI agent access to 67M Brazilian companies, free MCP server

What it looks like in practice

Setup in 60 seconds

Option A — Node (zero-install via npx)

Option B — Python

What tools the LLM gets

Use it as a plain SDK too

Typed errors, because life

Why bother building this

What's next

Links

Construindo um explorador de rede societária com grafos em Python

O modelo de dados

Modelagem do grafo

O traversal recursivo

traverse_company: o coração do algoritmo

expand_person: encontrando empresas conectadas

Proteções contra explosão combinatória

Detecção de Red Flags

Cache: essencial para grafos

Frontend: visualização do grafo

Casos de uso reais

Conclusão

Rate limiting com Redis para um SaaS freemium: a arquitetura por trás de 3 tiers

Por que Redis?

A chave do rate limit

O check: INCR + EXPIRE atômico

Edge case: e se o EXPIRE falhar?

Identificando o usuário: IP vs JWT

Determinando o plano: cache de 5 minutos

O Middleware FastAPI

Headers X-RateLimit

Rate limit separado para autenticação

O que não contar

Redirecionamento no frontend

Monitoramento

Conclusão

Next.js 15 com 55 milhões de páginas dinâmicas: SSR, SEO e performance

O problema: 55M páginas únicas

Arquitetura: Server Component puro

OpenGraph Image dinâmica

Code Splitting agressivo

Canonical URLs via Middleware

JSON-LD para Rich Results

API Proxy via Rewrites

Resultados de performance

Conclusão

Busca fuzzy em 55 milhões de registros: como pg_trgm salvou meu projeto

Por que não Elasticsearch?

O que é pg_trgm?

A estrutura das tabelas

Os indexes GIN

A query: UNION ALL com ILIKE

Por que ILIKE e não o operador % (similarity)?

O LIMIT por branch

Buscas numéricas: atalho por CNPJ

Cache com Redis

Hydratação: evitando N+1

Resultados de performance

Conclusão

Como importei 55 milhões de empresas para PostgreSQL em menos de 3 horas

O cenário

Tentativa 1: INSERT com ORM (12+ horas)

Tentativa 2: executemany com batches (4+ horas)

Solução final: COPY + Temp Tables (< 3 horas)

Passo 1: Otimizar a sessão

Passo 2: Dropar indexes antes, recriar depois

Passo 3: COPY via temp table + UPSERT

Passo 4: Paralelismo no download e importação

Resultados

Lições aprendidas

O resultado

Option A — Node (zero-install via `npx`)

Por que ILIKE e não o operador `%` (similarity)?

Tentativa 2: `executemany` com batches (4+ horas)