Мини-проект: API для интернет-магазина - IMI-DS

Привет! В этом проекте мы создадим бэкенд для интернет-магазина: товары, корзина, заказы и аутентификация. Это отличная возможность применить все знания FastAPI на практике и получить проект для портфолио.

В этом проекте мы создадим:

Управление товарами (CRUD)
Корзину покупок (добавление, удаление, просмотр)
Оформление заказов
Аутентификацию пользователей (регистрация, логин, JWT)
Тесты для API

Содержание

Что нужно знать перед началом
Основная часть
Структура проекта
Установка зависимостей
Модели данных (schemas.py)
Аутентификация (auth.py)
База данных (database.py)
Роутеры
Основное приложение (main.py)
Тесты (tests/test_api.py)
Задачи для доработки
Запуск и тестирование
Шпаргалка
Заключение
Что дальше?

Что нужно знать перед началом

Для этого проекта тебе понадобится:

Установленный FastAPI и Uvicorn
Базовое понимание JWT-аутентификации

Совет: Проект можно использовать как основу для реального магазина.

Основная часть

Структура проекта

shop_api/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   ├── auth.py
│   ├── crud.py
│   ├── database.py
│   └── routers/
│       ├── __init__.py
│       ├── products.py
│       ├── cart.py
│       └── orders.py
├── tests/
│   └── test_api.py
├── requirements.txt

Совет: Используй виртуальное окружение для изоляции зависимостей.

Установка зависимостей

pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart pytest httpx

requirements.txt:

fastapi==0.104.1
uvicorn==0.24.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
pytest==7.4.3
httpx==0.25.1

Модели данных (schemas.py)

from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List
from datetime import datetime

# ======== Пользователи ========
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=6)

class UserLogin(BaseModel):
    username: str
    password: str

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True

# ======== Товары ========
class ProductCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None
    price: float = Field(..., gt=0)
    stock: int = Field(..., ge=0)
    category: Optional[str] = None

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1)
    description: Optional[str] = None
    price: Optional[float] = Field(None, gt=0)
    stock: Optional[int] = Field(None, ge=0)
    category: Optional[str] = None

class ProductResponse(BaseModel):
    id: int
    name: str
    description: Optional[str]
    price: float
    stock: int
    category: Optional[str]

    class Config:
        from_attributes = True

# ======== Корзина ========
class CartItem(BaseModel):
    product_id: int
    quantity: int = Field(..., ge=1)

class CartResponse(BaseModel):
    items: List[CartItem]
    total: float

# ======== Заказы ========
class OrderCreate(BaseModel):
    delivery_address: str = Field(..., min_length=5)

class OrderItemResponse(BaseModel):
    product_id: int
    product_name: str
    quantity: int
    price: float

class OrderResponse(BaseModel):
    id: int
    user_id: int
    items: List[OrderItemResponse]
    total: float
    status: str
    delivery_address: str
    created_at: datetime

    class Config:
        from_attributes = True

Аутентификация (auth.py)

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import HTTPException, status, Depends
from fastapi.security import OAuth2PasswordBearer
from typing import Optional

# Настройки
SECRET_KEY = "your-secret-key-here"  # В продакшене используй env
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str) -> Optional[dict]:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        return None

def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    payload = decode_token(token)
    if payload is None:
        raise credentials_exception
    username = payload.get("sub")
    if username is None:
        raise credentials_exception
    return username

База данных (database.py)

from datetime import datetime
from typing import List, Dict, Optional
from schemas import ProductCreate, UserCreate, CartItem

# Хранилища
users_db: Dict[int, dict] = {}
products_db: Dict[int, dict] = {}
carts_db: Dict[int, dict] = {}
orders_db: Dict[int, dict] = {}

# Счётчики ID
user_id_counter = 1
product_id_counter = 1
cart_id_counter = 1
order_id_counter = 1

# ======== Пользователи ========
def get_users() -> List[dict]:
    return list(users_db.values())

def get_user(user_id: int) -> Optional[dict]:
    return users_db.get(user_id)

def get_user_by_username(username: str) -> Optional[dict]:
    for user in users_db.values():
        if user["username"] == username:
            return user
    return None

def get_user_by_email(email: str) -> Optional[dict]:
    for user in users_db.values():
        if user["email"] == email:
            return user
    return None

