Buscando API com solicitações.Sessão

Question

Escrevi um script que analisa a API na programação (terça a sábado), baixando tudo do dia anterior.

---

import requests
import pandas as pd
from datetime import date, timedelta

# # This is what I'd normally use, but since there would be no data today, 
# # I assign specific date myself 
# DATE = (date.today() - timedelta(days=1)).strftime("%Y-%m-%d")
DATE = "2020-10-23"
URL = "https://spending.gov.ua/portal-api/v2/api/transactions/page/" 


def fetch(session, params):
    next_page, last_page = 0, 0
    while next_page <= last_page:
        params["page"] = next_page
        data = session.get(URL, params=params).json()
        yield pd.json_normalize(data.get("transactions"))\
                .assign(page=params.get("page"))
        next_page, last_page = next_page+1, data["count"] // data["pageSize"]
                
        
def fetch_all():
    with requests.Session() as session:
        params = {"page": 0, "pageSize": 100, "startdate": DATE, "enddate": DATE}
        yield from fetch(session, params)
        
        
if __name__ == "__main__":
    data = fetch_all()
    pd.concat(data).to_csv(f"data/{DATE}.csv", index=False)

---

Estou pensando em algumas coisas.

Em primeiro lugar , se estou usando requests.Sessioncorretamente.

Li na documentação que:

O objeto Session permite que você mantenha certos parâmetros nas solicitações. ... Portanto, se você estiver fazendo várias solicitações ao mesmo host, a conexão TCP subjacente será reutilizada, o que pode resultar em um aumento significativo de desempenho.

Não tenho certeza se esse é o caso aqui, pois não notei nenhuma mudança no desempenho.

Em segundo lugar , se dividir o código em duas funções em vez de uma fosse uma boa ideia.

Aqui eu pensei que seria mais fácil de manter - a função subjacente fetchnão muda enquanto fetch_allpotencialmente poderia. Por exemplo, eu poderia alimentar um intervalo de datas em vez de uma única data, mudando fetch_allpara:

def fetch_all(date_range):
    with requests.Session() as session:
        for date in date_range:
            params = {"page": 0, "pageSize": 100, "startdate": date, "enddate": date}
            yield from fetch(session, params)

Além disso, o yielde yield from- poderiam ter usado .appende retornado uma lista. Não tenho certeza de qual abordagem é melhor.

Answer 1

Estou pensando em algumas coisas.

Em primeiro lugar , se estou usando requests.Sessioncorretamente.

Sim você é. Em uma de minhas outras análises , usar requests.Sessionda mesma maneira para iterar em uma API paginada quase reduziu pela metade o tempo total de execução.

Fiz alguns testes rápidos baixando as últimas 7 páginas (páginas 1625-1631) para "2020-10-23" e foi um pouco melhor do que fazer solicitações com requests.get:

• requests.get: 23,2 segundos
• requests.Session: 17,7 segundos

Em segundo lugar , se dividir o código em duas funções em vez de uma fosse uma boa ideia.

Acho que é bom dividi-lo em duas funções. Dito isso, tenho alguns comentários sobre as responsabilidades e interface de fetche como tirar melhor proveito de seus usos de yielde yield fromabaixo.

---

No geral, o código parece limpo e fácil de ler. Veja como acho que pode ser melhorado:

• Acho que todos os detalhes de baixo nível de como emitir solicitações para a API devem ser abstraídos do chamador de fetch. Ou seja, fetcha assinatura da função de deve ser semelhante a esta:

  def fetch(
      session: requests.Session,
      start_date: date,
      end_date: date,
      starting_page: int = 0,
      page_size: int = 100,
  ) -> Iterator[pd.DataFrame]:
      pass
  

  Portanto, agora, criar um apropriado paramsseria fetchresponsabilidade de, não fetch_allde. Observe também que start_datee end_datesão do tipo datetime.date, não str. Da mesma forma, fetch_allnão deve se preocupar com qual formato de serialização de string de data a API aceita; isso é fetchresponsabilidade.

