Game Tracker

Personal game library — Flask + SQLite + Steam API

[← Back to Portfolio]

Overview

A personal web app for tracking my game library, playtime, achievements, and session history. It runs as a separate Flask service on its own subdomain and stores everything in SQLite. Steam data syncs automatically — a background thread polls the Steam API every 30 minutes to keep hours current, and a scheduled job at 00:05 EST records a full snapshot each night to power the activity feed.

The database has five core tables: games, achievements, tags, daily_snapshots, and daily_game_snapshots. The snapshot tables are what make the daily feed work — by diffing consecutive snapshots you can reconstruct exactly which games were played on any given day and for how long. There's also a steam_import_status table that makes bulk imports resumable: if the import is interrupted mid-way, it picks up from where it left off rather than starting over.

Python Flask SQLite Steam API Threading Schedule

Code

// Database schema
The core schema. games stores the library; achievements links to it with a CASCADE delete. The two snapshot tables record nightly totals and per-game breakdowns separately so the daily diff query stays simple. steam_import_status tracks which app IDs have been imported so bulk imports are idempotent and resumable.
CREATE TABLE games (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    title           TEXT NOT NULL,
    platform        TEXT,
    status          TEXT,          -- 'Playing', 'Completed', 'Backlog', etc.
    rating          INTEGER,
    hours_played    REAL,
    steam_app_id    INTEGER,
    cover_url       TEXT,
    completion_date TEXT,
    is_favorite     INTEGER DEFAULT 0
);

CREATE TABLE achievements (
    id       INTEGER PRIMARY KEY AUTOINCREMENT,
    game_id  INTEGER,
    title    TEXT NOT NULL,
    description TEXT,
    date     TEXT,
    unlocked INTEGER DEFAULT 1,
    icon_url TEXT,
    FOREIGN KEY(game_id) REFERENCES games(id) ON DELETE CASCADE
);

CREATE TABLE daily_snapshots (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    date         TEXT NOT NULL UNIQUE,
    total_hours  REAL NOT NULL,
    games_played INTEGER NOT NULL
);

CREATE TABLE daily_game_snapshots (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    date         TEXT NOT NULL,
    game_id      INTEGER NOT NULL,
    game_title   TEXT NOT NULL,
    hours_played REAL NOT NULL,
    UNIQUE(date, game_id),
    FOREIGN KEY(game_id) REFERENCES games(id) ON DELETE CASCADE
);

CREATE TABLE steam_import_status (
    steam_app_id          INTEGER UNIQUE,
    game_imported         INTEGER DEFAULT 0,
    achievements_imported INTEGER DEFAULT 0,
    error_message         TEXT
);
// Games API — achievement aggregation in one query
Rather than doing N+1 queries, a single LEFT JOIN with conditional COUNTs returns every game alongside its achievement progress in one round-trip. The CASE expression avoids a division-by-zero when a game has no achievements at all.
# GET /api/games
cur.execute("""
    SELECT g.*,
           COUNT(CASE WHEN a.unlocked=1 THEN 1 END) AS unlocked_achievements,
           COUNT(a.id)                               AS total_achievements,
           CASE
             WHEN COUNT(a.id) > 0 THEN
               ROUND(COUNT(CASE WHEN a.unlocked=1 THEN 1 END) * 100.0
                     / COUNT(a.id), 1)
             ELSE 0
           END AS completion_percentage
    FROM games g
    LEFT JOIN achievements a ON g.id = a.game_id
    GROUP BY g.id
    ORDER BY g.is_favorite DESC, g.created_at DESC
""")
// Steam library import — resumable with per-game commits
The import endpoint handles libraries of any size. It sorts games by playtime descending (so the most-played titles import first), skips any game already in steam_import_status, and when fetching achievements commits per-game rather than in one big transaction — that way a timeout or API error mid-import doesn't lose all the work done so far. The mandatory 1-second sleep between achievement API calls is Steam's rate limit.
# fetch owned games, sorted most-played first
steam_games.sort(key=lambda x: x.get('playtime_forever', 0), reverse=True)

# check what's already been imported
cur.execute('SELECT steam_app_id, game_imported, achievements_imported FROM steam_import_status')
import_status = {row['steam_app_id']: row for row in cur.fetchall()}

