FastAPI и Flask: Отличия, производительность и примеры использования

from flask import Flask
app = Flask(__name__)
 
@app.route('/')
def hello_world():
    return 'Hello, World!'
 
if __name__ == '__main__':
    app.run()

from fastapi import FastAPI, Query
from pydantic import BaseModel
 
app = FastAPI()
 
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False
 
@app.get("/")
async def read_root():
    return {"Hello": "World"}
 
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = Query(None, max_length=50)):
    return {"item_id": item_id, "q": q}

# Пример кода FastAPI с асинхронными запросами к БД
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Асинхронный запрос к БД
    user = await database.fetch_one(
        query="SELECT * FROM users WHERE id = :id",
        values={"id": user_id}
    )
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# FastAPI с асинхронными запросами к внешним API
@app.get("/aggregate")
async def aggregate_data():
    # Выполняем несколько запросов параллельно
    tasks = [
        fetch_service_data(f"https://api.example.com/service/{i}")
        for i in range(1, 6)
    ]
    results = await asyncio.gather(*tasks)
    return {"results": results}
 
async def fetch_service_data(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

# Пример использования Flask с расширениями
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_migrate import Migrate
 
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/dbname'
db = SQLAlchemy(app)
migrate = Migrate(app, db)
login_manager = LoginManager(app)

# FastAPI автоматически генерирует документацию
from fastapi import FastAPI, Path, Query
from pydantic import BaseModel
 
app = FastAPI(
    title="My Awesome API",
    description="This API does awesome things",
    version="0.1.0"
)
 
class User(BaseModel):
    id: int
    name: str
    email: str
 
@app.get("/users/{user_id}", response_model=User)
async def get_user(
    user_id: int = Path(..., title="User ID", gt=0),
    include_details: bool = Query(False, title="Include details")
):
    """
    Retrieve a user by ID.
    
    This endpoint returns user information based on the provided user ID.
    """
    # ...

# Flask - идеален для быстрого прототипирования
from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
# Временное хранилище данных
products = []
 
@app.route('/products', methods=['GET'])
def get_products():
    return jsonify(products)
 
@app.route('/products', methods=['POST'])
def add_product():
    new_product = request.json
    products.append(new_product)
    return jsonify({"id": len(products)}), 201

# FastAPI - силен в высоконагруженных API с сложной валидацией
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, validator
import asyncio
 
app = FastAPI()
 
class OrderData(BaseModel):
    product_id: int
    quantity: int
    customer_id: str
    
    @validator('quantity')
    def quantity_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('Quantity must be positive')
        return v
 
@app.post("/orders/")
async def create_order(order: OrderData):
    # Асинхронно проверяем наличие товара
    inventory_status = await check_inventory(order.product_id, order.quantity)
    if not inventory_status['available']:
        raise HTTPException(status_code=400, detail="Not enough items in stock")
    
    # Асинхронно создаём заказ и уведомляем склад
    order_task = create_order_async(order)
    notification_task = notify_warehouse_async(order)
    
    await asyncio.gather(order_task, notification_task)
    return {"status": "success"}

# Flask версия
@app.route('/process_payment', methods=['POST'])
def process_payment():
    data = request.json
    # Валидация данных вручную
    if not data or 'amount' not in data or 'user_id' not in data:
        return jsonify({"error": "Invalid data"}), 400
    
    # Обработка платежа
    result = payment_service.process(data['user_id'], data['amount'])
    return jsonify(result)
 
# FastAPI версия
class PaymentData(BaseModel):
    user_id: str
    amount: float
    
    @validator('amount')
    def amount_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('Amount must be positive')
        return v
 
@app.post('/process_payment')
async def process_payment(payment: PaymentData):
    # Валидация происходит автоматически
    result = await payment_service.process_async(payment.user_id, payment.amount)
    return result

from flask import Flask, request, redirect
from flask_login import LoginManager, login_required, current_user
 
app = Flask(__name__)
app.secret_key = "супер-секретный-ключ"
login_manager = LoginManager(app)
 
@login_manager.user_loader
def load_user(user_id):
    return User.get(user_id)
 
@app.route('/protected')
@login_required
def protected():
    return f'Привет, {current_user.name}!'

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
 
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Недействительные учетные данные",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        return username
    except JWTError:
        raise credentials_exception
 
@app.get("/users/me")
async def read_users_me(current_user: str = Depends(get_current_user)):
    return {"username": current_user}

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
 
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:password@localhost/dbname'
db = SQLAlchemy(app)
 
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    email = db.Column(db.String(120), unique=True)
 
@app.route('/users/<username>')
def get_user(username):
    user = User.query.filter_by(username=username).first_or_404()
    return {'username': user.username, 'email': user.email}

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from .database import get_session, User
 
app = FastAPI()
 
@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User).where(User.id == user_id))
    user = result.scalars().first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
 
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:password@localhost/dbname'
app.config['SQLALCHEMY_POOL_SIZE'] = 10
app.config['SQLALCHEMY_MAX_OVERFLOW'] = 20
db = SQLAlchemy(app)

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
 
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    echo=False,
    pool_pre_ping=True
)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
async def get_db():
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()
 
