Какие знаешь ключевые этапы пути запроса в Backend приложении?

Какие знаешь ключевые этапы пути запроса в Backend приложении?

Понимание полного цикла жизни HTTP-запроса от клиента до БД и обратно — важнейший навык для любого backend разработчика.

Общая архитектура: Request → Response

Клиент (браузер/мобильное приложение)
    ↓
1. HTTP запрос отправляется
    ↓
2. Сетевой слой (TCP/IP, TLS)
    ↓
3. Load Balancer (если есть)
    ↓
4. Web Server (Nginx, Apache) или ASGI сервер
    ↓
5. MIDDLEWARE (аутентификация, логирование)
    ↓
6. ROUTING (определяем, какой handler вызвать)
    ↓
7. CONTROLLER/HANDLER (бизнес-логика)
    ↓
8. SERVICE LAYER (бизнес-правила)
    ↓
9. DATABASE ACCESS (ORM или SQL запросы)
    ↓
10. DATABASE (PostgreSQL, MySQL, etc.)
    ↓
(обратный путь с результатами)
    ↓
11. Response Serialization (JSON, XML)
    ↓
12. Response отправляется клиенту

Детальное описание каждого этапа

1. HTTP Запрос

GET /api/v1/users/123 HTTP/1.1
Host: api.example.com
Authorization: Bearer token_xyz
Content-Type: application/json
User-Agent: Mozilla/5.0

{"name": "John"}

Что здесь:

• Метод (GET, POST, PUT, DELETE)
• URL/Path (/api/v1/users/123)
• Headers (авторизация, тип контента)
• Body (данные)
• Query params (?page=1&limit=10)

2. Network Layer

ТCP handshake (3-way)
TLS/SSL encryption (если HTTPS)
Данные отправляются в пакетах

На уровне Python:

import socket

# На этом уровне работают фреймворки
# FastAPI, Flask, Django используют асинхронные/синхронные сокеты

3. Load Balancer

CloudFlare / AWS ELB / Nginx (reverse proxy)
    ↓
    ├─ Server 1 (если этот перегружен → пропускаем)
    ├─ Server 2 (выбираем это)
    └─ Server 3

Стратегии:

• Round-robin (по кругу)
• Least connections (к серверу с наименьшей нагрузкой)
• IP hash (одного клиента всегда на один сервер)
• Weighted (разные серверы разной мощности)

4. Web Server / ASGI

# FastAPI/Uvicorn пример
import uvicorn
from fastapi import FastAPI

app = FastAPI()

if __name__ == "__main__":
    # Uvicorn слушает на 0.0.0.0:8000
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Nginx перенаправляет на localhost:8000

ASGI сервер (Uvicorn):

• Принимает HTTP запрос
• Парсит headers и body
• Передаёт в приложение
• Отправляет response обратно

5. Middleware (Слой обработки запроса)

from fastapi import FastAPI, Request
from starlette.middleware.cors import CORSMiddleware
import time

app = FastAPI()

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

# Custom middleware для логирования
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    
    # Логируем входящий запрос
    print(f"📥 {request.method} {request.url.path}")
    
    # Передаём запрос дальше
    response = await call_next(request)
    
    # Логируем время обработки
    process_time = time.time() - start_time
    print(f"📤 {response.status_code} in {process_time:.3f}s")
    
    response.headers["X-Process-Time"] = str(process_time)
    return response

# Authentication middleware
@app.middleware("http")
async def authenticate(request: Request, call_next):
    token = request.headers.get("Authorization")
    
    if not token:
        return {"error": "No token"}
    
    # Проверяем токен (обычно с JWT)
    try:
        user = verify_jwt_token(token)
        request.state.user = user
    except:
        return {"error": "Invalid token"}
    
    response = await call_next(request)
    return response

Порядок middleware (снаружи внутрь):

CORS → Logging → Authentication → Rate Limiting → Compression → Handler

