Разработка приложений, использующих API сторонних сервисов

Раздел: Сетевые приложения -> Взаимодействие с внешними сервисами

Основы работы с API на Python

Как получить данные от внешнего сервиса с минимальными усилиями?

Наиболее эффективным решением для взаимодействия с REST API в Python является библиотека requests. Она предоставляет удобный интерфейс для отправки HTTP запросов и обработки ответов.

Пример базового GET запроса:


import requests

url = "https://api.github.com/users/octocat"
response = requests.get(url)
if response.status_code == 200:
    user_data = response.json()
    print(user_data['login'])
else:
    print(f"Ошибка: {response.status_code}")

Telegram python библиотека (библиотека python для telegram)

Пояснения:

  • Функция requests.get отправляет HTTP GET запрос.
  • Свойство status_code содержит код ответа сервера.
  • Метод .json() парсит ответ как JSON.

Типичная ошибка: ConnectionError возникает при отсутствии сети или неверном домене. Решение: обернуть запрос в try/except и проверять таймауты.

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

Библиотека httpx поддерживает как синхронные, так и асинхронные вызовы. Для установки: pip install httpx.

Синхронный пример:


import httpx

response = httpx.get('https://api.github.com/users/octocat')
data = response.json()
print(data['login'])

Python api программа (программа с использованием api на python)

Асинхронный пример с async/await:


import httpx
import asyncio

async def fetch_user():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.github.com/users/octocat')
        data = response.json()
        print(data['login'])

asyncio.run(fetch_user())

Пояснение: асинхронный клиент позволяет отправлять множество запросов параллельно без блокировки.

Проблема: при работе с httpx в асинхронном режиме необходимо управлять контекстным менеджером. Забытый async with вызовет ошибку времени выполнения.

Как организовать множество параллельных запросов с aiohttp?

Для высокопроизводительной асинхронной работы подходит библиотека aiohttp. Установка: pip install aiohttp.

Пример одновременного получения нескольких пользователей:


import aiohttp
import asyncio

async def fetch_user(session, url):
    async with session.get(url) as response:
        return await response.json()

async def main():
    urls = [f'https://api.github.com/users/{i}' for i in range(1, 6)]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_user(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for user in results:
            print(user.get('login'))

asyncio.run(main())

Пояснение: asyncio.gather запускает все задачи параллельно, что значительно ускоряет сбор данных.

Ошибка: ClientConnectorError возникает при попытке подключения к несуществующему хосту. Рекомендуется добавлять таймауты и обработку исключений.

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

Встроенная библиотека urllib.request позволяет отправлять HTTP запросы без сторонних зависимостей.

Пример GET запроса:


from urllib.request import urlopen
import json

url = "https://api.github.com/users/octocat"
with urlopen(url) as response:
    data = json.loads(response.read())
    print(data['login'])

Пояснение: urlopen возвращает файлоподобный объект. Для JSON используется модуль json. Недостаток: более громоздкая обработка ошибок и заголовков.

Частая проблема: отсутствие обработки HTTP ошибок (например, 404). Решение: использовать urlopen внутри try/except и проверять код ответа через метод getcode().

Как аутентифицироваться в API с использованием токенов?

Многие API требуют авторизации через Bearer токен. Пример с библиотекой requests:


import requests

headers = {
    'Authorization': 'Bearer YOUR_ACCESS_TOKEN'
}
response = requests.get('https://api.example.com/private', headers=headers)
if response.status_code == 200:
    print(response.json())
else:
    print(f"Авторизация не удалась: {response.status_code}")

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

Ошибка: 401 Unauthorized при неверном или просроченном токене. Решение: обновлять токен заранее и обрабатывать код 401 для автоматической переаутентификации.

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

Асинхронный сбор данных с пагинацией

Следующий пример показывает, как собрать все страницы с результатами из API GitHub, используя aiohttp и asyncio:

Пример 

import aiohttp
import asyncio

async def fetch_repos(session, page):
    url = f"https://api.github.com/users/octocat/repos?page={page}&per_page=10"
    async with session.get(url) as response:
        data = await response.json()
        return data

async def main():
    async with aiohttp.ClientSession() as session:
        # предположим, что на первой странице есть Link header для пагинации
        first_page = await fetch_repos(session, 1)
        all_repos = first_page
        # Если есть Link header с rel="next", продолжаем
        # Для простоты используем цикл по номерам страниц
        for page in range(2, 5):  # ограничим 4 страницами
            repos = await fetch_repos(session, page)
            if not repos:
                break
            all_repos.extend(repos)
        print(f"Собрано {len(all_repos)} репозиториев")

asyncio.run(main())
Собрано 40 репозиториев

Пояснение: пагинация часто реализуется через query-параметры page и per_page. Результат показывает суммарное количество записей.

Обработка rate limiting с повторными попытками

API могут ограничивать частоту запросов. Используем библиотеку tenacity для автоматических повторов:

Пример 

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type(requests.exceptions.HTTPError)
)
def fetch_with_retry(url):
    response = requests.get(url)
    if response.status_code == 429:
        raise requests.exceptions.HTTPError("Rate limit exceeded")
    response.raise_for_status()
    return response.json()

data = fetch_with_retry('https://api.example.com/data')
print(data)
{'key': 'value'}

Пояснение: декоратор @retry перехватывает исключения, ждёт с экспоненциальной задержкой и повторяет запрос до 5 раз.

Загрузка файла на сервер через multipart/form-data

Отправка изображения на сервер:

Пример 

import requests

url = 'https://httpbin.org/post'
files = {'file': ('photo.jpg', open('photo.jpg', 'rb'), 'image/jpeg')}
response = requests.post(url, files=files)
print(response.json()['files']['file'])
'... (содержимое файла в base64)'

Пояснение: параметр files принимает словарь, где ключ - имя поля, значение - кортеж (имя файла, объект файла, MIME-тип).

Работа с GraphQL API

GraphQL требует отправки POST запроса с телом, содержащим query:

Пример 

import requests

url = 'https://api.github.com/graphql'
headers = {'Authorization': 'Bearer YOUR_TOKEN'}
query = """
{
  user(login: "octocat") {
    name
    repositories(first: 3) {
      nodes {
        name
      }
    }
  }
}
"""
response = requests.post(url, json={'query': query}, headers=headers)
data = response.json()
print(data['data']['user']['name'])
The Octocat

Пояснение: GraphQL эндпоинт принимает JSON с полем query. Ответ содержит вложенную структуру.

Подключение к WebSocket API

Для работы в реальном времени используется библиотека websockets:

Пример 

import asyncio
import websockets

async def listen():
    async with websockets.connect('wss://echo.websocket.org') as ws:
        await ws.send('Привет, сервер!')
        response = await ws.recv()
        print(response)

asyncio.run(listen())
Привет, сервер!

Пояснение: WebSocket позволяет устанавливать двустороннее соединение. Пример использует эхо сервер для теста.

Программа с использованием API на Python - comments

En
Python api программа (python)