#!/usr/bin/env python3 """ Redis leaderboard demo server. Run this file and visit http://localhost:8080 to add players, change scores, inspect the top N entries, and view entries around a rank. """ from __future__ import annotations import argparse from html import escape from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json from pathlib import Path import sys from urllib.parse import parse_qs, urlparse sys.path.insert(0, str(Path(__file__).resolve().parent)) try: import redis from leaderboard import RedisLeaderboard except ImportError as exc: print(f"Error: {exc}") print("Make sure the 'redis' package is installed: pip install redis") sys.exit(1) SAMPLE_PLAYERS = [ { "user_id": "player-1", "score": 980, "metadata": { "name": "Avery", "description": "Steady climber who never wastes a turn.", }, }, { "user_id": "player-2", "score": 1310, "metadata": { "name": "Mina", "description": "Always finds a way into the top three.", }, }, { "user_id": "player-3", "score": 1175, "metadata": { "name": "Noah", "description": "Takes big swings and occasionally lands them.", }, }, { "user_id": "player-4", "score": 1435, "metadata": { "name": "Priya", "description": "Current pace-setter with a ruthless endgame.", }, }, { "user_id": "player-5", "score": 1080, "metadata": { "name": "Jules", "description": "Quietly consistent and hard to catch.", }, }, { "user_id": "player-6", "score": 1240, "metadata": { "name": "Rin", "description": "Moves fast after every weekly reset.", }, }, ] class LeaderboardDemoHandler(BaseHTTPRequestHandler): """Serve an interactive leaderboard demo page.""" leaderboard: RedisLeaderboard | None = None def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path in {"/", "/index.html"}: self._send_html(self._html_page()) return if parsed.path == "/api/state": self._handle_state_request(parsed.query) return self.send_error(404) def do_POST(self) -> None: parsed = urlparse(self.path) if parsed.path == "/api/players": self._handle_player_upsert() return if parsed.path == "/api/increment": self._handle_increment() return if parsed.path == "/api/config": self._handle_config_update() return if parsed.path == "/api/reset": self._handle_reset() return self.send_error(404) def _handle_state_request(self, query: str) -> None: params = parse_qs(query) top_count = self._parse_positive_int(params.get("top", ["5"])[0], default=5) around_rank = self._parse_positive_int(params.get("rank", ["3"])[0], default=3) around_count = self._parse_positive_int(params.get("around", ["5"])[0], default=5) top_entries = self.leaderboard.get_top(top_count) around_entries = self.leaderboard.get_around_rank(around_rank, around_count) self._send_json( { "leaderboard_key": self.leaderboard.key, "max_entries": self.leaderboard.max_entries, "top_count": top_count, "around_rank": around_rank, "around_count": around_count, "size": self.leaderboard.get_size(), "top_entries": top_entries, "around_entries": around_entries, }, 200, ) def _handle_player_upsert(self) -> None: params = self._read_form_data() user_id = params.get("user_id", [""])[0].strip() score = self._parse_float(params.get("score", [""])[0]) name = params.get("name", [""])[0].strip() description = params.get("description", [""])[0].strip() if not user_id: self._send_json({"error": "User ID is required."}, 400) return if score is None: self._send_json({"error": "Score must be a valid number."}, 400) return entry = self.leaderboard.upsert_user( user_id, score, { "name": name or user_id, "description": description or "No description provided.", }, ) self._send_json( { "message": "Player saved.", "entry": entry, "max_entries": self.leaderboard.max_entries, }, 200, ) def _handle_increment(self) -> None: params = self._read_form_data() user_id = params.get("user_id", [""])[0].strip() amount = self._parse_float(params.get("amount", [""])[0]) name = params.get("name", [""])[0].strip() description = params.get("description", [""])[0].strip() if not user_id: self._send_json({"error": "User ID is required."}, 400) return if amount is None: self._send_json({"error": "Increment must be a valid number."}, 400) return existing_metadata = self.leaderboard.get_user_metadata(user_id) metadata = existing_metadata or { "name": name or user_id, "description": description or "Created during score increment.", } if existing_metadata: if name: metadata["name"] = name if description: metadata["description"] = description entry = self.leaderboard.increment_score(user_id, amount, metadata) self._send_json( { "message": "Score updated.", "entry": entry, "max_entries": self.leaderboard.max_entries, }, 200, ) def _handle_config_update(self) -> None: params = self._read_form_data() max_entries = self._parse_positive_int( params.get("max_entries", [""])[0], default=None, ) if max_entries is None: self._send_json( {"error": "Max entries must be a whole number greater than 0."}, 400, ) return trimmed_user_ids = self.leaderboard.set_max_entries(max_entries) self._send_json( { "message": "Leaderboard limit updated.", "max_entries": self.leaderboard.max_entries, "trimmed_user_ids": trimmed_user_ids, }, 200, ) def _handle_reset(self) -> None: self._reset_demo_data() self._send_json( { "message": "Demo leaderboard reset.", "max_entries": self.leaderboard.max_entries, }, 200, ) def _reset_demo_data(self) -> None: self.leaderboard.clear() for player in SAMPLE_PLAYERS: self.leaderboard.upsert_user( player["user_id"], player["score"], player["metadata"], ) def _read_form_data(self) -> dict[str, list[str]]: content_length = int(self.headers.get("Content-Length", "0")) raw_body = self.rfile.read(content_length).decode("utf-8") return parse_qs(raw_body) def _parse_positive_int(self, raw_value: str, default: int | None) -> int | None: try: value = int(raw_value) except (TypeError, ValueError): return default if value < 1: return default return value def _parse_float(self, raw_value: str) -> float | None: try: return float(raw_value) except (TypeError, ValueError): return None def _send_json(self, payload: dict[str, object], status: int) -> None: self.send_response(status) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps(payload).encode("utf-8")) def _send_html(self, content: str) -> None: self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(content.encode("utf-8")) def _html_page(self) -> str: title = escape("Redis Leaderboard Demo") return f""" {title}

