Практическое руководство: создание веб-приложений на Python 3

Расширенные примеры веб-приложений на FastAPI

Пример 1. FastAPI с SQLAlchemy и SQLite

Установка:

pip install fastapi uvicorn sqlalchemy databases

Код приложения (main.py):

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class ItemDB(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)
    created_at = Column(DateTime, default=datetime.utcnow)

Base.metadata.create_all(bind=engine)

app = FastAPI()

class ItemCreate(BaseModel):
    name: str
    description: str

class ItemResponse(ItemCreate):
    id: int
    created_at: datetime
    class Config:
        orm_mode = True

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

@app.post("/items/", response_model=ItemResponse)
async def create_item(item: ItemCreate):
    db = next(get_db())
    db_item = ItemDB(name=item.name, description=item.description)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items/{item_id}", response_model=ItemResponse)
async def read_item(item_id: int):
    db = next(get_db())
    item = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

Результат (запуск и тест через curl):

$ uvicorn main:app --reload
INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

$ curl -X POST "http://localhost:8000/items/" -H "Content-Type: application/json" -d '{"name":"Тестовый предмет","description":"Описание"}'
{"id":1,"name":"Тестовый предмет","description":"Описание","created_at":"2025-04-01T12:00:00"}

$ curl "http://localhost:8000/items/1"
{"id":1,"name":"Тестовый предмет","description":"Описание","created_at":"2025-04-01T12:00:00"}

Пример 2. FastAPI с WebSocket

Код приложения с чатом через WebSocket:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    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/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client #{client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")

Результат (тестирование через JavaScript в браузере):

// Открытие соединения в консоли браузера:
let ws = new WebSocket("ws://localhost:8000/ws/1");
ws.onmessage = function(event) { console.log(event.data); };
ws.send("Привет, мир!");
// Вывод: Client #1: Привет, мир!

Пример 3. FastAPI с фоновыми задачами (BackgroundTasks)

Код приложения с отправкой email в фоне:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import time

app = FastAPI()

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(f"{time.ctime()}: {message}\n")

class Notification(BaseModel):
    email: str
    subject: str
    body: str

@app.post("/send-notification/")
async def send_notification(notification: Notification, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Email to {notification.email}: {notification.subject}")
    # Здесь можно добавить реальную отправку email
    return {"status": "Notification will be sent in background"}

Результат (запрос и содержимое файла log.txt):

$ curl -X POST "http://localhost:8000/send-notification/" -H "Content-Type: application/json" -d '{"email":"test@example.com","subject":"Привет","body":"Текст"}'
{"status":"Notification will be sent in background"}

$ cat log.txt
Wed Apr  1 12:05:00 2025: Email to test@example.com: Привет