Skip to content

Instantly share code, notes, and snippets.

@turicas
Last active August 13, 2022 23:04
Show Gist options
  • Save turicas/3e3621d61415e3453cd03a1997f7473f to your computer and use it in GitHub Desktop.
Save turicas/3e3621d61415e3453cd03a1997f7473f to your computer and use it in GitHub Desktop.
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://api.brasil.io/v1/"
def __init__(self, auth_token):
self.__auth_token = auth_token
@property
def headers(self):
return {
"User-Agent": "python-urllib/brasilio-client-0.1.0",
}
@property
def api_headers(self):
data = self.headers
data.update({"Authorization": f"Token {self.__auth_token}"})
return data
def api_request(self, path, query_string=None):
url = urljoin(self.base_url, path)
if query_string:
url += "?" + urlencode(query_string)
request = Request(url, headers=self.api_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 = {}
url = next_page
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 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`
@fernandosoares
Copy link

Boa noite. Gente, o dataset de casos está ok? A propriedade CONFIRMED está muito acima do normal por UF. Estou fazendo algo errado? Como posso pegar o numero de casos do dia anterior?

Copy link

ghost commented Aug 13, 2022

Quando faço a requisição fica aparecendo esta menssagem: {'message': 'As credenciais de autenticação não foram fornecidas ou estão inválidas. Acesse https://brasil.io/auth/tokens-api/ para gerenciar suas chaves de acesso a API ou nosso blog post com o passo-a-passo da autenticação em https://blog.brasil.io/2020/10/31/nossa-api-sera-obrigatoriamente-autenticada/'}. sendo que eu já tenho uma chave

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment