Разработка REST API клиентов на Python: базовые принципы и лучшие практики

python -m venv venv

# На Windows
venv\Scripts\activate

# На macOS/Linux
source venv/bin/activate

pip install requests python-dotenv

pip freeze > requirements.txt

API_KEY=your_api_key_here
API_BASE_URL=https://api.example.com

api_client/
├── .env # Конфиденциальные данные
├── requirements.txt # Зависимости проекта
├── config.py # Конфигурация (загрузка переменных окружения)
├── api/
│ ├── __init__.py
│ ├── client.py # Базовый API клиент
│ └── resources/ # Ресурсы API
│ ├── __init__.py
│ ├── users.py # Методы для работы с пользователями
│ └── products.py # Методы для работы с продуктами
└── tests/ # Тесты API клиента
├── __init__.py
├── test_users.py
└── test_products.py

import os
from dotenv import load_dotenv

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

# Получение переменных окружения
API_KEY = os.getenv("API_KEY")
API_BASE_URL = os.getenv("API_BASE_URL")

# Проверка наличия обязательных переменных
if not API_KEY:
raise ValueError("API_KEY is not set in the environment")
if not API_BASE_URL:
raise ValueError("API_BASE_URL is not set in the environment")

import requests
from requests.exceptions import RequestException
import logging
from config import API_BASE_URL, API_KEY

# Настройка логгера
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class APIClient:
"""Базовый класс для работы с REST API."""

def __init__(self, base_url=None, api_key=None, timeout=10):
"""
Инициализация клиента API.

Args:
base_url (str): Базовый URL API. По умолчанию берется из конфигурации.
api_key (str): API ключ для авторизации. По умолчанию берется из конфигурации.
timeout (int): Таймаут для запросов в секундах.
"""
self.base_url = base_url or API_BASE_URL
self.api_key = api_key or API_KEY
self.timeout = timeout
self.session = requests.Session()

# Установка дефолтных заголовков для всех запросов
self.session.headers.update({
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json',
'Accept': 'application/json'
})

def _build_url(self, endpoint):
"""Формирование полного URL для запроса."""
if endpoint.startswith('http'):
return endpoint

# Удаляем лишние слеши для корректного объединения
base = self.base_url.rstrip('/')
endpoint = endpoint.lstrip('/')
return f"{base}/{endpoint}"

def _handle_response(self, response):
"""
Обработка ответа от API.

Args:
response (requests.Response): Объект ответа.

Returns:
dict: Данные ответа в формате JSON.

Raises:
HTTPError: Если статус ответа указывает на ошибку.
"""
try:
response.raise_for_status()
return response.json()
except ValueError:
# Если ответ не содержит JSON
logger.warning(f"Response is not JSON: {response.text}")
return response.text

def request(self, method, endpoint, **kwargs):
"""
Выполнение HTTP-запроса к API.

Args:
method (str): HTTP метод (GET, POST, PUT, DELETE и т.д.).
endpoint (str): Эндпоинт API.
**kwargs: Дополнительные аргументы для requests.request.

Returns:
dict: Данные ответа в формате JSON.

Raises:
RequestException: При ошибке выполнения запроса.
"""
url = self._build_url(endpoint)

# Установка таймаута по умолчанию, если не указан
if 'timeout' not in kwargs:
kwargs['timeout'] = self.timeout

try:
logger.info(f"Making {method} request to {url}")
response = self.session.request(method, url, **kwargs)
return self._handle_response(response)
except RequestException as e:
logger.error(f"Request error: {e}")
raise

# Удобные методы-обертки для основных HTTP методов
def get(self, endpoint, **kwargs):
"""Выполнение GET-запроса."""
return self.request('GET', endpoint, **kwargs)

def post(self, endpoint, data=None, json=None, **kwargs):
"""Выполнение POST-запроса."""
return self.request('POST', endpoint, data=data, json=json, **kwargs)

def put(self, endpoint, data=None, json=None, **kwargs):
"""Выполнение PUT-запроса."""
return self.request('PUT', endpoint, data=data, json=json, **kwargs)

def patch(self, endpoint, data=None, json=None, **kwargs):
"""Выполнение PATCH-запроса."""
return self.request('PATCH', endpoint, data=data, json=json, **kwargs)