def create_user(data: UserCreate, hashed_password: str) -> dict:
    global user_id_counter
    user = {
        "id": user_id_counter,
        "username": data.username,
        "email": data.email,
        "hashed_password": hashed_password,
        "is_active": True,
        "created_at": datetime.now()
    }
    users_db[user_id_counter] = user
    user_id_counter += 1
    return user

# ======== Товары ========
def get_products() -> List[dict]:
    return list(products_db.values())

def get_product(product_id: int) -> Optional[dict]:
    return products_db.get(product_id)

def create_product(data: ProductCreate) -> dict:
    global product_id_counter
    product = {
        "id": product_id_counter,
        **data.model_dump(),
        "created_at": datetime.now()
    }
    products_db[product_id_counter] = product
    product_id_counter += 1
    return product

def update_product(product_id: int, data: dict) -> Optional[dict]:
    product = products_db.get(product_id)
    if not product:
        return None
    for key, value in data.items():
        if value is not None:
            product[key] = value
    return product

def delete_product(product_id: int) -> bool:
    if product_id not in products_db:
        return False
    del products_db[product_id]
    return True

# ======== Корзина ========
def get_cart(user_id: int) -> dict:
    if user_id not in carts_db:
        carts_db[user_id] = {"items": [], "user_id": user_id}
    return carts_db[user_id]

def add_to_cart(user_id: int, product_id: int, quantity: int) -> dict:
    cart = get_cart(user_id)
    product = get_product(product_id)
    if not product:
        return {"error": "Product not found"}

    # Проверка наличия
    if product["stock"] < quantity:
        return {"error": "Not enough stock"}

    # Ищем товар в корзине
    for item in cart["items"]:
        if item["product_id"] == product_id:
            item["quantity"] += quantity
            break
    else:
        cart["items"].append({"product_id": product_id, "quantity": quantity})

    # Обновляем остаток
    product["stock"] -= quantity
    return cart

def remove_from_cart(user_id: int, product_id: int) -> dict:
    cart = get_cart(user_id)
    for i, item in enumerate(cart["items"]):
        if item["product_id"] == product_id:
            product = get_product(product_id)
            if product:
                product["stock"] += item["quantity"]
            del cart["items"][i]
            break
    return cart

def clear_cart(user_id: int) -> dict:
    cart = get_cart(user_id)
    # Возвращаем товары на склад
    for item in cart["items"]:
        product = get_product(item["product_id"])
        if product:
            product["stock"] += item["quantity"]
    cart["items"] = []
    return cart

def calculate_cart_total(user_id: int) -> float:
    cart = get_cart(user_id)
    total = 0.0
    for item in cart["items"]:
        product = get_product(item["product_id"])
        if product:
            total += product["price"] * item["quantity"]
    return total

# ======== Заказы ========
def create_order(user_id: int, delivery_address: str) -> dict:
    cart = get_cart(user_id)
    if not cart["items"]:
        return {"error": "Cart is empty"}

    total = calculate_cart_total(user_id)
    order = {
        "id": len(orders_db) + 1,
        "user_id": user_id,
        "items": cart["items"].copy(),
        "total": total,
        "status": "created",
        "delivery_address": delivery_address,
        "created_at": datetime.now()
    }
    orders_db[order["id"]] = order
    cart["items"] = []  # Очищаем корзину
    return order

def get_orders(user_id: int) -> List[dict]:
    return [o for o in orders_db.values() if o["user_id"] == user_id]

def get_order(order_id: int) -> Optional[dict]:
    return orders_db.get(order_id)

Роутеры

routers/products.py:

from fastapi import APIRouter, Depends, HTTPException, status
from typing import List
from schemas import ProductCreate, ProductUpdate, ProductResponse
from database import get_products, get_product, create_product, update_product, delete_product
from auth import get_current_user

router = APIRouter(prefix="/products", tags=["products"])

@router.get("/", response_model=List[ProductResponse])
def list_products():
    return get_products()

