Telegram bot for generating dynamic QR codes — with per-code analytics, MaxMind geolocation, TON crypto payments, animated QR variants, and a multi-tier subscription model. Built on Clean Architecture and CQRS with a full Mediator pattern.
Users create QR codes that can be updated after generation: the printed code never changes, but the destination URL can be swapped at any time. Every scan is tracked — geolocation (country, city via MaxMind GeoIP2), device type, and timestamp. Users upgrade tiers to unlock more codes, higher limits, and animated QR styles. Payments go through TON blockchain.
The project strictly separates business logic from infrastructure using Clean Architecture and CQRS. All commands and queries go through a Mediator; handlers in application/ have zero dependency on aiogram or any database driver.
bot/
├── domain/
│ ├── entities/ # QrCode, User, Tariff — pure Python, no ORM
│ ├── value_objects/ # QrCodeId, UserId, RedirectUrl, ScanLocation
│ ├── events/ # QrCodeCreated, QrCodeScanned, TariffUpgraded
│ ├── services/ # Domain logic: QR generation policy, limit checks
│ ├── ports/ # Interfaces: IQrRepository, ICachePort, IEventBus
│ └── repositories/ # Abstract repository contracts
│
├── application/
│ ├── commands/ # CreateQrCommand, UpdateTargetCommand, DeleteQrCommand
│ ├── queries/ # GetQrDetailsQuery, GetClickStatsQuery, GetUserProfileQuery
│ ├── command_handlers/ # One handler per command — pure orchestration
│ ├── query_handlers/ # One handler per query
│ ├── event_handlers/ # Analytics, cache invalidation, history, logging
│ ├── mediator/ # Mediator + registry + pipeline behaviors
│ ├── services/ # DynamicQrResolver, QrGenerationService, RedirectUrlBuilder
│ └── dtos/ # QrCodeDto, UserDto, StatsDto
│
├── infrastructure/
│ ├── repositories/ # SQLAlchemy async implementations of domain ports
│ ├── cache/ # Redis-backed cache implementation
│ ├── geo/ # MaxMind GeoIP2 — country + city lookup
│ ├── qr/ # qrcode + Pillow — static and animated QR rendering
│ ├── web/ # aiohttp redirect server for dynamic URL resolution
│ └── event_bus.py # In-process and Redis-backed event bus
│
├── presentation/
│ ├── handlers/ # aiogram routers — dispatch to Mediator
│ └── keyboards/ # Inline keyboard factories
│
└── di/
└── container.py # DI container + module registration
| Layer | Technology |
|---|---|
| Bot framework | aiogram 3.24 |
| Database | PostgreSQL + asyncpg 0.30 + SQLAlchemy 2 + Alembic |
| Cache | Redis |
| QR generation | qrcode + Pillow (static + animated) |
| Geolocation | MaxMind GeoIP2 |
| Payments | TON SDK (tonsdk) |
| i18n | Fluent (fluent-runtime) |
| Configuration | Pydantic Settings |
| Linting | Ruff + mypy (strict on domain + application layers) |
| Tests | pytest + pytest-asyncio |
| Feature | Detail |
|---|---|
| Dynamic QR codes | Update destination URL post-creation — code image unchanged |
| Static + animated QR | Multiple styles: standard, eye variants, frame overlays, gradient fills |
| Per-code analytics | Scan count, geolocation (country + city), device type, timestamps |
| Scan history | Full per-code scan timeline |
| Multi-tier tariffs | Free / paid tiers with different code limits and style access |
| TON payments | Crypto payment for tier upgrades |
| Redirect server | aiohttp server resolves dynamic QR scans to current target URL |
| i18n | Fluent format, multiple locales |
| Feature flags | Runtime feature flag system per user |
| Event sourcing | Domain events persisted and replayed via Redis event bus |
Every operation in the bot is either a Command (mutation) or a Query (read). Both go through the Mediator, which resolves the handler, runs pipeline behaviors (logging, caching, validation), and returns the result. This means:
- Handlers contain zero Telegram or DB code — they call domain services and repository ports
- New behavior (rate limiting, audit logging) is added as a pipeline behavior, not scattered across handlers
- Unit tests mock the ports, not the database
aiogram handler
→ Mediator.send(CreateQrCommand(...))
→ pipeline behaviors (logging, cache check)
→ CreateQrCommandHandler
→ domain.QrGenerationService.generate(...)
→ IQrRepository.save(qr_code)
→ IEventBus.publish(QrCodeCreated(...))
→ QrEventHandler (cache invalidation)
→ AnalyticsEventHandler (write scan record)
→ HistoryEventHandler (append to history)
2600+ tests across unit, integration, and handler layers.
make test # full suite
make test-unit # unit tests only (fast, no DB)
make coverage # HTML coverage report → htmlcov/The domain and application layers are tested without any database or Telegram connection. Infrastructure tests use real PostgreSQL via Docker.
git clone https://github.com/bumbaRasch/qr-code-bot
cd qr-code-bot
cp .env.example .env
# Required: BOT_TOKEN, DB_HOST/DB_NAME/DB_USER/DB_PASSWORD, REDIS_HOST
docker compose up -d --buildBOT_TOKEN=your_telegram_bot_token
ADMIN_IDS=123456789
# Database
DB_HOST=localhost
DB_PORT=5432
DB_NAME=qr_bot
DB_USER=qr_bot_user
DB_PASSWORD=your_password
# Redis
REDIS_HOST=localhost
REDIS_PORT=6379
# Bot mode: polling (dev) or webhook (prod)
BOT_MODE=pollingSee .env.example for the full list including webhook configuration.
Mediator over direct handler calls — decouples presentation from application. Adding a new cross-cutting concern (e.g. rate limiting) means adding one pipeline behavior, not editing every handler.
Domain has no framework imports — bot/domain/ has zero dependency on aiogram, SQLAlchemy, or Redis. Entities and value objects are plain Python dataclasses. This makes the core logic portable and trivially testable.
Strict mypy on domain + application layers — pyproject.toml configures disallow_untyped_defs = true for bot.domain.* and bot.application.*. Infrastructure and presentation layers use relaxed settings.
Redis event bus — domain events are published synchronously in-process for fast handlers (cache invalidation) and asynchronously via Redis for slow handlers (analytics writes, history).