def delete(self, endpoint, **kwargs):
"""Выполнение DELETE-запроса."""
return self.request('DELETE', endpoint, **kwargs)

from ..client import APIClient

class UserAPI(APIClient):
"""Класс для работы с API пользователей."""

def get_users(self, page=1, limit=10):
"""
Получение списка пользователей с пагинацией.

Args:
page (int): Номер страницы.
limit (int): Количество результатов на странице.

Returns:
dict: Данные о пользователях.
"""
return self.get('users', params={'page': page, 'limit': limit})

def get_user(self, user_id):
"""
Получение данных конкретного пользователя по ID.

Args:
user_id (int): Идентификатор пользователя.

Returns:
dict: Данные пользователя.
"""
return self.get(f'users/{user_id}')

def create_user(self, user_data):
"""
Создание нового пользователя.

Args:
user_data (dict): Данные пользователя для создания.

Returns:
dict: Данные созданного пользователя.
"""
return self.post('users', json=user_data)

def update_user(self, user_id, user_data):
"""
Обновление данных пользователя.

Args:
user_id (int): Идентификатор пользователя.
user_data (dict): Обновленные данные пользователя.

Returns:
dict: Обновленные данные пользователя.
"""
return self.put(f'users/{user_id}', json=user_data)

def delete_user(self, user_id):
"""
Удаление пользователя.

Args:
user_id (int): Идентификатор пользователя.

Returns:
dict: Результат операции удаления.
"""
return self.delete(f'users/{user_id}')

# example.py
from api.resources.users import UserAPI

def main():
# Создаем экземпляр API-клиента для работы с пользователями
user_api = UserAPI()

# Получаем список пользователей
users = user_api.get_users(page=1, limit=5)
print(f"Users: {users}")

# Создаем нового пользователя
new_user = {
'name': 'John Doe',
'email': 'john.doe@example.com',
'age': 30
}
created_user = user_api.create_user(new_user)
print(f"Created user: {created_user}")

# Получаем данные пользователя по ID
user_id = created_user['id']
user = user_api.get_user(user_id)
print(f"User details: {user}")

# Обновляем пользователя
updated_data = {'age': 31}
updated_user = user_api.update_user(user_id, updated_data)
print(f"Updated user: {updated_user}")

# Удаляем пользователя
delete_result = user_api.delete_user(user_id)
print(f"Delete result: {delete_result}")

if __name__ == "__main__":
main()

import json
from dataclasses import dataclass
from typing import Any, Dict, Optional, Union

@dataclass
class APIResponse:
"""Класс для представления ответа API."""
status_code: int
data: Optional[Any] = None
error: Optional[str] = None
headers: Optional[Dict[str, str]] = None

@property
def success(self) -> bool:
"""Проверка успешности запроса."""
return 200 <= self.status_code < 300 and self.error is None

class APIClient:
# ... предыдущий код ...

def _parse_json(self, response):
"""
Преобразование ответа в JSON с обработкой возможных ошибок.

Args:
response (requests.Response): Объект ответа.

Returns:
Any: Данные в формате Python-объектов или None.
"""
if not response.content:
return None

try:
return response.json()
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON: {e}")
logger.debug(f"Response content: {response.text}")
return None

def _create_response(self, response):
"""
Создание объекта APIResponse на основе HTTP-ответа.

Args:
response (requests.Response): Объект ответа.

Returns:
APIResponse: Обработанный ответ API.
"""
data = self._parse_json(response)

# Проверка на ошибки в ответе
error = None
if not response.ok:
if data and isinstance(data, dict):
# Многие API возвращают сообщение об ошибке в определенном формате
error = data.get('error') or data.get('message') or response.reason
else:
error = response.reason

return APIResponse(
status_code=response.status_code,
data=data,
error=error,
headers=dict(response.headers)
)

def request(self, method, endpoint, **kwargs):
"""
Выполнение HTTP-запроса к API.

Args:
method (str): HTTP метод (GET, POST, PUT, DELETE и т.д.).
endpoint (str): Эндпоинт API.
**kwargs: Дополнительные аргументы для requests.request.

Returns:
APIResponse: Объект с результатами запроса.
"""
url = self._build_url(endpoint)

# Установка таймаута по умолчанию, если не указан
if 'timeout' not in kwargs:
kwargs['timeout'] = self.timeout

try:
logger.info(f"Making {method} request to {url}")
response = self.session.request(method, url, **kwargs)
return self._create_response(response)
except requests.RequestException as e:
logger.error(f"Request error: {e}")
return APIResponse(
status_code=0,
error=str(e)
)

from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
import json

@dataclass
class User:
id: int
name: str
email: str
created_at: datetime
updated_at: Optional[datetime] = None

@classmethod
def from_json(cls, data: dict):
"""Создание объекта User из JSON-данных."""
# Преобразуем строки дат в объекты datetime
created_at = datetime.fromisoformat(data['created_at'].replace('Z', '+00:00'))
updated_at = None
if data.get('updated_at'):
updated_at = datetime.fromisoformat(data['updated_at'].replace('Z', '+00:00'))

return cls(
id=data['id'],
name=data['name'],
email=data['email'],
created_at=created_at,
updated_at=updated_at
)

def to_json(self) -> str:
"""Преобразование объекта User в JSON-строку."""
data = {
'id': self.id,
'name': self.name,
'email': self.email,
'created_at': self.created_at.isoformat(),
}
if self.updated_at:
data['updated_at'] = self.updated_at.isoformat()

return json.dumps(data)

class UserAPI(APIClient):
# ... предыдущий код ...

def get_user(self, user_id) -> Optional[User]:
"""
Получение данных конкретного пользователя по ID.

Args:
user_id (int): Идентификатор пользователя.

Returns:
Optional[User]: Объект пользователя или None в случае ошибки.
"""
response = self.get(f'users/{user_id}')
if response.success and response.data:
return User.from_json(response.data)
return None

def get_users(self, page=1, limit=10) -> List[User]:
"""
Получение списка пользователей с пагинацией.

Args:
page (int): Номер страницы.
limit (int): Количество результатов на странице.

Returns:
List[User]: Список объектов пользователей.
"""
response = self.get('users', params={'page': page, 'limit': limit})
if response.success and response.data:
# Предполагаем, что данные возвращаются в виде списка пользователей
users = []
for user_data in response.data:
users.append(User.from_json(user_data))
return users
return []

import functools

def handle_api_response(model_class=None):
"""
Декоратор для обработки ответов API.

Args:
model_class: Класс модели для преобразования данных.

Returns:
Декорированная функция.
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
response = func(*args, **kwargs)

if not response.success:
logger.error(f"API error: {response.error}")
return None

if model_class and response.data:
# Если данные – список, преобразуем каждый элемент
if isinstance(response.data, list):
return [model_class.from_json(item) for item in response.data]
# Если данные – словарь, преобразуем в объект модели
return model_class.from_json(response.data)

return response.data
return wrapper
return decorator

class UserAPI(APIClient):
@handle_api_response(User)
def get_user(self, user_id):
return self.get(f'users/{user_id}')

@handle_api_response(User)
def get_users(self, page=1, limit=10):
return self.get('users', params={'page': page, 'limit': limit})

class APIClient:
"""Базовый класс для работы с REST API с поддержкой различных методов аутентификации."""

def __init__(self, base_url=None, timeout=10, auth_method=None, **auth_params):
"""
Инициализация клиента API.

Args:
base_url (str): Базовый URL API.
timeout (int): Таймаут для запросов в секундах.
auth_method (str): Метод аутентификации ('api_key', 'basic', 'bearer', 'oauth').
**auth_params: Параметры для выбранного метода аутентификации.
"""
self.base_url = base_url or API_BASE_URL
self.timeout = timeout
self.session = requests.Session()

# Применяем выбранный метод аутентификации
if auth_method:
self._setup_authentication(auth_method, **auth_params)

def _setup_authentication(self, auth_method, **auth_params):
"""
Настройка аутентификации для API-клиента.

