Делаем HTTP-запросы на Python с библиотеками requests, aiohttp и httpx

Многие сайты и сервисы умеют отвечать на запросы других приложений: отдавать погоду, количество товаров в магазине, популярные книги за неделю и что угодно ещё, если это предусмотрели разработчики сайтов. Но чтобы сайт нам что-то ответил, нужно уметь написать запрос, отправить его и потом знать, что с ним делать. Сегодня разберёмся со всем этим с помощью Python: посмотрим, какие есть библиотеки для этого и как с ними работать. Пригодится всем, кто хочет брать данные с сайтов напрямую без посредников.

Если вы пока не писали код на Python, почитайте статьи про то, как начать:

Как работают запросы и ответы по протоколу HTTP

Когда мы смотрим что-то в браузере, в этот момент наше устройство отправляет запросы на сервер. 

После того как мы набираем адрес и нажимаем Enter, браузер переделывает запрос в длинную строку, из которой браузер и сервер получают необходимую информацию: что хочет сделать клиент (то есть ваше устройство), на какой сервер послать запрос, что нужно учесть при выполнении. После этого сервер присылает соответствующий ответ.

Такое взаимодействие называется общением по HTTP-протоколу — Hyper Text Transfer Protocol. Сегодня разберём самые основы, а в следующий раз поговорим более детально.

Когда эта технология появилась, клиент мог только получать информацию от сервера. Сейчас возможностей стало больше, соответственно, видов HTTP-запросов тоже стало больше. Мы будем использовать эти:

На запросы сервер всегда выдаёт код статуса. Это трёхзначное число от 100 до 599, по которому можно понять, что произошло во время обработки запроса:

Ошибки на сервере показывают числа от 500 до 599.

Что будем делать сегодня

Мы возьмём три библиотеки, с которыми на Python можно отправлять HTTP-запросы: 

Мы напишем запросы с помощью каждой и посмотрим, как они станут работать. Для теста будем использовать бесплатный API postman-echo.com. Это специальный API для тренировки, поэтому в реальности там ничего не создаётся и не удаляется. Но ответы от сервера при правильно написанном коде будут такие, как будто всё сработало.

В чём отличие синхронных запросов от асинхронных. Синхронные запросы блокируют весь код. На время их выполнения программа останавливается и ждёт ответ от сервера. Асинхронные запросы требуют времени на запуск и после этого отдают управление программе. Поэтому можно запустить несколько запросов почти одновременно.

Чтобы засекать время работы с разными библиотеками, мы добавим во все скрипты встроенный в Python модуль time.

Устанавливаем библиотеки

Перед добавлением в скрипт нужно скачать все модули в командной строке или в среде разработки. Для установки в командной строке напишите:

pip install requests
pip install aiohttp
pip install httpx

Если вы привыкли всё устанавливать в среде разработки, нужно перейти во вкладку терминала и выполнить эти же команды там:

Создаём запросы с requests 

Создаём новый Python-файл и первым делом импортируем зависимости — саму библиотеку и модуль time:

# импортируем модуль time, чтобы можно было засекать время
import time
# импортируем библиотеку requests для запросов
import requests

У нас есть URL-адрес нашего API: https://postman-echo.com. Он понадобится для составления запросов, поэтому положим этот адрес в переменную-константу, которая обозначается заглавными буквами:

# пишем основной адрес, куда будем слать запросы
BASE_URL = "https://postman-echo.com"

Запросы выглядят как разные методы, например requests.get(). В скобках нужно указать URL, по которому выполняется запрос. Если мы хотим добавить или изменить информацию, вторым аргументом нужно написать, что именно добавляем — так мы будем делать в запросах POST и PUT. 

Ещё по желанию можно указать время таймаута в секундах. Тогда, если запрос будет выполняться слишком медленно и превысит указанное время, программа закроет его. Нам эта возможность сегодня не нужна, поэтому для GET-запроса мы укажем только адрес. Для этого к основному адресу нужно добавить строку /get. Можно набрать этот запрос в строке браузера и увидеть ответ от сервера в формате JSON:

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

Так выглядит GET-запрос:

# пишем функцию GET-запроса
def fetch_get():
   # объявляем переменную, в которой отправляем GET-запрос на нужный адрес
   response = requests.get(f"{BASE_URL}/get")
   # возвращаем статус-код от сервера и ответ в формате JSON
   return response.status_code, response.json()

Для запросов POST и PUT нужно указать JSON-данные, которые хотим добавить или обновить на сервере. Положим их в отдельные переменные data_to_post и data_to_put.

Ещё в POST и PUT другая переменная response: мы используем методы post() и put() и добавляем к основному адресу строки /post и /put. В остальном функции для запросов будут выглядеть так же:

# пишем функцию POST-запроса
def fetch_post():
   # объявляем переменную и указываем, что нужно добавить
   data_to_post = {"key": "value"}
   # объявляем переменную, в которой отправляем POST-запрос на нужный адрес
   response = requests.post(f"{BASE_URL}/post", json=data_to_post)
   # возвращаем статус-код от сервера и ответ в формате JSON
   return response.status_code, response.json()


# пишем функцию PUT-запроса
def fetch_put():
   # объявляем переменную и указываем, что нужно обновить
   data_to_put = {"key": "updated_value"}
   # объявляем переменную, в которой отправляем PUT-запрос на нужный адрес
   response = requests.put(f"{BASE_URL}/put", json=data_to_put)
   # возвращаем статус-код от сервера и ответ в формате JSON
   return response.status_code, response.json()

Метод DELETE отправляется на адрес данных, которые нужно удалить. URL-адрес может включать точный идентификатор для удаления. Такой запрос — https://example.com/users/1 — удалит первого пользователя в списке users. 

Наш API — для тестирования, поэтому у нас будет указан только специальный адрес для нужного ответа: https://postman-echo.com/del.

Так выглядит целиком:

# пишем функцию DELETE-запроса
def fetch_delete():
   # объявляем переменную, в которой отправляем DELETE-запрос на нужный адрес
   response = requests.delete(f"{BASE_URL}/delete")
   # возвращаем статус-код от сервера и ответ в формате JSON
   return response.status_code, response.json()

Все эти запросы мы запустим в основной функции. Для каждого выведем ответ и статус-код от сервера, а перед началом и после окончания зафиксируем время выполнения всех запросов.

Скрипт запросов с requests целиком:

# импортируем модуль time, чтобы можно было засекать время
import time
# импортируем библиотеку requests для запросов
import requests

# пишем основной адрес, куда будем слать запросы
BASE_URL = "https://postman-echo.com"


# пишем функцию GET-запроса
def fetch_get():
   # объявляем переменную, в которой отправляем GET-запрос на нужный адрес
   response = requests.get(f"{BASE_URL}/get")
   # возвращаем статус-код от сервера и ответ в формате JSON
   return response.status_code, response.json()


# пишем функцию POST-запроса
def fetch_post():
   # объявляем переменную и указываем, что нужно добавить
   data_to_post = {"key": "value"}
   # объявляем переменную, в которой отправляем POST-запрос на нужный адрес
   response = requests.post(f"{BASE_URL}/post", json=data_to_post)
   # возвращаем статус-код от сервера и ответ в формате JSON
   return response.status_code, response.json()


# пишем функцию PUT-запроса
def fetch_put():
   # объявляем переменную и указываем, что нужно обновить
   data_to_put = {"key": "updated_value"}
   # объявляем переменную, в которой отправляем PUT-запрос на нужный адрес
   response = requests.put(f"{BASE_URL}/put", json=data_to_put)
   # возвращаем статус-код от сервера и ответ в формате JSON
   return response.status_code, response.json()


# пишем функцию DELETE-запроса
def fetch_delete():
   # объявляем переменную, в которой отправляем DELETE-запрос на нужный адрес
   response = requests.delete(f"{BASE_URL}/delete")
   # возвращаем статус-код от сервера и ответ в формате JSON
   return response.status_code, response.json()


# пишем основную функцию
def main():
   # начинаем отсчёт времени
   start = time.perf_counter()

   # выполняем GET-запрос
   status_code, response_json = fetch_get()
   # выводим код ответа от сервера и содержимое ответа
   print(f"GET Status Code: {status_code}")
   print(f"GET Response: {response_json}")
   # выполняем POST-запрос
   status_code, response_json = fetch_post()
   # выводим код ответа от сервера и содержимое ответа
   print(f"POST Status Code: {status_code}")
   print(f"POST Response: {response_json}")
   # выполняем PUT-запрос
   status_code, response_json = fetch_put()
   # выводим код ответа от сервера и содержимое ответа
   print(f"PUT Status Code: {status_code}")
   print(f"PUT Response: {response_json}")
   # выполняем DELETE-запрос
   status_code, response_json = fetch_delete()
   # выводим код ответа от сервера и содержимое ответа
   print(f"DELETE Status Code: {status_code}")
   print(f"DELETE Response: {response_json}")

   # останавливаем отсчёт времени и выводим результат,
   # округляя до двух цифр после запятой
   end = time.perf_counter()
   print(f"Time taken: {end - start:.2f} seconds.")


# запускаем основную функцию
main()

Посмотрим на результат:

Объединяем запросы в один сеанс

В нашем коде есть проблема: для каждого запроса мы открываем соединение, а после ответа сервера закрываем. Если объединить все запросы одним сеансом, получится быстрее.

