Создание веб-сервисов на Python: обзор фреймворков и практические примеры

Расширенные примеры и нестандартные сценарии

FastAPI: работа с базой данных через SQLAlchemy и асинхронность

Пример подключения к PostgreSQL с использованием асинхронного SQLAlchemy. Модель и эндпоинты для CRUD операций.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, sessionmaker
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, Float, select
from pydantic import BaseModel

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

class ItemModel(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))
    price = Column(Float)

class ItemCreate(BaseModel):
    name: str
    price: float

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

app = FastAPI()

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.post("/items/", response_model=ItemCreate)
async def create_item(item: ItemCreate, db: AsyncSession = Depends(get_db)):
    new_item = ItemModel(name=item.name, price=item.price)
    db.add(new_item)
    await db.commit()
    await db.refresh(new_item)
    return new_item

@app.get("/items/")
async def read_items(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(ItemModel))
    items = result.scalars().all()
    return items

Результат: при отправке POST запроса с телом {"name": "laptop", "price": 999.99} сервер вернёт созданный объект. GET запрос вернёт список всех товаров.

POST /items/ -> 201
{
  "name": "laptop",
  "price": 999.99
}

GET /items/ -> 200
[
  {
    "id": 1,
    "name": "laptop",
    "price": 999.99
  }
]

FastAPI: валидация данных с использованием Pydantic и кастомные валидаторы

Добавление дополнительных проверок, например, проверка уникальности имени.

from pydantic import BaseModel, validator
from typing import Optional

class ItemCreate(BaseModel):
    name: str
    price: float = None

    @validator('name')
    def name_must_be_unique(cls, v):
        # Здесь может быть запрос к БД для проверки уникальности
        if v == "existing_name":
            raise ValueError('Item with this name already exists')
        return v
    
    @validator('price')
    def price_must_be_positive(cls, v):
        if v is not None and v <= 0:
            raise ValueError('Price must be positive')
        return v

При попытке создать элемент с существующим именем будет выброшена ошибка валидации 422.

Flask: интеграция с базой данных SQLite и ORM SQLAlchemy

Пример простого приложения с поддержкой CRUD.

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///items.db'
db = SQLAlchemy(app)

class Item(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)
    price = db.Column(db.Float, nullable=False)

    def serialize(self):
        return {'id': self.id, 'name': self.name, 'price': self.price}

@app.route('/items', methods=['GET'])
def get_items():
    items = Item.query.all()
    return jsonify([item.serialize() for item in items])

@app.route('/items', methods=['POST'])
def create_item():
    data = request.json
    new_item = Item(name=data['name'], price=data['price'])
    db.session.add(new_item)
    db.session.commit()
    return jsonify(new_item.serialize()), 201

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

Результат: при запуске создаётся таблица items. POST запрос добавляет запись.

aiohttp: middleware для логирования и обработки ошибок

from aiohttp import web
import logging

logging.basicConfig(level=logging.INFO)

@web.middleware
async def error_middleware(request, handler):
    try:
        response = await handler(request)
        return response
    except web.HTTPException as ex:
        raise
    except Exception as ex:
        logging.error(f"Unhandled error: {ex}")
        return web.json_response({'error': 'Internal server error'}, status=500)

async def handle(request):
    return web.json_response({'message': 'Success'})

app = web.Application(middlewares=[error_middleware])
app.router.add_get('/', handle)
web.run_app(app)

Результат: если в обработчике возникнет исключение, middleware вернёт JSON с ошибкой 500.

Django REST Framework: пагинация и фильтрация

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
    'PAGE_SIZE': 10
}

# views.py
from rest_framework import generics
from myapp.models import Item
from .serializers import ItemSerializer
from django_filters.rest_framework import DjangoFilterBackend

class ItemList(generics.ListAPIView):
    queryset = Item.objects.all()
    serializer_class = ItemSerializer
    filter_backends = [DjangoFilterBackend]
    filterset_fields = ['name', 'price']

Теперь GET /api/items/?name=laptop&page=2 вернёт вторую страницу с отфильтрованными записями.