Python async http request

Asynchronous HTTP requests in Python

In python, you can make HTTP request to API using the requests module
or native urllib3 module. However, requests and urllib3 are synchronous. It means that only one HTTP call can be made at a time in a single thread. Sometimes you have to make multiples HTTP call and synchronous code will perform baldy. To avoid this, you can use multi-threading or since python 3.4 asyncio module.

Test case

In order to show the difference of time between sync and async code, i made a script that read a file with 500 cities names and perform HTTP call to an API to retrieve information about location, population and so on from the city name.

Sync code performance

@timeit def fetch_all(cities): responses = [] with requests.session() as session: for city in cities: resp = session.get(f"https://geo.api.gouv.fr/communes?nom=city>&fields=nom,region&format=json&geometry=centr") responses.append(resp.json()) return responses 

Async code performance

async def fetch(session, url): """Execute an http call async Args: session: contexte for making the http call url: URL to call Return: responses: A dict like object containing http response """ async with session.get(url) as response: resp = await response.json() return resp async def fetch_all(cities): """ Gather many HTTP call made async Args: cities: a list of string Return: responses: A list of dict like object containing http response """ async with aiohttp.ClientSession() as session: tasks = [] for city in cities: tasks.append( fetch( session, f"https://geo.api.gouv.fr/communes?nom=city>&fields=nom,region&format=json&geometry=centr", ) ) responses = await asyncio.gather(*tasks, return_exceptions=True) return responses @timeit def run(cities): responses = asyncio.run(fetch_all(cities)) return responses 

Conclusion

As you can see, the async version is a lot faster than the sync version so if you run into a situation where your code is performing multiple I/O calls then you should consider concurrency to improve performance. However asynchronous version requires more work as you can see. If you want to see the threading version that works with the requests module and also see how to implement automatic retry and caching on your API call check out this tutorial.

Источник

Как я писал асинхронные веб-запросы на Python, или почему провайдер считает, что я бандит

На днях по работе потребовалось сделать утилиту, которая прямо вот из консоли ходит в апи нашего клауд сервиса и берет оттуда кое-какую информацию. Подробности что и зачем — вне этого рассказа. Принципиальный вопрос здесь другой — скорость. Скорость реально важна (порядок количества запросов — десятки и сотни). Потому что ждать — не кайф.
Здесь я хочу поделиться своим ресёрчем на тему запросов, как делать круто, а как нет. С примерами кода конечно. А так же рассказать, как я тупил.

Начнем, пожалуй, с классики

Последовательные синхронные запросы. Будем использовать всем известную либу requests и tqdm для красивого вывода в консоль. В качестве игрушечного примера выбрал первую-попавшуюся публичную апишку: https://catfact.ninja/ . Метрикой качества будет RPS (Request per second). Чем выше — тем соотвественно лучше.

import time import requests from tqdm import tqdm URL = 'https://catfact.ninja/' class Api: def __init__(self, url: str): self.url = url def http_get(self, path: str, times: int): content = [] for _ in tqdm(range(times), desc='Fetching data. ', colour='GREEN'): response = requests.get(self.url + path) content.append(response.json()) return content if __name__ == '__main__': N = 10 api = Api(URL) start_timestamp = time.time() print(api.http_get(path='fact/', times=N)) task_time = round(time.time() - start_timestamp, 2) rps = round(N / task_time, 1) print( f"| Requests: ; Total time: s; RPS: . |\n" )

Получаем следующий вывод в терминале:

Fetching data. 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 10/10 [00:06, , , , , , , , , ] | Requests: 10; Total time: 6.61 s; RPS: 1.5. | 

RPS — 1.5. Очень грустно. У меня еще и интернет не самый быстрый сейчас дома. Ну тут добавить нечего.

Что можно оптимизировать уже сейчас? Ответ: использовать requests.Session
Eсли делать несколько запросов к одному и тому же хосту, базовое TCP-соединение будет использоваться повторно, что приводит к значительному увеличению производительности. (цитата из документации requests)

Используем сессию

 def http_get_with_session(self, path: str, times: int): content = [] with requests.session() as session: for _ in tqdm(range(times), desc='Fetching data. ', colour='GREEN'): response = session.get(self.url + path) content.append(response.json()) return content

Немного изменив метод, и вызвав его, видим следующее:

Fetching data. 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 10/10 [00:02, , , , , , , , , ] | Requests: 10; Total time: 2.58 s; RPS: 3.9. | 

Почти 4 RPS, в сравнении с 1.5 уже прорыв.
Но не секрет, что для сокращения i/o time есть практика использования асихнронных/многопоточных программ. Это как раз такой случай, потому что во время ожидания ответа от сервера наша программа ничего не делает, хотя могла бы отправлять уже другой запрос, а потом другой и т.д. Попробуем реализовать асинхронный подход к решению кейса.

async / await

Для удобства вызовов сделаем функцию-оболочку:

def run_case(func, path, times): start_timestamp = time.time() asyncio.run(func(path, times)) task_time = round(time.time() - start_timestamp, 2) rps = round(times / task_time, 1) print( f"| Requests: ; Total time: s; RPS: . |\n" )

И собственно сама реализация метода (не забудьте поставить aiohttp, обычные реквесты не работают в асинхронной парадигме):

 async def async_http_get(self, path: str, times: int): async with aiohttp.ClientSession() as session: content = [] for _ in tqdm(range(times), desc='Async fetching data. ', colour='GREEN'): response = await session.get(url=self.url + path) content.append(await response.text(encoding='UTF-8')) return content

if __name__ == '__main__': N = 50 api = Api(URL) run_case(api.async_http_get, path='fact/', times=N)

Async fetching data. 100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 50/50 [00:12 

 async def async_gather_http_get(self, path: str, times: int): async with aiohttp.ClientSession() as session: tasks = [] for _ in tqdm(range(times), desc='Async gather fetching data. ', colour='GREEN'): tasks.append(asyncio.create_task(session.get(self.url + path))) responses = await asyncio.gather(*tasks) return [await r.json() for r in responses]

if __name__ == '__main__': N = 50 api = Api(URL) run_case(api.async_gather_http_get, path='fact/', times=N)

Async gather fetching data. 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 50/50 [00:00

| Requests: 200; Total time: 2.97 s; RPS: 67.3. |