Starlette – основа для FastAPI

онлайн тренажер по питону
Онлайн-тренажер Python для начинающих

Изучайте Python легко и без перегрузки теорией. Решайте практические задачи с автоматической проверкой, получайте подсказки на русском языке и пишите код прямо в браузере — без необходимости что-либо устанавливать.

Начать курс

Введение

Starlette — это современный, высокопроизводительный асинхронный веб-фреймворк для Python, который служит фундаментом для популярного FastAPI. Созданный специально для работы с ASGI (Asynchronous Server Gateway Interface), Starlette предоставляет разработчикам мощные инструменты для создания быстрых и масштабируемых веб-приложений. В этой статье мы подробно рассмотрим все аспекты работы с данным фреймворком, его возможности и преимущества.

Что такое Starlette

История и цели проекта

Starlette был создан Томом Кристенсеном (Tom Christie), известным разработчиком Django REST Framework. Проект появился в 2018 году с целью предоставить разработчикам легковесный, но мощный инструмент для создания асинхронных веб-приложений на Python. Основная философия Starlette заключается в предоставлении минималистичного, но функционального фреймворка, который не навязывает разработчику конкретные решения.

Основные особенности фреймворка

Starlette отличается рядом ключевых особенностей, которые делают его привлекательным для современной веб-разработки:

  • Полная поддержка ASGI — обеспечивает максимальную совместимость с современными асинхронными серверами
  • Нативная асинхронность — все компоненты фреймворка разработаны с учетом async/await синтаксиса
  • Минималистичная архитектура — предоставляет только необходимые компоненты, не перегружая функциональностью
  • Высокая производительность — демонстрирует отличные показатели скорости обработки запросов
  • Модульность — позволяет использовать только необходимые компоненты
  • Простота тестирования — включает встроенные инструменты для написания тестов

Как Starlette связан с FastAPI

Архитектура FastAPI: использование Starlette в качестве основы

FastAPI построен поверх Starlette, используя его как базовый слой для обработки HTTP-запросов, маршрутизации и middleware. Это позволяет FastAPI сосредоточиться на предоставлении дополнительных возможностей, таких как автоматическая валидация данных, генерация OpenAPI-документации и интеграция с Pydantic, при этом сохраняя всю производительность и гибкость Starlette.

Почему FastAPI выбрал Starlette как основу

Выбор Starlette в качестве основы для FastAPI обусловлен несколькими факторами:

  • Производительность — Starlette показывает отличные результаты в бенчмарках производительности
  • ASGI-совместимость — полная поддержка современного стандарта асинхронных веб-приложений
  • Стабильность архитектуры — хорошо продуманная и протестированная кодовая база
  • Активное сообщество — регулярные обновления и поддержка разработчиков

Преимущества асинхронности и производительности

Поддержка ASGI и асинхронных вызовов

ASGI (Asynchronous Server Gateway Interface) — это новый стандарт, пришедший на смену WSGI для асинхронных приложений. Starlette полностью совместим с ASGI, что позволяет:

  • Обрабатывать тысячи одновременных соединений
  • Эффективно работать с WebSocket-соединениями
  • Использовать современные асинхронные серверы как Uvicorn, Hypercorn
  • Интегрироваться с асинхронными базами данных и внешними API

Сравнение производительности с WSGI-фреймворками

В отличие от традиционных WSGI-фреймворков (Flask, Django), Starlette обеспечивает:

  • Неблокирующий ввод-вывод — запросы не блокируют друг друга
  • Высокая пропускная способность — способность обрабатывать больше запросов в секунду
  • Эффективное использование ресурсов — меньшее потребление памяти и CPU
  • Лучшая масштабируемость — более эффективная работа под высокой нагрузкой

Архитектура и компоненты Starlette

Основные компоненты фреймворка

Starlette состоит из нескольких ключевых компонентов:

  • Applications — основной класс приложения
  • Routing — система маршрутизации запросов
  • Requests/Responses — объекты для работы с HTTP-запросами и ответами
  • Middleware — промежуточные обработчики
  • Background Tasks — фоновые задачи
  • WebSockets — поддержка WebSocket-соединений
  • Static Files — обслуживание статических файлов
  • Templates — интеграция с шаблонизаторами

Жизненный цикл запроса