Redis leaderboard demo

This demo stores scores in a Redis sorted set and keeps player details in per-user hashes. You can inspect the top performers, look around a rank position, and trim the board to a fixed size.

Sorted set rankings Hash-based metadata Top N and around-rank queries

Add or update a player

Increment a score

Leaderboard settings

Leaderboard key: leaderboard:demo Stored entries: 0 Max kept: 100

Top entries

Rank User Score Metadata

Entries around rank

Rank User Score Metadata
""" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run the Redis leaderboard demo server.") parser.add_argument("--host", default="127.0.0.1", help="HTTP host to bind to") parser.add_argument("--port", type=int, default=8080, help="HTTP port to bind to") parser.add_argument("--redis-host", default="localhost", help="Redis host") parser.add_argument("--redis-port", type=int, default=6379, help="Redis port") parser.add_argument( "--max-entries", type=int, default=100, help="Maximum number of leaderboard entries to keep", ) return parser.parse_args() def main() -> None: args = parse_args() redis_client = redis.Redis( host=args.redis_host, port=args.redis_port, decode_responses=True, ) redis_client.ping() leaderboard = RedisLeaderboard( redis_client=redis_client, key="leaderboard:demo", max_entries=args.max_entries, ) LeaderboardDemoHandler.leaderboard = leaderboard leaderboard.clear() for player in SAMPLE_PLAYERS: leaderboard.upsert_user( player["user_id"], player["score"], player["metadata"], ) server = ThreadingHTTPServer((args.host, args.port), LeaderboardDemoHandler) print(f"Leaderboard demo server running at http://{args.host}:{args.port}") print(f"Connected to Redis at {args.redis_host}:{args.redis_port}") print(f"Keeping the top {leaderboard.max_entries} entries") print("Press Ctrl+C to stop.") try: server.serve_forever() except KeyboardInterrupt: print("\nStopping server.") finally: server.server_close() if __name__ == "__main__": main()