Практикум — сервис каталога на Python

ОБЯЗАТЕЛЬНОДЛЯ НОВИЧКОВ

Разработчику

Практикум, шаг 4 из 8. Реализация catalog-api. Контракт — шаг 2, DTO — шаг 3.

---

Подготовка проекта

mkdir OrderDesk\catalog-api
cd OrderDesk\catalog-api
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install fastapi uvicorn[standard] pydantic sqlalchemy python-jose[cryptography]

requirements.txt:

fastapi>=0.115.0
uvicorn[standard]>=0.32.0
pydantic>=2.0
sqlalchemy>=2.0
python-jose[cryptography]>=3.3

---

База данных (SQLite)

app/db.py — упрощённо:

from sqlalchemy import create_engine, Column, String, Integer, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime, timezone

import uuid

Base = declarative_base()

class ProductRow(Base):
    __tablename__ = "products"
    id = Column(String, primary_key=True, default=lambda: f"prod_{uuid.uuid4().hex[:8]}")
    sku = Column(String, unique=True, nullable=False)
    name = Column(String, nullable=False)
    price_cents = Column(Integer, nullable=False)
    stock_available = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

class ReservationRow(Base):
    __tablename__ = "reservations"
    id = Column(String, primary_key=True, default=lambda: f"res_{uuid.uuid4().hex[:8]}")
    product_id = Column(String, nullable=False)
    quantity = Column(Integer, nullable=False)
    order_ref = Column(String, nullable=False)
    idempotency_key = Column(String, unique=True, nullable=True)
    expires_at = Column(DateTime, nullable=False)

engine = create_engine("sqlite:///./catalog.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)

def init_db():
    Base.metadata.create_all(bind=engine)

---

Приложение FastAPI

app/main.py (основные маршруты):

from fastapi import FastAPI, Depends, HTTPException, Header, Request
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from datetime import datetime, timedelta, timezone
from decimal import Decimal

from .db import SessionLocal, init_db, ProductRow, ReservationRow
from .schemas import ProductCreate, ProductResponse, ReservationCreate, ReservationResponse
from .mapping import to_product_response

API_KEY = "dev-catalog-key-change-me"  # в проде — из переменной окружения

app = FastAPI(title="OrderDesk Catalog", version="1.0.0")

@app.on_event("startup")
def startup():
    init_db()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def require_api_key(x_api_key: str | None = Header(default=None)):
    if x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")

@app.get("/api/v1/products")
def list_products(page: int = 1, page_size: int = 20, db: Session = Depends(get_db)):
    rows = db.query(ProductRow).offset((page - 1) * page_size).limit(page_size).all()
    return [to_product_response(r) for r in rows]

@app.post("/api/v1/products", status_code=201)
def create_product(body: ProductCreate, db: Session = Depends(get_db)):
    row = ProductRow(
        sku=body.sku,
        name=body.name,
        price_cents=int(body.price * 100),
        stock_available=body.stock_available,
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return to_product_response(row)

@app.post("/api/v1/reservations", status_code=201, dependencies=[Depends(require_api_key)])
def create_reservation(
    body: ReservationCreate,
    db: Session = Depends(get_db),
    idempotency_key: str = Header(alias="Idempotency-Key"),
):
    existing = db.query(ReservationRow).filter_by(idempotency_key=idempotency_key).first()
    if existing:
        return _reservation_to_dto(existing)

    product = db.query(ProductRow).filter_by(id=body.product_id).first()
    if not product or product.stock_available < body.quantity:
        raise HTTPException(
            status_code=409,
            detail=f"Insufficient stock for {body.product_id}",
        )

    product.stock_available -= body.quantity
    res = ReservationRow(
        product_id=body.product_id,
        quantity=body.quantity,
        order_ref=body.order_ref,
        idempotency_key=idempotency_key,
        expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
    )
    db.add(res)
    db.commit()
    db.refresh(res)
    return _reservation_to_dto(res)

def _reservation_to_dto(row: ReservationRow) -> ReservationResponse:
    return ReservationResponse(
        reservationId=row.id,
        productId=row.product_id,
        quantity=row.quantity,
        expiresAt=row.expires_at,
    )

schemas.py — полные классы в шаге 3. Файл app/mapping.py:

from decimal import Decimal
from .schemas import ProductResponse

def to_product_response(row) -> ProductResponse:
    return ProductResponse(
        id=row.id,
        sku=row.sku,
        name=row.name,
        price=Decimal(row.price_cents) / 100,
        stockAvailable=row.stock_available,
        createdAt=row.created_at,
    )

---

Отмена резерва

DELETE /api/v1/reservations/{reservationId} возвращает остаток на склад — вызывается из orders-api при отмене заказа:

@app.delete("/api/v1/reservations/{reservation_id}", status_code=204,
            dependencies=[Depends(require_api_key)])
def delete_reservation(reservation_id: str, db: Session = Depends(get_db)):
    row = db.query(ReservationRow).filter_by(id=reservation_id).first()
    if not row:
        raise HTTPException(status_code=404, detail="Reservation not found")
    product = db.query(ProductRow).filter_by(id=row.product_id).first()
    if product:
        product.stock_available += row.quantity
    db.delete(row)
    db.commit()

---

Middleware X-Request-Id

import uuid

from starlette.middleware.base import BaseHTTPMiddleware

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        rid = request.headers.get("x-request-id") or str(uuid.uuid4())
        request.state.request_id = rid
        response = await call_next(request)
        response.headers["X-Request-Id"] = rid
        return response

app.add_middleware(RequestIdMiddleware)

---

Запуск и быстрая проверка

cd OrderDesk\catalog-api
uvicorn app.main:app --reload --port 8100

Документация Swagger: http://localhost:8100/docs

curl -X POST http://localhost:8100/api/v1/products `
  -H "Content-Type: application/json" `
  -d '{"sku":"T-1","name":"Test item","price":10,"stockAvailable":50}'

---

Чек-лист готовности catalog-api

• GET /api/v1/products возвращает список
• POST /api/v1/reservations без X-Api-Key → 401
• Повтор с тем же Idempotency-Key → тот же reservationId
• Резерв при нехватке остатка → 409
• DELETE резерва возвращает остаток на склад

---

Мини-тест на pytest (опционально)

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_reserve_idempotent():
    client.post("/api/v1/products", json={"sku":"T","name":"N","price":1,"stockAvailable":5})
    pid = client.get("/api/v1/products").json()[0]["id"]
    headers = {"X-Api-Key": "dev-catalog-key-change-me", "Idempotency-Key": "k1"}
    body = {"productId": pid, "quantity": 1, "orderRef": "ord_1"}
    r1 = client.post("/api/v1/reservations", json=body, headers=headers)
    r2 = client.post("/api/v1/reservations", json=body, headers=headers)
    assert r1.json()["reservationId"] == r2.json()["reservationId"]

Далее orders-api вызывает этот сервис: шаг 5.