Como exportar grandes volumes de dados via proxy sem recomeçar do zero após uma queda: guia passo a passo

Introdução: por que uma exportação longa quase sempre cai, e isso é normal

Se você já exportou via proxy alguns milhões de registros ou um arquivo de dezenas de gigabytes, conhece essa sensação. O script rodou seis horas, mostrava 83 por cento, e então caiu com erro de conexão. E tudo o que você tem é um arquivo incompleto e a certeza de que vai precisar começar de novo.

A primeira coisa a aceitar: uma exportação longa sempre cai. Não "às vezes", não "com rede ruim", mas sempre, se durar tempo suficiente. As causas são dezenas, e a maioria está fora do seu controle:

• O proxy troca o endereço IP externo. Em proxies móveis Proxeon isso é comportamento padrão: rotação por temporizador ou sob demanda. No momento da troca de IP, a conexão TCP aberta é encerrada, e o servidor de origem já enxerga outro cliente.
• O servidor de origem fecha a conexão pelo próprio timeout, reinicia, publica uma atualização ou simplesmente responde com erro 5xx.
• Seu próprio processo reinicia: atualização do sistema, disco cheio, erro no código em um registro atípico, um Ctrl+C acidental.
• O token de autorização expira, a sessão cai, o cursor de paginação fica obsoleto.
• O notebook entra em suspensão, o Wi-Fi troca de ponto, o provedor muda a rota.

Lutar contra cada uma dessas causas separadamente é inútil. A abordagem correta é outra: projetar a exportação de modo que uma queda em qualquer momento custe não seis horas, mas uma página ou um pedaço de arquivo. É exatamente a isso que este guia se dedica.

O que você vai ter no final

Depois de seguir as instruções, você terá:

• Uma função funcional de retomada de download a partir do meio via protocolo HTTP com o cabeçalho Range, com verificação de que o servidor suporta isso.
• Um esquema claro de checkpoints para exportações paginadas: o que exatamente salvar e onde guardar o estado, para que ele não se perca junto com o processo.
• Gravação idempotente de resultados, na qual recarregar a mesma página não cria duplicatas.
• Deduplicação que não consome toda a memória RAM com milhões de linhas.
• Carregador paralelo com fila de tarefas, repetição de tarefas falhas e limite de simultaneidade.
• Um esqueleto pronto de carregador robusto em Python, que você adapta para sua origem em uma hora.

Para quem é este guia

Para desenvolvedores e analistas que já sabem fazer requisições HTTP em Python e pelo menos uma vez perderam os resultados de uma exportação longa. Nível intermediário: explicamos o básico, mas não ensinamos a programar do zero. Leitores avançados encontrarão seções sobre armazenar hashes fora da memória e sobre escrita paralela segura em SQLite.

O que você precisa saber antes

• Python no nível de funções, loops, dicionários e tratamento de exceções.
• Fundamentos de HTTP: o que é método, cabeçalho, código de resposta, corpo.
• Noção geral de como configurar um proxy na biblioteca requests.

Um aviso à parte: não vamos abordar aqui códigos de resposta e estratégias de retry com espera exponencial. Existe um artigo dedicado ao erro 429 e retries. Neste guia o foco é outro: o estado da exportação e sua retomada. Retries respondem à pergunta "quando repetir a requisição", e nós respondemos à pergunta "de que ponto continuar o trabalho depois que as retentativas se esgotaram e o processo morreu".

Quanto tempo vai levar

Ler e rodar os exemplos em uma origem de teste: duas a três horas. Adaptar o esqueleto para sua API real ou servidor de arquivos: mais uma ou duas horas, dependendo de quão não padronizada é a paginação. No total, um dia de trabalho com folga.

Preparação inicial e conceitos básicos

Ferramentas e acessos

1. Instale o Python versão 3.11 ou mais recente. Em 2026, as linhas 3.12 e 3.13 são atuais, e todos os exemplos foram testados nelas. Para conferir a versão: abra o terminal e digite python --version. Se aparecer 3.11 ou superior, tudo certo.
2. Instale a biblioteca requests: pip install requests. A versão 2.32 ou mais recente é suficiente. O módulo sqlite3 vem na biblioteca padrão do Python, não precisa instalar nada à parte.
3. Obtenha acesso ao proxy. Abra o painel do Proxeon, escolha o canal desejado e copie quatro valores: host, porta, login e senha. Normalmente eles vêm reunidos em uma linha no formato http://USER:PASS@HOST:PORT. Essa linha você usará em todo o restante.
4. Coloque a linha do proxy em uma variável de ambiente, não no código. No Linux e macOS: export PROXY_URL=http://USER:PASS@HOST:PORT. No Windows PowerShell: $env:PROXY_URL='http://USER:PASS@HOST:PORT'. Assim você não vai commitar a senha no repositório por acidente.
5. Verifique se o proxy responde. Execute no terminal: curl -x $PROXY_URL -I https://api.example.com/, substituindo pelo endereço da sua origem. Você deve ver uma linha com o código de resposta, por exemplo HTTP/2 200. Se aparecer erro de autorização do proxy 407, confira o login e a senha.

Requisitos de sistema

Qualquer máquina com 2 GB de memória RAM livre e disco em que caibam o resultado da exportação mais 20 por cento de folga para índices do SQLite. Se você pretende carregar milhões de registros, o disco importa mais que a memória: toda a abordagem é construída sobre o princípio de que o estado vive no disco, não em variáveis do processo.

Backups

O arquivo de estado que você vai criar abaixo (nos exemplos, export.sqlite) se tornará o artefato mais valioso de todo o trabalho. Crie o hábito de copiá-lo antes de qualquer experimento com o código: cp export.sqlite export.sqlite.bak. Uma vez isso vai te poupar um dia inteiro de exportação.

Atenção: nunca edite o arquivo SQLite manualmente enquanto o carregador estiver rodando. Até a leitura por outro programa em modo inadequado pode bloquear a escrita e derrubar o processo. Se precisar ver o estado, pare o carregador ou use o modo WAL, que explicaremos na seção sobre paralelismo.

Termos-chave em linguagem simples

• Checkpoint — uma marca salva em disco indicando "até aqui tudo foi exportado e gravado". Após uma queda, o carregador lê o checkpoint e continua a partir dele.
• Cursor — uma string opaca que a API devolve junto com a página e que precisa ser enviada para obter a próxima página. Você não monta nem interpreta o cursor; apenas o repassa.
• Exportação paginada por deslocamento — quando você pede "página 37 com 500 registros". Esquema simples, mas ao inserir novos registros na origem as páginas se deslocam e surgem duplicatas ou lacunas.
• Exportação paginada por chave (keyset) — quando você pede "tudo com identificador maior que 184203, ordenado por identificador". Esquema mais robusto para retomada, se a origem o suportar.
• Idempotência — propriedade de uma operação em que executá-la novamente produz o mesmo resultado que executá-la uma vez. Você gravou a página duas vezes, mas no banco ela está só uma vez.
• Chave de deduplicação — valor pelo qual dois registros são considerados o mesmo. Idealmente, é o identificador da origem; se não houver, a chave é calculada como hash de campos estáveis.
• Cabeçalho Range — forma de pedir ao servidor HTTP que entregue não o arquivo inteiro, mas uma parte dele, por exemplo, os bytes de 1048576 até o fim.
• Semântica at-least-once — garantia de que cada registro será obtido pelo menos uma vez. Repetições são possíveis, faltas não. É exatamente isso que você terá depois deste guia, e as duplicatas serão removidas pela deduplicação.

Princípio central

Os sete passos abaixo se resumem a uma ideia: cada unidade de trabalho precisa ser atômica e repetível. Unidade de trabalho é ou um pedaço de arquivo, ou uma página da API, ou uma tarefa da fila. Atômica significa que o resultado e a marca de sua conclusão são salvos juntos. Repetível significa que, se a unidade for executada duas vezes, nada se quebra. Quando as duas propriedades estão atendidas, uma queda em qualquer ponto se torna inofensiva.

Passo 1: Retomada de download via HTTP com o cabeçalho Range

Objetivo da etapa: aprender a baixar um arquivo grande via proxy de modo que, após uma queda, o download continue do byte onde parou, e não do zero.

Como funciona

O protocolo HTTP permite que o cliente solicite parte de um recurso. Para isso, adiciona-se à requisição o cabeçalho Range: bytes=INÍCIO-, onde INÍCIO é o deslocamento em bytes. Se o servidor suporta requisições parciais, ele responde com o código 206 Partial Content e o cabeçalho Content-Range: bytes INÍCIO-FIM/TOTAL. Se não suporta, ele ignora o Range e entrega o arquivo inteiro com código 200. Sua tarefa é distinguir esses casos.

Saber antecipadamente se o servidor suporta retomada ajuda uma requisição HEAD: ela retorna apenas os cabeçalhos, sem corpo. Observe o Accept-Ranges: bytes. O valor none ou a ausência do cabeçalho geralmente indicam que não há retomada, embora alguns servidores processem Range corretamente de qualquer forma, então a verificação final é pelo código de resposta.

Instruções passo a passo

1. Faça uma requisição HEAD via proxy e guarde os cabeçalhos Accept-Ranges, Content-Length, ETag e Last-Modified. O ETag será útil para saber se o arquivo no servidor mudou entre suas tentativas.
2. Veja quantos bytes já existem no arquivo local. Se o arquivo não existe, considere zero.
3. Se o tamanho local já é igual ao Content-Length, o arquivo está completo, não é preciso fazer nada.
4. Se o tamanho local é maior que zero e o servidor anuncia suporte a ranges, adicione à requisição o cabeçalho Range com o tamanho atual. Adicione também If-Range com o ETag guardado: assim o servidor só devolverá a resposta parcial se o arquivo não mudou; caso contrário, devolverá o arquivo inteiro com código 200.
5. Envie obrigatoriamente Accept-Encoding: identity. Sem isso, o servidor pode aplicar compressão em tempo real, e os deslocamentos em bytes deixarão de coincidir com o seu arquivo.
6. Abra o arquivo local no modo de anexação ab se recebeu 206, ou no modo de sobrescrita wb se recebeu 200.
7. Leia o corpo em fluxo, em pedaços de 256 KB, e grave em disco. Não carregue a resposta inteira na memória.
8. Ao terminar, compare o tamanho final com o Content-Length. Se não coincidir, a conexão caiu silenciosamente e será preciso mais uma rodada.

Código funcional

import os
import requests

PROXY_URL = os.environ['PROXY_URL'] # string do painel Proxeon
PROXIES = {'http': PROXY_URL, 'https': PROXY_URL}


def probe(url):
r = requests.head(url, proxies=PROXIES, allow_redirects=True, timeout=30,
 headers={'Accept-Encoding': 'identity'})
r.raise_for_status()
return {
'ranges': r.headers.get('Accept-Ranges', 'none').lower(),
'length': int(r.headers.get('Content-Length', 0) or 0),
'etag': r.headers.get('ETag'),
}


def download_resumable(url, path):
meta = probe(url)
have = os.path.getsize(path) if os.path.exists(path) else 0
if meta['length'] and have >= meta['length']:
print('arquivo já está completo:', have, 'bytes')
return True
headers = {'Accept-Encoding': 'identity'}
if have > 0 and meta['ranges'] == 'bytes':
headers['Range'] = 'bytes=%d-' % have
if meta['etag']:
headers['If-Range'] = meta['etag']
with requests.get(url, headers=headers, proxies=PROXIES, stream=True,
 timeout=(30, 120)) as r:
if r.status_code == 206:
expected = 'bytes %d-' % have
if not r.headers.get('Content-Range', '').startswith(expected):
raise IOError('servidor entregou o range errado: ' + r.headers.get('Content-Range', ''))
mode = 'ab'
elif r.status_code == 200:
print('servidor entrega o arquivo inteiro, começando do zero')
mode = 'wb'
have = 0
elif r.status_code == 416:
raise IOError('range solicitado está fora do arquivo, verifique o tamanho local')
else:
r.raise_for_status()
with open(path, mode) as f:
for chunk in r.iter_content(chunk_size=256 * 1024):
if chunk:
f.write(chunk)
have += len(chunk)
if meta['length'] and have != meta['length']:
print('queda: obtidos %d de %d' % (have, meta['length']))
return False
return True


def download_until_done(url, path, max_rounds=50):
for i in range(max_rounds):
try:
if download_resumable(url, path):
return
except (requests.ConnectionError, requests.Timeout, IOError) as e:
print('rodada %d interrompida: %s' % (i + 1, type(e).__name__))
# pausa antes da próxima rodada: estratégia descrita no artigo sobre 429 e retries
raise RuntimeError('não foi possível completar o download em %d rodadas' % max_rounds)


if __name__ == '__main__':
download_until_done('https://files.example.com/export-2026.csv.gz', 'export-2026.csv.gz')

Repare na função download_until_done: ela não contém lógica de espera entre tentativas. Isso é proposital. Insira ali sua estratégia de pausas do artigo sobre retries; aqui só importa o ciclo "verificou o tamanho, retomou, verificou de novo".

Dica: se o arquivo é distribuído como arquivo compactado, não o descompacte em tempo real durante a retomada. Primeiro obtenha o arquivo completo, verifique o tamanho e, se o servidor fornecer soma de verificação, confira-a. Só então descompacte. Um gzip baixado parcialmente parece corrompido, e você perderá tempo procurando um erro inexistente.

Resultado esperado

Verificação: rode o script em um arquivo de pelo menos 200 MB e, após dez segundos, interrompa com Ctrl+C. Veja o tamanho do arquivo local, por exemplo 41 943 040 bytes. Rode o script de novo. No console não deve aparecer a linha "começando do zero", e o tamanho do arquivo deve continuar crescendo, sem voltar ao início. Ao final, o tamanho total deve coincidir exatamente com o Content-Length da requisição HEAD.

Possíveis problemas