Обработка запроса в Starlette проходит следующие этапы:

  1. Получение запроса — ASGI-сервер передает запрос в приложение
  2. Middleware — обработка запроса промежуточными обработчиками
  3. Маршрутизация — определение соответствующего обработчика
  4. Обработка — выполнение бизнес-логики
  5. Формирование ответа — создание HTTP-ответа
  6. Middleware — постобработка ответа
  7. Отправка ответа — передача ответа клиенту

Подробное изучение маршрутизации

Основы маршрутизации в Starlette

Starlette предоставляет гибкую систему маршрутизации, которая поддерживает различные типы маршрутов:

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route, Mount

async def homepage(request):
    return JSONResponse({'message': 'Главная страница'})

async def user_detail(request):
    user_id = request.path_params['user_id']
    return JSONResponse({'user_id': user_id})

app = Starlette(debug=True, routes=[
    Route("/", homepage),
    Route("/users/{user_id:int}", user_detail),
])

Типы маршрутов

Starlette поддерживает несколько типов маршрутов:

  • Route — обычные HTTP-маршруты
  • WebSocketRoute — маршруты для WebSocket-соединений
  • Mount — монтирование поддиректорий или под-приложений
  • Host — маршрутизация по доменам

Параметры маршрутов

Фреймворк поддерживает различные типы параметров в URL:

  • {param} — строковый параметр
  • {param:int} — числовой параметр
  • {param:float} — параметр с плавающей точкой
  • {param:path} — путь (может содержать слэши)
  • {param:uuid} — UUID-параметр

Работа с запросами и ответами

Объект Request

Starlette предоставляет богатый объект Request для работы с входящими запросами:

from starlette.requests import Request

async def handle_request(request: Request):
    # Получение данных из запроса
    method = request.method
    url = request.url
    headers = request.headers
    query_params = request.query_params
    path_params = request.path_params
    
    # Работа с телом запроса
    json_data = await request.json()
    form_data = await request.form()
    body = await request.body()
    
    return JSONResponse({'method': method, 'url': str(url)})

Типы ответов

Starlette предоставляет различные типы ответов для разных сценариев:

from starlette.responses import (
    JSONResponse, HTMLResponse, PlainTextResponse,
    RedirectResponse, FileResponse, StreamingResponse
)

async def json_response(request):
    return JSONResponse({'data': 'json'})

async def html_response(request):
    return HTMLResponse('<h1>HTML контент</h1>')

async def redirect_response(request):
    return RedirectResponse(url='/new-location')

async def file_response(request):
    return FileResponse('path/to/file.pdf')

Middleware в Starlette

Что такое middleware

Middleware в Starlette — это компоненты, которые обрабатывают запросы и ответы на уровне приложения. Они позволяют добавлять функциональность, которая применяется ко всем или определенным маршрутам.

Встроенные middleware

Starlette включает несколько полезных middleware:

  • CORSMiddleware — обработка CORS-запросов
  • SessionMiddleware — работа с сессиями
  • HTTPSRedirectMiddleware — перенаправление на HTTPS
  • TrustedHostMiddleware — защита от атак Host header
  • GZipMiddleware — сжатие ответов

Создание собственного middleware

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        
        # Обработка запроса
        response = await call_next(request)
        
        # Логирование
        process_time = time.time() - start_time
        logger.info(f"Обработка {request.method} {request.url} заняла {process_time:.2f}s")
        
        return response

app.add_middleware(LoggingMiddleware)

WebSocket и real-time функциональность

Основы работы с WebSocket

Starlette предоставляет нативную поддержку WebSocket-соединений:

from starlette.websockets import WebSocket
from starlette.routing import WebSocketRoute

async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Эхо: {data}")
    except WebSocketDisconnect:
        print("Клиент отключился")

app = Starlette(routes=[
    WebSocketRoute("/ws", websocket_endpoint)
])

Продвинутые возможности WebSocket

  • Broadcast messaging — отправка сообщений всем подключенным клиентам
  • Room management — группировка соединений по комнатам
  • Authentication — аутентификация WebSocket-соединений
  • Error handling — обработка ошибок и отключений

Фоновые задачи

Background Tasks

Starlette предоставляет механизм для выполнения фоновых задач:

from starlette.background import BackgroundTasks

def send_email(email: str, message: str):
    # Логика отправки email
    print(f"Отправка email на {email}: {message}")

async def create_user(request):
    user_data = await request.json()
    
    # Создание пользователя
    user = create_user_in_db(user_data)
    
    # Фоновая задача
    task = BackgroundTasks()
    task.add_task(send_email, user.email, "Добро пожаловать!")
    
    return JSONResponse(
        {"message": "Пользователь создан"}, 
        background=task
    )