Args:
auth_method (str): Метод аутентификации.
**auth_params: Параметры аутентификации.
"""
if auth_method == 'api_key':
api_key = auth_params.get('api_key') or API_KEY
key_name = auth_params.get('key_name', 'X-API-Key')
key_in = auth_params.get('in', 'header')

if key_in == 'header':
self.session.headers[key_name] = api_key
elif key_in == 'query':
self.session.params = {key_name: api_key}

elif auth_method == 'basic':
username = auth_params.get('username')
password = auth_params.get('password')
if username and password:
self.session.auth = (username, password)

elif auth_method == 'bearer':
token = auth_params.get('token')
if token:
self.session.headers['Authorization'] = f'Bearer {token}'

elif auth_method == 'oauth':
# Здесь можно добавить поддержку OAuth 2.0
pass

# Установка стандартных заголовков для JSON API
self.session.headers.update({
'Content-Type': 'application/json',
'Accept': 'application/json'
})

class APIError(Exception):
"""Базовый класс для исключений API-клиента."""
def __init__(self, message, status_code=None, response=None):
self.message = message
self.status_code = status_code
self.response = response
super().__init__(self.message)

class ConnectionError(APIError):
"""Ошибка соединения с API."""
pass

class AuthenticationError(APIError):
"""Ошибка аутентификации."""
pass

class ResourceNotFoundError(APIError):
"""Запрашиваемый ресурс не найден."""
pass

class ValidationError(APIError):
"""Ошибка валидации запроса."""
pass

class ServerError(APIError):
"""Внутренняя ошибка сервера API."""
pass

class APIClient:
# ... предыдущий код ...

def _handle_error_response(self, response):
"""
Обработка ошибок HTTP-ответа.

Args:
response (requests.Response): Объект ответа.

Raises:
Соответствующее исключение на основе кода статуса.
"""
error_data = None
try:
if response.content:
error_data = response.json()
except ValueError:
error_data = {'message': response.text}

error_message = 'Unknown error'
if error_data:
if isinstance(error_data, dict):
error_message = error_data.get('message') or error_data.get('error') or str(error_data)
else:
error_message = str(error_data)

if response.status_code == 401:
raise AuthenticationError(f"Authentication failed: {error_message}", response.status_code, response)
elif response.status_code == 403:
raise AuthenticationError(f"Permission denied: {error_message}", response.status_code, response)
elif response.status_code == 404:
raise ResourceNotFoundError(f"Resource not found: {error_message}", response.status_code, response)
elif 400 <= response.status_code < 500:
raise ValidationError(f"Request validation failed: {error_message}", response.status_code, response)
elif 500 <= response.status_code < 600:
raise ServerError(f"Server error: {error_message}", response.status_code, response)
else:
raise APIError(f"API error: {error_message}", response.status_code, response)

def request(self, method, endpoint, **kwargs):
"""
Выполнение HTTP-запроса к API с обработкой ошибок.

Args:
method (str): HTTP метод.
endpoint (str): Эндпоинт API.
**kwargs: Дополнительные аргументы для requests.request.

Returns:
Any: Данные ответа.

Raises:
APIError: При возникновении ошибки запроса.
"""
url = self._build_url(endpoint)

# Установка таймаута по умолчанию, если не указан
if 'timeout' not in kwargs:
kwargs['timeout'] = self.timeout

try:
logger.info(f"Making {method} request to {url}")
response = self.session.request(method, url, **kwargs)

if not response.ok:
self._handle_error_response(response)

# Если нет ошибок, возвращаем данные
return self._parse_json(response)

except requests.Timeout as e:
logger.error(f"Request timeout: {e}")
raise ConnectionError(f"Request timed out: {str(e)}")
except requests.ConnectionError as e:
logger.error(f"Connection error: {e}")
raise ConnectionError(f"Connection failed: {str(e)}")
except requests.RequestException as e:
logger.error(f"Request error: {e}")
raise APIError(f"Request failed: {str(e)}")
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON response: {e}")
raise APIError(f"Invalid JSON response: {str(e)}")

import time

class OAuth2Client(APIClient):
"""Клиент API с поддержкой OAuth 2.0 и автоматическим обновлением токенов."""

def __init__(self, base_url, client_id, client_secret, token_url, **kwargs):
"""
Инициализация OAuth 2.0 клиента.

Args:
base_url (str): Базовый URL API.
client_id (str): Идентификатор клиента OAuth.
client_secret (str): Секрет клиента OAuth.
token_url (str): URL для получения/обновления токена.
**kwargs: Дополнительные аргументы для APIClient.
"""
super().__init__(base_url, **kwargs)
self.client_id = client_id
self.client_secret = client_secret
self.token_url = token_url
self.access_token = None
self.refresh_token = None
self.token_expires_at = 0

def fetch_token(self, username, password):
"""
Получение токена OAuth 2.0 через flow "password".

