Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Exemplo de código para acessar dados do Brasil.IO
import csv
import gzip
import io
import json
from urllib.parse import urlencode, urljoin
from urllib.request import Request, urlopen
class BrasilIO:
base_url = "https://brasil.io/api/"
def __init__(self, auth_token):
self.__auth_token = auth_token
@property
def headers(self):
return {
"User-Agent": "python-urllib/brasilio-client-0.1.0",
"Authentication": f"Token {self.__auth_token}",
}
def request(self, path, query_string=None):
url = urljoin(self.base_url, path)
if query_string is not None:
url += "?" + urlencode(query_string)
request = Request(url, headers=self.headers)
response = urlopen(request)
return json.load(response)
def data(self, dataset_slug, table_name, filters=None):
url = f"dataset/{dataset_slug}/{table_name}/data/"
filters = filters or {}
filters["page"] = 1
finished = False
while not finished:
response = self.request(url, filters)
next_page = response.get("next", None)
for row in response["results"]:
yield row
filters["page"] += 1
finished = next_page is None
def download(self, dataset, table_name):
url = f"https://data.brasil.io/dataset/{dataset}/{table_name}.csv.gz"
request = Request(url, headers=self.headers)
response = urlopen(request)
return response
if __name__ == "__main__":
api = BrasilIO("meu-api-token")
dataset_slug = "covid19"
table_name = "caso_full"
# Para baixar o arquivo completo:
# Após fazer o download, você salvá-lo no disco ou percorrer o arquivo em
# memória. Para salvá-lo no disco:
response = api.download(dataset_slug, table_name)
with open(f"{dataset_slug}_{table_name}.csv.gz", mode="wb") as fobj:
fobj.write(response.read())
# TODO: o código acima pode ser melhorado de forma a não utilizar
# `response.read()` para não colocar todo oarquivo em memória e sim fazer
# streaming da resposta HTTP e salvar cada chunk diretamente no `fobj`.
# Caso queira percorrer o CSV em memória:
response = api.download(dataset_slug, table_name)
fobj = io.TextIOWrapper(gzip.GzipFile(fileobj=response), encoding="utf-8")
reader = csv.DictReader(fobj)
for row in tqdm(reader):
pass # faça algo com `row`
# Para navegar pela API:
filters = {"state": "PR", "is_last": True}
data = api.data(dataset_slug, table_name, filters)
for row in data:
pass # faça algo com `row`
@ramongss

This comment has been minimized.

Copy link

@ramongss ramongss commented Jul 6, 2020

Tentando ajudar e fazendo minha contribuição para esse projeto 📊.

Traduzi o código para R:

download_brasilio_table <- function(dataset, table_name){
  url <- sprintf("https://data.brasil.io/dataset/%s/%s.csv.gz", dataset, table_name)
  tmp <- tempfile()
  download.file(url, tmp)
  response <- read.csv(gzfile(tmp), encoding = "UTF-8")
  unlink(tmp)
  return(response)
}

# Passe o nome da tabela para a funcao, como "caso", "caso_full", "obito_cartorio":
data <- download_brasilio_table("covid19", "caso_full")

Quem quiser propor melhorias, esse código em R também está aqui.

Parabéns para Álvaro e todos os colabores 👏👏👏.

@fccoelho

This comment has been minimized.

Copy link

@fccoelho fccoelho commented Jul 9, 2020

estou pegando um erro forbidden ao tentar baixar o https://data.brasil.io/dataset/covid19/caso_full.csv.gz direto do Pandas, mas via wget funciona normalmente

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.