Como exportar grandes volumes de dados via proxy e não recomeçar depois de uma queda

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

Se você já rodou uma exportação grande de dados, conhece essa sensação. O processo rodou por horas, chegou aos noventa por cento e caiu. A conexão caiu, o servidor retornou erro, o notebook entrou em suspensão. E tudo precisa começar de novo. Este guia foi escrito para que isso não aconteça mais.

O que você vai conseguir no final. Você vai aprender a construir um carregador que sobrevive a quedas. Ele retoma arquivos do meio, lembra em qual página parou, não cria duplicatas ao repetir e consegue retomar mesmo depois de uma longa pausa. Você vai receber um esqueleto pronto em Python que pode adaptar para a sua tarefa.

Para quem é este guia. Para engenheiros, analistas e desenvolvedores que exportam dados de APIs, baixam arquivos grandes ou coletam resultados paginados via proxy. Nível intermediário. Você precisa entender o básico de HTTP e saber ler código em Python. Não é necessário conhecimento profundo de programação de redes.

O que você precisa saber antes. Python básico, o conceito de requisição e resposta HTTP, o que são cabeçalhos e códigos de status. Se você já trabalhou com a biblioteca requests, isso é suficiente.

Quanto tempo vai levar. A leitura e compreensão dos conceitos leva cerca de quarenta minutos. Montar um carregador funcional com nosso esqueleto leva de uma a três horas, dependendo da sua fonte de dados.

Uma observação importante sobre o tema. Não vamos abordar códigos de status como 429 e estratégias de repetição com atraso (backoff). Existe um material separado sobre isso. Aqui o foco é apenas um: o estado do processo e sua retomada. Como salvar o progresso, como não perder nem duplicar dados, como continuar de onde parou.

Dica: Tenha à mão um bloco de notas ou um arquivo separado onde vai anotar os parâmetros da sua fonte: ela suporta retomada, tem navegação paginada, qual é o formato do cursor dela. Essas anotações serão úteis em cada etapa.

Preparação inicial: ferramentas, acessos e ambiente

Antes de escrever código, vamos montar o ambiente de trabalho. Isso leva dez minutos, mas economiza horas de depuração.

O que instalar

1. Instale o Python versão 3.10 ou mais recente. Verifique a versão com o comando python --version no terminal.
2. Crie um ambiente virtual com o comando python -m venv venv, para que as dependências do projeto não se misturem com as do sistema.
3. Ative o ambiente. No Windows é venv\Scripts\activate, no macOS e Linux é source venv/bin/activate.
4. Instale a biblioteca para requisições HTTP com o comando pip install requests.
5. Para trabalhar mais rápido com o banco de estado, não é preciso instalar nada extra: o módulo sqlite3 já faz parte da biblioteca padrão do Python.

O que você precisa em termos de acesso

• Acesso à sua fonte de dados: URL, token ou chave de API, se for necessário.
• Um proxy da Proxeon com endereço, porta e dados de autenticação. Sem um proxy estável, a exportação resiliente perde sentido, porque é justamente o proxy que distribui a carga e torna as conexões previsíveis.
• Espaço em disco para o arquivo de estado e para os próprios dados exportados.

Testando o proxy Proxeon

1. Pegue a string de conexão no formato http://usuário:senha@endereço:porta.
2. Teste com uma requisição simples. No terminal, execute curl -x http://usuário:senha@endereço:porta https://api.ipify.org e confirme que retornou o IP do proxy, e não o seu próprio.

Dica: Salve a string de conexão do proxy em uma variável de ambiente, e não no código. Assim você não envia a senha para o sistema de controle de versão por acidente. No código, leia-a através de os.environ.

⚠️ Atenção: Trabalhe sempre apenas com fontes de dados às quais você tem acesso legal. Respeite os termos de uso do serviço e os limites estabelecidos pelo proprietário. O proxy Proxeon é destinado a trabalho de engenharia legal: distribuição de carga, estabilidade de conexões e exportação correta de dados.

✅ Verificação: O ambiente está pronto se o comando python -c "import requests, sqlite3" rodou sem erros e a requisição via proxy retornou o endereço do servidor proxy.

Conceitos básicos: dicionário da exportação resiliente em palavras simples

Antes de escrever código, vamos entender os termos-chave. Sem eles, os próximos passos vão soar como feitiçaria.

Retomada