У библиотеки requests есть объект requests.Session — он создаёт пул соединений, в котором сохраняются параметры запросов и временные файлы cookie. 

Чтобы всё работало, объект нужно передать в функции как аргумент. Но вызывать эти функции мы будем позже, поэтому сейчас нужно только указать, что аргумент будет объектом Session. Для этого указываем аргумент с любым названием и через двоеточие — тип аргумента:

def fetch_post(session: requests.Session) -> Any:
   data_to_post = {"key": "value"}
   response = session.post(f"{BASE_URL}/post", json=data_to_post)
   return response.status_code, response.json()

В основной функции мы создадим контекстный менеджер — конструкцию с ключевым словом with. Контекстный менеджер выполняет дополнительный код до и после основного кода внутри себя.

Такой код создаст единый пул запросов, а потом закроет соединение с сервером:

with requests.Session() as session:
   # выполняем GET-запрос
   status_code, response_json = fetch_get(session)
   # выполняем POST-запрос
   status_code, response_json = fetch_post(session)
   # выполняем PUT-запрос
   status_code, response_json = fetch_put(session)
   # выполняем DELETE-запрос
   status_code, response_json = fetch_delete(session)

Проверим выполнение и посмотрим на время:

Создаём запросы с помощью aiohttp

Даже с единым пулом соединений код в requests выполняется последовательно: каждый новый запрос выполняется только после ответа от предыдущего. Библиотека aiohttp даёт возможность отправить все запросы асинхронно, не дожидаясь ответов.

Асинхронные функции называются корутинами и получают в Python тип объекта не function, а coroutine. Для работы с ними понадобится встроенный модуль asyncio. Чтобы сделать из функции корутину, нужны две вещи:

Пул соединений на aiohttp создаётся похожим с requests образом. В корутины в качестве аргумента мы передаём объект aiohttp.ClientSession. Функции запросов будут немного отличаться, в каждой из них мы будем использовать асинхронный контекстный менеджер. Он выглядит так же, как обычный, только перед with добавляется слово async. 

Асинхронная функция для GET-запроса:

# пишем функцию GET-запроса
async def fetch_get(session: aiohttp.ClientSession):
   # открываем сеанс, отправляем запрос
   async with session.get(f"{BASE_URL}/get") as response:
       # возвращаем ответ от сервера
       return await response.json()

Чтобы отправить запросы разом, в asyncio есть несколько вариантов. Мы используем метод asyncio.gather(): он запустит все переданные в скобках задачи самым быстрым путём.

async with aiohttp.ClientSession() as session:
   # создаём список, куда последовательно сохраняем все ответы
   results = await asyncio.gather(
       # перечисляем асинхронные функции, которые нужно запустить
       fetch_get(session),
       fetch_post(session),
       fetch_put(session),
       fetch_delete(session)
   )

Ответы из функций вернутся в виде одного списка — упорядоченной коллекции элементов. Каждый ответ мы будем выводить по его номеру, начиная с 0:

print("GET:", results[0])
print("POST:", results[1])
print("PUT:", results[2])
print("DELETE:", results[3])

Весь код для создания запросов на aiohttp:

# импортируем библиотеку асинхронных запросов aiohttp
import aiohttp
# подключаем модуль для асинхронного кода asyncio
import asyncio
# импортируем модуль time, чтобы можно было засекать время
import time

# пишем основной адрес, куда будем слать запросы
BASE_URL = "https://postman-echo.com"


# пишем функцию GET-запроса
async def fetch_get(session: aiohttp.ClientSession):
   # открываем сеанс, отправляем запрос
   async with session.get(f"{BASE_URL}/get") as response:
       # возвращаем ответ от сервера
       return await response.json()


# пишем функцию POST-запроса
async def fetch_post(session: aiohttp.ClientSession):
   # создаём запись, которую хотим отправить
   data_to_post = {"key": "value"}
   # открываем сеанс, отправляем запрос
   async with session.post(f"{BASE_URL}/post", json=data_to_post) as response:
       # возвращаем ответ от сервера
       return await response.json()


# пишем функцию PUT-запроса
async def fetch_put(session: aiohttp.ClientSession):
   # создаём запись, на которую хотим заменить существующую запись
   data_to_put = {"key": "updated_value"}
   # открываем сеанс, отправляем запрос
   async with session.put(f"{BASE_URL}/put", json=data_to_put) as response:
       # возвращаем ответ от сервера
       return await response.json()


# пишем функцию DELETE-запроса
async def fetch_delete(session: aiohttp.ClientSession):
   # открываем сеанс, отправляем запрос
   async with session.delete(f"{BASE_URL}/delete") as response:
       # возвращаем ответ от сервера
       return await response.json()