Обработка исключений и ошибок

Встроенная обработка ошибок

Starlette предоставляет гибкую систему обработки исключений:

from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse

async def not_found_handler(request, exc):
    return JSONResponse(
        {"error": "Ресурс не найден"}, 
        status_code=404
    )

async def server_error_handler(request, exc):
    return JSONResponse(
        {"error": "Внутренняя ошибка сервера"}, 
        status_code=500
    )

app.add_exception_handler(404, not_found_handler)
app.add_exception_handler(500, server_error_handler)

Пользовательские исключения

class CustomException(Exception):
    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code

async def custom_exception_handler(request, exc):
    return JSONResponse(
        {"error": exc.message}, 
        status_code=exc.status_code
    )

app.add_exception_handler(CustomException, custom_exception_handler)

Статические файлы и шаблоны

Обслуживание статических файлов

from starlette.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="static"), name="static")
app.mount("/media", StaticFiles(directory="media"), name="media")

Интеграция с шаблонизаторами

from starlette.templating import Jinja2Templates

templates = Jinja2Templates(directory='templates')

async def template_response(request):
    context = {
        'request': request,
        'title': 'Моя страница',
        'users': get_users()
    }
    return templates.TemplateResponse('index.html', context)

Тестирование приложений Starlette

Основы тестирования

from starlette.testclient import TestClient
import pytest

@pytest.fixture
def client():
    return TestClient(app)

def test_homepage(client):
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_user(client):
    user_data = {"name": "Test User", "email": "test@example.com"}
    response = client.post("/users", json=user_data)
    assert response.status_code == 201

Асинхронное тестирование

import httpx
import pytest

@pytest.mark.asyncio
async def test_async_endpoint():
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/async-endpoint")
        assert response.status_code == 200

Безопасность в Starlette

CORS (Cross-Origin Resource Sharing)

from starlette.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Работа с HTTPS

from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware

app.add_middleware(HTTPSRedirectMiddleware)

Защита от атак

from starlette.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware, 
    allowed_hosts=["myapp.com", "*.myapp.com"]
)

Таблица методов и функций Starlette

Компонент Класс/Функция Описание Пример использования
Applications Starlette() Основной класс приложения app = Starlette(debug=True, routes=routes)
Routing Route() HTTP-маршрут Route("/users/{id}", get_user)
  WebSocketRoute() WebSocket-маршрут WebSocketRoute("/ws", websocket_endpoint)
  Mount() Монтирование под-приложений Mount("/api", sub_app)
  Host() Маршрутизация по доменам Host("api.example.com", routes=api_routes)
Requests Request.method HTTP-метод запроса request.method
  Request.url URL запроса request.url
  Request.headers Заголовки запроса request.headers["Authorization"]
  Request.query_params Query-параметры request.query_params["page"]
  Request.path_params Параметры пути request.path_params["user_id"]
  Request.json() JSON-данные data = await request.json()
  Request.form() Данные формы form = await request.form()
  Request.body() Тело запроса body = await request.body()
Responses JSONResponse() JSON-ответ JSONResponse({"key": "value"})
  HTMLResponse() HTML-ответ HTMLResponse("<h1>Hello</h1>")
  PlainTextResponse() Текстовый ответ PlainTextResponse("Hello")
  RedirectResponse() Перенаправление RedirectResponse(url="/new-url")
  FileResponse() Файловый ответ FileResponse("file.pdf")
  StreamingResponse() Потоковый ответ StreamingResponse(generate_data())
Middleware BaseHTTPMiddleware Базовый класс middleware class Custom(BaseHTTPMiddleware)
  CORSMiddleware CORS-поддержка app.add_middleware(CORSMiddleware)
  SessionMiddleware Сессии app.add_middleware(SessionMiddleware)
  HTTPSRedirectMiddleware HTTPS-перенаправление app.add_middleware(HTTPSRedirectMiddleware)
  TrustedHostMiddleware Защита Host header app.add_middleware(TrustedHostMiddleware)
  GZipMiddleware Сжатие ответов app.add_middleware(GZipMiddleware)
WebSockets WebSocket.accept() Принять соединение await websocket.accept()
  WebSocket.receive_text() Получить текст data = await websocket.receive_text()
  WebSocket.send_text() Отправить текст await websocket.send_text("Hello")
  WebSocket.receive_json() Получить JSON data = await websocket.receive_json()
  WebSocket.send_json() Отправить JSON await websocket.send_json({"key": "value"})
  WebSocket.close() Закрыть соединение await websocket.close()
