Como armazenar respostas da API de CPF em PostgreSQL com Python

Para armazenar respostas da API de CPF em PostgreSQL com Python, você consulta o endpoint GET https://api.cpfhub.io/cpf/{CPF} com o header x-api-key, recebe o JSON com nome, data de nascimento e gênero, e persiste esses dados em uma tabela PostgreSQL usando psycopg2. Nunca guarde o CPF em texto plano — armazene o hash SHA-256 e registre base legal e finalidade conforme a LGPD exige. A latência da API é de aproximadamente 900ms, então implemente cache no banco para evitar consultas desnecessárias.

Schema do banco de dados

CREATE TABLE consultas_cpf (
    id SERIAL PRIMARY KEY,
    cpf_hash VARCHAR(64) NOT NULL,
    nome VARCHAR(255),
    data_nascimento DATE,
    genero CHAR(1),
    verificado BOOLEAN DEFAULT FALSE,
    base_legal VARCHAR(100) NOT NULL,
    finalidade VARCHAR(255) NOT NULL,
    consultado_em TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    expira_em TIMESTAMP WITH TIME ZONE,
    UNIQUE(cpf_hash, finalidade)
);

CREATE INDEX idx_cpf_hash ON consultas_cpf(cpf_hash);
CREATE INDEX idx_consultado_em ON consultas_cpf(consultado_em);

Pontos importantes:

• cpf_hash — Armazenamos o hash SHA-256 do CPF, nunca o CPF em texto plano.

• base_legal e finalidade — Exigidos pela LGPD para justificar o tratamento.

• expira_em — Para implementar política de retenção. Consulte a documentação do PostgreSQL sobre tipos de data e tempo para detalhes sobre TIMESTAMP WITH TIME ZONE.

Consultar a API e armazenar

import os
import hashlib
import requests
import psycopg2
from datetime import datetime, timezone, timedelta

CPFHUB_API_KEY = os.environ['CPFHUB_API_KEY']
DB_URL = os.environ['DATABASE_URL']

def consultar_e_armazenar(cpf: str, base_legal: str, finalidade: str) -> dict:
    # 1. Gerar hash do CPF
    cpf_hash = hashlib.sha256(cpf.encode()).hexdigest()

    # 2. Verificar se ja existe no banco (cache)
    conn = psycopg2.connect(DB_URL)
    cur = conn.cursor()

    cur.execute(
    "SELECT nome, data_nascimento, genero FROM consultas_cpf "
    "WHERE cpf_hash = %s AND expira_em > NOW()",
    (cpf_hash,)
    )
    existente = cur.fetchone()

    if existente:
    conn.close()
    return {
    'source': 'cache',
    'nome': existente[0],
    'nascimento': str(existente[1]),
    'genero': existente[2]
    }

    # 3. Consultar API
    url = f'https://api.cpfhub.io/cpf/{cpf}'
    headers = {
    'x-api-key': CPFHUB_API_KEY,
    'Accept': 'application/json'
    }

    response = requests.get(url, headers=headers, timeout=10)
    resultado = response.json()

    if not resultado.get('success'):
    conn.close()
    return {'error': 'CPF nao encontrado'}

    dados = resultado['data']

    # 4. Armazenar no banco
    expira_em = datetime.now(timezone.utc) + timedelta(days=90)

    cur.execute(
    "INSERT INTO consultas_cpf "
    "(cpf_hash, nome, data_nascimento, genero, verificado, base_legal, finalidade, expira_em) "
    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s) "
    "ON CONFLICT (cpf_hash, finalidade) DO UPDATE SET "
    "nome = EXCLUDED.nome, consultado_em = NOW(), expira_em = EXCLUDED.expira_em",
    (
    cpf_hash,
    dados['name'],
    f"{dados['year']}-{dados['month']:02d}-{dados['day']:02d}",
    dados['gender'],
    True,
    base_legal,
    finalidade,
    expira_em
    )
    )

    conn.commit()
    conn.close()

    return {
    'source': 'api',
    'nome': dados['name'],
    'nascimento': dados['birthDate'],
    'genero': dados['gender']
    }

