Работа с API в языке Python: инструменты и подходы

Раздел: Прочее -> API и запросы

Введение в API запросы на Python

API (Application Programming Interface) позволяет программам обмениваться данными. В Python существует несколько библиотек для выполнения HTTP-запросов. Наиболее популярное и эффективное решение для синхронных запросов - библиотека requests. Она предоставляет удобный интерфейс для работы с методами GET, POST, PUT, DELETE и обработки ответов.

Базовое решение: библиотека requests

Установка: pip install requests

import requests

# GET запрос
response = requests.get('https://api.github.com/users/octocat')
print(response.status_code)  # 200
print(response.json()['login'])  # octocat

Api python (api в python)

Пояснение: response.status_code содержит код ответа, response.json() десериализует JSON в словарь Python. Типичная проблема: игнорирование проверки статуса. Решение: использовать response.raise_for_status().

Частая ошибка: потеря соединения при большом количестве запросов

При выполнении множества запросов без повторного использования соединения может возникнуть ошибка ConnectionError. Используйте объект Session:

with requests.Session() as session:
    for i in range(10):
        resp = session.get('https://httpbin.org/get')
        print(resp.status_code)

Сессия автоматически управляет пулом соединений и cookie.

POST запрос с данными

payload = {'key1': 'value1', 'key2': 'value2'}
response = requests.post('https://httpbin.org/post', data=payload)
print(response.json())

Для отправки JSON используйте параметр json:

import json
data = {'username': 'test', 'password': 'pass'}
response = requests.post('https://httpbin.org/post', json=data)
print(response.json()['json'])

Как сделать асинхронные запросы для ускорения работы?

Вариант A: httpx - современная замена requests с поддержкой async/await

Установка: pip install httpx. httpx поддерживает как синхронный, так и асинхронный режимы.

import httpx
async def fetch(url):
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.status_code

# Запуск асинхронно
import asyncio
async def main():
    status = await fetch('https://httpbin.org/get')
    print(status)
asyncio.run(main())

При работе с асинхронным кодом часто забывают использовать await, что приводит к предупреждению или ошибке. Всегда проверяйте, что вызовы HTTP сопровождаются await внутри асинхронной функции.

Как выполнять запросы без сторонних библиотек?

Вариант B: urllib - встроенная библиотека

from urllib.request import urlopen
from urllib.parse import urlencode

params = urlencode({'q': 'python'})
url = f'https://httpbin.org/get?{params}'
with urlopen(url) as response:
    data = response.read()
    print(data.decode('utf-8')[:100])

Недостаток: больше кода для обработки ошибок и JSON. Типичная проблема: нужно закрывать соединение вручную, но контекстный менеджер решает это.

Как сделать параллельные асинхронные запросы с максимальной производительностью?

Вариант C: aiohttp - асинхронная библиотека для высоконагруженных систем

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.json()

async def main():
    urls = ['https://api.github.com/users/octocat'] * 5
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(r['login'])

asyncio.run(main())

Важно: создавать одну сессию для всех запросов, иначе производительность упадет. Типичная ошибка: не закрывать сессию - используйте async with.

Расширенные примеры работы с API

Пример 1: Пагинация через requests с обработкой ссылок

Пример 
import requests

def fetch_all_pages(base_url):
    results = []
    url = base_url
    while url:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        results.extend(data.get('items', []))
        # Извлечение URL следующей страницы из заголовка Link
        link_header = response.headers.get('Link', '')
        if 'rel="next"' in link_header:
            # Парсинг простого формата ; rel="next"
            parts = link_header.split(',')
            for part in parts:
                if 'rel="next"' in part:
                    url = part.split(';')[0].strip(' <>')
                    break
        else:
            url = None
    return results

# Пример GitHub API
items = fetch_all_pages('https://api.github.com/search/repositories?q=python&per_page=2')
print(f'Всего получено: {len(items)} репозиториев')
Всего получено: 10 репозиториев (зависит от количества)

Пример 2: Отправка файла через multipart/form-data с httpx

Пример 
import httpx

# Создаем тестовый файл
with open('test.txt', 'w') as f:
    f.write('Hello World')

async def upload_file():
    async with httpx.AsyncClient() as client:
        files = {'file': open('test.txt', 'rb')}
        response = await client.post('https://httpbin.org/post', files=files)
        print(response.json()['files'])

import asyncio
asyncio.run(upload_file())
{'file': 'Hello World'}

Пример 3: Обработка rate limiting с повторными попытками (retry)

Пример 
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()
retries = Retry(total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retries)
session.mount('https://', adapter)
session.mount('http://', adapter)

try:
    response = session.get('https://httpbin.org/status/429')
    print(response.status_code)
except requests.exceptions.RetryError as e:
    print('Исчерпаны попытки:', e)
После 3 попыток с задержкой будет выброшено исключение RetryError

Пример 4: Асинхронные запросы с aiohttp и обработка ошибок

Пример 
import aiohttp
import asyncio

async def fetch_with_retry(session, url, retries=3):
    for attempt in range(retries):
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                if resp.status == 200:
                    return await resp.json()
                else:
                    print(f'Попытка {attempt+1}: статус {resp.status}')
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            print(f'Попытка {attempt+1}: ошибка {e}')
            await asyncio.sleep(0.5 * (attempt + 1))
    return None

async def main():
    urls = ['https://api.github.com/users/octocat', 'https://httpbin.org/delay/5']
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_retry(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for r in results:
            if r:
                print(r.get('login') or r)

asyncio.run(main())
octocat
None (из-за таймаута) или данные из /delay/5 при успехе

Пример 5: Использование urllib с OAuth2 авторизацией (код без библиотек)

Пример 
from urllib.request import Request, urlopen
from urllib.parse import urlencode

# Предположим, у нас есть токен доступа
token = 'your_access_token'
headers = {'Authorization': f'Bearer {token}'}
req = Request('https://api.example.com/data', headers=headers, method='GET')
with urlopen(req) as resp:
    data = resp.read()
    print(data.decode()[:200])
Ответ в JSON формате (если сервер поддерживает)

При использовании urllib необходимо вручную обрабатывать кодировку и ошибки HTTP, что увеличивает объем кода.

API в Python - comments

En
Api python (python)