@router.get("/{product_id}", response_model=ProductResponse)
def get_product_by_id(product_id: int):
    product = get_product(product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

@router.post("/", response_model=ProductResponse, status_code=201)
def create_new_product(product: ProductCreate, current_user: str = Depends(get_current_user)):
    # Здесь можно добавить проверку прав (например, только админ)
    return create_product(product)

@router.put("/{product_id}", response_model=ProductResponse)
def update_product_by_id(product_id: int, product: ProductUpdate, current_user: str = Depends(get_current_user)):
    updated = update_product(product_id, product.model_dump(exclude_unset=True))
    if not updated:
        raise HTTPException(status_code=404, detail="Product not found")
    return updated

@router.delete("/{product_id}", status_code=204)
def delete_product_by_id(product_id: int, current_user: str = Depends(get_current_user)):
    if not delete_product(product_id):
        raise HTTPException(status_code=404, detail="Product not found")
    return None

routers/auth.py:

from fastapi import APIRouter, HTTPException, status, Depends
from fastapi.security import OAuth2PasswordRequestForm
from schemas import UserCreate, UserResponse, UserLogin, Token
from database import get_user_by_username, get_user_by_email, create_user
from auth import hash_password, verify_password, create_access_token

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/register", response_model=UserResponse, status_code=201)
def register(user: UserCreate):
    if get_user_by_username(user.username):
        raise HTTPException(status_code=400, detail="Username already exists")
    if get_user_by_email(user.email):
        raise HTTPException(status_code=400, detail="Email already exists")

    hashed = hash_password(user.password)
    db_user = create_user(user, hashed)
    return UserResponse(
        id=db_user["id"],
        username=db_user["username"],
        email=db_user["email"],
        is_active=db_user["is_active"],
        created_at=db_user["created_at"]
    )

@router.post("/login", response_model=Token)
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = get_user_by_username(form_data.username)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    if not verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    access_token = create_access_token(data={"sub": user["username"]})
    return {"access_token": access_token, "token_type": "bearer"}

routers/cart.py:

from fastapi import APIRouter, Depends, HTTPException
from schemas import CartItem, CartResponse
from database import get_cart, add_to_cart, remove_from_cart, clear_cart, calculate_cart_total, get_product
from auth import get_current_user

router = APIRouter(prefix="/cart", tags=["cart"])

@router.get("/", response_model=CartResponse)
def get_user_cart(current_user: str = Depends(get_current_user)):
    cart = get_cart(current_user["id"])
    total = calculate_cart_total(current_user["id"])
    return {"items": cart["items"], "total": total}

@router.post("/add")
def add_item_to_cart(item: CartItem, current_user: str = Depends(get_current_user)):
    cart = add_to_cart(current_user["id"], item.product_id, item.quantity)
    if "error" in cart:
        raise HTTPException(status_code=400, detail=cart["error"])
    return cart

@router.delete("/remove/{product_id}")
def remove_item_from_cart(product_id: int, current_user: str = Depends(get_current_user)):
    cart = remove_from_cart(current_user["id"], product_id)
    return cart

@router.delete("/clear")
def clear_user_cart(current_user: str = Depends(get_current_user)):
    cart = clear_cart(current_user["id"])
    return {"message": "Cart cleared"}

routers/orders.py:

from fastapi import APIRouter, Depends, HTTPException, status
from typing import List
from schemas import OrderCreate, OrderResponse
from database import create_order, get_orders, get_order, get_cart
from auth import get_current_user

router = APIRouter(prefix="/orders", tags=["orders"])

@router.post("/", response_model=OrderResponse, status_code=201)
def create_new_order(order: OrderCreate, current_user: str = Depends(get_current_user)):
    result = create_order(current_user["id"], order.delivery_address)
    if "error" in result:
        raise HTTPException(status_code=400, detail=result["error"])
    return result

@router.get("/", response_model=List[OrderResponse])
def list_orders(current_user: str = Depends(get_current_user)):
    return get_orders(current_user["id"])

@router.get("/{order_id}", response_model=OrderResponse)
def get_order_by_id(order_id: int, current_user: str = Depends(get_current_user)):
    order = get_order(order_id)
    if not order:
        raise HTTPException(status_code=404, detail="Order not found")
    if order["user_id"] != current_user["id"]:
        raise HTTPException(status_code=403, detail="Not your order")
    return order

Основное приложение (main.py)

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from routers import products, auth, cart, orders

app = FastAPI(
    title="Shop API",
    description="API для интернет-магазина",
    version="1.0.0"
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# Роутеры
app.include_router(auth.router)
app.include_router(products.router)
app.include_router(cart.router)
app.include_router(orders.router)

@app.get("/")
def root():
    return {"message": "Welcome to Shop API!"}

Тесты (tests/test_api.py)

import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.database import users_db, products_db, carts_db, orders_db

client = TestClient(app)

class TestShopAPI:
    def setup_method(self):
        users_db.clear()
        products_db.clear()
        carts_db.clear()
        orders_db.clear()

    def test_register_user(self):
        response = client.post("/auth/register", json={
            "username": "testuser",
            "email": "test@example.com",
            "password": "secret123"
        })
        assert response.status_code == 201
        data = response.json()
        assert data["username"] == "testuser"
        assert "id" in data

    def test_login_user(self):
        # Регистрация
        client.post("/auth/register", json={
            "username": "testuser",
            "email": "test@example.com",
            "password": "secret123"
        })
        # Логин
        response = client.post("/auth/login", data={
            "username": "testuser",
            "password": "secret123"
        })
        assert response.status_code == 200
        assert "access_token" in response.json()

    def test_create_product(self):
        # Регистрация и логин
        client.post("/auth/register", json={
            "username": "testuser",
            "email": "test@example.com",
            "password": "secret123"
        })
        login = client.post("/auth/login", data={
            "username": "testuser",
            "password": "secret123"
        })
        token = login.json()["access_token"]

        # Создание товара
        response = client.post("/products", json={
            "name": "Test Product",
            "price": 99.99,
            "stock": 10
        }, headers={"Authorization": f"Bearer {token}"})
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "Test Product"
        assert data["price"] == 99.99

    def test_add_to_cart(self):
        # Регистрация и логин
        client.post("/auth/register", json={
            "username": "testuser",
            "email": "test@example.com",
            "password": "secret123"
        })
        login = client.post("/auth/login", data={
            "username": "testuser",
            "password": "secret123"
        })
        token = login.json()["access_token"]

        # Создание товара
        product = client.post("/products", json={
            "name": "Test Product",
            "price": 99.99,
            "stock": 10
        }, headers={"Authorization": f"Bearer {token}"})
        product_id = product.json()["id"]

        # Добавление в корзину
        response = client.post("/cart/add", json={
            "product_id": product_id,
            "quantity": 2
        }, headers={"Authorization": f"Bearer {token}"})
        assert response.status_code == 200
        assert len(response.json()["items"]) == 1
        assert response.json()["items"][0]["quantity"] == 2

Задачи для доработки

Задача 1. Добавь роли пользователей (admin, user).

Задача 2. Добавь поиск товаров по названию и категории.

Задача 3. Добавь пагинацию для списка товаров.

Задача 4. Добавь проверку на отрицательное количество при добавлении в корзину.

Задача 5. Добавь возможность отмены заказа.

Ответы:

Задача 1

# Добавь поле "role" в модель User и проверку в эндпоинтах

Задача 2

@app.get("/products/search")
def search_products(q: str):
    products = get_products()
    return [p for p in products if q.lower() in p["name"].lower()]

Задача 3

@app.get("/products")
def list_products(skip: int = 0, limit: int = 10):
    return get_products()[skip:skip+limit]

Задача 4

def add_to_cart(user_id: int, product_id: int, quantity: int):
    if quantity <= 0:
        return {"error": "Quantity must be positive"}
    # ...

Задача 5

@app.post("/orders/{order_id}/cancel")
def cancel_order(order_id: int, current_user: str = Depends(get_current_user)):
    order = get_order(order_id)
    if not order:
        raise HTTPException(status_code=404, detail="Order not found")
    if order["user_id"] != current_user["id"]:
        raise HTTPException(status_code=403, detail="Not your order")
    order["status"] = "cancelled"
    return {"message": "Order cancelled"}

Запуск и тестирование

Запуск:

uvicorn app.main:app --reload

Тесты:

pytest tests/

Документация:

Swagger: http://localhost:8000/docs
ReDoc: http://localhost:8000/redoc

Шпаргалка

Эндпоинт	Метод	Назначение
`/auth/register`	POST	Регистрация
`/auth/login`	POST	Логин
`/products`	GET/POST	Список/создание товаров
`/products/{id}`	GET/PUT/DELETE	CRUD товара
`/cart`	GET	Корзина
`/cart/add`	POST	Добавить в корзину
`/cart/remove/{id}`	DELETE	Удалить из корзины
`/cart/clear`	DELETE	Очистить корзину
`/orders`	POST	Оформление заказа
`/orders`	GET	Список заказов