• O servidor sempre devolve 200 em vez de 206. Isso significa que a retomada não é suportada. A única saída para essa origem é baixar o arquivo inteiro de uma vez com timeout maior, ou buscar na origem um formato alternativo de exportação em partes, por exemplo, divisão por datas.
• Não há cabeçalho Content-Length. O servidor entrega o arquivo em modo chunked sem declarar o tamanho. Não é possível verificar a completude pelo tamanho, nem retomar: Range exige deslocamentos conhecidos. Combine com a origem ou use uma soma de verificação, se for publicada.
• Resposta 416 na primeira rodada. O arquivo local é maior que o arquivo no servidor. O arquivo no servidor mudou e ficou menor. Apague o arquivo local e comece de novo.
• O tamanho coincide, mas o arquivo está corrompido. Muito provavelmente houve uma queda no meio com código 200 e sobrescrita do zero, seguida de anexação. Recrie o arquivo. Para que isso não se repita, armazene o ETag em um arquivo separado ao lado e compare antes de cada rodada.

Passo 2: Checkpoints para exportações paginadas

Objetivo da etapa: salvar o estado da exportação de modo que, após qualquer queda, o processo continue a partir da última página gravada com sucesso.

O que salvar

O checkpoint mínimo depende do tipo de paginação da origem. Vamos ver três casos.

1. Paginação por cursor. A API devolve, junto com os dados, um campo como next_cursor. Salve exatamente ele. É o caso mais simples: o cursor já contém tudo o que o servidor precisa para continuar.
2. Paginação por número de página ou deslocamento. Salve o número da última página totalmente gravada e o tamanho da página. Lembre-se de que, se registros forem inseridos na origem durante a exportação, os deslocamentos mudam, então a deduplicação do passo 4 é obrigatória.
3. Paginação por chave. Salve o identificador do último registro gravado. Ao retomar, solicite tudo que for maior que esse identificador. O esquema não teme inserções nem pausas longas.

Independentemente do tipo de paginação, vale acrescentar ao checkpoint campos auxiliares: o identificador do último registro (mesmo no esquema por cursor, é uma âncora de reserva caso o cursor expire), contadores de páginas e linhas para acompanhar o progresso, hora de início da exportação e hora da última atualização.

Onde guardar o estado

Há duas opções viáveis, e ambas são melhores que variáveis em memória.

Opção A: arquivo JSON com substituição atômica

Serve quando os resultados são gravados em arquivos separados, e não em banco. A principal armadilha: se você escrever o estado direto no arquivo de destino e o processo cair no meio da escrita, terá um JSON truncado que não será lido. A solução é escrever em um arquivo temporário ao lado e renomeá-lo por cima do principal. A operação de renomeação dentro do mesmo sistema de arquivos é atômica.

import json
import os
import tempfile


def save_state(path, state):
directory = os.path.dirname(os.path.abspath(path))
fd, tmp = tempfile.mkstemp(dir=directory, prefix='.state-')
with os.fdopen(fd, 'w', encoding='utf-8') as f:
json.dump(state, f, ensure_ascii=False)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)


def load_state(path, default):
if not os.path.exists(path):
return dict(default)
with open(path, encoding='utf-8') as f:
return json.load(f)

Opção B: tabela no SQLite junto aos dados

É a opção preferida quando você armazena os registros em banco. O checkpoint é atualizado na mesma transação em que as linhas da página são inseridas. Ou grava-se tanto os dados quanto a marca, ou nada. Uma discrepância do tipo "dados existem, marca não" não acontece por princípio.

import sqlite3

con = sqlite3.connect('export.sqlite')
con.executescript('''
CREATE TABLE IF NOT EXISTS records(
id TEXT PRIMARY KEY,
payload TEXT NOT NULL,
fetched_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS checkpoint(
job TEXT PRIMARY KEY,
cursor TEXT,
last_id TEXT,
pages INTEGER NOT NULL DEFAULT 0,
updated_at TEXT
);
''')


def commit_page(job, rows, next_cursor, pages, ts):
with con: # uma transação por página
con.executemany(
'INSERT OR IGNORE INTO records(id, payload, fetched_at) VALUES (?, ?, ?)',
[(str(r['id']), json.dumps(r, ensure_ascii=False), ts) for r in rows])
con.execute(
'INSERT INTO checkpoint(job, cursor, last_id, pages, updated_at) VALUES (?, ?, ?, ?, ?) '
'ON CONFLICT(job) DO UPDATE SET cursor=excluded.cursor, last_id=excluded.last_id, '
'pages=excluded.pages, updated_at=excluded.updated_at',
(job, next_cursor, str(rows[-1]['id']), pages, ts))

Ordem das operações

Guarde a regra: primeiro os dados, depois o checkpoint, e de preferência na mesma transação. Se a transação for inviável (por exemplo, dados gravados em arquivos), a ordem é exatamente essa: gravou o arquivo da página, sincronizou em disco, depois atualizou o estado. Se houver queda entre essas duas ações, você obterá o recarregamento de uma página, o que é seguro graças ao passo 3. A ordem inversa causará a perda de uma página, e isso já é perda de dados.

Dica: guarde no checkpoint não o cursor atual, mas o cursor da próxima página, devolvido pelo servidor. Assim, ao retomar, você já pede o que ainda não tem, sem requisição extra da página já obtida.

Resultado esperado

Verificação: comece a exportação, espere dez páginas e encerre o processo à força. Abra o banco com sqlite3 export.sqlite e execute SELECT pages, last_id FROM checkpoint;. Você deve ver o número 10 e o identificador. Em seguida execute SELECT count(*) FROM records; e confirme que o número de registros é igual a dez vezes o tamanho da página. Rode o carregador de novo: a primeira mensagem no console deve ser algo como "início: páginas 10".

Possíveis problemas

• Erro "database is locked". Outro processo está com a conexão aberta. Feche todas as janelas do sqlite3 e outras ferramentas que abriram o arquivo. Para trabalho multithread, ative o modo WAL, veja o passo 5.
• O cursor foi salvo, mas os dados não. Você atualizou o checkpoint fora da transação que grava os dados. Volte ao código acima e confirme que as duas operações estão dentro de um mesmo bloco with con:.
• O JSON de estado ficou vazio ou corrompido. Você escreveu direto no arquivo, sem arquivo temporário e substituição. Use a função save_state na íntegra.

Passo 3: Idempotência na gravação dos resultados

Objetivo da etapa: fazer com que o reprocessamento de qualquer página não gere duplicatas nem quebre os dados.

Por que a repetição é inevitável

Depois do passo 2, você já viu o cenário em que uma página é gravada duas vezes: o processo caiu após inserir os dados, mas antes de atualizar o checkpoint. Além disso, repetições vêm da paginação por deslocamento quando a origem muda, de workers paralelos que receberam a mesma tarefa após um restart, e simplesmente de um reinício manual "por precaução". Combater repetições no lado da requisição é inútil. O correto é tornar a própria gravação à prova de repetição.

Escolha da chave de deduplicação

