tgex-backend/CLAUDE.md

CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

Project Overview

This is a FastAPI + Telegram Bot backend application for managing Telegram channel integrations. The project uses:

• FastAPI for HTTP API endpoints
• aiogram for Telegram bot functionality
• Tortoise ORM (async) for database operations
• PostgreSQL with asyncpg driver
• aerich for database migrations
• uv for Python package management

Development Commands

Environment Setup

# Install dependencies (using uv)
uv sync

# Start PostgreSQL (via Docker Compose)
docker-compose up -d

# Run database migrations
make migrate-up  # or: export $(cat .env | xargs) && aerich upgrade

Running the Application

# Start the FastAPI server
uv run uvicorn src.main:app --host 0.0.0.0 --port 8000 --reload

Code Quality

# Run linter (Ruff)
uv run ruff check .

# Format code
uv run ruff format .

# Type checking (mypy with strict mode)
uv run mypy .

Testing

# Run all tests
uv run pytest

# Run a single test file
uv run pytest tests/test_file.py

# Run a specific test
uv run pytest tests/test_file.py::test_function_name

Database Migrations

# Create a new migration (auto-generated based on model changes)
make migrate-create  # or: source .env && aerich migrate

# Apply migrations
make migrate-up  # or: source .env && aerich upgrade

# Rollback one migration
make migrate-down  # or: source .env && aerich downgrade

Important Migration Workflow:

1. For simple model changes: Let aerich auto-generate the migration:

   source .env && aerich migrate --name "descriptive_name"
   

2. For complex refactorings (table renames, data migration, etc.):

   • FIRST: Update domain models to reflect the new structure
   • THEN: Run aerich migrate to auto-generate the migration file
   • FINALLY: Edit the generated file to add data migration logic

   Example workflow:

   # 1. Update models in src/domain/
   # 2. Generate migration (will create base ALTER TABLE statements)
   source .env && aerich migrate --name "refactor_tables"
   # 3. Edit the generated file in migrations/models/ to add:
   #    - Data migration SQL (INSERT ... SELECT)
   #    - Proper ordering of operations
   #    - Comments explaining complex logic
   

3. For custom constraints (CHECK, custom indexes, etc.):

   source .env && aerich migrate --name "add_check_constraint" --empty
   # Then edit and add: # ruff: noqa and # mypy: ignore-errors at the top
   

Migration Best Practices:

• Always test migrations on a copy of production data
• Use transactions (RUN_IN_TRANSACTION = True)
• Add comments explaining complex data transformations
• For table renames with data migration:
  1. Rename old table to old_*
  2. Create new table structure
  3. Migrate data with INSERT INTO new SELECT ... FROM old
  4. Drop old table last
• Include proper downgrade logic (even if it loses some data)

Architecture

Layered Architecture

The codebase follows a clean architecture pattern with clear separation of concerns:

1. Domain Layer (src/domain/)

   • Tortoise ORM models that represent database entities
   • All models inherit from domain.base.TimestampedModel which provides:
     • Auto-generated id (UUID primary key)
     • Timestamps: created_at, updated_at, deleted_at
     • Timezone-aware datetime fields
   • Use class Meta: table = 'table_name' for explicit table naming

2. Use Case Layer (src/usecase/)

   • Business logic functions organized by feature
   • Each use case is a standalone function with signature: async def use_case_name(self: 'Usecase', ...) -> ReturnType
   • Depends on Protocol interfaces (Database, TelegramBotWriter, JWTEncoder)
   • Use cases are assembled in the Usecase dataclass for dependency injection
   • IMPORTANT RULES:
     • Один use case отвечает за полноценный сценарий «от триггера до завершения»
     • Use case НЕ должен вызывать другие use cases (общую логику выносить в хелперы/сервисы)
     • Use case содержит ВСЮ бизнес-логику, включая управление состояниями диалога
     • Use case может отправлять сообщения пользователям через self.telegram_bot protocol
     • Всегда оборачивать операции БД в async with self.database.transaction()

3. Adapter Layer (src/adapter/)

   • Concrete implementations of protocol interfaces:
     • Postgres: Database operations (implements Database protocol)
     • Telegram: Telegram bot integration (implements TelegramWriter protocol)
     • JWT: Token encoding/decoding (implements JWTEncoder protocol)