6. Routing (Определение handler'а)

from fastapi import FastAPI

app = FastAPI()

# Динамическая маршрутизация
@app.get("/users/{user_id}")  # Path parameter
async def get_user(user_id: int, skip: int = 0, limit: int = 10):  # Query params
    # handler вызывается с распарсенными параметрами
    pass

@app.post("/users/")
async def create_user(user_data: UserSchema):  # Body parameter
    pass

@app.put("/users/{user_id}")
async def update_user(user_id: int, user_data: UserSchema):
    pass

# Внутри FastAPI
# 1. Ищет matching route
# 2. Парсит path parameters
# 3. Парсит query parameters
# 4. Валидирует с Pydantic
# 5. Вызывает handler

7. Controller/Handler (Точка входа)

from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session

@app.get("/api/v1/users/{user_id}")
async def get_user(
    user_id: int,
    db: Session = Depends(get_db),  # Dependency injection
    current_user: User = Depends(get_current_user)  # Аутентификация
):
    """Handler — тонкий слой, вызывает use case"""
    
    if not current_user:
        raise HTTPException(status_code=401, detail="Not authenticated")
    
    # Вызываем use case из service layer
    user = await UserService.get_user(user_id, db)
    
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    # Возвращаем данные (FastAPI сериализует в JSON)
    return user

8. Service Layer (Бизнес-логика)

from sqlalchemy.orm import Session

class UserService:
    @staticmethod
    async def get_user(user_id: int, db: Session):
        """Получить пользователя с правами"""
        
        # Получаем пользователя
        user = db.query(User).filter(User.id == user_id).first()
        
        if not user:
            return None
        
        # Проверяем права доступа
        if user.is_deleted:
            return None  # Скрываем удалённых пользователей
        
        # Обогащаем данные (например, считаем подписчиков)
        user.followers_count = db.query(Follow).filter(
            Follow.following_id == user_id
        ).count()
        
        return user
    
    @staticmethod
    async def create_user(user_data: UserCreateSchema, db: Session):
        """Создание с валидацией"""
        
        # Проверяем уникальность
        existing = db.query(User).filter(User.email == user_data.email).first()
        if existing:
            raise ValueError("Email already exists")
        
        # Хешируем пароль
        from passlib.context import CryptContext
        pwd_context = CryptContext(schemes=["bcrypt"])
        hashed_password = pwd_context.hash(user_data.password)
        
        # Создаём пользователя
        user = User(
            name=user_data.name,
            email=user_data.email,
            hashed_password=hashed_password
        )
        
        db.add(user)
        db.commit()
        db.refresh(user)
        
        return user

9. Repository/DAO Layer (Доступ к БД)

from sqlalchemy.orm import Session
from sqlalchemy import select

class UserRepository:
    """Абстракция доступа к БД"""
    
    def __init__(self, db: Session):
        self.db = db
    
    def get_by_id(self, user_id: int):
        return self.db.query(User).filter(User.id == user_id).first()
    
    def get_by_email(self, email: str):
        return self.db.query(User).filter(User.email == email).first()
    
    def create(self, user: User):
        self.db.add(user)
        self.db.commit()
        self.db.refresh(user)
        return user
    
    def delete(self, user_id: int):
        user = self.get_by_id(user_id)
        if user:
            self.db.delete(user)
            self.db.commit()

10. Database Query

# ORM query (SQLAlchemy)
user = db.query(User).filter(User.id == 123).first()

# Генерирует SQL:
# SELECT * FROM users WHERE id = 123 LIMIT 1;

# Отправляется на PostgreSQL/MySQL
# БД выполняет query, возвращает результат
# ORM маппирует результат в объект User

Порядок выполнения SQL:

1. Parse (проверка синтаксиса)
2. Compile (оптимизация плана)
3. Execute (выполнение)
4. Fetch (возврат результатов)

11. Response Creation

from pydantic import BaseModel

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    followers_count: int
    
    class Config:
        from_attributes = True  # SQLAlchemy интеграция

# FastAPI автоматически сериализует
return UserResponse.from_orm(user)  # User SQLAlchemy объект → JSON

# Генерирует:
# {"id": 123, "name": "John", "email": "john@example.com", "followers_count": 5}

12. Response отправляется

JSON body (200 OK)
↓
Content-Type: application/json
Content-Length: 245
Cache-Control: no-cache
↓
Oтправляется клиенту
↓
Браузер парсит и отображает

Полный пример: Request → Response

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
import asyncio

app = FastAPI()

# 1. DATABASE CONNECTION
engine = create_engine("postgresql://user:pass@localhost/db")
SessionLocal = sessionmaker(bind=engine)

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

# 2. MODELS
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    email = Column(String(100), unique=True)

# 3. SCHEMAS (для валидации input/output)
class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    
    class Config:
        from_attributes = True

# 4. SERVICE LAYER
class UserService:
    @staticmethod
    def create_user(user_data: UserCreate, db: Session) -> User:
        db_user = User(name=user_data.name, email=user_data.email)
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user

# 5. HANDLER
@app.post("/api/v1/users", response_model=UserResponse)
async def create_user_handler(
    user_data: UserCreate,
    db: Session = Depends(get_db)
):
    return UserService.create_user(user_data, db)

# 6. RUN
# uvicorn main:app --reload

# REQUEST FLOW:
# POST /api/v1/users
# Body: {"name": "John", "email": "john@example.com"}
#   ↓ Middleware (логирование)
#   ↓ Routing (находит create_user_handler)
#   ↓ Pydantic валидация (UserCreate)
#   ↓ Dependency injection (get_db → Session)
#   ↓ Handler (вызывает UserService.create_user)
#   ↓ ORM (INSERT INTO users ...)
#   ↓ Database (PostgreSQL)
#   ↓ Response serialization (UserResponse)
#   ↓ JSON response
# ← 201 Created {"id": 1, "name": "John", "email": "john@example.com"}

Типичные ошибки на пути

# ❌ 1. Медленный запрос в handler
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).get(user_id)  # Медленно
    posts = [p for p in user.posts]  # N+1 запрос!

# ✓ Правильно
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).options(
        joinedload(User.posts)  # JOIN вместо N+1
    ).get(user_id)

# ❌ 2. Блокирующий код в async handler
@app.get("/users/{user_id}")
async def get_user(user_id: int):  # async!
    time.sleep(5)  # ❌ Блокирует весь event loop

# ✓ Правильно
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    await asyncio.sleep(5)  # ✓ Асинхронная задержка

# ❌ 3. N+1 запросы
users = db.query(User).all()
for user in users:
    print(user.posts)  # 1 + n запросов!

# ✓ Правильно
users = db.query(User).options(joinedload(User.posts)).all()  # 1 запрос

Заключение

Полный путь запроса состоит из:

1. Network → TCP/TLS
2. Load Balancer → выбор сервера
3. Middleware → логирование, аутентификация
4. Routing → определение handler
5. Handler → вызов сервиса
6. Service → бизнес-логика
7. Repository → доступ к БД
8. Database → выполнение SQL
9. Response → сериализация + отправка

Оптимизация на каждом этапе может дать 100x ускорение!