1. Há identificador na origem. Use-o. É o campo id, uuid, order_number ou similar que a origem garante ser único. Se você exporta de várias origens para uma única tabela, use chave composta: nome da origem mais identificador.
2. Não há identificador, mas há um conjunto de campos que, juntos, determinam o registro. Por exemplo, para uma linha de tabela de preços, é o SKU mais o depósito mais a data. Monte a chave com esses campos, normalizando-os: converta strings para o mesmo caso, remova espaços nas pontas, coloque datas em formato único.
3. Não há nada estável. Então a chave passa a ser o hash do registro inteiro após canonização. Esse caso é detalhado no passo 4. Lembre-se de que, se a origem altera o registro (atualiza o preço), o hash muda e você obterá as duas versões. Às vezes é exatamente o que se quer, às vezes não.

Atenção: não use como chave o número sequencial da linha na resposta ou o número da página. Esses valores mudam a qualquer alteração na origem, e a deduplicação vira uma geradora de duplicatas.

Inserção idempotente no banco

No SQLite e na maioria dos bancos relacionais existe uma construção que ou ignora o conflito de chave primária, ou atualiza a linha existente. A primeira variante, INSERT OR IGNORE, você já viu no passo 2. Ela serve quando os registros são imutáveis. A segunda variante é necessária se a origem pode atualizar registros e você quer a versão mais recente:

def upsert_rows(con, rows, ts):
con.executemany(
'INSERT INTO records(id, payload, fetched_at) VALUES (?, ?, ?) '
'ON CONFLICT(id) DO UPDATE SET payload=excluded.payload, fetched_at=excluded.fetched_at',
[(str(r['id']), json.dumps(r, ensure_ascii=False), ts) for r in rows])

Gravação idempotente em arquivos

Se o resultado deve ficar em arquivos, e não no banco, aplique o mesmo princípio: uma página corresponde a um arquivo com nome determinístico. O nome depende dos parâmetros da página, não do horário nem de um contador. Antes de baixar, verifique se o arquivo final existe; se existir, pule a página. Escreva em um nome temporário e renomeie ao concluir, como na função save_state.

def page_path(base_dir, job, cursor_or_page):
safe = str(cursor_or_page).replace('/', '_').replace(':', '_')[:120]
return os.path.join(base_dir, job, 'page-%s.jsonl' % safe)


def write_page_idempotent(path, rows):
if os.path.exists(path):
return False # a página já existe, não precisa regravar
os.makedirs(os.path.dirname(path), exist_ok=True)
tmp = path + '.part'
with open(tmp, 'w', encoding='utf-8') as f:
for r in rows:
f.write(json.dumps(r, ensure_ascii=False))
f.write(chr(10))
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)
return True

Arquivos com a extensão .part remanescentes de uma queda podem ser apagados com segurança na inicialização: por definição, são incompletos.

Dica: em exportação por deslocamento, não confie apenas em "o arquivo existe, então a página está pronta". Verifique também se o número de linhas no arquivo equivale ao tamanho da página (exceto a última). Um arquivo vazio ou curto com nome existente é melhor rebaixar e baixar de novo.

Resultado esperado

Verificação: chame a função de gravação da mesma página três vezes seguidas. Em seguida execute SELECT count(*) FROM records;. O número deve ser o tamanho de uma página, e não o triplo. Na variante em arquivos, deve haver exatamente um arquivo de página no diretório e nenhum arquivo .part.

Passo 4: Deduplicação de resultados sem estourar a memória

Objetivo da etapa: filtrar registros repetidos em um fluxo de milhões de linhas, sem manter todas as chaves na memória RAM.

Hash do registro

Quando o registro não tem identificador, a chave passa a ser o hash de seu conteúdo. Para que registros iguais gerem o mesmo hash, o conteúdo precisa ser canonizado: ordenar as chaves do dicionário, remover espaços extras, fixar separadores. Caso contrário, o mesmo registro, recebido com outra ordem de campos, gera um hash diferente.

import hashlib
import json


def record_key(rec, fields=None):
src = rec if fields is None else {k: rec.get(k) for k in fields}
canon = json.dumps(src, sort_keys=True, ensure_ascii=False, separators=(',', ':'))
return hashlib.blake2b(canon.encode('utf-8'), digest_size=16).digest()

A função devolve 16 bytes. Isso é suficiente: a probabilidade de colisão aleatória para centenas de milhões de registros é desprezível. O parâmetro fields permite calcular o hash apenas por campos estáveis, excluindo, por exemplo, a hora da última atualização, que muda a cada requisição.

Por que um conjunto em memória não funciona em milhões

A primeira ideia que surge é criar seen = set() e ir acumulando chaves. Vamos fazer as contas. Um objeto bytes de 16 bytes ocupa em Python cerca de 49 bytes mais os próprios dados, ou seja, uns 65 bytes. Um slot no set, considerando o fator de carga, adiciona mais uns 30 bytes. Chegamos a cerca de 95 bytes por chave. Em 10 milhões de registros são cerca de 950 MB; em 50 milhões, quase 5 GB. E o principal: após reiniciar o processo, o set fica vazio e toda a deduplicação recomeça da estaca zero.

Três formas de não inflar a memória

1. Guardar as chaves no próprio banco. O caminho mais simples e confiável. Se a chave é a chave primária da tabela records, a deduplicação já foi feita pela construção INSERT OR IGNORE do passo 3. O índice vive no disco, sobrevive a reinícios, e o SQLite faz cache das páginas quentes do índice. Para 10 milhões de chaves de 16 bytes, o índice ocupa aproximadamente 400-500 MB em disco, mas não em memória.
2. Tabela separada de chaves vistas sem rowid. É necessária se os dados em si você grava não no SQLite, mas, por exemplo, em arquivos. Nesse caso, o SQLite é usado apenas como um conjunto compacto em disco.
3. Conjunto comprimido em memória como pré-filtro. Variante avançada: truncar o hash para 8 bytes e armazená-lo como número inteiro em um array ordenado, ou usar um filtro de Bloom. A memória cai várias vezes, mas surge a probabilidade de falso positivo. Por isso, esse pré-filtro é usado só para descartar rapidamente registros comprovadamente novos, e a checagem final é sempre feita no banco.

Implementação do conjunto em disco

class DiskSeen:
def __init__(self, con):
self.con = con
con.execute('CREATE TABLE IF NOT EXISTS seen(key BLOB PRIMARY KEY) WITHOUT ROWID')

def filter_new(self, rows):
keyed = [(record_key(r), r) for r in rows]
keys = [k for k, _ in keyed]
placeholders = ','.join('?' * len(keys))
known = {row[0] for row in self.con.execute(
'SELECT key FROM seen WHERE key IN (%s)' % placeholders, keys)}
fresh = [(k, r) for k, r in keyed if k not in known]
# deduplicação dentro da própria página
unique = {}
for k, r in fresh:
unique.setdefault(k, r)
return unique

def remember(self, keys):
self.con.executemany('INSERT OR IGNORE INTO seen(key) VALUES (?)', [(k,) for k in keys])

Chame filter_new antes de gravar a página, e remember dentro da mesma transação que grava os dados e o checkpoint. Assim, após uma queda, o conjunto de chaves vistas, os dados e a marca de progresso estarão sempre consistentes entre si.