app = FastAPI()
 
@app.get("/items/{item_id}")
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
    # Работа с базой данных
    ...

from flask import Flask
from flask_pymongo import PyMongo
 
app = Flask(__name__)
app.config["MONGO_URI"] = "mongodb://localhost:27017/mydatabase"
mongo = PyMongo(app)
 
@app.route("/users")
def get_users():
    users = mongo.db.users.find()
    return {"users": list(users)}

from fastapi import FastAPI, Depends
from motor.motor_asyncio import AsyncIOMotorClient
 
app = FastAPI()
 
async def get_database():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    try:
        yield client.mydatabase
    finally:
        client.close()
 
@app.get("/users")
async def get_users(db = Depends(get_database)):
    users = await db.users.find().to_list(100)
    return {"users": users}

from functools import wraps
from flask import request, abort
 
def require_api_key(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        api_key = request.headers.get('X-API-Key')
        if not api_key or not is_valid_key(api_key):
            abort(401)
        return f(*args, **kwargs)
    return decorated_function
 
@app.route('/api/data')
@require_api_key
def get_data():
    # Защищенный эндпоинт
    return {"data": "секретные данные"}

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security.api_key import APIKeyHeader
from starlette.status import HTTP_403_FORBIDDEN
 
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
 
async def get_api_key(api_key: str = Security(api_key_header)):
    if not api_key or not is_valid_key(api_key):
        raise HTTPException(
            status_code=HTTP_403_FORBIDDEN, detail="Could not validate API key"
        )
    return api_key
 
@app.get("/api/data")
async def get_data(api_key: str = Depends(get_api_key)):
    # Защищенный эндпоинт
    return {"data": "секретные данные"}

from flask import Flask
from flask_caching import Cache
 
app = Flask(__name__)
cache = Cache(app, config={'CACHE_TYPE': 'redis', 'CACHE_REDIS_URL': 'redis://localhost:6379/0'})
 
@app.route('/users')
@cache.cached(timeout=60)  # Кэшируем на 60 секунд
def get_users():
    # Представим, что это дорогостоящий запрос к БД
    users = db.session.query(User).all()
    return jsonify([user.to_dict() for user in users])

from fastapi import FastAPI, Depends
import aioredis
import ujson
 
app = FastAPI()
 
async def get_redis():
    redis = aioredis.from_url("redis://localhost", encoding="utf-8", decode_responses=True)
    try:
        yield redis
    finally:
        await redis.close()
 
@app.get("/users")
async def get_users(redis: aioredis.Redis = Depends(get_redis)):
    # Проверяем кэш
    cached_data = await redis.get("users:all")
    if cached_data:
        return ujson.loads(cached_data)
    
    # Если в кэше нет данных, получаем их и кэшируем
    users = await db.fetch_all("SELECT * FROM users")
    await redis.set("users:all", ujson.dumps(users), ex=60)  # 60 секунд
    return users

# В обработчике создания нового пользователя
async def create_user(user_data, redis):
    # Сохраняем пользователя в БД
    user_id = await db.execute("INSERT INTO users ... RETURNING id", user_data)
    
    # Инвалидируем кэш списка всех пользователей
    await redis.delete("users:all")
    
    return user_id

from flask import Flask
from flask_socketio import SocketIO, emit
 
app = Flask(__name__)
socketio = SocketIO(app)
 
@socketio.on('connect')
def handle_connect():
    print('Клиент подключился')
 
@socketio.on('message')
def handle_message(data):
    print('Получено сообщение:', data)
    emit('response', {'data': f'Вы отправили: {data}'})
 
if __name__ == '__main__':
    socketio.run(app)

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 
class ConnectionManager:
    def __init__(self):
        self.active_connections = []
 
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
 
    async def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
 
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)
 
