sidra-fetcher: Cliente Python para a API do IBGE/SIDRA

Biblioteca Python para buscar e processar dados e metadados das APIs oficiais do IBGE — Agregados v3 e SIDRA. Fornece acesso tipado via dataclasses a pesquisas, agregados, períodos, territórios, variáveis e classificações, com clientes síncrono e assíncrono.

Instalação

pip install sidra-fetcher

Com uv:

uv add sidra-fetcher

Requisitos: Python 3.12+

Interface de Linha de Comando (CLI)

O sidra-fetcher inclui uma interface de linha de comando para exploração rápida de metadados.

Uso Autônomo

# Listar todas as pesquisas
sidra-fetcher list pesquisas

# Listar agregados de uma pesquisa (ex: 73)
sidra-fetcher list agregados 73

# Ver metadados detalhados de um agregado (ex: 1612)
sidra-fetcher info 1612

# Listar períodos disponíveis
sidra-fetcher periods 1612

# Baixar TODOS os dados de um agregado (ver seção dedicada abaixo)
sidra-fetcher download 1705 --dry-run
sidra-fetcher download 1705 -o ./sidra_data

Integração com quantilica-cli

Se o quantilica-cli estiver instalado no mesmo ambiente, o sidra-fetcher será detectado automaticamente como um plugin:

quantilica sidra info 1612
quantilica sidra download 1705 -o ./sidra_data

Uso Rápido

from sidra_fetcher.fetcher import SidraClient

with SidraClient() as client:
    # Listar todas as pesquisas e seus agregados
    index = client.get_indice_pesquisas_agregados()
    print(index[0].nome, "->", index[0].agregados[0].nome)

    # Metadados completos do agregado 1705 (IPCA-15)
    agregado = client.get_agregado(1705)
    print(agregado.nome)
    print(f"Períodos: {len(agregado.periodos)}")
    print(f"Localidades: {len(agregado.localidades)}")

API Python

SidraClient (síncrono)

from sidra_fetcher.fetcher import SidraClient

client = SidraClient(timeout=60)

Todos os métodos abaixo também funcionam via context manager: with SidraClient() as client:.

get_indice_pesquisas_agregados()

Retorna todas as pesquisas com o índice de agregados aninhado.

index = client.get_indice_pesquisas_agregados()
# -> list[IndicePesquisaAgregados]

for pesquisa in index:
    print(pesquisa.id, pesquisa.nome)
    for agregado in pesquisa.agregados:
        print("  ", agregado.id, agregado.nome)

get_agregado_metadados(agregado_id)

Retorna os metadados completos de um agregado — variáveis, classificações, níveis territoriais e periodicidade.

meta = client.get_agregado_metadados(1705)
# -> Agregado

print(meta.id, meta.nome)
print(meta.periodicidade.frequencia)   # ex: "mensal"
print([v.nome for v in meta.variaveis])
print([c.nome for c in meta.classificacoes])

get_agregado_periodos(agregado_id)

Retorna os períodos disponíveis de um agregado, com metadados temporais analisados.

periodos = client.get_agregado_periodos(1705)
# -> list[Periodo]

for p in periodos:
    print(p.id, p.frequencia, p.data_inicio, p.data_fim)
    # ex: "202312  mensal  2023-12-01  2023-12-31"

Veja Análise de Períodos para a lista completa de campos do objeto Periodo.

get_agregado_localidades(agregado_id, localidades_nivel)

Retorna as localidades filtradas por um ou mais códigos de nível territorial (ex: "N1" para Brasil, "N3" para estados, "N6" para municípios).

localidades = client.get_agregado_localidades(1705, "N3")
# -> list[Localidade]

for loc in localidades:
    print(loc.id, loc.nome, loc.nivel.id)

get_agregado(agregado_id)

Método de conveniência: busca metadados, períodos e todas as localidades declaradas em uma única chamada.

agregado = client.get_agregado(1705)
# -> Agregado (com .periodos e .localidades preenchidos)

get_acervo(acervo)

Busca uma listagem de "acervo" (coleção). Use AcervoEnum para selecionar a coleção desejada.

from sidra_fetcher.agregados import AcervoEnum

assuntos = client.get_acervo(AcervoEnum.ASSUNTO)
variaveis = client.get_acervo(AcervoEnum.VARIAVEL)