Retomada é continuar o download de um arquivo a partir do byte onde ele foi interrompido. Em vez de baixar o arquivo de novo, você pede ao servidor apenas o pedaço que falta. Funciona através do cabeçalho HTTP Range.

Checkpoint

Checkpoint é um ponto de progresso salvo. Pense no salvamento de um jogo de computador. Se algo der errado, você volta ao último salvamento, e não ao início do jogo. Na exportação, o checkpoint guarda em qual página ou registro você parou.

Cursor

Cursor é uma marca que a API te dá para que você possa pedir a próxima leva de dados. Muitas vezes é uma string como eyJvZmZzZXQiOjEwMH0. Você a envia de volta, e o servidor entende de onde continuar.

Idempotência

Idempotência é a propriedade de uma operação em que sua repetição não altera o resultado. Se você gravou duas vezes a mesma linha com a mesma chave, no final a linha é uma, e não duas. Isso é proteção contra duplicatas em repetições.

Deduplicação

Deduplicação é o descarte de registros repetidos. Mesmo com trabalho cuidadoso, o mesmo objeto pode chegar duas vezes. A deduplicação garante que no seu conjunto final ele permaneça apenas uma vez.

Princípio fundamental

Um carregador resiliente se baseia em uma única ideia: o progresso precisa ser salvo constantemente, e não apenas no final. Qualquer etapa pode ser a última antes da queda. Ou seja, após cada pedaço bem-sucedido do trabalho, o estado deve ser gravado em disco. Assim, retomar é simplesmente ler o estado e continuar.

Dica: Memorize a regra das três perguntas para qualquer exportação. Primeira: onde eu parei? Segunda: como não duplicar o que já recebi? Terceira: o que vai expirar enquanto eu estiver fora? As respostas a elas é que constituem a resiliência.

Etapa 1: Retomada via HTTP com o cabeçalho Range

Objetivo da etapa. Aprender a baixar um arquivo grande de modo que, após uma queda, você continue a partir do byte não baixado, e não do zero.

Como funciona

O HTTP permite solicitar não o arquivo inteiro, mas uma parte dele. Para isso, adiciona-se o cabeçalho Range à requisição. Por exemplo, Range: bytes=1048576- significa: me envie tudo a partir do byte número 1048576. Mas primeiro é preciso confirmar que o servidor sabe fazer isso.

1. Envie uma requisição HEAD ou um GET comum para o arquivo e observe os cabeçalhos da resposta.
2. Procure o cabeçalho Accept-Ranges. Se o valor for bytes, o servidor suporta retomada.
3. Se o cabeçalho não existir ou o valor for none, a retomada é impossível. Nesse caso, será preciso baixar o arquivo inteiro em uma única tentativa ou procurar uma fonte alternativa.

Verificando o suporte à retomada

Aqui está o código que verifica se o servidor consegue entregar partes do arquivo.

import requests
def supports_resume(url, proxies):
resp = requests.head(url, proxies=proxies, timeout=30, allow_redirects=True)
accept = resp.headers.get("Accept-Ranges", "none")
total = resp.headers.get("Content-Length")
return accept.lower() == "bytes", total

Retomando o arquivo do meio

Agora o código principal. Ele verifica quantos bytes já foram baixados localmente e pede ao servidor apenas o restante.

import os
import requests
def download_resumable(url, dest, proxies):
already = 0
if os.path.exists(dest):
already = os.path.getsize(dest)
headers = {}
if already > 0:
headers["Range"] = f"bytes={already}-"
mode = "ab" if already > 0 else "wb"
with requests.get(url, headers=headers, proxies=proxies,
 stream=True, timeout=60) as r:
if already > 0 and r.status_code == 200:
mode = "wb"
already = 0
with open(dest, mode) as f:
for chunk in r.iter_content(chunk_size=65536):
if chunk:
f.write(chunk)
return os.path.getsize(dest)

Vamos entender os pontos importantes. Se o servidor retornou o status 206, ele entregou de fato parte do arquivo e a escrita incremental vai funcionar corretamente. Se o servidor retornou 200 apesar do cabeçalho Range, significa que ele ignorou a retomada e está entregando o arquivo inteiro. Nesse caso, mudamos para o modo de sobrescrita completa, para não colar o pedaço antigo ao novo e corromper o arquivo.

⚠️ Atenção: Nunca escreva dados no modo ab se não tiver certeza de que o servidor respondeu com o código 206. Caso contrário, você vai obter um arquivo corrompido, onde o início é o resto da tentativa anterior e a continuação é o novo arquivo completo. Esse arquivo abrirá com erro, e você perderá tempo procurando a causa.

