-
-
Save turicas/3e3621d61415e3453cd03a1997f7473f to your computer and use it in GitHub Desktop.
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` |
@ibandim123 Você pode user os query parameters limit
e last_available_confirmed
, mas info aqui
Python é muito ruim de visualizar qualquer coisa, estou tentando consumir no React mas o CORS ta impedindo mesmo meu header estando assim:
Authorization: "Token <meutoken>",
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Headers": "*",
"User-Agent": "*",
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?
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
Consegui resolver aqui, abaixo segue as telas do Insomnia para ilustrar e ajudar a quem precisar.
Nesta parte, atenção ao uso de aspas duplas, foi um detalhe que penei aqui.