REST API в Python: техники взаимодействия с внешними сервисами

pip install requests

# Создание виртуального окружения
python -m venv api_env

# Активация (Windows)
api_env\Scripts\activate

# Активация (Linux/MacOS)
source api_env/bin/activate

# Установка необходимых библиотек
pip install requests python-dotenv requests-oauthlib

import requests

# Базовый GET-запрос
response = requests.get('https://api.github.com/user/repos')

# Проверка статус-кода
if response.status_code == 200:
# Вывод результата
print(response.json())
else:
print(f"Ошибка: {response.status_code}")

# Данные для отправки
new_item = {
"name": "Python API Project",
"description": "REST API integration example",
"public": True
}

# POST-запрос с JSON-данными
response = requests.post(
'https://api.example.com/items',
json=new_item
)

# Проверка успешного создания
if response.status_code == 201:
print(f"Создан элемент с ID: {response.json().get('id')}")

# PUT обычно заменяет весь ресурс
updated_data = {"name": "Updated Name", "description": "New description"}
response = requests.put(
'https://api.example.com/items/42',
json=updated_data
)

# PATCH частично обновляет ресурс
partial_update = {"name": "Modified Name"}
response = requests.patch(
'https://api.example.com/items/42',
json=partial_update
)

# DELETE-запрос
response = requests.delete('https://api.example.com/items/42')

# Проверка успешного удаления
if response.status_code == 204:
print("Ресурс успешно удален")

# Пример с использованием всех параметров
response = requests.get(
'https://api.example.com/search',
params={'q': 'python', 'limit': 10},
headers={'Authorization': 'Bearer YOUR_TOKEN'},
timeout=5,
allow_redirects=True,
verify=True
)

import requests

response = requests.get('https://api.github.com/users/python')
user_data = response.json()

# Доступ к данным как к словарю
print(f"Имя: {user_data['name']}")
print(f"Подписчики: {user_data['followers']}")
print(f"Репозитории: {user_data['public_repos']}")

# Пример сложной структуры
response = requests.get('https://api.github.com/repos/python/cpython/commits')
commits = response.json()

# Извлечение информации из вложенных объектов
for commit in commits[:5]: # Первые 5 коммитов
author = commit['commit']['author']['name']
message = commit['commit']['message'].split('\n')[0] # Первая строка сообщения
date = commit['commit']['author']['date']
print(f"{date} – {author}: {message}")

# Поиск репозиториев с заданными параметрами
params = {
'q': 'language:python',
'sort': 'stars',
'order': 'desc',
'per_page': 10,
'page': 1
}

response = requests.get('https://api.github.com/search/repositories', params=params)
repos = response.json()

print(f"Найдено {repos['total_count']} репозиториев")
for repo in repos['items']:
print(f"{repo['name']} – ⭐ {repo['stargazers_count']}")

# Безопасное извлечение данных с проверкой наличия ключей
def get_safe_value(data, keys, default=None):
"""
Безопасно извлекает значение из вложенной структуры данных

Args:
data (dict): исходный словарь
keys (list): список ключей для последовательного доступа
default: значение по умолчанию, если путь не существует

Returns:
Значение по указанному пути или значение по умолчанию
"""
current = data
for key in keys:
if isinstance(current, dict) and key in current:
current = current[key]
else:
return default
return current

# Использование
user_location = get_safe_value(user_data, ['location'], 'Не указано')
company_name = get_safe_value(user_data, ['company'], 'Не указано')

def get_all_pages(url, params=None):
"""Получает все страницы данных из API с пагинацией"""
if params is None:
params = {}

all_results = []
page = 1

while True:
params['page'] = page
response = requests.get(url, params=params)

if response.status_code != 200:
break

page_results = response.json()

# Проверка, пуст ли ответ или закончились страницы
if not page_results or (isinstance(page_results, list) and len(page_results) == 0):
break

all_results.extend(page_results)
page += 1

return all_results

# Отправка данных формы
form_data = {
'username': 'python_user',
'password': 'secure_password'
}
response = requests.post('https://api.example.com/login', data=form_data)

# Отправка JSON-данных
json_data = {
'user': {
'username': 'python_user',
'email': 'user@example.com',
'preferences': {
'theme': 'dark',
'notifications': True
}
}
}
response = requests.post('https://api.example.com/users', json=json_data)

import pandas as pd

# Получение данных через API
response = requests.get('https://api.example.com/data')
data = response.json()

# Преобразование в DataFrame для анализа
df = pd.DataFrame(data)

# Базовый анализ
print(df.describe())

# Фильтрация
filtered_data = df[df['value'] > 100]

# Агрегация
summary = df.groupby('category').agg({'value': ['mean', 'sum', 'count']})

import requests
from requests.auth import HTTPBasicAuth

response = requests.get(
'https://api.example.com/data',
auth=HTTPBasicAuth('username', 'password')
)

# Альтернативный синтаксис
response = requests.get(
'https://api.example.com/data',
auth=('username', 'password')
)

# Через параметр запроса
params = {
'api_key': 'your_api_key_here',
'query': 'data'
}
response = requests.get('https://api.example.com/search', params=params)