Valores disponíveis no AcervoEnum:

Baixando Todos os Dados de um Agregado

A API SIDRA limita cada requisição ao endpoint /values a 100.000 valores (SIDRA_API_VALUES_LIMIT, produto das quantidades selecionadas em cada dimensão: localidades × variáveis × categorias por classificação × períodos). O sidra-fetcher descobre automaticamente todas as dimensões de um agregado (via metadados) e divide o download em quantas requisições forem necessárias para nunca ultrapassar esse limite — sem o usuário precisar saber de antemão o tamanho da tabela.

Isso é um recurso ad-hoc: basta um agregado_id arbitrário, ao contrário do sidra-sql, que exige um pipeline curado (fetch.toml) por tabela. A ordem de divisão (chunking) é, em prioridade: período → lote de localidades → variável → combinação de categorias de classificação — cada nível territorial do agregado (administrativo, especial, ibge) é baixado separadamente.

Planejar antes de baixar

from sidra_fetcher.fetcher import SidraClient
from sidra_fetcher.download import describe_download_plan

with SidraClient() as client:
    chunks = client.plan_dados_agregado(1705)
    resumo = describe_download_plan(chunks)
    print(resumo)
    # {'n_requests': 192, 'n_valores': 513792,
    #  'por_nivel': {'N1': {...}, 'N6': {...}}}

Cada item de chunks é um DownloadChunk(nivel_territorial, parametro, n_valores) — parametro é o Parametro de sidra_fetcher.sidra pronto para virar URL.

Baixar os dados

Três formas, dependendo do tamanho da tabela:

# Tabelas pequenas: mescla tudo em uma lista única (mantém 1 cabeçalho).
linhas = client.get_dados_agregado(1705)

# Tabelas grandes: um chunk por vez, sem bufferizar tudo em memória.
for chunk, linhas in client.iter_dados_agregado(1705):
    ...

# Grava em disco: um NDJSON + manifest (SHA-256) por nível territorial.
paths = client.download_dados_agregado(1705, "./sidra_data", max_workers=4)

download_dados_agregado grava dados_{nivel}.ndjson (uma linha JSON por registro — a primeira é o cabeçalho, mantido uma única vez mesmo quando o download exige múltiplos requests) e dados_{nivel}.ndjson.manifest.json (mesma convenção de proveniência de save_agregado/load_agregado) para cada nível territorial efetivamente baixado.

Todos os métodos aceitam restringir a seleção (por padrão, baixam tudo):

client.plan_dados_agregado(
    1705,
    niveis_territoriais=["N3", "N6"],   # padrão: todos os níveis do agregado
    variaveis=["355"],                   # padrão: todas
    periodos=["202301", "202302"],       # padrão: todos
    classificacoes={"315": ["7169"]},    # padrão: todas as categorias
)

Via CLI

# Só mostra o plano (requests e valores estimados por nível), sem baixar.
sidra-fetcher download 1705 --dry-run

# Baixa de fato, com concorrência e uma pequena pausa entre requests.
sidra-fetcher download 1705 -o ./sidra_data --niveis N3,N6 --max-workers 4 --delay 0.5

# Via quantilica-cli (mesmos argumentos)
quantilica sidra download 1705 -o ./sidra_data

AsyncSidraClient (assíncrono)

Equivalente assíncrono do SidraClient. Todos os métodos são corrotinas. get_agregado busca metadados e períodos concorrentemente via asyncio.gather. Também expõe plan_dados_agregado (mesmo planejamento de chunking, sem I/O extra) — o download em si (get_dados_agregado/iter_dados_agregado/download_dados_agregado) está disponível apenas no SidraClient síncrono.

import asyncio
from sidra_fetcher.fetcher import AsyncSidraClient

async def main():
    async with AsyncSidraClient() as client:
        index = await client.get_indice_pesquisas_agregados()
        agregado = await client.get_agregado(1705)
        print(len(agregado.periodos))

asyncio.run(main())

Análise de Períodos

get_agregado_periodos analisa as strings de período retornadas pela API e as converte em objetos Periodo estruturados, com frequência detectada e intervalo de datas calculado automaticamente.

Campos do objeto Periodo

Tipos de frequência

As constantes de frequência também são importáveis:

from sidra_fetcher.periodos import (
    FREQUENCIA_MENSAL,
    FREQUENCIA_TRIMESTRAL,
    FREQUENCIA_TRIMESTRE_MOVEL,
    FREQUENCIA_SEMESTRAL,
    FREQUENCIA_ANUAL,
    FREQUENCIA_PLURIANUAL,
    FREQUENCIA_NAO_RECONHECIDA,
)

Construtor de URL SIDRA

A classe Parametro constrói e analisa URLs de requisição SIDRA (/values).

from sidra_fetcher.sidra import Parametro, Formato, Precisao

params = Parametro(
    agregado="1705",
    territorios={"3": ["all"]},   # todos os estados
    variaveis=["4099"],            # taxa de desemprego
    periodos=["202301", "202302"],
    classificacoes={"2": ["6794"]},
    formato=Formato.A,
    decimais={"": Precisao.M},
)

print(params.url())
# https://apisidra.ibge.gov.br/values/t/1705/n3/all/v/4099/p/202301,202302/c2/6794/h/y/f/a/d/m

Parâmetros do Parametro

Analisar uma URL SIDRA existente

from sidra_fetcher.sidra import parameter_from_url, parse_url

url = "https://apisidra.ibge.gov.br/values/t/1705/n3/all/v/4099/p/all/h/y/f/a/d/m"

params = parameter_from_url(url)
print(params.agregado)     # "1705"
print(params.territorios)  # {"3": ["all"]}

parsed = parse_url(url)
print(parsed["aggregate"])    # "1705"
print(parsed["territories"])  # {"3": ["all"]}
print(parsed["periods"])      # ["all"]

Utilitários de Leitura e Achatamento

Salvar e carregar metadados de agregados

from sidra_fetcher.reader import save_agregado, load_agregado

# Salvar em JSON
save_agregado(agregado, "agregado_1705.json")

# Carregar do JSON (períodos são reanalisados automaticamente)
agregado = load_agregado("agregado_1705.json")

Achatar metadados para análise

flatten_aggregate_metadata transforma a estrutura hierárquica variável × classificação em uma sequência de dicts — uma linha por combinação única de variável + categoria:

from sidra_fetcher.reader import flatten_aggregate_metadata

raw_meta = client.get("https://servicodados.ibge.gov.br/api/v3/agregados/1705/metadados")

for row in flatten_aggregate_metadata(raw_meta):
    print(row["agregado"], row["D4N"], row.get("C5N"), row["MN"])

Cada linha contém:

flatten_surveys_metadata achata o índice de pesquisas em uma lista de dicts {pesquisa_id, pesquisa, agregado_id, agregado}:

from sidra_fetcher.reader import flatten_surveys_metadata

raw_index = client.get("https://servicodados.ibge.gov.br/api/v3/agregados")
rows = flatten_surveys_metadata(raw_index)

Utilitários de Tamanho e Estatísticas

from sidra_fetcher.stats import calculate_aggregate

stats = calculate_aggregate(agregado)
print(stats)
# {
#   "pesquisa_id": "...",
#   "agregado_id": 1705,
#   "n_localidades": 27,
#   "n_variaveis": 1,
#   "n_classificacoes": 1,
#   "n_dimensoes": 2,
#   "n_periodos": 84,
#   "period_size": 54,    # linhas por período
#   "total_size": 4536,   # total estimado de linhas
#   ...
# }

Estrutura do Projeto

src/sidra_fetcher/
├── __init__.py          — logger do pacote
├── agregados.py         — dataclasses e construtores de URL para a API Agregados
├── download.py          — planejamento (chunking) e download completo de um agregado
├── fetcher.py           — SidraClient e AsyncSidraClient
├── periodos.py          — análise de strings de período e detecção de frequência
├── reader.py            — parsers JSON → dataclass e achatamento de metadados
├── sidra.py             — construtor de URL SIDRA (Parametro) e parsers
└── stats.py             — estatísticas de tamanho e dimensões

Desenvolvimento

git clone https://github.com/Quantilica/sidra-fetcher.git
cd sidra-fetcher
uv sync --extra dev
uv run ruff check src/ tests/
uv run ruff format src/ tests/
uv run python -m unittest discover -v tests

Licença

MIT — veja LICENSE.