4. Controller Layer (src/controller/)

   • HTTP controllers (src/controller/http/): FastAPI route handlers
   • Telegram callbacks (src/controller/telegram_callback/): Telegram event handlers
   • ВАЖНО: Controllers - это тонкий слой роутинга
     • Контроллер только извлекает данные из событий (HTTP request / Telegram update)
     • Валидирует базовые параметры (user exists, message not empty)
     • Вызывает соответствующий use case
     • НЕ содержит бизнес-логику
     • НЕ обращается к базе данных напрямую
     • НЕ отправляет сообщения пользователям
   • Все handlers регистрируются через декораторы на telegram_callback_router
   • Handlers импортируются в __init__.py для auto-регистрации

5. DTO Layer (src/dto/)

   • Pydantic models for request/response validation
   • Separate from domain models to decouple API contracts from database schema

Key Architectural Patterns

Dependency Injection via Protocols:

• Use cases depend on Protocol interfaces, not concrete implementations
• Allows for easy testing and swapping implementations
• See src/usecase/__init__.py for protocol definitions

Transaction Management:

• Database uses context-aware transactions via DatabaseBase.transaction()
• Session stored in ContextVar (_session_ctx) for implicit session access
• Always use async with database.transaction() for database operations
• The session is accessible via database.session within transaction context

Shared Base Classes:

• shared/datebase_base.py: Base database class with connection pooling and migration checking
• shared/telegram_base.py: Base Telegram bot class with polling lifecycle
• shared/logger/: Structured logging with JSON and console formatters

Configuration:

• All config in src/config.py using Pydantic Settings
• Environment variables loaded from .env file
• Nested config via double underscore: DB__URL, TELEGRAM__TOKEN

Application Lifecycle

1. Startup (in src/main.py):

   • Logger initialized
   • Adapters instantiated (Postgres, Telegram, JWT)
   • Usecase dataclass created with adapter dependencies
   • FastAPI lifespan context:
     • Connect to database (includes migration check)
     • Start Telegram bot polling

2. Request Flow:

   • HTTP: FastAPI route → Controller → Usecase → Adapter → Database
   • Telegram: Event → Telegram callback → Usecase → Adapter → Database

3. Shutdown:

   • Stop Telegram bot polling
   • Close database connections

Important Implementation Details

Database Session Management:

• DO NOT create sessions manually
• Use async with database.transaction() which handles begin/commit/rollback
• Access session via database.session property within transaction context
• Session lifecycle is managed by ContextVars for thread-safety

Telegram Bot Integration:

• Bot runs in polling mode (not webhook)
• Routers registered in src/controller/telegram_callback/
• Base class handles graceful shutdown of polling task

Authentication:

• JWT-based authentication for HTTP endpoints
• Login flow: Telegram bot → Login token → JWT access token
• Use get_current_user dependency for protected routes

Code Style:

• Single quotes for strings (configured in Ruff)
• Line length: 120 characters
• Python 3.13+ syntax
• Strict mypy typing enabled

Telegram Bot Patterns

State Management:

• Состояния хранятся в БД через модель TelegramState
• Используется TelegramStateEnum для типизации состояний (не строки!)
• Database protocol методы: get_telegram_state(), set_telegram_state(), clear_telegram_state()

Controller Pattern (Telegram):

from aiogram import F
from aiogram.filters import Command
from aiogram.types import Message

@telegram_callback_router.message(Command('command_name'))
async def cmd_handler(message: Message) -> None:
    if not message.from_user:
        log.error('Failed to get user data')
        return

    usecase = dependencies.get_usecase()
    await usecase.some_use_case(
        telegram_id=message.from_user.id,
        chat_id=message.chat.id
    )

Use Case Pattern (Telegram):

from src import domain

async def some_use_case(self: 'Usecase', telegram_id: int, chat_id: int) -> None:
    async with self.database.transaction():
        # 1. Получить/проверить пользователя
        user = await self.database.get_user(telegram_id=telegram_id)
        if not user:
            await self.telegram_bot.send_message('❌ Не авторизован', chat_id=chat_id)
            return

        # 2. Получить состояние (если нужно)
        state = await self.database.get_telegram_state(telegram_id)

        # 3. Бизнес-логика
        # ...

        # 4. Обновить состояние
        await self.database.set_telegram_state(
            telegram_id=telegram_id,
            state=domain.TelegramStateEnum.SOME_STATE,
            context={'key': 'value'}
        )

        # 5. Отправить сообщение
        await self.telegram_bot.send_message('✅ Успешно', chat_id=chat_id)

