Практическое руководство по работе с API на языке Python

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

import requests
headers = {'Authorization': 'Bearer YOUR_TOKEN'}
response = requests.get('https://api.example.com/protected', headers=headers)
print(response.json())

{'id': 1, 'name': 'John'}

Аутентификация через Bearer токен: заголовок Authorization передаёт токен, сервер проверяет его.

import requests
import json
payload = {'username': 'test', 'password': 'secret'}
response = requests.post('https://api.example.com/login', json=payload)
if response.status_code == 200:
    token = response.json().get('token')
    print('Token:', token)
else:
    print('Error:', response.status_code)

Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

Отправка JSON-данных через POST и извлечение токена из ответа.

import requests
response = requests.get('https://api.example.com/large-file', stream=True)
with open('output.txt', 'wb') as f:
    for chunk in response.iter_content(chunk_size=8192):
        f.write(chunk)
print('File saved')

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

from concurrent.futures import ThreadPoolExecutor
import requests
urls = ['https://api.example.com/1', 'https://api.example.com/2', 'https://api.example.com/3']
def fetch(url):
    return requests.get(url).json()
with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(fetch, urls))
print(results)

[{'id': 1}, {'id': 2}, {'id': 3}]

Параллельные запросы с использованием пула потоков ускоряют получение данных из нескольких источников.

from tenacity import retry, stop_after_attempt, wait_exponential
import requests
@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_data(url):
    response = requests.get(url)
    response.raise_for_status()
    return response.json()
data = fetch_data('https://api.example.com/unstable')
print(data)

Библиотека tenacity реализует повторные попытки с экспоненциальной задержкой, что удобно для нестабильных сервисов.

import requests
url = 'https://api.example.com/upload'
files = {'file': open('example.txt', 'rb')}
response = requests.post(url, files=files)
print(response.status_code, response.json())

Загрузка файла через multipart/form-data с помощью параметра files.

import requests
params = {'page': 1, 'limit': 10, 'sort': 'name'}
response = requests.get('https://api.example.com/items', params=params)
print(response.url)
print(response.json())

https://api.example.com/items?page=1&limit=10&sort=name
[{'name': 'A'}, {'name': 'B'}]

Параметры запроса передаются через словарь params, автоматически формируя строку запроса.

import http.client
conn = http.client.HTTPSConnection("api.example.com")
headers = {'Authorization': 'Bearer token', 'Accept': 'application/json'}
conn.request("GET", "/data", headers=headers)
response = conn.getresponse()
data = response.read()
print(data.decode())
conn.close()

Низкоуровневое использование http.client для максимального контроля.

import asyncio
import aiohttp
async def fetch(session, url, retries=2):
    for attempt in range(retries):
        try:
            async with session.get(url) as resp:
                if resp.status == 200:
                    return await resp.json()
                else:
                    print(f"Attempt {attempt+1} failed with status {resp.status}")
        except aiohttp.ClientError as e:
            print(f"Error: {e}")
    return None
async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, f'https://api.example.com/data/{i}') for i in range(5)]
        results = await asyncio.gather(*tasks)
        print(results)
asyncio.run(main())

Асинхронный сбор данных с обработкой ошибок и повторными попытками внутри каждой задачи.

import httpx
cert = ("/path/to/cert.pem", "/path/to/key.pem")
response = httpx.get('https://api.example.com/secure', cert=cert)
print(response.text)

Использование клиентских сертификатов в httpx для аутентификации по TLS.