Background Tasks BackgroundTasks() Фоновые задачи tasks = BackgroundTasks()
  BackgroundTasks.add_task() Добавить задачу tasks.add_task(func, arg1, arg2)
Static Files StaticFiles() Статические файлы StaticFiles(directory="static")
Templates Jinja2Templates() Шаблоны Jinja2 templates = Jinja2Templates(directory="templates")
  TemplateResponse() Ответ с шаблоном templates.TemplateResponse("index.html", context)
Testing TestClient() Тестовый клиент client = TestClient(app)
Exceptions HTTPException() HTTP-исключение HTTPException(status_code=404, detail="Not found")
  WebSocketException() WebSocket-исключение WebSocketException(code=1000)

Интеграция с базами данных

Асинхронные ORM

Starlette отлично работает с асинхронными ORM:

# Tortoise ORM
from tortoise.contrib.starlette import register_tortoise

register_tortoise(
    app,
    db_url="sqlite://db.sqlite3",
    modules={"models": ["app.models"]},
    generate_schemas=True,
    add_exception_handlers=True,
)

# SQLAlchemy с databases
from databases import Database

database = Database("postgresql://user:pass@localhost/dbname")

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

Оптимизация производительности

Кеширование

from starlette.middleware.base import BaseHTTPMiddleware
import time

class CacheMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, cache_time=300):
        super().__init__(app)
        self.cache = {}
        self.cache_time = cache_time

    async def dispatch(self, request, call_next):
        cache_key = f"{request.method}:{request.url}"
        
        if cache_key in self.cache:
            cached_response, timestamp = self.cache[cache_key]
            if time.time() - timestamp < self.cache_time:
                return cached_response
        
        response = await call_next(request)
        self.cache[cache_key] = (response, time.time())
        return response

Пулы соединений

import aioredis
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    # Инициализация
    app.state.redis = aioredis.from_url("redis://localhost")
    app.state.db_pool = await create_db_pool()
    
    yield
    
    # Очистка
    await app.state.redis.close()
    await app.state.db_pool.close()

app = Starlette(lifespan=lifespan)

Мониторинг и логирование

Структурированное логирование

import structlog
from starlette.middleware.base import BaseHTTPMiddleware

logger = structlog.get_logger()

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start_time = time.time()
        
        try:
            response = await call_next(request)
            duration = time.time() - start_time
            
            logger.info(
                "Request completed",
                method=request.method,
                url=str(request.url),
                status_code=response.status_code,
                duration=duration
            )
            
            return response
        except Exception as e:
            logger.error(
                "Request failed",
                method=request.method,
                url=str(request.url),
                error=str(e)
            )
            raise

Метрики производительности

from prometheus_client import Counter, Histogram, generate_latest

REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint'])
REQUEST_DURATION = Histogram('http_request_duration_seconds', 'HTTP request duration')

class MetricsMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start_time = time.time()
        
        response = await call_next(request)
        
        REQUEST_COUNT.labels(
            method=request.method,
            endpoint=request.url.path
        ).inc()
        
        REQUEST_DURATION.observe(time.time() - start_time)
        
        return response

# Эндпоинт для метрик
async def metrics(request):
    return PlainTextResponse(generate_latest())

Развертывание в продакшене

Конфигурация Uvicorn

# gunicorn.conf.py
bind = "0.0.0.0:8000"
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
max_requests = 1000
max_requests_jitter = 100

Docker-контейнеризация

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

Kubernetes-деплой

apiVersion: apps/v1
kind: Deployment
metadata:
  name: starlette-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: starlette-app
  template:
    metadata:
      labels:
        app: starlette-app
    spec:
      containers:
      - name: app
        image: starlette-app:latest
        ports:
        - containerPort: 8000
        env:
        - name: DATABASE_URL
          value: "postgresql://user:pass@postgres:5432/db"

Расширения и плагины

Популярные расширения

  • Starlette-Admin — административная панель
  • Starlette-OAuth — интеграция с OAuth-провайдерами
  • Starlette-Prometheus — метрики Prometheus
  • Starlette-WTF — интеграция с WTForms
  • Starlette-GraphQL — поддержка GraphQL

Создание собственного расширения

from starlette.applications import Starlette
from starlette.middleware.base import BaseHTTPMiddleware