Dica: a consulta com IN para 500 valores roda pelo índice em milissegundos. Não verifique as chaves uma a uma em loop: isso é dezenas de vezes mais lento por causa do custo de cada consulta.

Resultado esperado

Verificação: monte uma página de teste com 500 registros, sendo 100 repetidos duas vezes dentro da própria página e outros 100 já presentes na tabela seen. A função filter_new deve devolver exatamente 300 registros. Após reiniciar o processo, os mesmos 500 registros devem gerar zero novos.

Possíveis problemas

• As duplicatas ainda passam. Confira a canonização: provavelmente há um campo com o horário da requisição ou uma ordem aleatória de elementos na lista. Exclua esse campo com o parâmetro fields ou ordene listas aninhadas antes do hash.
• Inserção lenta após alguns milhões de linhas. O índice deixou de caber no cache. Aumente o cache do SQLite com PRAGMA cache_size=-200000 (isso equivale a 200 MB) e garanta que as inserções vão em lotes em uma única transação por página, e não linha a linha.

Passo 5: Paralelismo sem perdas

Objetivo da etapa: acelerar a exportação com vários workers simultâneos via proxy, de modo que a queda de qualquer um deles não perca tarefas nem quebre o banco.

Quando paralelizar é possível e quando não é

A paginação por cursor é, por natureza, sequencial: o próximo cursor só é conhecido após receber a página anterior. Não é possível paralelizá-la diretamente. Mas quase sempre é possível dividir a exportação em shards independentes: por dias, por categorias, por regiões, pelos primeiros caracteres do identificador. Cada shard é exportado de forma sequencial com o próprio checkpoint, e os shards correm em paralelo. A paginação por deslocamento e por chave, com limites conhecidos, é paralelizável diretamente: tarefas do tipo "páginas de 1 a 100" ou "identificadores de 0 a 100000".

Fila de tarefas em disco

Uma fila em memória morre junto com o processo. Por isso, as tarefas ficam em uma tabela com status:

• pending — aguardando execução;
• running — tomada por um worker;
• done — executada e gravada;
• failed — tentativas esgotadas, requer atenção humana.

Na inicialização, o carregador primeiro converte todas as tarefas de running de volta para pending: se estão nesse status, o processo anterior morreu no meio do trabalho. Em seguida, os workers pegam as pending.

Um único ponto de escrita

O SQLite permite vários leitores simultâneos, mas apenas um escritor por vez. O padrão mais simples e seguro: os workers só baixam e retornam dados, e toda a gravação no banco é feita pela thread principal. Sem bloqueios no código, sem "database is locked". Adicionalmente, ative o modo WAL, para que a leitura de estado por outro processo não atrapalhe a escrita.

Limite de simultaneidade

Limite o número de workers por duas coisas. A primeira são as capacidades do proxy: se no painel do Proxeon você tem vários canais, faz sentido manter um ou dois workers por canal, para que a rotação de IP de um canal não derrube as conexões de todos os threads ao mesmo tempo. A segunda é a gentileza com a origem: mesmo sem limites formais, dez threads paralelos em uma API pequena vão gerar carga, e você receberá recusas. Comece com três ou quatro workers e aumente conforme observa a taxa de erros.

Código do carregador paralelo

import json
import sqlite3
from concurrent.futures import ThreadPoolExecutor, as_completed


def init_tasks(con):
con.execute('PRAGMA journal_mode=WAL')
con.executescript('''
CREATE TABLE IF NOT EXISTS tasks(
task_id TEXT PRIMARY KEY,
params TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
''')
with con:
con.execute('UPDATE tasks SET status=? WHERE status=?', ('pending', 'running'))


def enqueue(con, tasks):
with con:
con.executemany('INSERT OR IGNORE INTO tasks(task_id, params) VALUES (?, ?)',
[(t['task_id'], json.dumps(t['params'])) for t in tasks])


def claim(con, limit):
rows = con.execute('SELECT task_id, params FROM tasks WHERE status=? LIMIT ?',
 ('pending', limit)).fetchall()
with con:
con.executemany('UPDATE tasks SET status=? WHERE task_id=?',
[('running', r[0]) for r in rows])
return [(r[0], json.loads(r[1])) for r in rows]


def run_parallel(con, fetch_fn, write_fn, workers=4, max_attempts=5):
init_tasks(con)
with ThreadPoolExecutor(max_workers=workers) as pool:
while True:
batch = claim(con, workers * 2)
if not batch:
break
futures = {pool.submit(fetch_fn, params): task_id for task_id, params in batch}
for fut in as_completed(futures):
task_id = futures[fut]
try:
rows = fut.result()
except Exception as e:
with con:
con.execute(
'UPDATE tasks SET attempts=attempts+1, last_error=?, '
'status=CASE WHEN attempts+1 >= ? THEN ? ELSE ? END WHERE task_id=?',
(str(e)[:500], max_attempts, 'failed', 'pending', task_id))
continue
with con: # dados e status da tarefa na mesma transação
write_fn(con, rows)
con.execute('UPDATE tasks SET status=? WHERE task_id=?', ('done', task_id))
failed = con.execute('SELECT count(*) FROM tasks WHERE status=?', ('failed',)).fetchone()[0]
print('fila vazia, tarefas com erro:', failed)

A função fetch_fn faz a requisição via proxy e retorna a lista de registros. Ela roda em thread e não toca no banco. A função write_fn é chamada na thread principal dentro de uma transação e faz a inserção idempotente do passo 3. Uma tarefa que falha volta automaticamente para pending e será retomada no próximo ciclo do claim; depois de esgotar as tentativas, ela recebe o status failed, e você resolve isso manualmente.

Atenção: não passe o objeto de conexão sqlite3 para os workers. A conexão está atrelada à thread em que foi criada, e tentar usá-la de outra thread resultará em erro ou, pior, em corrupção silenciosa dos dados. Ou cada thread tem a própria conexão, ou, como no exemplo acima, nenhuma.

Dica: use um objeto requests.Session por worker com um endereço de proxy próprio. Se você tem vários canais no Proxeon, distribua-os entre os workers em rodízio: worker 0 pega o canal 0, worker 1 pega o canal 1, e assim por diante. Assim, a queda de conexão em um canal afeta apenas um thread.

Resultado esperado

Verificação: coloque 100 tarefas na fila, inicie quatro workers e mate o processo após meio minuto. Execute SELECT status, count(*) FROM tasks GROUP BY status;. Você verá algumas done, algumas running e o restante pending. Inicie de novo: as running devem desaparecer na inicialização, e ao final todas as tarefas devem estar em done, exceto as que realmente falharam e ficaram em failed com o texto do erro em last_error.

Passo 6: Retomada após uma pausa longa

Objetivo da etapa: retomar corretamente uma exportação interrompida por várias horas ou dias e não esbarrar em estado obsoleto.

O que fica obsoleto

Retomar após dez segundos e após uma semana são tarefas diferentes. Depois de uma longa pausa, parte do estado salvo deixa de ser válido.