Inline Keyboard Pattern:

from aiogram.types import InlineKeyboardButton

# В use case:
buttons = [
    [InlineKeyboardButton(text='Кнопка', callback_data='callback_id:value')]
    for item in items
]
await self.telegram_bot.send_message_with_inline_keyboard(
    'Выберите опцию:', chat_id=chat_id, buttons=buttons
)

Callback Query Handler:

@telegram_callback_router.callback_query(F.data.startswith('prefix:'))
async def callback_handler(callback: CallbackQuery) -> None:
    if not callback.from_user or not callback.data:
        return

    usecase = dependencies.get_usecase()
    await usecase.handle_callback(
        telegram_id=callback.from_user.id,
        chat_id=callback.message.chat.id,
        callback_data=callback.data,
        message_id=callback.message.message_id
    )
    await callback.answer()  # Убрать "loading" на кнопке

Text Message Filters:

# Обрабатывать только не-команды
@telegram_callback_router.message(F.text & ~F.text.startswith('/'))
async def text_handler(message: Message) -> None:
    # Этот handler НЕ сработает для команд типа /start
    ...

Важные правила для Telegram handlers:

• Специфичные фильтры (Command) регистрировать РАНЬШЕ общих (F.text)
• Использовать ~F.text.startswith('/') для исключения команд в текстовых handlers
• Всегда проверять message.from_user перед использованием
• Callback handlers должны вызывать await callback.answer() в конце
• НЕ импортировать aiogram типы в usecase layer (только через TYPE_CHECKING в Protocol)

Adding New Protocol Methods

When adding new functionality that requires protocol methods:

1. Define Protocol in usecase layer (src/usecase/__init__.py):

if typing.TYPE_CHECKING:
    from aiogram.types import InlineKeyboardButton  # Импорт только для типов

class TelegramBotWriter(typing.Protocol):
    async def send_message_with_inline_keyboard(
        self, text: str, chat_id: int, buttons: list[list['InlineKeyboardButton']]
    ) -> None: ...

2. Implement in adapter (src/adapter/telegram_bot.py):

from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup

class TelegramBot(TelegramBase):
    async def send_message_with_inline_keyboard(
        self, text: str, chat_id: int, buttons: list[list[InlineKeyboardButton]]
    ) -> None:
        keyboard = InlineKeyboardMarkup(inline_keyboard=buttons)
        await self.bot.send_message(chat_id=chat_id, text=text, reply_markup=keyboard)

3. Use in use case:

from aiogram.types import InlineKeyboardButton  # Можно импортировать в use case

buttons = [[InlineKeyboardButton(text='Text', callback_data='data')]]
await self.telegram_bot.send_message_with_inline_keyboard('Message', chat_id, buttons)

Типизация Enum в Domain:

• Используйте enum.StrEnum для строковых enum (Python 3.11+)
• Enum значения в lowercase с underscores
• Используйте fields.CharEnumField() в Tortoise ORM

import enum
from tortoise import fields
from .base import TimestampedModel

class TelegramStateEnum(enum.StrEnum):
    CREATIVE_WAITING_CHANNEL = 'creative_waiting_channel'
    CREATIVE_WAITING_NAME = 'creative_waiting_name'

class TelegramState(TimestampedModel):
    state = fields.CharEnumField(TelegramStateEnum, max_length=26)

    class Meta:
        table = 'telegram_state'

Database Constraints:

• Foreign keys: Use fields.ForeignKeyField() with on_delete=fields.CASCADE
• Unique constraints: Use unique=True or unique_together in Meta
• CHECK constraints: Cannot be defined in Tortoise models - create manual migrations
• Indexes: Use index=True on fields or define in Meta

Example with CHECK constraint:

# Create empty migration
export $(cat .env | xargs) && aerich migrate --name "add_check_constraint" --empty

Then edit the created file:

# migrations/models/1_xxx_add_check_constraint.py
# ruff: noqa
# mypy: ignore-errors
from tortoise import BaseDBAsyncClient

RUN_IN_TRANSACTION = True

async def upgrade(db: BaseDBAsyncClient) -> str:
    return """
        ALTER TABLE "table_name"
            ADD CONSTRAINT "constraint_name"
            CHECK (your_condition);
    """

async def downgrade(db: BaseDBAsyncClient) -> str:
    return """
        ALTER TABLE "table_name"
            DROP CONSTRAINT IF EXISTS "constraint_name";
    """