Database Layer

Database Layer

app/database/
├── db.py                           → asyncpg pool singleton
├── maintenance.py                  → DB size, table counts, cleanup, schema version
├── repositories.py                 → compatibility facade (re-exports all modules)
├── spotify_repository.py           → compatibility facade (re-exports spotify module)
└── repository_modules/
    ├── users.py                    → upsert_user(), get_user_language(), set_user_language()
    ├── searches.py                 → save_search(), get_search_history()
    ├── tracks.py                   → save_track(), get_track_by_deezer_id(), get_tracks_by_artist()
    ├── favorites.py                → add_favorite(), remove_favorite(), get_favorite_tracks()
    ├── errors.py                   → save_error(), get_recent_errors(), clear_errors()
    ├── search_cache.py             → save_search_cache(), get_cached_search() (24h TTL)
    ├── spotify.py                  → get_spotify_data_by_deezer_id(), update_spotify_data_for_track()
    └── admin_audit.py              → save_admin_audit(), get_recent_admin_audit()

Connection Pool

File: app/database/db.py

Module-level singleton asyncpg.Pool. Never call asyncpg.create_pool() directly.

Function	Description
init_db_pool()	Create pool on startup. Idempotent — safe to call multiple times.
close_db_pool()	Close pool on shutdown. Resets singleton.
get_pool()	Returns active pool. Raises RuntimeError if not initialized.

Pool is initialized in app/main.py before the bot starts and closed in the finally block.

Schema — Alembic

Schema is owned by migrations/versions/. The Dockerfile CMD runs alembic upgrade head before python run.py — idempotent, safe on every container restart.

Never alter schema by modifying Python code. Create a new revision:

alembic revision -m "describe change"
# then write raw SQL in upgrade()/downgrade() via op.execute(...)

Existing migrations:

Revision	Description
bf5069cbb1be	Baseline schema (users, searches, tracks, favorites, errors)
63ab83bf2873	Add search_cache table (v3.1.2)
cf55a191898c	Add admin_audit table (v3.5.0)
a3f81c2d47e9	Index search_cache.created_at (v3.7.8)
b7d92e4f1a03	Index tracks(artist, rank DESC); drop two indexes duplicated by UNIQUE (v3.7.10)

Repository Pattern

Transactions

Almost nothing needs one: every write is a single idempotent statement, and the cleanup helpers are re-runnable. The exception is cleanup_search_history, which issues one DELETE per user — it runs in a transaction, with its before/after counts inside it so a concurrent insert cannot skew them.

All repository functions take an asyncpg.Pool (or Connection) as the first argument. No ORM models — raw SQL via pool.fetchrow(), pool.fetch(), pool.execute().

Always import through the facade:

# ✅ correct
from app.database.repositories import save_track, get_track_by_deezer_id

# ❌ wrong — breaks facade contract
from app.database.repository_modules.tracks import save_track

Key Repositories

users.py

upsert_user(pool, user_id, username, lang) — INSERT ... ON CONFLICT DO UPDATE. Called on every bot interaction to keep user record fresh. God node in the graph (40 edges).

search_cache.py

get_cached_search(pool, query_normalized, source) → cached result or None if older than 24h. save_search_cache(pool, query_normalized, source, results) → upsert. normalize_query(q) — lowercase + strip for cache key.

errors.py

log_and_save_error(pool, ...) — structured error logging to DB + Python logger in one call. God node (40 edges). Used as global catch-all in aiogram's error handler.

Maintenance

File: app/database/maintenance.py

Used by admin /maintenance and /stats commands:

• get_database_size_bytes(pool) — PostgreSQL pg_database_size()
• get_table_count(pool, table) — row count for any table
• cleanup_old_errors(pool, keep_n) — prune error log
• cleanup_search_history(pool, days) — delete old search rows
• get_schema_version() — returns running app version (not a DB query — Alembic owns versioning)