1. Sessões e cookies. As sessões do servidor geralmente duram de algumas horas a um dia. Cookies salvos depois disso vão gerar respostas 401 ou redirecionamento para a tela de login. Solução: na inicialização, fazer uma autenticação completa, em vez de restaurar cookies de arquivo.
2. Tokens de acesso. Tokens OAuth vivem cerca de uma hora, às vezes menos. Se você tem refresh token, atualize o access token antes de começar e programe atualizações durante o trabalho, sem esperar pela recusa.
3. Cursores de paginação. Muitas APIs limitam a vida útil do cursor a minutos ou horas. Cursor expirado retorna erro 400 com mensagem de cursor inválido. Por isso, no passo 2 guardamos uma âncora de reserva: o identificador do último registro. Se a origem suporta filtro por identificador ou por data de alteração, monte uma nova consulta a partir dessa âncora. Se não suporta, será preciso reiniciar o shard desde o começo, e a deduplicação do passo 4 descartará o que já foi obtido.
4. Conteúdo do arquivo no servidor. Para a retomada do passo 1, é crítico que o arquivo não tenha mudado. Compare o ETag atual com o guardado antes de cada rodada; se não coincidir, comece o arquivo de novo.
5. Configurações do proxy. Em uma semana, o painel do Proxeon pode ter mudado a porta, a senha, ou o canal pode ter expirado. Verifique o proxy com uma requisição de teste antes de começar a esvaziar a fila.
6. O conjunto de dados em si. Se a exportação durar uma semana e a origem tiver adicionado e removido registros nesse período, o resultado será uma mistura de estados em momentos diferentes. Para muitas tarefas isso é aceitável. Se não for, guarde no checkpoint a hora de início e, após concluir, faça uma passagem incremental separada pelos registros alterados depois desse horário.

Verificação pré-voo

Reúna todas as verificações em uma função executada na inicialização, antes de qualquer trabalho real. Ela ou ajusta o estado, ou para o carregador com uma mensagem clara.

def preflight(session, state, probe_url):
# 1. proxy está vivo e autorizado
r = session.head(probe_url, timeout=20)
if r.status_code == 407:
raise SystemExit('o proxy recusou o login ou a senha, verifique os dados no painel do Proxeon')
# 2. token de acesso está atualizado
refresh_access_token(session)
# 3. cursor ainda é válido
if state['cursor']:
test = session.get(API_BASE + '/orders', params={'cursor': state['cursor'], 'limit': 1}, timeout=30)
if test.status_code == 400 and 'cursor' in test.text.lower():
print('cursor expirou, mudando para a âncora por last_id =', state['last_id'])
state['cursor'] = None
state['resume_after_id'] = state['last_id']
# 4. lembrete sobre a idade da exportação
print('exportação iniciada em', state['started_at'], 'páginas gravadas', state['pages'])
return state

A função refresh_access_token depende da sua origem: normalmente é uma requisição POST com o refresh token, após a qual você atualiza o cabeçalho Authorization na sessão. O campo resume_after_id é então usado na função de obter página como filtro "identificador maior que o informado".

Dica: guarde o refresh token e a senha do proxy não no checkpoint, mas em variáveis de ambiente ou em um arquivo separado de segredos com permissões restritas. O checkpoint você vai copiar, enviar a colegas e anexar a relatórios de erro; segredos ali são desnecessários.

Resultado esperado

Verificação: corrompa o cursor na tabela checkpoint com UPDATE checkpoint SET cursor='broken'; e inicie o carregador. No console deve aparecer a linha sobre a mudança para a âncora por last_id, e a exportação deve continuar sem falhar. O número de registros ao final deve coincidir com uma execução de controle, sem a corrupção do cursor.

Passo 7: Esqueleto pronto de carregador robusto em Python

Objetivo da etapa: reunir tudo dos passos anteriores em um único arquivo, que pode ser iniciado, interrompido, iniciado de novo e produzir o resultado completo sem duplicatas.

Estrutura do esqueleto

• Configuração via variáveis de ambiente: endereço do proxy Proxeon, endereço da API, token, nome da tarefa.
• Classe Store: SQLite com tabelas de registros e checkpoint, uma transação por página.
• Função de chave de registro para idempotência.
• Função de obter página via proxy.
• Loop principal com retomada por checkpoint e recriação da sessão após queda.

Código completo

# resumable_loader.py
import hashlib
import json
import os
import sqlite3
import time
from datetime import datetime, timezone

import requests

PROXY_URL = os.environ['PROXY_URL']# http://USER:PASS@HOST:PORT do painel Proxeon
API_BASE = os.environ.get('API_BASE', 'https://api.example.com')
API_TOKEN = os.environ.get('API_TOKEN', '')
DB_PATH = os.environ.get('DB_PATH', 'export.sqlite')
JOB = os.environ.get('JOB', 'orders-2026')
PAGE_SIZE = 500
MAX_ATTEMPTS = 8
FIELDS = ('cursor', 'last_id', 'pages', 'rows', 'started_at')


def now():
return datetime.now(timezone.utc).isoformat()


def record_key(rec):
canon = json.dumps({'id': rec['id']}, sort_keys=True, separators=(',', ':'))
return hashlib.blake2b(canon.encode('utf-8'), digest_size=16).digest()


class Store:
def __init__(self, path):
self.con = sqlite3.connect(path)
self.con.execute('PRAGMA journal_mode=WAL')
self.con.executescript('''
CREATE TABLE IF NOT EXISTS records(
key BLOB PRIMARY KEY,
payload TEXT NOT NULL,
fetched_at TEXT NOT NULL
) WITHOUT ROWID;
CREATE TABLE IF NOT EXISTS checkpoint(
job TEXT PRIMARY KEY,
cursor TEXT,
last_id TEXT,
pages INTEGER NOT NULL,
rows INTEGER NOT NULL,
started_at TEXT,
updated_at TEXT
);
''')

def load(self, job):
row = self.con.execute(
'SELECT cursor, last_id, pages, rows, started_at FROM checkpoint WHERE job=?',
(job,)).fetchone()
if row is None:
return {'cursor': None, 'last_id': None, 'pages': 0, 'rows': 0, 'started_at': now()}
return dict(zip(FIELDS, row))

def commit_page(self, job, rows, state):
ts = now()
with self.con:
self.con.executemany(
'INSERT OR IGNORE INTO records(key, payload, fetched_at) VALUES (?, ?, ?)',
[(record_key(r), json.dumps(r, ensure_ascii=False), ts) for r in rows])
self.con.execute(
'INSERT INTO checkpoint(job, cursor, last_id, pages, rows, started_at, updated_at) '
'VALUES (?, ?, ?, ?, ?, ?, ?) '
'ON CONFLICT(job) DO UPDATE SET cursor=excluded.cursor, last_id=excluded.last_id, '
'pages=excluded.pages, rows=excluded.rows, updated_at=excluded.updated_at',
(job, state['cursor'], state['last_id'], state['pages'], state['rows'],
 state['started_at'], ts))

def unique_count(self):
return self.con.execute('SELECT count(*) FROM records').fetchone()[0]


