Mengambil API dengan request.Session

Question

Saya menulis skrip yang mem-parsing API sesuai jadwal (Selasa-Sabtu), mengunduh semuanya untuk hari sebelumnya.

---

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)

---

Di sini saya bertanya-tanya tentang beberapa hal.

Pertama , jika saya menggunakan requests.Sessiondengan benar.

Saya membaca di dokumentasi bahwa:

Objek Session memungkinkan Anda mempertahankan parameter tertentu di seluruh permintaan. ... Jadi jika Anda membuat beberapa permintaan ke host yang sama, koneksi TCP yang mendasarinya akan digunakan kembali, yang dapat menghasilkan peningkatan kinerja yang signifikan.

Saya tidak yakin apakah itu yang terjadi di sini karena saya tidak melihat adanya perubahan dalam kinerja.

Kedua , jika memecah kode menjadi dua fungsi, bukan satu adalah ide yang bagus.

Di sini saya pikir akan lebih mudah untuk mempertahankannya - fungsi yang mendasarinya fetchtidak berubah sementara fetch_allberpotensi bisa. Misalnya, saya bisa memberi makan rentang tanggal alih-alih tanggal menghanguskan, mengubah fetch_allmenjadi:

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)

Dan juga, yielddan yield from- bisa saja menggunakan .appenddan mengembalikan daftar sebagai gantinya. Tidak yakin pendekatan mana yang lebih baik.

Answer 1

Di sini saya bertanya-tanya tentang beberapa hal.

Pertama , jika saya menggunakan requests.Sessiondengan benar.

Ya, kamu. Dalam salah satu ulasan saya yang lain , menggunakan requests.Sessioncara yang sama untuk iterasi pada API paginasi hampir mengurangi separuh waktu eksekusi total.

Saya melakukan beberapa pengujian cepat dengan mengunduh 7 halaman terakhir (halaman 1625-1631) untuk "2020-10-23" dan hasilnya sedikit lebih baik daripada membuat permintaan dengan requests.get:

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

Kedua , jika memecah kode menjadi dua fungsi, bukan satu adalah ide yang bagus.

Saya pikir tidak masalah untuk memecahnya menjadi dua fungsi. Karena itu, saya memiliki beberapa komentar tentang tanggung jawab dan antarmuka fetchdan cara memanfaatkan penggunaan Anda di bawah yielddan yield fromdi bawah dengan lebih baik.

---

Secara keseluruhan kodenya terlihat bersih dan mudah dibaca. Inilah cara saya pikir itu dapat ditingkatkan:

• Saya pikir semua detail tingkat rendah tentang cara mengeluarkan permintaan ke API harus disarikan dari pemanggil fetch. Artinya, fetchtanda tangan fungsi akan terlihat seperti ini:

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

  Jadi sekarang membuat yang sesuai paramsakan menjadi fetchtanggung jawab, bukan fetch_all. Perhatikan juga bahwa start_datedan end_dateadalah tipe datetime.date, bukan str. Demikian pula, fetch_alltidak perlu khawatir dengan format serialisasi string tanggal yang diterima API; ini adalah fetchtanggung jawab.

• Di dalam fetch, daripada mempertahankan variabel next_pagedan last_pagepada setiap permintaan, saya pikir akan lebih baik untuk menghitung jumlah total halaman (n) hanya sekali dengan permintaan pertama (halaman k), kemudian gunakan loop for untuk halaman 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)
  

  Imbalannya di sini adalah ada duplikasi kecil kode karena permintaan pertama ditangani sedikit berbeda, tetapi sekarang kami telah mendelegasikan tanggung jawab perulangan nomor halaman ke loop for.

• Saya merekomendasikan menambahkan event hook ke sessionobjek sehingga selalu memanggil raise_for_status()objek respons. Ini memastikan bahwa semua permintaan yang dibuat dengan sesi meningkat requests.HTTPErrorjika server memberi kami respons 4xx atau 5xx, dan mencegah kami mengonversi .json()data respons kesalahan menjadi kerangka data:

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

• Saat ini program menggabungkan semua dataframe di memori sebelum mengekspornya ke file CSV. Untuk mengambil keuntungan dari fetch_allmenjadi Iterator[pd.DataFrame], saya pikir akan lebih baik untuk menulis setiap dataframe ke CSV segera, jadi kita tidak perlu terus dalam memori lebih lama dari yang diperlukan:

  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"
      )
  

---

Versi refactored:

#!/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"
        )