for i, game in enumerate(steam_games):
    app_id = game['appid']

    # skip if fully imported
    status = import_status.get(app_id)
    if status and status['game_imported'] == 1:
        continue

    hours = round(game.get('playtime_forever', 0) / 60, 1)
    cover  = f"https://cdn.cloudflare.steamstatic.com/steam/apps/{app_id}/header.jpg"
    cur.execute(
        "INSERT INTO games (title, platform, status, hours_played, steam_app_id, cover_url)"
        " VALUES (?,?,?,?,?,?)",
        (game['name'], 'PC', 'Playing', hours, app_id, cover)
    )
    game_id = cur.lastrowid
    cur.execute('INSERT OR REPLACE INTO steam_import_status (steam_app_id, game_imported) VALUES (?,1)',
                (app_id,))

    if import_achievements:
        achievements = get_steam_achievements(app_id)
        for ach in achievements:
            cur.execute(
                'INSERT INTO achievements (game_id, title, description, date, unlocked, icon_url)'
                ' VALUES (?,?,?,?,?,?)',
                (game_id, ach['name'], ach['description'],
                 ach['unlock_date'], ach['achieved'], ach['icon'])
            )
        cur.execute('UPDATE steam_import_status SET achievements_imported=1 WHERE steam_app_id=?', (app_id,))
        conn.commit()              # commit per-game so a mid-import failure loses minimal work
        if i < len(steam_games) - 1:
            time.sleep(1)          # Steam rate limit: 1 req/sec for achievement endpoint
// Auto-completion detection
When syncing a game from Steam, if every achievement is unlocked the game's status is automatically set to Completed and the completion date is derived from the latest achievement unlock date. This means 100% games self-manage their status without manual intervention.
steam_achievements_list = get_steam_achievements(app_id)

unlocked = sum(1 for a in steam_achievements_list if a.get('achieved', 0))
total    = len(steam_achievements_list)

if total > 0 and unlocked == total:
    # derive completion date from the latest achievement unlock
    dates = [a['unlock_date'] for a in steam_achievements_list if a['unlock_date']]
    try:
        completion_date = max(
            datetime.strptime(d, '%Y-%m-%d') for d in dates
        ).strftime('%Y-%m-%d')
    except (ValueError, TypeError):
        completion_date = datetime.now().strftime('%Y-%m-%d')

    cur.execute(
        'UPDATE games SET status=?, completion_date=? WHERE id=?',
        ('Completed', completion_date, game_id)
    )
// Daily snapshot — record and diff
At midnight the scheduler upserts a summary row and a full per-game breakdown. The activity feed is computed by comparing yesterday's snapshot to today's — any game whose hours increased shows up as "played today". Games with no previous snapshot entry show as first-time plays.
def record_daily_snapshot(self):
    date_str = self.get_current_date_est().isoformat()
    conn = sqlite3.connect(self.db_path)
    cur  = conn.cursor()

    cur.execute('SELECT SUM(hours_played) FROM games WHERE hours_played IS NOT NULL')
    total_hours = cur.fetchone()[0] or 0
    cur.execute('SELECT COUNT(*) FROM games WHERE hours_played > 0')
    games_count = cur.fetchone()[0]

    # upsert daily summary
    cur.execute('SELECT id FROM daily_snapshots WHERE date=?', (date_str,))
    if cur.fetchone():
        cur.execute('UPDATE daily_snapshots SET total_hours=?, games_played=? WHERE date=?',
                    (total_hours, games_count, date_str))
        cur.execute('DELETE FROM daily_game_snapshots WHERE date=?', (date_str,))
    else:
        cur.execute('INSERT INTO daily_snapshots (date, total_hours, games_played) VALUES (?,?,?)',
                    (date_str, total_hours, games_count))

    # per-game breakdown for diff queries
    for game in cur.execute(
        'SELECT id, title, hours_played, cover_url FROM games WHERE hours_played > 0'
    ):
        cur.execute(
            'INSERT INTO daily_game_snapshots (date, game_id, game_title, hours_played, cover_url)'
            ' VALUES (?,?,?,?,?)',
            (date_str, game['id'], game['title'], game['hours_played'], game['cover_url'])
        )
    conn.commit()


def get_games_played_on_date(self, date_str):
    'Diff today vs yesterday to find which games had hours added.'
    cur.execute('SELECT game_id, hours_played FROM daily_game_snapshots WHERE date=?', (date_str,))
    today = {r['game_id']: r['hours_played'] for r in cur.fetchall()}

    prev_date = (datetime.strptime(date_str, '%Y-%m-%d').date() - timedelta(days=1)).isoformat()
    cur.execute('SELECT game_id, hours_played FROM daily_game_snapshots WHERE date=?', (prev_date,))
    yesterday = {r['game_id']: r['hours_played'] for r in cur.fetchall()}

    result = []
    for game_id, hours in today.items():
        delta = hours - yesterday.get(game_id, 0)
        if delta > 0.01:              # ignore sub-minute rounding noise
            result.append({'game_id': game_id, 'hours_added': round(delta, 1)})

    result.sort(key=lambda x: x['hours_added'], reverse=True)
    return result
