REST API - это архитектурный стиль взаимодействия компонентов распределённого приложения в сети. Python располагает функциональным набором инструментов для работы с REST API и основная библиотека для этого - requests. Она настолько хорошо спроектирована, что стала практически стандартом де-факто среди Python-разработчиков.
| Python | 1
2
3
4
5
6
| import requests
# Простой GET-запрос
response = requests.get('https://api.example.com/data')
if response.status_code == 200:
data = response.json() |
|
- aiohttp для асинхронных запросов,
- httpx как современная альтернатива requests,
- urllib3 для низкоуровневого контроля.
Основы взаимодействия с REST API
Структура URL в REST API играет ключевую роль в организации взаимодействия. Типичный URL состоит из базового адреса и конечной точки (эндпоинта). Например, если мы работаем с API книжного магазина, URL может выглядеть так: https://api.bookstore.com/v1/books. Здесь /v1 указывает версию API, а /books - конкретный ресурс.
Для работы с API Python предлагает библиотеку requests, которая существенно упрощает взаимодействие с веб-сервисами. Она предоставляет интуитивно понятный интерфейс для выполнения HTTP-запросов:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| import requests
# GET-запрос для получения списка книг
response = requests.get('https://api.bookstore.com/v1/books')
# POST-запрос для создания новой книги
new_book = {
'title': 'Python для профессионалов',
'author': 'Иван Петров',
'price': 999
}
response = requests.post('https://api.bookstore.com/v1/books', json=new_book)
# PUT-запрос для обновления информации о книге
updated_info = {'price': 899}
response = requests.put('https://api.bookstore.com/v1/books/123', json=updated_info)
# DELETE-запрос для удаления книги
response = requests.delete('https://api.bookstore.com/v1/books/123') |
|
| Python | 1
2
3
4
5
6
| headers = {
'Authorization': 'Bearer your_access_token',
'Accept': 'application/json',
'User-Agent': 'MyApp/1.0'
}
response = requests.get('https://api.bookstore.com/v1/books', headers=headers) |
|
| Python | 1
2
3
4
5
6
7
8
| params = {
'author': 'Иван Петров',
'price_min': 500,
'price_max': 1000,
'page': 1,
'per_page': 20
}
response = requests.get('https://api.bookstore.com/v1/books', params=params) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
| import xmltodict
import yaml
# Работа с XML
xml_response = requests.get('https://api.example.com/xml-endpoint')
data = xmltodict.parse(xml_response.text)
# Работа с YAML
yaml_response = requests.get('https://api.example.com/yaml-endpoint')
data = yaml.safe_load(yaml_response.text) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
| response = requests.get('https://api.bookstore.com/v1/books/123')
if response.status_code == 200:
book = response.json()
print(f"Найдена книга: {book['title']}")
elif response.status_code == 404:
print("Книга не найдена")
elif response.status_code == 401:
print("Требуется авторизация")
else:
print(f"Произошла ошибка: {response.status_code}") |
|
| Python | 1
2
3
4
5
| response = requests.get('https://api.bookstore.com/v1/books', stream=True)
for line in response.iter_lines():
if line:
book = json.loads(line)
process_book(book) |
|
При работе с большими наборами данных через API часто требуется реализовать пагинацию. Вот пример реализации постраничной загрузки с автоматическим переходом по страницам:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| def fetch_all_books():
page = 1
all_books = []
while True:
response = requests.get(
'https://api.bookstore.com/v1/books',
params={'page': page, 'per_page': 100}
)
if response.status_code != 200:
break
books = response.json()
if not books:
break
all_books.extend(books)
page += 1
return all_books |
|
| Python | 1
2
3
4
5
6
7
8
9
10
| from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10)
)
def get_book_with_retry(book_id):
response = requests.get(f'https://api.bookstore.com/v1/books/{book_id}')
response.raise_for_status()
return response.json() |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
| class APIError(Exception):
def __init__(self, status_code, message):
self.status_code = status_code
self.message = message
super().__init__(f"API Error {status_code}: {message}")
class APIClient:
def __init__(self, base_url):
self.base_url = base_url
self.session = requests.Session()
def handle_response(self, response):
if response.status_code >= 500:
raise APIError(response.status_code, "Ошибка сервера")
elif response.status_code == 404:
raise APIError(response.status_code, "Ресурс не найден")
elif response.status_code == 401:
raise APIError(response.status_code, "Ошибка аутентификации")
elif response.status_code >= 400:
raise APIError(response.status_code, "Ошибка клиента")
return response.json()
def get_book(self, book_id):
response = self.session.get(f"{self.base_url}/books/{book_id}")
return self.handle_response(response) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| import aiohttp
import asyncio
async def fetch_book(session, book_id):
async with session.get(f'https://api.bookstore.com/v1/books/{book_id}') as response:
return await response.json()
async def fetch_multiple_books(book_ids):
async with aiohttp.ClientSession() as session:
tasks = [fetch_book(session, book_id) for book_id in book_ids]
return await asyncio.gather(*tasks)
# Использование
book_ids = [1, 2, 3, 4, 5]
books = asyncio.run(fetch_multiple_books(book_ids)) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| class TokenManager:
def __init__(self, client_id, client_secret):
self.client_id = client_id
self.client_secret = client_secret
self.access_token = None
self.refresh_token = None
self.expires_at = None
def is_token_expired(self):
return not self.expires_at or datetime.now() > self.expires_at
def refresh_access_token(self):
if self.is_token_expired():
response = requests.post(
'https://api.bookstore.com/v1/token/refresh',
data={
'client_id': self.client_id,
'client_secret': self.client_secret,
'refresh_token': self.refresh_token
}
)
token_data = response.json()
self.access_token = token_data['access_token']
self.refresh_token = token_data['refresh_token']
self.expires_at = datetime.now() + timedelta(seconds=token_data['expires_in']) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
@dataclass
class Author:
id: int
name: str
email: Optional[str] = None
@dataclass
class Book:
id: int
title: str
author: Author
published_at: datetime
price: float
@classmethod
def from_api_response(cls, data):
return cls(
id=data['id'],
title=data['title'],
author=Author(**data['author']),
published_at=datetime.fromisoformat(data['published_at']),
price=float(data['price'])
) |
|
Взаимодействие с google api (youtube)
Добрый день. Хочу написать простой скрипт, который бы собирал статистику по каналам на ютьюбе. Но для начала решил попробовать запустить хотя бы... Взаимодействие с API blockchain
Подскажите, как взаимодействовать с http://www.blockchain.com. Мне нужно снять все транзакции с адреса, мой код выводит только последние 50... Взаимодействие по API с тестовым контуром федресурса
Добрый день! Я написал код, используя инструкцию по подключению к тестовому контуру... Python. Взаимодействие между подпрограммами
Господа форумчане!
Подскажите, как реализовать последнюю процедуру (getpropMark()) в задаче:
Задача
Класс StudentScore для генерации...
Практические примеры и техники
Аутентификация - один из ключевых аспектов работы с API. Существует несколько распространённых методов аутентификации, каждый со своими особенностями. Простейший вариант - использование API-ключа:
| Python | 1
2
3
4
| class APIClient:
def __init__(self, api_key):
self.session = requests.Session()
self.session.headers.update({'X-API-Key': api_key}) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| from oauthlib.oauth2 import WebApplicationClient
import requests
class OAuth2Client:
def __init__(self, client_id, client_secret, auth_url, token_url):
self.client = WebApplicationClient(client_id)
self.client_secret = client_secret
self.auth_url = auth_url
self.token_url = token_url
def get_auth_url(self, redirect_uri, scope):
return self.client.prepare_request_uri(
self.auth_url,
redirect_uri=redirect_uri,
scope=scope
)
def fetch_token(self, authorization_response, redirect_uri):
token = self.client.fetch_token(
self.token_url,
authorization_response=authorization_response,
client_secret=self.client_secret,
redirect_uri=redirect_uri
)
return token |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| from datetime import datetime
from decimal import Decimal
import json
class APIEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Decimal):
return str(obj)
if hasattr(obj, 'to_dict'):
return obj.to_dict()
return super().default(obj)
class APIDecoder(json.JSONDecoder):
def __init__(self, *args, **kwargs):
super().__init__(object_hook=self.object_hook, *args, **kwargs)
def object_hook(self, dct):
if 'timestamp' in dct:
try:
dct['timestamp'] = datetime.fromisoformat(dct['timestamp'])
except ValueError:
pass
return dct |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| class BookStoreAPI:
def __init__(self, base_url, api_key):
self.books = BookEndpoint(base_url, api_key)
self.authors = AuthorEndpoint(base_url, api_key)
self.orders = OrderEndpoint(base_url, api_key)
class BaseEndpoint:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({'X-API-Key': api_key})
def _make_request(self, method, path, **kwargs):
url = f"{self.base_url}/{path}"
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
return response.json()
class BookEndpoint(BaseEndpoint):
def get_book(self, book_id):
return self._make_request('GET', f'books/{book_id}')
def create_book(self, data):
return self._make_request('POST', 'books', json=data) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| import aiohttp
import asyncio
from typing import List, Dict
class AsyncBookStoreAPI:
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url
self.headers = {'X-API-Key': api_key}
async def fetch_books(self, book_ids: List[int]) -> List[Dict]:
async with aiohttp.ClientSession(headers=self.headers) as session:
tasks = [
self._fetch_book(session, book_id)
for book_id in book_ids
]
return await asyncio.gather(*tasks)
async def _fetch_book(self, session: aiohttp.ClientSession, book_id: int) -> Dict:
url = f"{self.base_url}/books/{book_id}"
async with session.get(url) as response:
response.raise_for_status()
return await response.json() |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| from urllib3 import PoolManager
from urllib3.util import Retry
class ConnectionPool:
def __init__(self, pool_size=10, retries=3):
self.retry_strategy = Retry(
total=retries,
backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504]
)
self.pool = PoolManager(
maxsize=pool_size,
retries=self.retry_strategy,
timeout=5.0
)
def request(self, method, url, **kwargs):
return self.pool.request(method, url, **kwargs) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| class APIException(Exception):
def __init__(self, message, status_code=None, response=None):
self.message = message
self.status_code = status_code
self.response = response
super().__init__(message)
class AuthenticationError(APIException): pass
class RateLimitError(APIException): pass
class ResourceNotFoundError(APIException): pass
class ValidationError(APIException): pass
def handle_api_error(response):
if response.status_code == 401:
raise AuthenticationError("Ошибка аутентификации")
elif response.status_code == 429:
raise RateLimitError("Превышен лимит запросов")
elif response.status_code == 404:
raise ResourceNotFoundError("Ресурс не найден")
elif response.status_code == 422:
raise ValidationError("Ошибка валидации данных")
elif response.status_code >= 500:
raise APIException("Ошибка сервера", status_code=response.status_code) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
| from typing import List, Optional
from dataclasses import dataclass
from datetime import datetime
@dataclass
class Book:
id: int
title: str
authors: List[str]
price: float
published: datetime
@classmethod
def from_dict(cls, data: dict):
return cls(
id=data['id'],
title=data['title'],
authors=data['authors'],
price=float(data['price']),
published=datetime.fromisoformat(data['published'])
)
class BookConverter:
@staticmethod
def to_model(data: dict) -> Book:
return Book.from_dict(data)
@staticmethod
def to_dict(book: Book) -> dict:
return {
'id': book.id,
'title': book.title,
'authors': book.authors,
'price': str(book.price),
'published': book.published.isoformat()
} |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
| from functools import wraps
from datetime import datetime, timedelta
import pickle
class Cache:
def __init__(self, ttl_seconds=300):
self.cache = {}
self.ttl = timedelta(seconds=ttl_seconds)
def __call__(self, func):
@wraps(func)
async def wrapper(*args, **kwargs):
key = pickle.dumps((args, kwargs))
if key in self.cache:
data, timestamp = self.cache[key]
if datetime.now() - timestamp < self.ttl:
return data
result = await func(*args, **kwargs)
self.cache[key] = (result, datetime.now())
return result
return wrapper
class BookAPI:
def __init__(self, base_url: str):
self.base_url = base_url
@Cache(ttl_seconds=300)
async def get_book(self, book_id: int):
async with aiohttp.ClientSession() as session:
url = f"{self.base_url}/books/{book_id}"
async with session.get(url) as response:
return await response.json() |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
| class Field:
def __init__(self, name=None, required=True):
self.name = name
self.required = required
def __get__(self, instance, owner):
if instance is None:
return self
return instance._data.get(self.name)
def __set__(self, instance, value):
instance._data[self.name] = value
class Model:
def __init__(self, **kwargs):
self._data = {}
fields = self._get_fields()
for name, field in fields.items():
value = kwargs.get(field.name or name)
if field.required and value is None:
raise ValueError(f"Field {name} is required")
setattr(self, name, value)
@classmethod
def _get_fields(cls):
return {
name: field for name, field in cls.__dict__.items()
if isinstance(field, Field)
}
class Author(Model):
id = Field()
name = Field()
email = Field(required=False)
class Book(Model):
id = Field()
title = Field()
author = Field()
price = Field() |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
| from typing import Any, Type
from decimal import Decimal
class Validator:
def __init__(self, field_type: Type, min_value=None, max_value=None, pattern=None):
self.field_type = field_type
self.min_value = min_value
self.max_value = max_value
self.pattern = pattern
def validate(self, value: Any) -> Any:
if not isinstance(value, self.field_type):
try:
value = self.field_type(value)
except (ValueError, TypeError):
raise ValidationError(f"Expected {self.field_type.__name__}")
if self.min_value is not None and value < self.min_value:
raise ValidationError(f"Value must be >= {self.min_value}")
if self.max_value is not None and value > self.max_value:
raise ValidationError(f"Value must be <= {self.max_value}")
if self.pattern and not self.pattern.match(str(value)):
raise ValidationError(f"Value does not match pattern {self.pattern}")
return value
class BookValidator:
price = Validator(Decimal, min_value=Decimal('0.01'))
title = Validator(str, pattern=re.compile(r'^[\w\s]{1,100}$'))
@classmethod
def validate(cls, data: dict) -> dict:
validated = {}
for field, validator in cls.__dict__.items():
if isinstance(validator, Validator):
value = data.get(field)
if value is not None:
validated[field] = validator.validate(value)
return validated |
|
Продвинутые стратегии и лучшие практики
При работе с нестабильными соединениями важно реализовать надёжный механизм повторных попыток. Вот пример реализации с экспоненциальной задержкой:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| from time import sleep
from random import uniform
class RetryHandler:
def __init__(self, max_retries=3, base_delay=1):
self.max_retries = max_retries
self.base_delay = base_delay
def execute(self, func, *args, **kwargs):
last_exception = None
for attempt in range(self.max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
delay = self.base_delay * (2 ** attempt) + uniform(0, 0.1)
sleep(delay)
raise last_exception |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| from collections import OrderedDict
from threading import Lock
from time import time
class APICache:
def __init__(self, max_size=1000, ttl=300):
self.cache = OrderedDict()
self.max_size = max_size
self.ttl = ttl
self.lock = Lock()
def get(self, key):
with self.lock:
if key in self.cache:
value, timestamp = self.cache[key]
if time() - timestamp <= self.ttl:
self.cache.move_to_end(key)
return value
else:
del self.cache[key]
return None
def set(self, key, value):
with self.lock:
if len(self.cache) >= self.max_size:
self.cache.popitem(last=False)
self.cache[key] = (value, time())
self.cache.move_to_end(key) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| import hashlib
import hmac
from base64 import b64encode
class SecurityMiddleware:
def __init__(self, secret_key):
self.secret_key = secret_key.encode()
def sign_request(self, method, url, body=None):
timestamp = str(int(time()))
message = f"{method}{url}{timestamp}"
if body:
message += hashlib.sha256(body.encode()).hexdigest()
signature = hmac.new(
self.secret_key,
message.encode(),
hashlib.sha256
).digest()
return {
'X-Timestamp': timestamp,
'X-Signature': b64encode(signature).decode()
}
def verify_response(self, response):
if not response.headers.get('X-Server-Signature'):
raise SecurityError("Missing server signature")
# Проверка подписи сервера |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| from dataclasses import dataclass
from typing import Dict, Any
import yaml
@dataclass
class APIEndpoint:
method: str
path: str
params: Dict[str, Any]
response_type: Any
class SwaggerClientGenerator:
def __init__(self, spec_path):
with open(spec_path) as f:
self.spec = yaml.safe_load(f)
def generate_client(self):
endpoints = {}
for path, methods in self.spec['paths'].items():
for method, details in methods.items():
endpoint = APIEndpoint(
method=method,
path=path,
params=details.get('parameters', {}),
response_type=self._get_response_type(details)
)
endpoints[details['operationId']] = endpoint
return endpoints |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| class MockResponse:
def __init__(self, status_code, json_data, headers=None):
self.status_code = status_code
self._json_data = json_data
self.headers = headers or {}
def json(self):
return self._json_data
def raise_for_status(self):
if self.status_code >= 400:
raise requests.HTTPError(f"HTTP {self.status_code}")
class APIMocker:
def __init__(self):
self.responses = {}
def mock(self, method, url, status_code=200, json_data=None, headers=None):
self.responses[(method, url)] = MockResponse(
status_code, json_data, headers
)
def get_response(self, method, url):
return self.responses.get((method, url)) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| class APIClient:
"""
Клиент для работы с API сервиса
Attributes:
base_url: Базовый URL API
api_key: Ключ для аутентификации
timeout: Таймаут запросов в секундах
Raises:
AuthenticationError: При ошибках аутентификации
RateLimitError: При превышении лимита запросов
APIError: При прочих ошибках API
"""
def __init__(self, base_url, api_key, timeout=30):
self.base_url = base_url
self.session = self._create_session(api_key)
self.timeout = timeout
def _create_session(self, api_key):
session = requests.Session()
session.headers.update({
'Authorization': f'Bearer {api_key}',
'User-Agent': 'MyAPI-Client/1.0'
})
return session |
|
Посоветуйте литературу, где описывалось бы взаимодействие Python и IIS/Apache
Книг полно по Питону. Но сложно найти целостную информацию от А до Я. Интересует подробное описание механизма взаимодействия Flask/Django с... Взаимодействие Python и Excel
Помогите с написанием скрипта. Мне нужно, чтобы скрипт автоматически заполнял Эксель таблицу вводимыми мною данными. При том, чтобы он форматировал... Python socket Взаимодействие клиента и сервера в разных сетях wifi
Есть сервер и клиент. При подключении клиента к серверу на разных устройствах которые находятся в одной сети обмен данными происходит успешно. При... Python Selenium, запретить взаимодействие пользователю с браузером
Есть ли способ программно запрещать/разрешать пользователю взаимодействие с браузером? К примеру чтобы пользователь видел браузер, что там... Selenium Python взаимодействие со всплывающим окном
Добрый день!
Подскажите, как взаимодействовать со всплывающим окном?
При нажатии кнопки "Отмена" выскакивает окно - Alert
Контекстное... Python 3.10 Взаимодействие с накопителями в Linux
Скорее всего я плохо гуглю, но не один час просидел со следующей проблемой:
Средствами PyQt5 создано окно, выполняющее не суть важно какой... Не могу настроить webhook для бота WhatsApp использую API с сайта api-messenger.com
Ситуация следующая.
На сайте сказано:
При сохранении адреса, система проверят адрес, создавая POST запрос с заголовком Apimessenger,... Выводить текущие данные о погоде с использованием API сайта https://www.weatherbit.io/api. Данные о погоде должны быть в
С помощью Python нужно вывысти текущие данные о погоде с использованием API сайта https://www.weatherbit.io/api Данные о погоде должны быть выведены... Api python wargaming
Как отсылать запрос и получать ответ через api... Python C/Api
Python Api это функции для взаимодействия с интерпретатором? или библиотека для компиляции интерпретатора внутри внутри c++? Можно ли скомпилировать... Python и Google API
Коллеги, добрый вечер. Не могу разобраться. По сей день не было опыта работы с API, столкнулся впервые. Два вопроса. Вопрос первый.
Код по... Python, VK.Api
Питон ругаеться на аксесс-токен, хотя он был задан правильный.
|
Я не гений, но могу предположить что изначальный Ваш код:
