Получение API с помощью request.Session

Oct 26 2020

Я написал скрипт, который анализирует API по расписанию (вторник-суббота), скачивая все за предыдущий день.

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)

Здесь меня интересует пара вещей.

Во-первых , если я requests.Sessionправильно использую .

Я прочитал в документации, что:

Объект Session позволяет сохранять определенные параметры в запросах. ... Таким образом, если вы делаете несколько запросов к одному и тому же хосту, базовое TCP-соединение будет повторно использовано, что может привести к значительному увеличению производительности.

Я не уверен, так ли это, потому что я не заметил никаких изменений в производительности.

Во-вторых , было бы неплохо разделить код на две функции вместо одной.

Здесь я подумал, что будет проще поддерживать - основная функция fetchне меняется, хотя fetch_allпотенциально может. Например, я мог бы указать диапазон дат вместо одной даты, изменив его fetch_allна:

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)

Кроме того , yieldи yield from- мог бы использоваться .appendи возвратил список вместо. Не уверен, какой подход лучше.

Ответы

3 Setris Oct 27 2020 at 08:45

Здесь меня интересует пара вещей.

Во-первых , если я requests.Sessionправильно использую .

Да Вы. В одном из моих других обзоров использование requests.Sessionаналогичного способа для итерации разбитого на страницы API почти вдвое сократило общее время выполнения.

Я провел небольшое тестирование, загрузив последние 7 страниц (страницы 1625-1631) для «2020-10-23», и это было немного лучше, чем отправка запросов с помощью requests.get:

requests.get: 23,2 секунды
requests.Session: 17,7 секунды

Во-вторых , было бы неплохо разделить код на две функции вместо одной.

Я думаю, что это нормально, если он разделен на две функции. Тем не менее, у меня есть несколько комментариев об обязанностях и интерфейсе, fetchа также о том, как лучше воспользоваться вашим использованием yieldи yield fromниже.

В целом код выглядит чистым и легко читается. Вот как, я думаю, это можно улучшить:

Я думаю, что все низкоуровневые детали того, как отправлять запросы к API, должны быть абстрагированы от вызывающего fetch. То есть fetchподпись функции должна выглядеть примерно так:
```
def fetch(
    session: requests.Session,
    start_date: date,
    end_date: date,
    starting_page: int = 0,
    page_size: int = 100,
) -> Iterator[pd.DataFrame]:
    pass
```
Так что теперь создание подходящего paramsбыло бы fetchобязанностью России, а не России fetch_all. Также обратите внимание, что start_dateи end_dateотносятся к типу datetime.date, а не str. Точно так же fetch_allне следует беспокоиться о том, какой формат сериализации строки даты принимает API; это fetchответственность.

Внутри fetch, вместо того, чтобы поддерживать переменные next_pageи last_pageдля каждого запроса, я думаю, было бы лучше вычислить общее количество страниц (n) только один раз с первым запросом (страница k), а затем использовать цикл for для страниц k + 1 .. п-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)

Компромисс здесь в том, что есть небольшое дублирование кода, потому что первый запрос обрабатывается немного иначе, но теперь мы делегировали ответственность за итерацию номера страницы циклу for.