Args:
username (str): Имя пользователя.
password (str): Пароль пользователя.

Returns:
bool: Успешность получения токена.
"""
try:
response = requests.post(self.token_url, data={
'grant_type': 'password',
'client_id': self.client_id,
'client_secret': self.client_secret,
'username': username,
'password': password
})
response.raise_for_status()
token_data = response.json()

self.access_token = token_data['access_token']
self.refresh_token = token_data.get('refresh_token')
expires_in = token_data.get('expires_in', 3600)
self.token_expires_at = time.time() + expires_in

# Обновляем заголовок Authorization
self.session.headers['Authorization'] = f'Bearer {self.access_token}'
return True

except (requests.RequestException, KeyError) as e:
logger.error(f"Failed to fetch token: {e}")
return False

def refresh_access_token(self):
"""
Обновление токена доступа с использованием refresh_token.

Returns:
bool: Успешность обновления токена.
"""
if not self.refresh_token:
return False

try:
response = requests.post(self.token_url, data={
'grant_type': 'refresh_token',
'client_id': self.client_id,
'client_secret': self.client_secret,
'refresh_token': self.refresh_token
})
response.raise_for_status()
token_data = response.json()

self.access_token = token_data['access_token']
# Некоторые сервера также обновляют refresh_token
if 'refresh_token' in token_data:
self.refresh_token = token_data['refresh_token']

expires_in = token_data.get('expires_in', 3600)
self.token_expires_at = time.time() + expires_in

# Обновляем заголовок Authorization
self.session.headers['Authorization'] = f'Bearer {self.access_token}'
return True

except (requests.RequestException, KeyError) as e:
logger.error(f"Failed to refresh token: {e}")
return False

def request(self, method, endpoint, **kwargs):
"""
Выполнение HTTP-запроса с автоматическим обновлением токена.

Args:
method (str): HTTP метод.
endpoint (str): Эндпоинт API.
**kwargs: Дополнительные аргументы для requests.request.

Returns:
Any: Данные ответа.
"""
# Проверяем, не истекает ли токен
if self.access_token and time.time() > self.token_expires_at – 60:
logger.info("Access token is expiring soon, refreshing...")
if not self.refresh_access_token():
raise AuthenticationError("Failed to refresh access token")

try:
return super().request(method, endpoint, **kwargs)
except AuthenticationError as e:
# Если получили 401, попробуем обновить токен и повторить запрос
if e.status_code == 401 and self.refresh_token:
logger.info("Authentication failed, trying to refresh token...")
if self.refresh_access_token():
# Повторяем запрос с новым токеном
return super().request(method, endpoint, **kwargs)
# Если не удалось обновить токен или ошибка не 401, пробрасываем исключение
raise

import random
import time

def retry_with_backoff(retries=3, backoff_factor=0.3, status_forcelist=(500, 502, 504)):
"""
Декоратор для повторных попыток выполнения запроса с экспоненциальным отступом.

Args:
retries (int): Максимальное количество повторных попыток.
backoff_factor (float): Множитель для расчёта времени ожидания.
status_forcelist (tuple): Коды статусов, при которых выполнять повторные попытки.

Returns:
Декорированная функция.
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
retry_count = 0
while True:
try:
return func(*args, **kwargs)
except (ConnectionError, ServerError) as e:
# Повторяем попытку только для определённых ошибок
if isinstance(e, APIError) and e.status_code not in status_forcelist:
raise

retry_count += 1
if retry_count > retries:
raise

# Рассчитываем время ожидания с джиттером (случайное значение)
sleep_time = backoff_factor * (2 ** (retry_count – 1)) * (0.5 + random.random())
logger.warning(f"Retrying request in {sleep_time:.2f} seconds... (attempt {retry_count} of {retries})")
time.sleep(sleep_time)
return wrapper
return decorator

# Применение декоратора к методу request
class APIClient:
# ... предыдущий код ...

@retry_with_backoff()
def request(self, method, endpoint, **kwargs):
# ... предыдущий код ...