Розробка бекенду сайту на Python (FastAPI)
FastAPI — сучасний Python-фреймворк, який будує API навколо типів. Ви оголошуєте функцію з type hints, а FastAPI автоматично генерує валідацію через Pydantic, OpenAPI-документацію та JSON Schema. Без ручної документації, без окремих валідаторів — все виводиться з типів.
Продуктивність FastAPI на асинхронних операціях (I/O) порівнюється з Node.js. Для CPU-bound задач — окреме питання, там потрібні process pool або вивантаження в Celery.
Основи та структура ендпоінтів
from fastapi import FastAPI, Depends, HTTPException, Query, Path, status
from pydantic import BaseModel, EmailStr, Field
from typing import Optional, List
import uvicorn
app = FastAPI(
title="My API",
version="1.0.0",
docs_url="/api/docs",
redoc_url="/api/redoc"
)
class ProductCreate(BaseModel):
name: str = Field(..., min_length=2, max_length=255)
price: float = Field(..., gt=0)
category_id: int
description: Optional[str] = None
class ProductResponse(BaseModel):
id: int
name: str
price: float
category_id: int
class Config:
from_attributes = True # дозволяє створювати з ORM-об'єктів
@app.get('/api/v1/products', response_model=List[ProductResponse])
async def list_products(
page: int = Query(1, ge=1),
limit: int = Query(20, ge=1, le=100),
category_id: Optional[int] = Query(None),
db: AsyncSession = Depends(get_db)
):
offset = (page - 1) * limit
query = select(Product).offset(offset).limit(limit)
if category_id:
query = query.where(Product.category_id == category_id)
result = await db.execute(query)
return result.scalars().all()
@app.post('/api/v1/products', response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
async def create_product(
body: ProductCreate,
current_user: User = Depends(require_role('admin')),
db: AsyncSession = Depends(get_db)
):
product = Product(**body.model_dump())
db.add(product)
await db.commit()
await db.refresh(product)
return product
Dependency Injection
DI в FastAPI працює через Depends — один з найкращих механізмів в Python-екосистемі:
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
# Підключення до БД на запит
async_engine = create_async_engine(settings.DATABASE_URL, pool_size=10)
async def get_db():
async with AsyncSession(async_engine) as session:
try:
yield session
except Exception:
await session.rollback()
raise
finally:
await session.close()
# Аутентифікація
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/api/auth/token')
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> User:
try:
payload = jwt.decode(token, settings.JWT_SECRET, algorithms=['HS256'])
user_id: int = payload.get('sub')
except JWTError:
raise HTTPException(status_code=401, detail='Invalid token')
user = await db.get(User, user_id)
if not user or not user.is_active:
raise HTTPException(status_code=401, detail='Inactive user')
return user
def require_role(*roles: str):
async def checker(user: User = Depends(get_current_user)) -> User:
if user.role not in roles:
raise HTTPException(status_code=403, detail='Insufficient permissions')
return user
return checker
SQLAlchemy 2.0 (async)
FastAPI добре працює з SQLAlchemy 2.0 в async-режимі:
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, Numeric, ForeignKey, DateTime, func
class Base(DeclarativeBase):
pass
class Product(Base):
__tablename__ = 'products'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(255))
slug: Mapped[str] = mapped_column(String(255), unique=True)
price: Mapped[float] = mapped_column(Numeric(10, 2))
category_id: Mapped[int | None] = mapped_column(ForeignKey('categories.id'), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
category: Mapped['Category'] = relationship(back_populates='products', lazy='selectin')
lazy='selectin' для зв'язків — кращий вибір в async-режимі, уникає N+1 без явних join.
Фонові задачі
Прості задачи — через BackgroundTasks, важкі — через Celery:
from fastapi import BackgroundTasks
import asyncio
# Легкі async-задачи прямо в запросі
@app.post('/api/orders/{order_id}/confirm')
async def confirm_order(
order_id: int,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db)
):
order = await get_order_or_404(order_id, db)
order.status = 'confirmed'
await db.commit()
# Не чекаємо завершення
background_tasks.add_task(send_confirmation_email, order.user.email, order_id)
background_tasks.add_task(update_inventory, order.items)
return {'status': 'confirmed'}
# Важкі задачи — Celery
from celery import Celery
celery = Celery(__name__, broker=settings.REDIS_URL)
@celery.task(name='generate_report')
def generate_report(user_id: int, date_range: dict):
# CPU-інтенсивна обробка
pass
Middleware та CORS
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*']
)
@app.middleware('http')
async def add_process_time(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
response.headers['X-Process-Time'] = str(round(duration * 1000, 2))
return response
WebSocket
from fastapi import WebSocket, WebSocketDisconnect
class ConnectionManager:
def __init__(self):
self.active: dict[int, list[WebSocket]] = {}
async def connect(self, user_id: int, ws: WebSocket):
await ws.accept()
self.active.setdefault(user_id, []).append(ws)
async def broadcast_to_user(self, user_id: int, message: dict):
for ws in self.active.get(user_id, []):
await ws.send_json(message)
manager = ConnectionManager()
@app.websocket('/ws/notifications')
async def notifications_ws(
websocket: WebSocket,
token: str = Query(...),
):
user = await verify_ws_token(token)
await manager.connect(user.id, websocket)
try:
while True:
await websocket.receive_text() # тримаємо з'єднання
except WebSocketDisconnect:
manager.disconnect(user.id, websocket)
Розгортання
FastAPI запускається через Uvicorn або Gunicorn з Uvicorn workers:
# Production: кілька воркерів
gunicorn app.main:app \
-w 4 \
-k uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 30 \
--keepalive 5
Терміни розробки
- Каркас + моделі + auth — 4–7 днів
- API endpoints + валідація — 1–2 тижні
- Інтеграції та фонові задачи — 1–2 тижні
- Тести (pytest + httpx AsyncClient) — 5–7 днів
API для сайту середнього масштабу: 4–8 тижнів. FastAPI виграє, коли команда знає Python та потрібна автодокументація — Swagger/ReDoc генерується з коду без зусиль.