Установка и настройка окружения

Быстрый старт: установка FastStream, настройка проекта, запуск брокеров в Docker и первый hello-world за 5 минут.

#Требования

Python 3.10+ (требуется поддержка type hints и match/case)
Docker и Docker Compose для запуска брокеров
pip или poetry для установки зависимостей

Проверьте версии:

python --version  # 3.10 или выше
docker --version
docker-compose --version

#Установка FastStream

#Через pip

Базовая установка (без брокеров):

pip install faststream

Установка с поддержкой конкретного брокера:

# RabbitMQ
pip install faststream[rabbit]

# Kafka
pip install faststream[kafka]

# Redis
pip install faststream[redis]

# NATS
pip install faststream[nats]

# Все брокеры сразу
pip install faststream[all]

#Через poetry (рекомендуется)

poetry new faststream-app
cd faststream-app

poetry add faststream
poetry add faststream-rabbit  # или faststream-kafka, faststream-redis, faststream-nats

pyproject.toml:

[tool.poetry.dependencies]
python = "^3.10"
faststream = "^0.5.0"
faststream-rabbit = "^0.5.0"  # или другой брокер
pydantic = "^2.0"

#Запуск брокеров в Docker

Создайте docker-compose.yml для локальной разработки:

version: '3.8'

services:
  # RabbitMQ
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"   # AMQP
      - "15672:15672" # Management UI
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

  # Apache Kafka (с Zookeeper)
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  # Redis
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  # NATS
  nats:
    image: nats:2.10
    ports:
      - "4222:4222"   # Client port
      - "8222:8222"   # Monitoring port

Запуск:

docker-compose up -d

Проверка:

docker-compose ps

# RabbitMQ UI: http://localhost:15672 (guest/guest)
# Kafka: localhost:9092
# Redis: redis-cli -h localhost -p 6379
# NATS: nats://localhost:4222

#Первый проект: Hello World

#Структура проекта

faststream-app/
├── app/
│   ├── __init__.py
│   ├── main.py          # Точка входа
│   ├── broker.py        # Конфигурация брокера
│   └── handlers.py      # Обработчики сообщений
├── docker-compose.yml
├── pyproject.toml
└── README.md

#Конфигурация брокера

app/broker.py:

from faststream.rabbit import RabbitBroker

broker = RabbitBroker("amqp://guest:guest@localhost:5672/")

Для других брокеров:

# Kafka
from faststream.kafka import KafkaBroker
broker = KafkaBroker("localhost:9092")

# Redis
from faststream.redis import RedisBroker
broker = RedisBroker("redis://localhost:6379")

# NATS
from faststream.nats import NatsBroker
broker = NatsBroker("nats://localhost:4222")

#Обработчик сообщений

app/handlers.py:

from pydantic import BaseModel
from app.broker import broker

class Message(BaseModel):
    name: str

@broker.subscriber("hello")
async def handle_hello(msg: Message):
    print(f"Hello, {msg.name}!")
    await broker.publish({"response": f"Hi, {msg.name}!"}, "responses")

#Точка входа

app/main.py:

from faststream import FastStream
from app.broker import broker
from app.handlers import handle_hello  # Важно: импортировать для регистрации

app = FastStream(broker)

if __name__ == "__main__":
    import asyncio
    
    async def main():
        await app.start()
        
        # Опубликовать тестовое сообщение
        await broker.publish({"name": "World"}, "hello")
        
        # Дать время на обработку
        await asyncio.sleep(1)
        await app.close()
    
    asyncio.run(main())

#Запуск

python -m app.main

Вывод:

Hello, World!

#CLI: запуск через faststream

FastStream предоставляет CLI для запуска приложений:

faststream run app.main:app

С автоперезагрузкой при разработке:

faststream run app.main:app --reload

Проверка здоровья:

faststream check app.main:app

#Project Template: faststream init

FastStream может создать шаблон проекта:

faststream init my-project
cd my-project
pip install -e .

Структура:

my-project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── router.py
│   └── settings.py
├── tests/
│   └── test_app.py
├── pyproject.toml
└── README.md

#Конфигурация через переменные окружения

Создайте .env:

BROKER_URL=amqp://guest:guest@localhost:5672/
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
REDIS_URL=redis://localhost:6379
NATS_URL=nats://localhost:4222

app/settings.py:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    broker_url: str = "amqp://guest:guest@localhost:5672/"
    kafka_servers: str = "localhost:9092"
    redis_url: str = "redis://localhost:6379"
    nats_url: str = "nats://localhost:4222"
    
    class Config:
        env_file = ".env"

settings = Settings()

app/broker.py:

from faststream.rabbit import RabbitBroker
from app.settings import settings

broker = RabbitBroker(settings.broker_url)

#Логирование

Настройте логирование для отладки:

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

logger = logging.getLogger(__name__)

Включите логирование FastStream:

# Подробные логи
FASTSTREAM_LOG_LEVEL=DEBUG faststream run app.main:app

#AsyncAPI: автогенерация документации

FastStream генерирует AsyncAPI-спецификацию:

faststream asyncapi app.main:app -o asyncapi.json

Просмотр в браузере:

faststream asyncapi app.main:app --serve

Откройте http://localhost:8000 — увидите интерактивную документацию по всем publishers и subscribers.

#Тестовое окружение

Для тестов используйте TestClient (не требует брокера):

tests/test_app.py:

import pytest
from faststream.testing import TestClient
from app.main import app

@pytest.mark.asyncio
async def test_hello():
    async with TestClient(app) as client:
        await client.publish({"name": "Test"}, "hello")
        # assertions

Запуск тестов:

pytest tests/ -v

#Частые проблемы

#Ошибка подключения к брокеру

ConnectionError: Could not connect to RabbitMQ

Решение: Убедитесь, что брокер запущен:

docker-compose ps
docker-compose logs rabbitmq

#Порт занят

Error: Port 5672 is already in use

Решение: Остановите другие процессы или измените порт в docker-compose.yml.

#Модуль не найден

ModuleNotFoundError: No module named 'faststream.rabbit'

Решение: Установите нужный экстра:

pip install faststream[rabbit]

faststream-app/ ├── app/ │ ├── __init__.py │ ├── main.py # Точка входа │ ├── broker.py # Конфигурация брокера │ └── handlers.py # Обработчики сообщений ├── docker-compose.yml ├── pyproject.toml └── README.md

my-project/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── router.py │ └── settings.py ├── tests/ │ └── test_app.py ├── pyproject.toml └── README.md