def make_session():
s = requests.Session()
s.proxies = {'http': PROXY_URL, 'https': PROXY_URL}
s.headers['User-Agent'] = 'resumable-loader/1.0'
if API_TOKEN:
s.headers['Authorization'] = 'Bearer ' + API_TOKEN
return s


def fetch_page(session, cursor):
params = {'limit': PAGE_SIZE}
if cursor:
params['cursor'] = cursor
r = session.get(API_BASE + '/orders', params=params, timeout=(15, 90))
r.raise_for_status()
body = r.json()
return body['items'], body.get('next_cursor')


def run():
store = Store(DB_PATH)
state = store.load(JOB)
print('início: páginas %d, linhas %d, únicas no banco %d'
 % (state['pages'], state['rows'], store.unique_count()))
session = make_session()
attempts = 0
while True:
try:
items, next_cursor = fetch_page(session, state['cursor'])
attempts = 0
except (requests.ConnectionError, requests.Timeout, requests.HTTPError) as e:
attempts += 1
if attempts > MAX_ATTEMPTS:
print('tentativas esgotadas, estado salvo, execute novamente mais tarde')
raise
print('queda (%s), tentativa %d de %d' % (type(e).__name__, attempts, MAX_ATTEMPTS))
time.sleep(min(60, 2 ** attempts)) # escolha de pausas descrita no artigo sobre 429 e retries
session = make_session() # nova sessão: a conexão via proxy é recriada
continue
if not items:
break
state['pages'] += 1
state['rows'] += len(items)
state['last_id'] = str(items[-1]['id'])
state['cursor'] = next_cursor
store.commit_page(JOB, items, state)
if state['pages'] % 20 == 0:
print('páginas %d, linhas %d' % (state['pages'], state['rows']))
if next_cursor is None:
break
print('pronto: páginas %d, linhas obtidas %d, únicas no banco %d'
 % (state['pages'], state['rows'], store.unique_count()))


if __name__ == '__main__':
run()

Como adaptar para sua origem

1. Substitua o caminho /orders e os nomes dos campos items, next_cursor, id pelos que sua API devolve. São três lugares nas funções fetch_page e record_key.
2. Se a origem usa paginação por deslocamento, troque o parâmetro cursor por page e calcule o próximo valor como state['pages'] + 1. No checkpoint, guarde o número da página em vez do cursor.
3. Se a paginação é por chave, passe um parâmetro como after_id a partir de state['last_id'] e remova o trabalho com cursor.
4. Se precisar de paralelismo, mova fetch_page para fetch_fn do passo 5 e use commit_page como write_fn. Divida a exportação em shards e preencha a fila de tarefas.
5. Adicione a função preflight do passo 6 antes do loop principal.

Verificação do resultado: checklist

Antes de rodar o carregador em um volume real de várias horas, passe-o por esta lista. Cada item leva alguns minutos, e juntos eles garantem que a exportação noturna não perderá dados.

1. Teste de interrupção. Inicie o carregador, pressione Ctrl+C em 30 segundos. Inicie de novo. A primeira linha da saída deve mostrar um número de páginas diferente de zero, e não "páginas 0".
2. Teste de duplicatas. Reduza PAGE_SIZE para 10, interrompa o carregador cinco vezes seguidas em momentos aleatórios. Ao final, compare o número de registros únicos com o número de linhas obtidas: os únicos devem ser menores ou iguais, e em paginação por cursor sem inserções na origem praticamente iguais.
3. Teste de consistência. Após qualquer interrupção, execute duas consultas: SELECT rows FROM checkpoint; e SELECT count(*) FROM records;. A diferença entre elas não deve exceder o tamanho de uma página. Se exceder, checkpoint e dados não são gravados na mesma transação.
4. Teste de proxy. Temporariamente desative a variável PROXY_URL ou informe uma senha errada. O carregador deve falhar na primeira requisição com um erro claro, e não travar nem começar a acessar a origem diretamente.
5. Teste de cursor expirado. Corrompa o cursor no banco, como descrito no passo 6, e confirme que a alternância para a âncora funciona.
6. Teste de disco. Verifique o tamanho do arquivo export.sqlite após mil páginas e multiplique pelo número esperado de páginas. Confirme que haverá espaço em disco com folga de 20 por cento.

Verificação: o indicador de sucesso é quando, após três interrupções intencionais e três reinícios, o total de registros únicos coincide com o obtido em uma execução contínua na mesma origem, e no console nunca apareceu a linha sobre começar do zero.

Recursos adicionais e otimização

• Progresso e estimativa de tempo. Se o total de registros é conhecido, exiba percentual e estimativa de tempo restante a cada vinte páginas. Isso é útil para você e para distinguir um travamento de um trabalho lento.
• Compressão do payload. Para dezenas de milhões de registros, o JSON em texto ocupa muito espaço. Comprima o campo payload com zlib.compress antes de gravar e armazene como BLOB. A economia costuma ser de três a oito vezes.
• Migração para banco em servidor. O esquema com checkpoint na mesma transação dos dados se transfere para PostgreSQL quase sem mudanças. A construção ON CONFLICT é suportada lá, e a restrição de "um escritor" desaparece.
• Exportações incrementais. Guarde a hora de início de cada tarefa e, após a exportação completa, execute uma tarefa separada com o filtro "alterado após". Assim você mantém uma cópia atualizada sem rebaixar tudo de novo.
• Processo separado por shard. Em vez de threads, você pode iniciar vários exemplares do script com valores diferentes de JOB e arquivos DB_PATH diferentes, e reunir os resultados no final. É mais fácil de depurar e elimina completamente a questão da escrita concorrente.
• Métricas de quedas. Registre cada queda com o tipo de exceção e o horário. Depois de um dia você verá que as quedas se agrupam em torno dos intervalos de rotação de IP no canal Proxeon e poderá ajustar o intervalo de rotação à duração das suas requisições.

Erros típicos e soluções

Abaixo estão situações que quase todo mundo encontra nas primeiras execuções. Formato: problema, causa, solução.