# Через заголовок
headers = {
'X-API-Key': 'your_api_key_here'
}
response = requests.get('https://api.example.com/data', headers=headers)

from requests_oauthlib import OAuth2Session

# Настройка OAuth2 клиента
client_id = 'your_client_id'
client_secret = 'your_client_secret'
redirect_uri = 'https://your-app.com/callback'

# Создание сессии OAuth
oauth = OAuth2Session(client_id, redirect_uri=redirect_uri)

# Получение URL авторизации
authorization_url, state = oauth.authorization_url('https://api.service.com/oauth/authorize')
print(f'Перейдите по ссылке для авторизации: {authorization_url}')

# После авторизации пользователь будет перенаправлен с кодом авторизации
authorization_response = input('Введите полный URL после перенаправления: ')

# Обмен кода на токен
token = oauth.fetch_token(
'https://api.service.com/oauth/token',
authorization_response=authorization_response,
client_secret=client_secret
)

# Использование токена для доступа к ресурсам
response = oauth.get('https://api.service.com/resource')

import requests
import time
import logging
from requests.exceptions import RequestException, HTTPError, Timeout, ConnectionError

logging.basicConfig(level=logging.INFO, format='%(asctime)s – %(levelname)s – %(message)s')
logger = logging.getLogger(__name__)

def api_request_with_error_handling(url, method='get', max_retries=3, **kwargs):
"""
Выполняет запрос к API с комплексной обработкой ошибок

Args:
url (str): URL для запроса
method (str): HTTP-метод ('get', 'post', 'put', 'delete')
max_retries (int): Максимальное число повторных попыток
**kwargs: Дополнительные параметры для requests

Returns:
dict: JSON-ответ или None в случае неустранимой ошибки
"""
retry_count = 0

while retry_count < max_retries:
try:
# Выбор HTTP-метода
request_method = getattr(requests, method.lower())
response = request_method(url, **kwargs)

# Проверка статус-кода
response.raise_for_status()

# Возвращаем данные при успехе
return response.json()

except ConnectionError as e:
# Сетевые ошибки
retry_count += 1
wait_time = 2 ** retry_count
logger.warning(f"Ошибка соединения: {e}. Повтор {retry_count}/{max_retries} через {wait_time} сек.")
time.sleep(wait_time)

except Timeout as e:
# Тайм-ауты
retry_count += 1
wait_time = 2 ** retry_count
logger.warning(f"Тайм-аут: {e}. Повтор {retry_count}/{max_retries} через {wait_time} сек.")
time.sleep(wait_time)

except HTTPError as e:
# HTTP-ошибки
status_code = e.response.status_code

if status_code == 401:
logger.error("Ошибка аутентификации. Проверьте учетные данные.")
# Здесь можно добавить логику для обновления токенов
return None

elif status_code == 403:
logger.error("Доступ запрещен. Недостаточно прав.")
return None

elif status_code == 404:
logger.error(f"Ресурс не найден: {url}")
return None

elif status_code == 429:
# Rate limiting
retry_count += 1
retry_after = int(e.response.headers.get('Retry-After', 60))
logger.warning(f"Превышен лимит запросов. Повтор через {retry_after} сек.")
time.sleep(retry_after)

elif 500 <= status_code < 600:
# Серверные ошибки
retry_count += 1
wait_time = 2 ** retry_count
logger.warning(f"Серверная ошибка: {status_code}. Повтор {retry_count}/{max_retries} через {wait_time} сек.")
time.sleep(wait_time)

else:
logger.error(f"HTTP ошибка: {e}")
return None

except ValueError as e:
# Ошибки декодирования JSON
logger.error(f"Ошибка обработки JSON: {e}")
return None

except RequestException as e:
# Другие ошибки requests
logger.error(f"Ошибка запроса: {e}")
return None

logger.error(f"Превышено максимальное число попыток ({max_retries})")
return None

# Пример использования
result = api_request_with_error_handling(
'https://api.example.com/data',
headers={'Authorization': 'Bearer TOKEN'},
timeout=10
)

if result:
print("Данные успешно получены:", result)
else:
print("Не удалось получить данные")

import os
from dotenv import load_dotenv

# Загрузка переменных из файла .env
load_dotenv()

# Получение чувствительных данных из переменных окружения
api_key = os.getenv('API_KEY')
client_secret = os.getenv('CLIENT_SECRET')

# Использование в запросе
headers = {'Authorization': f'Bearer {api_key}'}
response = requests.get('https://api.example.com/data', headers=headers)

import time
from functools import wraps

def rate_limited(max_per_minute):
"""
Декоратор для ограничения частоты запросов

Args:
max_per_minute (int): Максимальное число запросов в минуту

Returns:
function: Декорированная функция
"""
interval = 60.0 / max_per_minute
last_called = [0\.0]

def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() – last_called[0]
remaining = interval – elapsed

if remaining > 0:
time.sleep(remaining)

result = func(*args, **kwargs)
last_called[0] = time.time()
return result

return wrapper
return decorator

# Использование декоратора
@rate_limited(10) # Не более 10 запросов в минуту
def fetch_data(resource_id):
return requests.get(f'https://api.example.com/resources/{resource_id}')