manager = ConnectionManager()
 
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Кто-то написал: {data}")
    except WebSocketDisconnect:
        await manager.disconnect(websocket)

myapp/
  __init__.py  # Фабрика приложения
  config.py    # Конфигурация
  models/      # Модели данных
  views/
    __init__.py
    auth.py    # Blueprint авторизации
    api.py     # Blueprint API
  services/    # Бизнес-логика
  utils/       # Вспомогательные функции

myapp/
  __init__.py
  config.py
  models/      # Pydantic и ORM модели
  routers/     # API маршрутизация
    __init__.py
    auth.py
    items.py
  dependencies/  # Зависимости
  services/    # Бизнес-логика
  utils/

@app.errorhandler(404)
def not_found_error(error):
    return render_template('404.html'), 404
 
@app.errorhandler(500)
def internal_error(error):
    db.session.rollback()  # Откатываем сессию в случае ошибки
    return render_template('500.html'), 500

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
 
app = FastAPI()
 
class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name
 
@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Упс! {exc.name} приносит извинения!"},
    )
 
@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "bad":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

@app.route('/users', methods=['POST'])
def create_user():
    data = request.json
    if not data:
        return jsonify({"error": "Empty request"}), 400
    
    # Валидация полей
    if 'username' not in data or not isinstance(data['username'], str):
        return jsonify({"error": "Username is required and must be a string"}), 400
    
    if 'email' not in data or not re.match(r"[^@]+@[^@]+\.[^@]+", data['email']):
        return jsonify({"error": "Valid email is required"}), 400
    
    # Если всё хорошо, создаем пользователя
    user = User(username=data['username'], email=data['email'])
    db.session.add(user)
    db.session.commit()
    return jsonify(user.to_dict()), 201

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr, validator
from sqlalchemy.orm import Session
 
app = FastAPI()
 
class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str
    
    @validator('username')
    def username_must_be_valid(cls, v):
        if len(v) < 3:
            raise ValueError('Username must be at least 3 characters')
        return v
 
@app.post("/users/", response_model=UserRead)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(**user.dict(exclude={"password"}))
    db_user.hashed_password = get_password_hash(user.password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# Flask + SQLAlchemy реализация
class ProductRepository:
  def __init__(self, session):
      self.session = session
  
  def get_by_id(self, product_id):
      return self.session.query(Product).filter(Product.id == product_id).first()
  
  def get_all(self, limit=100, offset=0):
      return self.session.query(Product).limit(limit).offset(offset).all()
  
  def create(self, name, price, category_id):
      product = Product(name=name, price=price, category_id=category_id)
      self.session.add(product)
      self.session.commit()
      return product
 
# Использование в контроллере
@app.route('/products/<int:product_id>')
def get_product(product_id):
  repo = ProductRepository(db.session)
  product = repo.get_by_id(product_id)
  if not product:
      abort(404)
  return jsonify(product.to_dict())

# FastAPI + SQLAlchemy async реализация
class ProductRepository:
  def __init__(self, session):
      self.session = session
  
  async def get_by_id(self, product_id):
      result = await self.session.execute(select(Product).where(Product.id == product_id))
      return result.scalars().first()
  
  async def get_all(self, limit=100, offset=0):
      result = await self.session.execute(select(Product).limit(limit).offset(offset))
      return result.scalars().all()
  
  async def create(self, name, price, category_id):
      product = Product(name=name, price=price, category_id=category_id)
      self.session.add(product)
      await self.session.commit()
      return product
 
# Использование в контроллере
@app.get('/products/{product_id}', response_model=ProductRead)
async def get_product(product_id: int, session: AsyncSession = Depends(get_session)):
  repo = ProductRepository(session)
  product = await repo.get_by_id(product_id)
  if not product:
      raise HTTPException(status_code=404, detail="Product not found")
  return product

from fastapi.testclient import TestClient
from unittest.mock import MagicMock, patch
 
# Мокаем зависимость базы данных
async def override_get_db():
  # Возвращаем мок сессии БД
  return MagicMock()
 
app.dependency_overrides[get_db] = override_get_db
 
client = TestClient(app)
 
def test_read_product():
  # Настраиваем поведение мока
  app.dependency_overrides[get_db]().execute.return_value.scalars.return_value.first.return_value = {
      "id": 1,
      "name": "Test Product",
      "price": 29.99
  }
  
  response = client.get("/products/1")
  assert response.status_code == 200
  assert response.json() == {"id": 1, "name": "Test Product", "price": 29.99}