1. Problema: após reiniciar, a exportação sempre começa do zero. Causa: o checkpoint é gravado em memória ou em um arquivo que não sobrevive à queda, ou o carregador não o lê na inicialização. Solução: garanta que a primeira ação na função run é store.load, e que o estado é atualizado após cada página dentro da transação.
2. Problema: o banco tem uma vez e meia mais registros que a origem. Causa: a chave de deduplicação é instável: entrou nela o horário da requisição, o número da página, ou um campo com ordem aleatória. Solução: calcule a chave apenas pelo identificador da origem ou por uma lista explícita de campos estáveis via parâmetro fields.
3. Problema: o banco tem menos registros que a origem, embora a exportação tenha terminado sem erros. Causa: o checkpoint era atualizado antes de gravar os dados, e após uma queda a página foi pulada. Ou a paginação por deslocamento, ao excluir registros na origem, deslocou as páginas para trás. Solução: mude a ordem para "dados, depois checkpoint" na mesma transação; para origens com exclusões, migre para paginação por chave.
4. Problema: a retomada do download gera um arquivo compactado corrompido, mesmo com tamanho coincidente. Causa: o servidor uma vez respondeu 200 em vez de 206, o arquivo foi sobrescrito parcialmente e depois anexado. Solução: guarde o ETag ao lado do arquivo, apague o arquivo quando mudar; verifique se o cabeçalho Content-Range corresponde ao deslocamento solicitado.
5. Problema: erro "database is locked" em trabalho paralelo. Causa: várias threads escrevem no SQLite ao mesmo tempo, ou a conexão foi passada entre threads. Solução: um único ponto de escrita na thread principal, workers apenas baixam; modo WAL; conexão criada na thread em que é usada.
6. Problema: depois de uma hora, todas as requisições passam a retornar 401. Causa: o token de acesso expirou. Solução: atualize o token antes do vencimento, e ao receber 401 dispare a atualização e repita a requisição uma vez, sem considerar isso uma queda.
7. Problema: a memória RAM cresce até vários gigabytes. Causa: o conjunto de chaves vistas ou a lista de todos os registros fica na memória do processo. Solução: deduplicação pela chave primária no banco ou pela tabela seen; gravação de dados por página, sem acumular.
8. Problema: as quedas ocorrem rigorosamente a cada poucos minutos. Causa: coincidem com o intervalo de rotação de IP no canal do proxy. Solução: é uma situação padrão, o carregador precisa sobreviver a ela. Se as requisições são longas, ajuste o intervalo de rotação no painel do Proxeon para ser bem maior que o tempo típico de uma requisição, ou use rotação sob demanda entre páginas.

FAQ: perguntas frequentes sobre exportação robusta

É obrigatório usar SQLite se o resultado final precisa ser CSV?

Não, mas é conveniente. O SQLite aqui faz o papel de armazenamento confiável do estado e do conjunto de chaves vistas. O CSV final você exporta com um único comando da tabela records após concluir. Se quiser dispensar o banco por completo, use a variante em arquivos do passo 3 com um arquivo por página e um checkpoint JSON com substituição atômica.

Com que frequência salvar o checkpoint: a cada página ou menos?

A cada página. Uma transação SQLite com algumas centenas de linhas leva milissegundos, o que é desprezível comparado à requisição de rede via proxy. A economia com checkpoints raros não compensa o risco de perder dezenas de páginas.

O que fazer se a API não fornece nem cursor nem identificadores, apenas números de página?

Trabalhe por número de página, guarde-o no checkpoint e inclua obrigatoriamente a deduplicação por hash do conteúdo do registro. Aceite que, em origens com alterações ativas, parte dos registros pode ser pulada por causa do deslocamento de páginas. Para dados críticos, faça uma segunda passagem em ordem inversa de páginas: o que foi pulado na primeira passagem tem alta probabilidade de aparecer na segunda.

É possível retomar o download em várias threads, cada uma com um range diferente?

Sim, se o servidor suporta Range. Divida o arquivo em pedaços de 50 a 100 MB, cada pedaço é uma tarefa da fila do passo 5 com seu arquivo temporário, e ao concluir todos os pedaços, una-os na ordem correta. Verifique cada pedaço por tamanho e o arquivo inteiro por soma de verificação, se houver.

Quantos workers colocar ao trabalhar via proxy?

Comece com três ou quatro em um canal Proxeon e observe a taxa de erros na tabela tasks. Se os erros forem menos de um por cento, adicione mais dois. Se os erros aumentarem, reduza. Mais de dez threads em um canal raramente traz ganho: você esbarra na largura de banda do canal ou na paciência da origem.

Vale a pena guardar os cookies da sessão entre execuções?

Normalmente não. Uma nova autenticação na inicialização leva segundos e é mais confiável do que restaurar cookies com tempo de vida desconhecido. Exceção: se a origem limita o número de logins por dia. Nesse caso, guarde os cookies, mas, ao primeiro 401 ou redirecionamento para login, descarte-os e autentique-se de novo.

Como saber que a exportação terminou de fato, e não caiu silenciosamente?

Para arquivos: o tamanho é igual ao Content-Length e a soma de verificação coincide. Para API: recebeu-se uma página sem next_cursor ou uma página vazia, e o número de linhas corresponde ao total, se a origem informa. Grave no checkpoint uma flag explícita de conclusão, para que uma nova execução não inicie um novo ciclo.

O que fazer com as tarefas em status failed?

Olhe o campo last_error. Se forem erros de rede, basta devolver as tarefas para pending com um UPDATE e rodar o carregador de novo. Se forem erros de parsing, significa que há registros de formato não padrão na origem: corrija o código e reinicie. Nunca apague failed em silêncio, esse é o único indício do que falta na exportação.

É possível usar esta abordagem com uma biblioteca assíncrona em vez de requests?

Sim, os princípios são os mesmos: unidade de trabalho atômica, checkpoint junto aos dados, gravação idempotente, fila em disco. Muda apenas o transporte. A única nuance: mantenha a escrita no SQLite síncrona e sequencial, e deixe o paralelismo no nível das requisições de rede.

Conclusão

Você percorreu o caminho do arquivo incompleto e do reinício nervoso até um carregador para o qual a queda é indiferente. Vamos fixar o que foi feito.

• Entendemos a retomada via HTTP: verificação de Accept-Ranges por HEAD, cabeçalho Range com o tamanho atual do arquivo, distinção entre os códigos 206 e 200, proteção contra troca de arquivo via ETag e If-Range.
• Construímos checkpoints para exportações paginadas: cursor, número de página ou identificador do último registro, mais contadores auxiliares, tudo na mesma transação dos dados.
• Tornamos a gravação idempotente por meio de chave primária e da construção INSERT OR IGNORE ou ON CONFLICT DO UPDATE, e para arquivos com nomes determinísticos e substituição atômica.
• Organizamos a deduplicação em disco, para que milhões de chaves não vivam na memória e sobrevivam a reinícios.
• Adicionamos paralelismo com fila de tarefas no SQLite, devolução automática de tarefas falhas e um único ponto de escrita.
• Previmos a retomada após pausa longa: atualização de tokens, nova autenticação, troca do cursor expirado por âncora de identificador, verificação do proxy Proxeon antes de começar.
• Reunimos tudo em um esqueleto funcional, que se adapta a uma origem específica com a troca de três ou quatro linhas.

O que fazer a seguir

Pegue o esqueleto do passo 7 e rode-o em um volume real pequeno, digamos dez mil registros. Passe pelo checklist da seção de verificação. Só depois inicie a exportação completa durante a noite. De manhã, ou você verá a linha "pronto" com contadores coincidentes, ou a linha sobre tentativas esgotadas com o estado preservado, e aí basta executar o script novamente.

Para onde evoluir

O próximo nível são as exportações incrementais por hora de alteração em vez de varreduras completas, a migração do estado para banco em servidor para várias máquinas, e uma estratégia de retries bem pensada considerando códigos de resposta, tema de um artigo específico sobre 429 e retries. A combinação de retries bem feitos desse artigo com o estado robusto deste guia produz um carregador que você pode deixar rodando uma semana sem abrir o terminal.

E por último. A queda de uma exportação longa não é um acidente, é uma situação de trabalho que você agora sabe tratar. Boas exportações.