Работа с JSON API

JSON в Python

JSON — стандартный формат обмена данными в веб-API. Python имеет встроенный модуль json:

import json
from pathlib import Path

# Python dict -> JSON string
data = {"name": "Алексей", "age": 30, "hobbies": ["программирование", "чтение"]}
json_str = json.dumps(data, ensure_ascii=False, indent=2)
print(json_str)

# JSON string -> Python dict
parsed = json.loads(json_str)
print(parsed["name"])  # Алексей

# Read/write JSON files
Path("data.json").write_text(
    json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8",
)
loaded = json.loads(Path("data.json").read_text(encoding="utf-8"))

REST API — основы

import httpx

BASE_URL = "https://jsonplaceholder.typicode.com"

def create_post(title: str, body: str, user_id: int) -> dict:
    response = httpx.post(f"{BASE_URL}/posts", json={"title": title, "body": body, "userId": user_id})
    response.raise_for_status()
    return response.json()

def get_posts(user_id: int | None = None) -> list[dict]:
    params = {"userId": user_id} if user_id else None
    response = httpx.get(f"{BASE_URL}/posts", params=params)
    response.raise_for_status()
    return response.json()

def delete_post(post_id: int) -> bool:
    response = httpx.delete(f"{BASE_URL}/posts/{post_id}")
    return response.status_code == 200

Модели данных с dataclasses

Работать с dict неудобно — нет автодополнения, нет проверки типов:

from dataclasses import dataclass

@dataclass
class Address:
    street: str
    city: str
    zipcode: str

    @classmethod
    def from_dict(cls, data: dict) -> "Address":
        return cls(street=data["street"], city=data["city"], zipcode=data["zipcode"])

@dataclass
class User:
    id: int
    name: str
    email: str
    phone: str
    address: Address

    @classmethod
    def from_dict(cls, data: dict) -> "User":
        return cls(
            id=data["id"], name=data["name"], email=data["email"],
            phone=data["phone"], address=Address.from_dict(data["address"]),
        )

# Usage
import httpx
response = httpx.get("https://jsonplaceholder.typicode.com/users/1")
user = User.from_dict(response.json())
print(f"{user.name} ({user.email})")
print(f"Город: {user.address.city}")

Пагинация

Offset-based

import httpx

def fetch_all_pages(base_url: str, per_page: int = 20) -> list[dict]:
    """Fetch all pages from paginated API."""
    all_items: list[dict] = []
    page = 1

    with httpx.Client(timeout=10.0) as client:
        while True:
            response = client.get(base_url, params={"_page": page, "_limit": per_page})
            response.raise_for_status()
            items = response.json()

            if not items:
                break

            all_items.extend(items)
            total = int(response.headers.get("x-total-count", 0))
            if len(all_items) >= total or len(items) < per_page:
                break
            page += 1

    return all_items

posts = fetch_all_pages("https://jsonplaceholder.typicode.com/posts", per_page=10)
print(f"Всего: {len(posts)} постов")

Cursor-based

import httpx

def fetch_with_cursor(url: str, max_pages: int = 100) -> list[dict]:
    """Fetch data using cursor-based pagination."""
    all_items: list[dict] = []
    cursor: str | None = None

    with httpx.Client(timeout=10.0) as client:
        for _ in range(max_pages):
            params: dict = {"limit": 50}
            if cursor:
                params["cursor"] = cursor

            response = client.get(url, params=params)
            response.raise_for_status()
            data = response.json()

            items = data.get("items", [])
            all_items.extend(items)

            cursor = data.get("next_cursor")
            if not cursor or not items:
                break

    return all_items

Валидация с Pydantic

from pydantic import BaseModel, Field, field_validator

class UserResponse(BaseModel):
    id: int
    name: str = Field(min_length=1, max_length=100)
    email: str
    phone: str | None = None

    @field_validator("email")
    @classmethod
    def validate_email(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("Некорректный email")
        return v.lower()

class PostResponse(BaseModel):
    id: int
    title: str
    body: str
    user_id: int = Field(alias="userId")

    model_config = {"populate_by_name": True}

# Parse and validate
import httpx
response = httpx.get("https://jsonplaceholder.typicode.com/posts/1")
post = PostResponse.model_validate(response.json())
print(f"Пост #{post.id}: {post.title}")

# Serialize back to JSON
json_str = post.model_dump_json(by_alias=True, indent=2)

Асинхронный API-клиент

import asyncio
import httpx
from dataclasses import dataclass

@dataclass
class AsyncAPIClient:
    base_url: str
    token: str | None = None
    timeout: float = 10.0

    async def _request(self, method: str, path: str, **kwargs) -> dict:
        headers = {"Accept": "application/json"}
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"

        async with httpx.AsyncClient(base_url=self.base_url, timeout=self.timeout) as client:
            response = await client.request(method, path, headers=headers, **kwargs)
            response.raise_for_status()
            return response.json()

    async def get(self, path: str, params: dict | None = None) -> dict:
        return await self._request("GET", path, params=params)

    async def post(self, path: str, data: dict) -> dict:
        return await self._request("POST", path, json=data)

async def main() -> None:
    client = AsyncAPIClient(base_url="https://jsonplaceholder.typicode.com")

    # Concurrent requests
    user, posts, todos = await asyncio.gather(
        client.get("/users/1"),
        client.get("/posts", params={"userId": 1}),
        client.get("/todos", params={"userId": 1}),
    )
    print(f"Пользователь: {user['name']}, постов: {len(posts)}, задач: {len(todos)}")

asyncio.run(main())