Dica: Baixe não direto no arquivo de destino, mas em um arquivo temporário com extensão .part. Quando o download terminar completamente, renomeie-o para o nome final. Assim você nunca vai confundir um arquivo pronto com um arquivo incompleto.

Controle de integridade

Após o download completo, é bom verificar que o arquivo não foi corrompido. Se o servidor enviou o cabeçalho Content-Length, compare-o com o tamanho real do arquivo em disco. Se os tamanhos coincidirem, o arquivo chegou inteiro.

def verify_size(dest, expected):
if expected is None:
return True
return os.path.getsize(dest) == int(expected)

✅ Verificação: Interrompa o download no meio, fechando o programa. Execute-o de novo. Nos logs, você deve ver que a requisição foi enviada com o cabeçalho Range e que o arquivo foi complementado, em vez de começar de novo. O tamanho final coincide com o esperado.

Etapa 2: Checkpoints para exportações paginadas

Objetivo da etapa. Configurar o salvamento de progresso para APIs que entregam dados em páginas, para que após uma queda você continue a partir da página certa.

O que exatamente salvar

Um arquivo é retomado por bytes, mas a exportação paginada segue uma lógica bem diferente. Aqui não há bytes, há páginas e registros. Ou seja, o checkpoint precisa guardar outra coisa.

• Cursor se a API funciona com cursores. Essa é a opção mais confiável, porque o cursor já sabe de onde continuar.
• Número da página ou deslocamento se a API funciona com offset e limit. Guarde o número da última página processada com sucesso.
• Identificador do último registro se for possível ordenar por ID ou data crescente. Então a próxima requisição pede registros com ID maior que o salvo.
• Contador de registros processados para controle e relatório.

Onde armazenar o estado

Você tem três opções principais, do simples ao confiável.

1. Arquivo JSON. O mais simples. Você grava um dicionário com cursor e contador em um arquivo após cada página. Serve para exportações únicas, não paralelas.
2. Banco SQLite. Mais confiável. Oferece transações, então o estado não se corrompe se houver queda no momento da gravação. Bom quando há muitos dados e é preciso deduplicação.
3. Banco de dados externo. Para exportações grandes e distribuídas, quando vários processos dividem o mesmo trabalho.

Salvando o checkpoint em JSON

import json
import os
def save_checkpoint(path, cursor, page, last_id, count):
tmp = path + ".tmp"
data = {
"cursor": cursor,
"page": page,
"last_id": last_id,
"count": count,
}
with open(tmp, "w") as f:
json.dump(data, f)
os.replace(tmp, path)
def load_checkpoint(path):
if not os.path.exists(path):
return {"cursor": None, "page": 0, "last_id": None, "count": 0}
with open(path) as f:
return json.load(f)

Repare no truque do arquivo temporário. Gravamos em um arquivo com sufixo .tmp e depois o renomeamos atomicamente via os.replace. Isso protege contra a situação em que o programa cai exatamente durante a gravação do checkpoint. O checkpoint antigo permanece intacto, em vez de virar um JSON pela metade que não pode ser lido.

⚠️ Atenção: Nunca grave o checkpoint diretamente no mesmo arquivo por cima do antigo sem um arquivo temporário. Uma queda no meio da gravação vai deixar você com um checkpoint corrompido, e a retomada se torna impossível. A substituição atômica resolve esse problema por completo.

Dica: Salve o checkpoint só depois que os dados da página forem realmente gravados no armazenamento. A ordem é: recebeu a página, gravou os dados, depois atualizou o checkpoint. Se você inverter a ordem, em caso de queda vai pular uma página e perder dados.

Loop principal com checkpoints

def paginate(fetch_page, save_data, cp_path, proxies):
cp = load_checkpoint(cp_path)
cursor = cp["cursor"]
count = cp["count"]
while True:
items, next_cursor = fetch_page(cursor, proxies)
if not items:
break
save_data(items)
count += len(items)
last_id = items[-1].get("id")
save_checkpoint(cp_path, next_cursor, cp["page"] + 1,
last_id, count)
cursor = next_cursor
if next_cursor is None:
break
return count

✅ Verificação: Inicie a exportação, deixe-a processar algumas páginas e interrompa. Abra o arquivo de checkpoint e confirme que há um cursor atualizado e um contador. Inicie de novo: a exportação deve continuar a partir do cursor salvo, e não da primeira página.

