API-клиент на Python: выбор инструмента и практические примеры
Основы работы с HTTP в Python
Для взаимодействия с веб-API в Python существует несколько библиотек. Каждая из них подходит для разных сценариев: от простых запросов до асинхронных высоконагруженных систем. Ниже рассмотрены основные варианты с примерами кода и пояснениями.
Как быстро и удобно отправлять HTTP-запросы с помощью requests?
Библиотека requests - де-факто стандарт для синхронных HTTP-запросов в Python. Она предоставляет интуитивный API, автоматическую обработку JSON, сессии и многое другое.
Пример GET-запроса с параметрами
import requests
url = 'https://api.github.com/search/repositories'
params = {'q': 'requests', 'sort': 'stars'}
response = requests.get(url, params=params)
print(response.status_code)
data = response.json()
print(data['items'][0]['full_name'])
Api client python (api-клиент на python)
Здесь params автоматически преобразуются в строку запроса. Метод json() парсит ответ. Статус-код позволяет проверить успешность запроса.
POST-запрос с JSON
import requests
url = 'https://jsonplaceholder.typicode.com/posts'
payload = {'title': 'foo', 'body': 'bar', 'userId': 1}
response = requests.post(url, json=payload)
print(response.json())
Python get https (https-запросы в python)
Параметр json автоматически сериализует словарь и устанавливает заголовок Content-Type: application/json.
Обработка ошибок и таймауты
import requests
from requests.exceptions import ConnectionError, Timeout
try:
response = requests.get('https://httpbin.org/delay/5', timeout=3)
response.raise_for_status()
except ConnectionError:
print('Не удалось подключиться к серверу')
except Timeout:
print('Таймаут запроса')
except requests.exceptions.HTTPError as e:
print(f'Ошибка HTTP: {e}')
Python get api (получение данных через api в python)
raise_for_status() генерирует исключение для кодов 4xx/5xx. Таймаут предотвращает зависание программы.
Использование сессий для сохранения кук и заголовков
import requests
session = requests.Session()
session.headers.update({'User-Agent': 'MyApp/1.0'})
session.get('https://httpbin.org/cookies/set?name=value')
response = session.get('https://httpbin.org/cookies')
print(response.json())
Сессии автоматически сохраняют куки между запросами. Это удобно для работы с API, требующими аутентификации.
Типичные проблемы: отсутствие проверки статуса, игнорирование таймаутов, утечка соединений при отсутствии менеджера контекста. Решение - использовать with session: или всегда закрывать сессию.
Как обойтись без сторонних библиотек, используя urllib?
Встроенный модуль urllib не требует установки, но имеет более громоздкий API.
GET-запрос с urllib.request
import urllib.request
import urllib.parse
url = 'https://api.github.com/search/repositories?' + urllib.parse.urlencode({'q': 'urllib'})
req = urllib.request.Request(url, headers={'User-Agent': 'Python'})
with urllib.request.urlopen(req, timeout=5) as response:
data = response.read().decode('utf-8')
print(response.status)
print(data[:200])
Необходимо вручную кодировать параметры, читать ответ и декодировать. Для JSON требуется json.loads().
POST-запрос с данными
import urllib.request
import json
url = 'https://jsonplaceholder.typicode.com/posts'
data = json.dumps({'title': 'foo', 'body': 'bar', 'userId': 1}).encode('utf-8')
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
with urllib.request.urlopen(req) as response:
print(response.read().decode())
Проблемы: больше кода, нет автоматической обработки ошибок, нет сессий, сложнее работа с куками. Для простых скриптов может быть оправдано, но в серьёзных проектах лучше использовать requests.
Как получить асинхронный клиент с современным API, используя httpx?
Библиотека httpx поддерживает как синхронный, так и асинхронный режимы, а также HTTP/2.
Синхронный запрос
import httpx
with httpx.Client() as client:
response = client.get('https://httpbin.org/get', params={'key': 'value'})
print(response.status_code, response.json())
Асинхронный запрос
import httpx
import asyncio
async def fetch():
async with httpx.AsyncClient() as client:
response = await client.get('https://httpbin.org/get')
print(response.json())
asyncio.run(fetch())
httpx поддерживает время ожидания, повторные попытки, прокси. Синтаксис похож на requests, что упрощает миграцию.
Ошибка: при асинхронном использовании легко забыть await или не использовать менеджер контекста. Важно также устанавливать таймауты, иначе приложение может зависнуть.
Как создать асинхронный клиент для высоконагруженных приложений с aiohttp?
aiohttp - мощная асинхронная библиотека, часто используемая с asyncio для тысяч одновременных запросов.
Простой GET-запрос
import aiohttp
import asyncio
async def main():
async with aiohttp.ClientSession() as session:
async with session.get('https://httpbin.org/get') as resp:
print(resp.status)
data = await resp.json()
print(data)
asyncio.run(main())
POST с JSON
import aiohttp
import asyncio
async def main():
async with aiohttp.ClientSession() as session:
async with session.post('https://jsonplaceholder.typicode.com/posts', json={'title': 'test'}) as resp:
print(await resp.json())
asyncio.run(main())
aiohttp требует явного управления сессией и ответом через async with. Поддерживает пул соединений, куки, таймауты.
Проблемы: сложнее отладка асинхронного кода, возможны блокировки при смешении синхронного и асинхронного. Также aiohttp может быть избыточен для простых проектов.
Как работать с HTTP на низком уровне, используя http.client?
Модуль http.client - встроенный, позволяет управлять каждым аспектом запроса, но требует много ручного кода.
Пример GET-запроса
import http.client
conn = http.client.HTTPSConnection('api.github.com')
conn.request('GET', '/search/repositories?q=python', headers={'User-Agent': 'Python'})
response = conn.getresponse()
print(response.status, response.reason)
data = response.read().decode()
print(data[:200])
conn.close()
Нужно вручную создавать соединение, отправлять запрос, читать ответ. Нет автоматического парсинга JSON, нет сессий.
Ошибки: утечка соединений при отсутствии close(), сложность работы с куками и переадресацией. Используется редко, только когда требуется полный контроль.
Расширенные примеры использования API-клиентов
В этом разделе представлены более сложные сценарии, которые могут понадобиться при работе с реальными API.
Пагинация с requests
Многие API возвращают данные постранично. Пример для GitHub API, который использует ссылки в заголовке Link.
import requests
url = 'https://api.github.com/search/repositories?q=python&per_page=2'
response = requests.get(url, headers={'User-Agent': 'MyApp'})
data = response.json()
print('Первая страница:', len(data['items']))
# Извлекаем ссылку на следующую страницу из заголовка Link
link_header = response.headers.get('Link')
if link_header:
# Формат: ; rel="next"
import re
match = re.search(r'<(https?://[^>]+)>; rel="next"', link_header)
if match:
next_url = match.group(1)
response2 = requests.get(next_url, headers={'User-Agent': 'MyApp'})
data2 = response2.json()
print('Вторая страница:', len(data2['items']))
Первая страница: 2 Вторая страница: 2
Автоматические повторные попытки при сбоях (retry)
Использование requests.adapters.HTTPAdapter для встроенной логики повторения.
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('http://', adapter)
session.mount('https://', adapter)
try:
response = session.get('https://httpbin.org/status/500', timeout=5)
print(response.status_code)
except requests.exceptions.RetryError:
print('Исчерпаны попытки')
(после 3 попыток будет исключение RetryError или успех, если сервер ответит 200)
Загрузка файла с помощью POST
Отправка файла с использованием multipart/form-data (например, загрузка на сервер).
import requests
url = 'https://httpbin.org/post'
with open('example.txt', 'w') as f:
f.write('Hello, world!')
with open('example.txt', 'rb') as f:
files = {'file': ('example.txt', f, 'text/plain')}
response = requests.post(url, files=files)
print(response.json()['files'])
{'file': 'Hello, world!'}
Асинхронные запросы с aiohttp и ограничением параллелизма
Выполнение множества запросов с контролем числа одновременных соединений.
import aiohttp
import asyncio
async def fetch(session, url):
async with session.get(url) as resp:
return await resp.text()
async def main():
urls = [f'https://httpbin.org/delay/1' for _ in range(10)]
connector = aiohttp.TCPConnector(limit=3) # не более 3 параллельных соединений
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks)
print(f'Получено {len(results)} ответов')
asyncio.run(main())
Получено 10 ответов
Работа с куками в urllib
Использование http.cookiejar для сохранения кук.
import urllib.request
from http.cookiejar import CookieJar
cookie_jar = CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookie_jar))
opener.open('https://httpbin.org/cookies/set?name=value')
response = opener.open('https://httpbin.org/cookies')
print(response.read().decode())
{
"cookies": {
"name": "value"
}
}
Обработка ошибок и логирование в httpx
Настройка логирования всех запросов с помощью встроенного механизма событий.
import httpx
import logging
logging.basicConfig(level=logging.INFO)
client = httpx.Client(event_hooks={'request': [lambda request: logging.info(f'Request: {request.method} {request.url}')]})
response = client.get('https://httpbin.org/get')
print(response.status_code)
INFO:root:Request: GET https://httpbin.org/get 200