Política de retenção (LGPD)

A LGPD exige que dados pessoais não sejam mantidos além do necessário. Implemente limpeza automática:

def limpar_expirados():
    conn = psycopg2.connect(DB_URL)
    cur = conn.cursor()

    cur.execute("DELETE FROM consultas_cpf WHERE expira_em < NOW()")
    removidos = cur.rowcount

    conn.commit()
    conn.close()

    print(f'{removidos} registros expirados removidos')

Execute diariamente via cron job ou scheduler.

Consultar registros para auditoria

def buscar_auditoria(cpf: str) -> list:
    cpf_hash = hashlib.sha256(cpf.encode()).hexdigest()

    conn = psycopg2.connect(DB_URL)
    cur = conn.cursor()

    cur.execute(
    "SELECT nome, verificado, base_legal, finalidade, consultado_em "
    "FROM consultas_cpf WHERE cpf_hash = %s ORDER BY consultado_em DESC",
    (cpf_hash,)
    )

    registros = []
    for row in cur.fetchall():
    registros.append({
    'nome': row[0],
    'verificado': row[1],
    'base_legal': row[2],
    'finalidade': row[3],
    'consultado_em': row[4].isoformat()
    })

    conn.close()
    return registros

Boas práticas

• Nunca armazene CPF em texto plano — Use hash SHA-256.

• Documente base legal e finalidade — Exigência da LGPD.

• Defina prazo de retenção — Use expira_em e limpeza automática.

• Use connection pooling — Em produção, use psycopg2.pool ou SQLAlchemy.

• Criptografe o banco — Habilite TDE ou criptografia a nível de coluna para dados sensíveis.

• Controle de acesso — Restrinja quem pode consultar a tabela.

Perguntas frequentes

Como o cache no PostgreSQL reduz o custo de consultas à CPFHub.io?

Cada consulta à API consome uma cota do plano. Ao armazenar o resultado no banco com um campo expira_em, você evita chamar a API para o mesmo CPF dentro do período de validade — servindo direto do banco. Como a latência da API é de ~900ms, o cache também melhora o tempo de resposta da sua aplicação.

O que acontece quando o limite de consultas do plano gratuito é atingido?

A API não bloqueia e não retorna erro. Consultas acima das 50 mensais do plano gratuito (ou das 1.000 do plano Pro) continuam sendo atendidas e geram cobrança de R$0,15 por consulta adicional. Por isso, implementar cache no PostgreSQL é especialmente útil para controlar gastos em aplicações com consultas repetidas.

Como registrar base legal e finalidade conforme a LGPD exige?

A Lei Geral de Proteção de Dados (Lei 13.709/2018) exige que o tratamento de dados pessoais tenha uma base legal documentada (consentimento, obrigação legal, legítimo interesse, entre outras) e uma finalidade específica e declarada. No schema proposto, as colunas base_legal e finalidade cumprem esse papel — registre valores como "consentimento" e "verificacao_cadastral" em cada inserção.

Posso usar este padrão com outro banco relacional, como MySQL ou SQLite?

A lógica de hash, cache e retenção funciona em qualquer banco relacional. A instrução ON CONFLICT ... DO UPDATE é específica do PostgreSQL (upsert). Para MySQL, substitua por INSERT ... ON DUPLICATE KEY UPDATE; para SQLite, use INSERT OR REPLACE. O driver Python muda (mysql-connector-python ou sqlite3 nativo), mas o padrão geral se mantém.

Conclusão

Armazenar respostas da API de CPF em PostgreSQL com Python permite cache eficiente, auditoria completa e conformidade LGPD. Usando hash de CPF, políticas de retenção e base legal documentada, sua aplicação estará preparada para auditorias e regulamentações. A CPFHub.io oferece plano gratuito com 50 consultas/mês para você começar.

Cadastre-se em cpfhub.io