Etapa 3: Idempotência, para que a repetição não crie duplicatas

Objetivo da etapa. Fazer com que uma nova execução ou a repetição de uma requisição específica não resulte em registros idênticos no seu armazenamento.

Por que surgem duplicatas

Imagine: você recebeu uma página de dados, gravou-a em arquivo, mas o programa caiu antes de atualizar o checkpoint. Na próxima execução, você vai pedir a mesma página de novo. Os dados chegam novamente e são gravados pela segunda vez. É assim que nascem as duplicatas. Isso é consequência inevitável das quedas, e é preciso combatê-lo no nível da arquitetura.

Chave de deduplicação

A principal ferramenta da idempotência é a chave de deduplicação. É um campo ou combinação de campos que identifica um registro de forma inequívoca. A escolha correta da chave resolve metade do problema.

• ID natural. Se o registro tem um identificador único da fonte, use-o. É a chave ideal.
• Combinação de campos. Se não há um ID único, monte a chave a partir de vários campos estáveis. Por exemplo, email mais data de cadastro.
• Hash do conteúdo. Se não há campos estáveis de jeito nenhum, calcule um hash de todo o registro. Essa é a opção extrema, porque qualquer alteração de campo vai gerar uma nova chave.

Gravação sem duplicatas via UPSERT

Se você armazena o resultado em SQLite ou outro banco, use a inserção com ignorar conflito. Assim, uma nova gravação com a mesma chave simplesmente não faz nada.

import sqlite3
def init_db(path):
conn = sqlite3.connect(path)
conn.execute(
"CREATE TABLE IF NOT EXISTS records ("
"dedup_key TEXT PRIMARY KEY, payload TEXT)"
)
conn.commit()
return conn
def save_records(conn, items):
rows = [(item["id"], json.dumps(item)) for item in items]
conn.executemany(
"INSERT OR IGNORE INTO records (dedup_key, payload) "
"VALUES (?, ?)", rows
)
conn.commit()

O detalhe-chave aqui é o PRIMARY KEY no campo dedup_key. O próprio banco vai recusar a inserção repetida com a mesma chave, porque INSERT OR IGNORE engole o conflito em silêncio. Você não precisa verificar manualmente se esse registro já existe. O banco faz isso por você e faz rápido.

Dica: Escolha a chave de deduplicação uma vez no início do projeto e registre-a na documentação. Trocar a chave no meio da exportação significa que registros antigos e novos deixarão de se corresponder, e as duplicatas vão aparecer mesmo assim. A estabilidade da chave é mais importante que a sua elegância.

✅ Verificação: Execute a exportação duas vezes seguidas no mesmo intervalo de dados. Conte o número de linhas no banco com o comando SELECT COUNT(*) FROM records. O número deve ser o mesmo após a primeira e após a segunda execução.

Etapa 4: Deduplicação de resultados sem inflar a memória

Objetivo da etapa. Descartar registros repetidos em milhões de linhas, sem carregar toda a memória do computador com um monte de chaves já vistas.

Abordagem ingênua e seu problema

A deduplicação mais simples: manter na memória um conjunto set de todas as chaves vistas. Para cada novo registro, verificar se a chave está no conjunto. Funciona muito bem em centenas de milhares de linhas. Mas em milhões e dezenas de milhões, o conjunto cresce e devora gigabytes de memória RAM. O programa fica lento ou trava.

Solução um: confiar no banco

O jeito mais simples e confiável em grandes volumes é não armazenar o que já foi visto na memória de jeito nenhum, e deixar a verificação para o banco através da PRIMARY KEY, como fizemos na etapa anterior. O banco armazena o índice em disco, e não na memória do seu processo. Ele vai lidar com dezenas de milhões de chaves sem pesar na sua memória RAM.

Solução dois: hash do registro

Quando não há chave natural, calcule um hash compacto do registro. O hash ocupa um espaço fixo e pequeno, independentemente do tamanho do próprio registro.