# пишем основную функцию
async def main():
   # начинаем отсчёт времени
   start = time.perf_counter()

   # собираем все наши запросы и запускаем все почти одновременно: все асинхронные
   # функции запускаются друг за другом, не дожидаясь выполнения предыдущей
   async with aiohttp.ClientSession() as session:
       # создаём список, куда последовательно сохраняем все ответы
       results = await asyncio.gather(
           # перечисляем асинхронные функции, которые нужно запустить
           fetch_get(session),
           fetch_post(session),
           fetch_put(session),
           fetch_delete(session)
       )

       # выводим результаты каждого запроса на экран
       print("GET:", results[0])
       print("POST:", results[1])
       print("PUT:", results[2])
       print("DELETE:", results[3])

   # останавливаем отсчёт времени и выводим результат,
   # округляя до двух цифр после запятой
   end = time.perf_counter()
   print(f"Time taken: {end - start:.2f} seconds.")


# запускаем основную функцию
asyncio.run(main())

Проверим:

Создаём запросы с помощью httpx

Популярность библиотеки requests вызвана двумя вещами:

Работать с aiohttp сложнее, хотя асинхронные запросы с ней работают быстрее.

Httpx работает так же просто, как requests, при этом с ней тоже можно делать асинхронные запросы. В чём будут отличия от requests в нашем случае:

response = await client.get(f"{BASE_URL}/get")

Код скрипта запросов:

# подключаем модуль для асинхронного кода asyncio
import asyncio
# импортируем модуль time, чтобы можно было засекать время
import time
# импортируем библиотеку запросов httpx
import httpx

# пишем основной адрес, куда будем слать запросы
BASE_URL = "https://postman-echo.com"


# пишем функцию GET-запроса
async def fetch_get(client: httpx.AsyncClient):
   # объявляем переменную, в которой отправляем асинхронный GET-запрос на нужный адрес
   response = await client.get(f"{BASE_URL}/get")
   # возвращаем ответ с сервера в формате JSON
   return response.json()


# пишем функцию POST-запроса
async def fetch_post(client: httpx.AsyncClient):
   # объявляем переменную и указываем, что нужно добавить
   data_to_post = {"key": "value"}
   # объявляем переменную, в которой отправляем асинхронный POST-запрос на нужный адрес
   response = await client.post(f"{BASE_URL}/post", json=data_to_post)
   # возвращаем ответ с сервера в формате JSON
   return response.json()


# пишем функцию PUT-запроса
async def fetch_put(client: httpx.AsyncClient):
   # объявляем переменную и указываем, что нужно обновить
   data_to_put = {"key": "updated_value"}
   # объявляем переменную, в которой отправляем асинхронный PUT-запрос на нужный адрес
   response = await client.put(f"{BASE_URL}/put", json=data_to_put)
   # возвращаем ответ с сервера в формате JSON
   return response.json()


# пишем функцию DELETE-запроса
async def fetch_delete(client: httpx.AsyncClient):
   # объявляем переменную, в которой отправляем асинхронный DELETE-запрос на нужный адрес
   response = await client.delete(f"{BASE_URL}/delete")
   # возвращаем ответ с сервера в формате JSON
   return response.json()


# пишем основную функцию
async def main():
   # начинаем отсчёт времени
   start = time.perf_counter()

   # собираем все наши запросы и запускаем все почти одновременно: все асинхронные
   # функции запускаются друг за другом, не дожидаясь выполнения предыдущей
   async with httpx.AsyncClient() as client:
       tasks = [
           fetch_get(client),
           fetch_post(client),
           fetch_put(client),
           fetch_delete(client),
       ]
       # создаём список, куда последовательно сохраняем все ответы
       results = await asyncio.gather(*tasks)

   # выводим результаты каждого запроса на экран
   print("GET:", results[0])
   print("POST:", results[1])
   print("PUT:", results[2])
   print("DELETE:", results[3])

   # останавливаем отсчёт времени и выводим результат,
   # округляя до двух цифр после запятой
   end = time.perf_counter()
   print(f"Time taken: {end - start:.2f} seconds.")


# запускаем основную функцию
asyncio.run(main())

Запускаем и смотрим на результат:

И всё-таки, что в итоге лучше?

Может показаться, что разница в несколько секунд между синхронными и асинхронными запросами незначительная. Но выгода становится ощутимее с количеством запросов: с каждым последовательным запросом время работы возрастает, а асинхронные запросы обрабатываются почти одновременно. 

Сто последовательных запросов будут работать в сто раз медленнее, чем один. А на сто асинхронных запросов понадобится примерно столько же времени, сколько на один — при условии, что все они будут выполняться на разных системах.

При написании запросов нужно запомнить несколько вещей:

👉 Короче, простой ответ такой: всё зависит от асинхронности и того, насколько она вам важна.