class CustomExtension:
    def __init__(self, app: Starlette, config: dict):
        self.app = app
        self.config = config
        self.setup_middleware()
        self.setup_routes()
    
    def setup_middleware(self):
        self.app.add_middleware(CustomMiddleware)
    
    def setup_routes(self):
        self.app.add_route("/extension-endpoint", self.handler)
    
    async def handler(self, request):
        return JSONResponse({"extension": "active"})

Сравнение с другими фреймворками

Starlette vs Flask

Характеристика Starlette Flask
Асинхронность Нативная поддержка Ограниченная
Производительность Очень высокая Средняя
Размер Минималистичный Компактный
Экосистема Растущая Зрелая
Кривая обучения Средняя Низкая

Starlette vs Sanic

Характеристика Starlette Sanic
Стабильность Высокая Средняя
ASGI-совместимость Полная Ограниченная
Документация Отличная Хорошая
Тестирование Встроенное Внешние инструменты

Starlette vs AIOHTTP

Характеристика Starlette AIOHTTP
Простота использования Высокая Средняя
Производительность Отличная Хорошая
Middleware Удобное API Сложное API
WebSocket Простая интеграция Требует настройки

Использование в микросервисной архитектуре

Service Discovery

import consul

class ServiceRegistry:
    def __init__(self):
        self.consul = consul.Consul()
    
    async def register_service(self, name, host, port):
        self.consul.agent.service.register(
            name=name,
            service_id=f"{name}-{host}-{port}",
            address=host,
            port=port,
            check=consul.Check.http(f"http://{host}:{port}/health")
        )
    
    async def discover_service(self, name):
        services = self.consul.health.service(name, passing=True)[1]
        return [(s['Service']['Address'], s['Service']['Port']) 
                for s in services]

Circuit Breaker

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED
    
    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is open")
        
        try:
            result = await func(*args, **kwargs)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise
    
    def on_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED
    
    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

Часто задаваемые вопросы

Можно ли использовать Starlette без FastAPI?

Да, Starlette — это полноценный самостоятельный фреймворк. Вы можете создавать на нем приложения любой сложности без использования FastAPI. Starlette предоставляет все необходимые инструменты для создания современных веб-приложений.

Чем Starlette отличается от Flask?

Основные отличия заключаются в архитектуре и производительности. Starlette работает асинхронно и использует ASGI, что позволяет обрабатывать тысячи одновременных соединений. Flask работает синхронно с WSGI и менее эффективен при высоких нагрузках. Starlette также имеет встроенную поддержку WebSocket и современных веб-стандартов.

Что такое ASGI и зачем он нужен?

ASGI (Asynchronous Server Gateway Interface) — это стандарт интерфейса между асинхронными Python-приложениями и веб-серверами. Он пришел на смену WSGI и обеспечивает поддержку асинхронного кода, WebSocket-соединений, HTTP/2 и других современных протоколов.

Подходит ли Starlette для больших проектов?

Да, Starlette отлично подходит для крупных проектов. Его использует FastAPI, который применяется в продакшене такими компаниями как Microsoft, Netflix, Uber. Модульная архитектура и высокая производительность делают его пригодным для высоконагруженных систем.

Какие библиотеки часто используют вместе со Starlette?

Популярные библиотеки включают: Pydantic для валидации данных, SQLAlchemy или Tortoise ORM для работы с базами данных, Jinja2 для шаблонов, Databases для асинхронной работы с БД, Redis для кеширования, и Uvicorn как ASGI-сервер.

Что лучше: писать на чистом Starlette или использовать FastAPI?

Выбор зависит от задач. Если вам нужно API с автоматической валидацией данных, генерацией документации и типизацией — используйте FastAPI. Если вы хотите полный контроль над архитектурой и минимальные зависимости — выбирайте Starlette.

Как Starlette обеспечивает безопасность?

Starlette предоставляет встроенные middleware для безопасности: CORSMiddleware для CORS, HTTPSRedirectMiddleware для принудительного HTTPS, TrustedHostMiddleware для защиты от атак на Host header. Также поддерживается интеграция с системами аутентификации и авторизации.

Можно ли использовать Starlette для GraphQL API?

Да, Starlette хорошо интегрируется с GraphQL-библиотеками, такими как Graphene или Strawberry. Существуют готовые расширения для интеграции GraphQL в Starlette-приложения.

Как оптимизировать производительность Starlette-приложения?