import hashlib
import json
def record_hash(item):
raw = json.dumps(item, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()

O parâmetro sort_keys=True aqui é crítico. Ele garante que registros com o mesmo conteúdo gerem o mesmo hash, mesmo que os campos tenham vindo em ordem diferente. Sem essa ordenação, dois objetos idênticos podem receber hashes diferentes e passar como registros distintos.

Solução três: filtro de Bloom para economizar memória

Se você realmente precisa de uma verificação rápida em memória em volumes enormes, use o filtro de Bloom. É uma estrutura que ocupa pouco espaço e responde rápido se já vimos a chave ou se com certeza não vimos. Ela tem uma particularidade: pode ocasionalmente dizer erroneamente que a chave já existia, quando não existia. Por isso o filtro de Bloom é usado como um descarte prévio rápido, e a verificação final fica com o banco.

1. Verificamos a chave com o filtro de Bloom.
2. Se o filtro diz que com certeza não vimos, gravamos direto no banco.
3. Se o filtro diz que possivelmente vimos, fazemos a verificação exata no banco.

⚠️ Atenção: Não tente deduplicar dezenas de milhões de linhas com um conjunto comum na memória. Em um notebook típico, isso vai levar à exaustão de memória e à queda do processo no meio da exportação. Transfira a carga para o disco via banco ou use um filtro de Bloom.

Dica: Se você exporta dados em lotes e dentro de um lote podem existir duplicatas, deduplique o lote na memória com um conjunto comum antes de gravar no banco. O lote é pequeno, a memória não sofre, e vão menos inserções desnecessárias para o banco.

✅ Verificação: Execute a deduplicação em um grande conjunto de teste com repetições intencionais. Verifique que o número final de registros únicos está correto e que o consumo de memória do processo permanece estável, sem crescer linearmente com o número de linhas.

Etapa 5: Paralelismo sem perdas

Objetivo da etapa. Acelerar a exportação com requisições paralelas, sem perder nenhuma tarefa e repetindo corretamente as que falharam.

Fila de tarefas

A base do paralelismo seguro é a fila de tarefas. Você divide o trabalho antecipadamente em pedaços independentes. Por exemplo, uma lista de páginas ou intervalos. Coloca-os em uma fila. Vários workers pegam tarefas da fila, executam e depositam o resultado. Se um worker cai, sua tarefa pode voltar para a fila e ser entregue a outro.

Limitação da simultaneidade

Não se pode disparar um número infinito de requisições paralelas. Isso vai sobrecarregar a fonte e o seu proxy. A abordagem correta é limitar o número de workers simultâneos a um valor razoável. Comece com um número pequeno e aumente, observando a estabilidade.

from concurrent.futures import ThreadPoolExecutor, as_completed
def run_parallel(tasks, worker, proxies, max_workers=5):
results = []
failed = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
future_map = {
pool.submit(worker, t, proxies): t for t in tasks
}
for future in as_completed(future_map):
task = future_map[future]
try:
results.append(future.result())
except Exception:
failed.append(task)
return results, failed

Repetição das tarefas que falharam

A lista failed coletada não são dados perdidos, mas a lista do que precisa ser repetido. Após a primeira passagem, você roda as tarefas que falharam mais uma vez. Normalmente isso basta para terminar o restante.

def run_with_retry(tasks, worker, proxies, rounds=3):
remaining = tasks
for _ in range(rounds):
done, remaining = run_parallel(remaining, worker, proxies)
if not remaining:
break
return remaining

O papel do proxy Proxeon no paralelismo. No trabalho paralelo, o proxy distribui as conexões, o que torna a exportação mais estável e previsível. Cada worker trabalha através de sua própria conexão, e a carga não se concentra em um único ponto.

⚠️ Atenção: Na gravação paralela em um único arquivo ou em um único checkpoint surgem condições de corrida. Dois workers podem sobrescrever o estado um do outro. Grave resultados apenas em banco com transações ou use um arquivo separado para cada worker, e monte o checkpoint consolidado em uma thread separada.

Dica: Faça as tarefas pequenas e independentes. Se uma tarefa cobre um intervalo muito grande, sua queda vai descartar muito trabalho. Tarefas pequenas se repetem de forma barata e quase imperceptível.

✅ Verificação: Inicie uma exportação paralela e derrube artificialmente parte dos workers. Após os rounds de repetição, a lista remaining deve ficar vazia, e o conjunto final de dados deve estar completo. Compare o número de registros obtidos com o esperado.

Etapa 6: Retomada após uma longa pausa

Objetivo da etapa. Continuar corretamente a exportação se muito tempo passou entre as tentativas, e entender o que pode ter expirado nesse período.

O que expira com o tempo

Uma queda de um minuto e uma pausa de um dia são situações diferentes. Ao longo de uma longa pausa, parte do seu estado pode se tornar inválido.

• Sessão. Muitos serviços mantêm a sessão por tempo limitado. Após uma longa pausa, o servidor vai esquecê-la, e as requisições começarão a retornar erro de autorização.
• Token de acesso. Tokens de API costumam ter tempo de vida em minutos ou horas. Um token expirado precisa ser renovado antes de continuar.
• Cursor. Alguns cursores duram pouco. Se o cursor expirou, será preciso começar do ponto estável mais próximo, por exemplo pelo identificador do último registro.
• Os próprios dados. Durante a pausa, podem ter surgido novos registros na fonte ou os antigos podem ter mudado. Isso afeta os deslocamentos na navegação paginada por offset.

Estratégia de retomada segura

1. Ao iniciar, verifique a idade do checkpoint. Se estiver antigo, esteja preparado para o fato de que parte do estado ficou desatualizada.
2. Renove o token de acesso e crie uma nova sessão antes da primeira requisição. Não confie nas antigas.
3. Prefira a retomada pelo identificador do último registro, e não pelo número da página. O ID é estável, enquanto o número da página se desloca se os dados mudaram.
4. Faça uma requisição de teste com o cursor salvo. Se ela retornar erro de cursor inválido, mude para a retomada por last_id.

def resume(cp, fetch_by_id, fetch_by_cursor, proxies):
if cp["cursor"]:
try:
return fetch_by_cursor(cp["cursor"], proxies)
except CursorExpired:
pass
return fetch_by_id(cp["last_id"], proxies)

Por que a retomada por ID é mais confiável. Imagine que você parou na página 50 de uma ordenação por data. Enquanto você estava fora, novos registros foram adicionados no início. Agora a página 50 contém dados completamente diferentes, e você vai pular parte dos registros. A retomada pelo identificador do último registro não sofre com isso: você simplesmente pede tudo o que for maior que o ID salvo.

Dica: Salve sempre no checkpoint tanto o cursor quanto o identificador do último registro ao mesmo tempo. O cursor é mais rápido, mas o ID é a sua corda de segurança caso o cursor expire durante uma longa pausa.

✅ Verificação: Pare a exportação, espere o suficiente para que o token ou o cursor expirem, e inicie de novo. O carregador deve renovar o token, detectar o cursor expirado e continuar pelo identificador, sem perda e sem duplicação de registros.

Etapa 7: Esqueleto pronto de um carregador resiliente em Python

Objetivo da etapa. Reunir tudo o que foi estudado em um único esqueleto funcional, que você vai adaptar para a sua fonte de dados.

Abaixo está reunido um esqueleto que combina checkpoints, deduplicação via banco, renovação de token e retomada. Você substitui as funções de obtenção de página pelas suas, de acordo com a API específica.

import os
import json
import sqlite3
import requests
class ResilientLoader:
def __init__(self, cp_path, db_path, proxies):
self.cp_path = cp_path
self.proxies = proxies
self.conn = sqlite3.connect(db_path)
self.conn.execute(
"CREATE TABLE IF NOT EXISTS records ("
"dedup_key TEXT PRIMARY KEY, payload TEXT)"
)
self.conn.commit()
def load_cp(self):
if not os.path.exists(self.cp_path):
return {"cursor": None, "last_id": None, "count": 0}
with open(self.cp_path) as f:
return json.load(f)
def save_cp(self, cp):
tmp = self.cp_path + ".tmp"
with open(tmp, "w") as f:
json.dump(cp, f)
os.replace(tmp, self.cp_path)
def save_records(self, items):
rows = [(str(i["id"]), json.dumps(i)) for i in items]
self.conn.executemany(
"INSERT OR IGNORE INTO records "
"(dedup_key, payload) VALUES (?, ?)", rows
)
self.conn.commit()
def run(self, fetch_page):
cp = self.load_cp()
while True:
items, next_cursor = fetch_page(
cp["cursor"], cp["last_id"], self.proxies
)
if not items:
break
self.save_records(items)
cp["count"] += len(items)
cp["last_id"] = items[-1]["id"]
cp["cursor"] = next_cursor
self.save_cp(cp)
if next_cursor is None:
break
return cp["count"]

Exemplo de função de obtenção de página para a sua fonte. Aqui você implementa a lógica da requisição e a interpretação da resposta.

def fetch_page(cursor, last_id, proxies):
params = {"limit": 100}
if cursor:
params["cursor"] = cursor
elif last_id:
params["after_id"] = last_id
r = requests.get(
"https://example-source/api/records",
params=params, proxies=proxies, timeout=60
)
r.raise_for_status()
data = r.json()
return data["items"], data.get("next_cursor")

Iniciar todo o mecanismo é simples.

proxies = {
"http": os.environ["PROXEON_URL"],
"https": os.environ["PROXEON_URL"],
}
loader = ResilientLoader("state.json", "out.db", proxies)
total = loader.run(fetch_page)
print("Total de registros:", total)

Dica: Adicione ao loop um log a cada cem registros: hora, contador, cursor atual. Assim você vai ver o progresso e entender facilmente se a exportação travou em um ponto.

✅ Verificação: Inicie o esqueleto em uma fonte real, interrompa no meio, inicie de novo. O número final de registros após a retomada vai coincidir com o número total de registros da fonte, e uma nova execução não vai aumentar o contador de linhas únicas.

Verificação do resultado: checklist da exportação resiliente

Percorra esta lista. Se todos os itens forem cumpridos, o seu carregador é de fato resiliente.

• A retomada de arquivo continua a partir do byte não baixado, e não do zero.
• O carregador lida corretamente com o caso em que o servidor ignora o cabeçalho Range.
• O checkpoint é salvo após cada página processada, e não apenas no final.
• O checkpoint é gravado atomicamente via arquivo temporário e renomeação.
• Uma nova execução não cria duplicatas no conjunto final.
• A deduplicação não cresce em memória linearmente com o número de linhas.
• Os workers paralelos não perdem as tarefas que falharam e as repetem.
• Após uma longa pausa, o token é renovado e o cursor expirado é substituído pela retomada por ID.

Como testar

1. Execute uma exportação completa de um conjunto pequeno e anote o número de registros.
2. Execute de novo no mesmo conjunto e confirme que o número não mudou.
3. Interrompa a exportação em pontos diferentes: no início, no meio, perto do fim.
4. Após cada interrupção, reinicie e verifique que o resultado é o mesmo.

Indicadores de sucesso. O número de registros únicos é estável entre execuções. O consumo de memória não cresce descontroladamente. A retomada sempre continua a partir do ponto salvo. Não há arquivos corrompidos após a retomada.

Erros típicos e suas soluções

Problema: o arquivo não abre após a retomada. Causa: os dados foram escritos no modo ab, embora o servidor tenha respondido com o código 200 e entregado o arquivo inteiro. Solução: verifique o status da resposta; em caso de 200, mude para a sobrescrita completa do arquivo do zero.

Problema: após uma queda, a exportação começa da primeira página. Causa: o checkpoint era salvo apenas no final do trabalho ou não era salvo. Solução: salve o checkpoint após cada página processada, logo após a gravação dos dados.

Problema: o checkpoint não pode ser lido, o JSON está corrompido. Causa: o programa caiu durante a gravação diretamente no arquivo de destino. Solução: grave em um arquivo temporário e substitua atomicamente via os.replace.

Problema: há duplicatas no conjunto final. Causa: não há chave de deduplicação ou ela é instável. Solução: defina uma PRIMARY KEY em uma chave confiável e use INSERT OR IGNORE.

Problema: o processo cai por falta de memória em grandes volumes. Causa: todas as chaves vistas são mantidas em um conjunto na memória. Solução: transfira a verificação de unicidade para o banco ou aplique um filtro de Bloom.

Problema: após uma longa pausa, as requisições retornam erro de autorização. Causa: o token ou a sessão expiraram durante a pausa. Solução: renove o token e crie uma nova sessão a cada início.

Problema: após a pausa, alguns registros foram pulados ou duplicados. Causa: a retomada foi feita pelo número da página, e os dados na fonte mudaram. Solução: retome pelo identificador do último registro, e não por offset.

Problema: os workers paralelos perdem parte dos dados. Causa: vários workers gravavam no mesmo checkpoint e sobrescreviam uns aos outros. Solução: grave os resultados em banco com transações, e não em um arquivo de estado compartilhado.

Recursos adicionais e otimização

Gravação em lote

Não grave no banco linha por linha. Junte um lote de várias centenas de registros e insira tudo de uma vez via executemany. Isso acelera a gravação em grandes volumes várias vezes.

Commit periódico

Chame commit não a cada gravação, mas a cada poucas centenas de linhas. Commit frequente demais deixa o banco lento. Raro demais arrisca perder mais dados em caso de queda. Encontre o equilíbrio para a sua carga.

Relatório de progresso

Adicione uma estimativa do tempo restante. Sabendo a velocidade de processamento das páginas e o número total de registros, você calcula quanto ainda falta esperar. Isso é útil em exportações longas.

Armazenamentos separados de dados brutos e processados

Mantenha as respostas brutas separadas dos registros processados. Se mais tarde você mudar a lógica de interpretação, não vai precisar baixar os dados de novo. Basta reprocessar as respostas brutas com o novo parser.

Dica: Configure o proxy Proxeon para que as conexões sejam estáveis durante toda a exportação. Uma conexão estável reduz o número de quedas, o que significa que o seu carregador entra em retomada com menos frequência e trabalha mais rápido.

FAQ: perguntas frequentes sobre exportação resiliente

Como saber se o servidor suporta retomada de arquivo? Envie uma requisição HEAD e observe o cabeçalho Accept-Ranges. O valor bytes indica suporte. A ausência do cabeçalho ou o valor none indicam que a retomada é impossível.

O que fazer se a API não fornece cursor, apenas páginas? Salve o número da página e, se possível, o identificador do último registro. Retome preferencialmente pelo ID, porque os números de página se deslocam quando os dados mudam.

Com que frequência salvar o checkpoint? Após cada página processada e gravada com sucesso. Assim, em caso de queda, você perde no máximo uma página de trabalho, e não a exportação inteira.

É possível deduplicar sem banco de dados? Em volumes pequenos, sim, com um conjunto comum na memória. Em milhões de linhas, isso é perigoso por causa da memória. Melhor usar um banco com PRIMARY KEY ou um filtro de Bloom.

O que escolher como chave de deduplicação? O ID único natural da fonte, se existir. Se não, uma combinação de campos estáveis. Em último caso, o hash de todo o registro com ordenação de chaves.

Por que a retomada por ID é mais confiável do que por número de página? Porque os dados na fonte podem mudar. Novos registros deslocam as páginas, e por número você vai pular ou duplicar dados. O ID não depende disso.

Quantos workers paralelos configurar? Comece com um número pequeno e aumente, observando a estabilidade e os limites da fonte. Paralelismo excessivo prejudica mais do que ajuda.

Como armazenar a string de conexão do proxy com segurança? Em uma variável de ambiente, e não no código. Leia-a via os.environ. Assim a senha não vai para o sistema de controle de versão.

O que fazer se o cursor expirou durante uma longa pausa? Capture o erro de cursor inválido e mude para a retomada pelo identificador do último registro salvo.

É preciso verificar a integridade do arquivo baixado? Sim. Compare o tamanho real do arquivo com o cabeçalho Content-Length. Se o servidor fornecer uma soma de verificação, verifique-a também.

Conclusão: o que você sabe agora e para onde ir

Você percorreu o caminho de uma exportação frágil, que desmorona na primeira queda, até um carregador resiliente. Agora você tem todas as ferramentas para que uma queda deixe de ser uma catástrofe e se torne uma situação comum de trabalho.

O que você dominou. A retomada de arquivos via cabeçalho Range com verificação de Accept-Ranges. O salvamento de progresso via checkpoints atômicos. Idempotência no nível da chave de deduplicação. Deduplicação em milhões de linhas sem inflar a memória. Paralelismo com repetição das tarefas que falharam. Retomada após uma longa pausa com renovação de tokens e substituição de cursores expirados. E o mais importante, um esqueleto pronto em Python que une tudo isso.

O que fazer depois. Pegue sua fonte de dados real e adapte a função de obtenção de página. Comece com um volume pequeno, depure a retomada nas interrupções e depois escale. Configure um proxy Proxeon estável, para que as conexões sejam previsíveis durante toda a exportação.

Para onde evoluir. Estude mais a fundo a gravação em lote e os filtros de Bloom. Adicione monitoramento de progresso e estimativa de tempo. Separe o armazenamento de dados brutos e processados. Gradualmente

Sobre o autor

Roman Melnikov

Technical Writer and System Administrator

Experiência profissional: Technical writer and DevOps engineer with 9 years of experience. Created over 50 detailed guides on system configuration and administration. His instructions helped thousands of professionals successfully solve technical tasks. Popular author on Habr and YouTube.

Formação: Bauman Moscow State Technical University. Information Systems and Technologies

Especialização:

Technical Documentation DevOps System Administration Linux Docker and Kubernetes CI/CD Infrastructure Automation Cloud Technologies System Monitoring Bash and Python Scripting