---
description: Rules for working with the Python backend (duty_teller/)
globs:
  - duty_teller/**
  - alembic/**
  - tests/**
---

# Backend — Python

## Package layout

```
duty_teller/
├── main.py / run.py          # Entry point: bot + uvicorn
├── config.py                  # Settings from env vars (python-dotenv)
├── cache.py                   # TTL caches with pattern invalidation
├── api/                       # FastAPI app, routes, auth, ICS endpoints
│   ├── app.py                 # FastAPI app creation, route registration, static mount
│   ├── dependencies.py        # get_db_session, require_miniapp_username, get_validated_dates
│   ├── telegram_auth.py       # initData HMAC validation
│   ├── calendar_ics.py        # External ICS fetch & parse
│   └── personal_calendar_ics.py
├── db/                        # Database layer
│   ├── models.py              # SQLAlchemy ORM models (Base)
│   ├── repository.py          # CRUD functions (receive Session)
│   ├── schemas.py             # Pydantic response schemas
│   └── session.py             # session_scope, get_engine, get_session
├── handlers/                  # Telegram bot command/message handlers
│   ├── commands.py            # /start, /help, /set_phone, /calendar_link, /set_role
│   ├── common.py              # is_admin_async, invalidate_is_admin_cache
│   ├── errors.py              # Global error handler
│   ├── group_duty_pin.py      # Pinned duty message in group chats
│   └── import_duty_schedule.py
├── i18n/                      # Translations
│   ├── core.py                # get_lang, t()
│   ├── lang.py                # normalize_lang
│   └── messages.py            # MESSAGES dict (ru/en)
├── importers/                 # File parsers
│   └── duty_schedule.py       # parse_duty_schedule
├── services/                  # Business logic (receives Session)
│   ├── import_service.py      # run_import
│   └── group_duty_pin_service.py
└── utils/                     # Shared helpers
    ├── dates.py
    ├── handover.py
    ├── http_client.py
    └── user.py
```

## Imports

- Use absolute imports from the `duty_teller` package: `from duty_teller.db.repository import get_or_create_user`.
- Never import handler modules from services or repository — dependency flows
  downward: handlers → services → repository → models.

## DB access pattern

Handlers are async; SQLAlchemy sessions are synchronous. Two patterns:

### In bot handlers — `session_scope` + `run_in_executor`

```python
def do_work():
    with session_scope(config.DATABASE_URL) as session:
        # synchronous DB code here
        ...

await asyncio.get_running_loop().run_in_executor(None, do_work)
```

### In FastAPI endpoints — dependency injection

```python
@router.get("/api/duties")
def get_duties(session: Session = Depends(get_db_session)):
    ...
```

`get_db_session` yields a `session_scope` context — FastAPI closes it after the request.

## Handler patterns

### Admin-only commands

```python
if not await is_admin_async(update.effective_user.id):
    await update.message.reply_text(t(lang, "import.admin_only"))
    return
```

`is_admin_async` is cached for 60 s and invalidated on role changes.

### i18n in handlers

```python
lang = get_lang(update.effective_user)
text = t(lang, "start.greeting")
```

`t(lang, key, **kwargs)` substitutes `{placeholders}` and falls back to English → raw key.

### Error handling

The global `error_handler` (registered via `app.add_error_handler`) logs the
exception and sends a generic localized error reply.

## Service layer

- Services receive a `Session` — they never open their own.
- Services call repository functions, never handler code.
- After mutations that affect cached data, call `invalidate_duty_related_caches()`.

## Cache invalidation

Three TTL caches in `cache.py`:

| Cache | TTL | Invalidation trigger |
|-------|-----|---------------------|
| `ics_calendar_cache` | 600 s | After duty import |
| `duty_pin_cache` | 90 s | After duty import |
| `is_admin_cache` | 60 s | After `set_user_role` |

- `invalidate_duty_related_caches()` clears ICS and pin caches (call after any duty mutation).
- `invalidate_is_admin_cache(telegram_user_id)` clears a single admin entry.
- Pattern invalidation: `cache.invalidate_pattern(key_prefix_tuple)`.

## Alembic migrations

- Config in `pyproject.toml` under `[tool.alembic]`, script location: `alembic/`.
- **Naming convention:** `NNN_description.py` — sequential zero-padded number (`001`, `002`, …).
- Revision IDs are the string number: `revision = "008"`, `down_revision = "007"`.
- Run: `alembic -c pyproject.toml upgrade head` / `downgrade -1`.
- `entrypoint.sh` runs `alembic upgrade head` before starting the app in Docker.

## Configuration

- All config comes from environment variables, loaded with `python-dotenv`.
- `duty_teller/config.py` exposes module-level constants (`BOT_TOKEN`, `DATABASE_URL`, etc.)
  built from `Settings.from_env()`.
- Never hardcode secrets or URLs — always use `config.*` constants.
- Key variables: `BOT_TOKEN`, `DATABASE_URL`, `MINI_APP_BASE_URL`, `HTTP_HOST`, `HTTP_PORT`,
  `ADMIN_USERNAMES`, `ALLOWED_USERNAMES`, `DUTY_DISPLAY_TZ`, `DEFAULT_LANGUAGE`.

## Code style

- Formatter: Black (line-length 120).
- Linter: Ruff (`ruff check duty_teller tests`).
- Follow PEP 8 and Google Python Style Guide for docstrings.
- Max line length: 120 characters.