• Dentro fetch, em vez de manter variáveis next_pagee last_pageem cada solicitação, acho que seria melhor calcular o número total de páginas (n) apenas uma vez com a primeira solicitação (página k) e, em seguida, usar um loop for para as páginas k + 1. n-1:

  def to_dataframe(json_data: Dict[str, Any], page: int) -> pd.DataFrame:
      return pd.json_normalize(json_data["transactions"]).assign(page=page)
  
  
  def fetch(
      session: requests.Session,
      start_date: date,
      end_date: date,
      starting_page: int = 0,
      page_size: int = 100,
  ) -> Iterator[pd.DataFrame]:
      params = {
          "startdate": start_date.isoformat(),
          "enddate": end_date.isoformat(),
          "page": starting_page,
          "pageSize": page_size,
      }
  
      data = session.get(URL, params=params).json()
      page_count = math.ceil(data["count"] / data["pageSize"])
      last_page = page_count - 1
      if starting_page > last_page:
          return
      print(f"{starting_page} / {last_page}")
      yield to_dataframe(data, starting_page)
  
      for page in range(starting_page + 1, page_count):
          params["page"] = page
          data = session.get(URL, params=params).json()
          print(f"{page} / {last_page}")
          yield to_dataframe(data, page)
  

  A desvantagem aqui é que há uma pequena duplicação de código porque a primeira solicitação é tratada de maneira um pouco diferente, mas agora delegamos a responsabilidade da iteração do número da página ao loop for.

• Recomendo adicionar um gancho de evento ao sessionobjeto para que ele sempre chame raise_for_status()o objeto de resposta. Isso garante que todas as solicitações feitas com a sessão aumentem requests.HTTPErrorse o servidor nos der uma resposta 4xx ou 5xx e nos impede de converter os .json()dados de uma resposta de erro em um dataframe:

  session.hooks["response"].append(
      lambda r, *args, **kwargs: r.raise_for_status()
  )
  

• Atualmente, o programa está combinando todos os dataframes na memória antes de exportá-los para um arquivo CSV. Para aproveitar a vantagem de fetch_allser um Iterator[pd.DataFrame], acho que seria melhor gravar cada dataframe no CSV imediatamente, de modo que não precisemos mantê-lo na memória por mais tempo do que o necessário:

  output_path = Path(f"data/{DATE}.csv")
  output_path.unlink(missing_ok=True)
  data = fetch_all()
  for i, dataframe in enumerate(data):
      write_header = True if i == 0 else False
      dataframe.to_csv(
          output_path, header=write_header, index=False, mode="a"
      )
  

---

Versão refatorada:

#!/usr/bin/env python3

import math
from datetime import date, timedelta
from pathlib import Path
from typing import Any, Dict, Iterator

import pandas as pd  # type: ignore
import requests

# # This is what I'd normally use, but since there would be no data today,
# # I assign specific date myself
# DATE = date.today() - timedelta(days=1)
DATE = date.fromisoformat("2020-10-23")
URL = "https://spending.gov.ua/portal-api/v2/api/transactions/page/"


def to_dataframe(json_data: Dict[str, Any], page: int) -> pd.DataFrame:
    return pd.json_normalize(json_data["transactions"]).assign(page=page)


def fetch(
    session: requests.Session,
    start_date: date,
    end_date: date,
    starting_page: int = 0,
    page_size: int = 100,
) -> Iterator[pd.DataFrame]:
    params = {
        "startdate": start_date.isoformat(),
        "enddate": end_date.isoformat(),
        "page": starting_page,
        "pageSize": page_size,
    }

    data = session.get(URL, params=params).json()
    page_count = math.ceil(data["count"] / data["pageSize"])
    last_page = page_count - 1
    if starting_page > last_page:
        return
    print(f"{starting_page} / {last_page}")
    yield to_dataframe(data, starting_page)

    for page in range(starting_page + 1, page_count):
        params["page"] = page
        data = session.get(URL, params=params).json()
        print(f"{page} / {last_page}")
        yield to_dataframe(data, page)


def fetch_all() -> Iterator[pd.DataFrame]:
    with requests.Session() as session:
        session.hooks["response"].append(
            lambda r, *args, **kwargs: r.raise_for_status()
        )
        yield from fetch(session, start_date=DATE, end_date=DATE)


if __name__ == "__main__":
    output_path = Path(f"data/{DATE}.csv")
    output_path.unlink(missing_ok=True)
    data = fetch_all()
    for i, dataframe in enumerate(data):
        write_header = True if i == 0 else False
        dataframe.to_csv(
            output_path, header=write_header, index=False, mode="a"
        )