// Activity feed — live sessions + historical diffs
The feed combines three data sources in one endpoint. Live sessions are detected without waiting for the midnight cron: a single LEFT JOIN compares current hours against the most recent snapshot row, and any game whose hours increased by more than 0.09 (≈5 min) shows up as "live". Historical sessions come from a self-join on daily_game_snapshots comparing each date's row to the previous day's — the same diff logic as get_games_played_on_date but done in SQL for the whole 30-day window at once. The final Python pass deduplicates (live games suppress their same-day snapshot entry), merges achievement groups, and sorts live items to the top.
# Live sessions: current hours vs. latest snapshot, no cron needed
cur.execute('''
    SELECT g.id as game_id, g.title as game_title, g.cover_url,
           g.hours_played as current_hours,
           COALESCE(dgs.hours_played, 0) as snapshot_hours,
           g.hours_played - COALESCE(dgs.hours_played, 0) as hours_added
    FROM games g
    LEFT JOIN (
        SELECT game_id, hours_played
        FROM daily_game_snapshots
        WHERE date = (SELECT MAX(date) FROM daily_game_snapshots)
    ) dgs ON g.id = dgs.game_id
    WHERE g.hours_played > COALESCE(dgs.hours_played, 0) + 0.09
    ORDER BY hours_added DESC
''')

# Historical sessions: self-join on dgs, diff each day vs. the previous day
cur.execute('''
    SELECT dgs1.date, dgs1.game_id, dgs1.game_title, dgs1.cover_url,
           dgs1.hours_played as total_hours,
           dgs1.hours_played - COALESCE(dgs2.hours_played, 0) as hours_added
    FROM daily_game_snapshots dgs1
    LEFT JOIN daily_game_snapshots dgs2
        ON dgs1.game_id = dgs2.game_id
        AND dgs2.date = date(dgs1.date, '-1 day')
    WHERE dgs1.date >= date('now', '-30 days')
      AND (dgs1.hours_played - COALESCE(dgs2.hours_played, 0)) > 0.09
    ORDER BY dgs1.date DESC, hours_added DESC
    LIMIT 100
''')

# Recent achievement unlocks
cur.execute('''
    SELECT a.title as ach_title, a.date, a.icon_url,
           g.id as game_id, g.title as game_title, g.cover_url
    FROM achievements a JOIN games g ON a.game_id = g.id
    WHERE a.unlocked = 1 AND a.date >= date('now', '-30 days')
    ORDER BY a.date DESC
''')

# Python: group achievements, deduplicate live vs. historical, sort
live_game_ids = {s['game_id'] for s in live_sessions}
feed = []
for s in hist_sessions:
    # suppress today's snapshot entry for games already shown as live
    if s['date'] == today and s['game_id'] in live_game_ids:
        continue
    feed.append({'type': 'session', 'live': False, **s})

# sort: live items first, then newest date
feed.sort(key=lambda x: (not x.get('live'), x['date']), reverse=True)
// Background threads — scheduler and Steam poller
Two daemon threads run alongside the Flask app. The scheduler fires the nightly snapshot job (after first updating Steam hours so the snapshot is fresh). The poller keeps hours current between snapshots — it waits 60 seconds on startup to let the app finish initialising before making any API calls.
def setup_daily_scheduler(tracker):
    def job():
        update_all_steam_hours_sync()   # sync Steam before snapshotting
        sync_new_steam_games()          # import any games added since last run
        tracker.record_daily_snapshot()

    schedule.every().day.at('05:05').do(job)   # 00:05 EST = 05:05 UTC

    def run_scheduler():
        while True:
            schedule.run_pending()
            time.sleep(60)

    threading.Thread(target=run_scheduler, daemon=True).start()


def setup_steam_poller(interval_minutes=30):
    def poll():
        time.sleep(60)                 # startup delay — don't hit Steam during boot
        while True:
            try:
                update_all_steam_hours_sync()
            except Exception as e:
                logger.error(f"Steam poll error: {e}")
            time.sleep(interval_minutes * 60)

    threading.Thread(target=poll, daemon=True, name='steam-poller').start()

Links

[Open Tracker] [Technical Paper →] [← Portfolio]