Для оптимизации можно использовать: кеширование ответов, пулы соединений к базе данных, оптимизацию запросов, использование CDN для статики, правильную настройку ASGI-сервера, мониторинг производительности и профилирование кода.

Поддерживает ли Starlette Server-Sent Events (SSE)?

Да, через StreamingResponse можно легко реализовать SSE для отправки данных в реальном времени от сервера к клиенту. Это полезно для уведомлений, обновлений статуса и других real-time функций.

Примеры реального использования

Создание REST API

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from starlette.middleware.cors import CORSMiddleware

# Модель данных
users = [
    {"id": 1, "name": "John Doe", "email": "john@example.com"},
    {"id": 2, "name": "Jane Smith", "email": "jane@example.com"}
]

# Обработчики
async def get_users(request):
    return JSONResponse(users)

async def get_user(request):
    user_id = int(request.path_params['user_id'])
    user = next((u for u in users if u['id'] == user_id), None)
    if user:
        return JSONResponse(user)
    return JSONResponse({"error": "User not found"}, status_code=404)

async def create_user(request):
    data = await request.json()
    new_user = {
        "id": len(users) + 1,
        "name": data["name"],
        "email": data["email"]
    }
    users.append(new_user)
    return JSONResponse(new_user, status_code=201)

# Приложение
app = Starlette(routes=[
    Route("/users", get_users, methods=["GET"]),
    Route("/users", create_user, methods=["POST"]),
    Route("/users/{user_id:int}", get_user, methods=["GET"]),
])

app.add_middleware(CORSMiddleware, allow_origins=["*"])

Создание WebSocket чата

from starlette.applications import Starlette
from starlette.websockets import WebSocket
from starlette.routing import WebSocketRoute
from starlette.responses import HTMLResponse
from starlette.routing import Route
import json

# Хранилище подключений
connections = set()

async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    connections.add(websocket)
    
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            
            # Отправка сообщения всем подключенным клиентам
            for connection in connections.copy():
                try:
                    await connection.send_text(json.dumps({
                        "user": message["user"],
                        "message": message["message"],
                        "timestamp": message.get("timestamp")
                    }))
                except:
                    connections.remove(connection)
    
    except Exception:
        connections.remove(websocket)

async def chat_page(request):
    return HTMLResponse("""
    <!DOCTYPE html>
    <html>
    <head>
        <title>Starlette Chat</title>
    </head>
    <body>
        <div id="messages"></div>
        <input type="text" id="messageInput" placeholder="Введите сообщение...">
        <button onclick="sendMessage()">Отправить</button>
        
        <script>
            const ws = new WebSocket('ws://localhost:8000/ws');
            
            ws.onmessage = function(event) {
                const data = JSON.parse(event.data);
                const messages = document.getElementById('messages');
                messages.innerHTML += `<p><strong>${data.user}:</strong> ${data.message}</p>`;
            };
            
            function sendMessage() {
                const input = document.getElementById('messageInput');
                const message = {
                    user: 'User',
                    message: input.value,
                    timestamp: new Date().toISOString()
                };
                ws.send(JSON.stringify(message));
                input.value = '';
            }
        </script>
    </body>
    </html>
    """)

app = Starlette(routes=[
    Route("/", chat_page),
    WebSocketRoute("/ws", websocket_endpoint),
])

Заключение

Starlette представляет собой мощный и современный фреймворк для создания высокопроизводительных веб-приложений на Python. Его асинхронная архитектура, ASGI-совместимость и минималистичный дизайн делают его идеальным выбором для различных типов проектов — от простых API до сложных микросервисных архитектур.

Основные преимущества Starlette включают высокую производительность, гибкость архитектуры, отличную документацию и активное сообщество разработчиков. Фреймворк предоставляет все необходимые инструменты для создания современных веб-приложений, включая поддержку WebSocket, middleware, фоновых задач и интеграцию с различными базами данных и системами.

Понимание принципов работы Starlette не только поможет вам создавать более эффективные приложения, но и даст глубокое понимание того, как работает FastAPI под капотом. Это знание особенно ценно при разработке высоконагруженных систем, где каждая миллисекунда имеет значение.

Starlette продолжает активно развиваться, регулярно получая обновления и новые функции. Его влияние на экосистему Python-фреймворков трудно переоценить — он задал новые стандарты производительности и архитектуры для современных веб-приложений. Изучение и использование Starlette — это инвестиция в будущее ваших проектов и профессиональное развитие как разработчика Python.

Новости