Build a game backend with Valkey in Python

mkdir valkey-arcade && cd valkey-arcadepython3 -m venv .venv && source .venv/bin/activatepip install valkey

export VALKEY_URL='rediss://default:<password>@inst-<id>.eu.kvnest.com:6380'

import osimport time import valkey client = valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True) print(client.ping())  # True # How long does a round trip from your machine take?start = time.perf_counter()for _ in range(100):    client.ping()print(f"Average round trip: {(time.perf_counter() - start) * 10:.2f} ms")

python level_1.py

import jsonimport osimport time import valkey client = valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True) # A save slot: a key holds any string, JSON included.client.set("save:ada", json.dumps({"level": 3, "lives": 2}))print(json.loads(client.get("save:ada"))) # A shield power-up that wears off by itself after 5 seconds.client.set("powerup:ada:shield", "on", ex=5)print("Shield TTL:", client.ttl("powerup:ada:shield"), "seconds") time.sleep(6)print("Shield after 6 seconds:", client.get("powerup:ada:shield"))  # None

python level_2.py

import jsonimport osimport time import valkey client = valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True)  def load_profile_from_database(player_id):    """Stands in for a slow database query or API call."""    time.sleep(0.3)    return {"id": player_id, "name": "Ada", "favourite_game": "Nest Invaders"}  def get_profile(player_id):    """Cache-aside: ask Valkey first; on a miss, load and keep it for 60 seconds."""    key = f"cache:profile:{player_id}"    cached = client.get(key)    if cached:        return json.loads(cached)     profile = load_profile_from_database(player_id)    client.set(key, json.dumps(profile), ex=60)    return profile  client.delete("cache:profile:ada")  # start with a cold cachefor call in ["1st", "2nd", "3rd"]:    start = time.perf_counter()    get_profile("ada")    print(f"{call} call: {(time.perf_counter() - start) * 1000:.1f} ms")

python level_3.py

import osfrom concurrent.futures import ThreadPoolExecutor import valkey client = valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True) # 1,000 players finish a game at the same moment, on 20 threads.with ThreadPoolExecutor(max_workers=20) as pool:    list(pool.map(lambda _: client.incr("stats:games_played"), range(1000)))print("Games played:", client.get("stats:games_played")) # Coins go up and down by any amount.client.incrby("coins:ada", 50)client.decrby("coins:ada", 20)print("Ada's coins:", client.get("coins:ada"))

python level_4.py

import os import valkey client = valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True) client.zadd("leaderboard", {"ada": 4200, "linus": 3100, "grace": 5600}) # gt=True: a score only replaces a lower personal best.client.zadd("leaderboard", {"ada": 6100}, gt=True)client.zadd("leaderboard", {"grace": 1000}, gt=True)  # ignored, 5600 is higher # The top 3, highest first, with their scores.top = client.zrange("leaderboard", 0, 2, desc=True, withscores=True)for i, (name, score) in enumerate(top, start=1):    print(f"{i}. {name} {score:.0f}") # Ranks start at 0, so add 1 for humans.rank = client.zrevrank("leaderboard", "linus")print(f"Linus is #{rank + 1}")

python level_5.py

import os import valkey client = valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True) # One key per player, one field per attribute.client.hset("player:ada", mapping={"name": "Ada", "country": "DE", "level": 1, "coins": 30}) # Change single fields without reading the rest.client.hincrby("player:ada", "level", 1)  # level up!client.hset("player:ada", "title", "Cache Cadet") print(client.hgetall("player:ada"))print("Level:", client.hget("player:ada", "level"))

python level_6.py

import os import valkey client = valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True)  def allow_submit(player):    """At most 5 score submissions per player in each 10-second window."""    key = f"ratelimit:{player}"    # A pipeline sends both commands in one MULTI/EXEC. nx=True sets the    # expiry only if there is none yet, so the window starts at the first one.    pipe = client.pipeline()    pipe.incr(key)    pipe.expire(key, 10, nx=True)    count, _ = pipe.execute()    return count <= 5  for i in range(1, 8):    print(f"Submission {i}:", "accepted" if allow_submit("mallory") else "slow down!")

python level_7.py

import os import valkey client = valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True) # A subscription takes over a connection; pubsub() opens its own.feed = client.pubsub()feed.subscribe("arcade:events")feed.get_message(timeout=1)  # wait for the subscription to be confirmed listeners = client.publish("arcade:events", "ada set a new high score: 6100")client.publish("arcade:events", "grace joined the game")print(f"Delivered to {listeners} listener(s)") for _ in range(2):    message = feed.get_message(timeout=1)    print(f"📣 [{message['channel']}] {message['data']}") feed.close()

python level_8.py

import osimport time import valkey client = valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True) feed = client.pubsub()feed.subscribe(**{"arcade:events": lambda message: print(f"  📣 {message['data']}")})feed.get_message(timeout=1)  # wait for the subscription to be confirmedlistener = feed.run_in_thread(sleep_time=0.01)  def submit_score(player, score):    """Every level so far, in one function."""    limit_key = f"ratelimit:{player}"    pipe = client.pipeline()    pipe.incr(limit_key)    pipe.expire(limit_key, 10, nx=True)    attempts, _ = pipe.execute()    if attempts > 5:        return f"{player}: rejected, too many submissions"     client.hincrby(f"player:{player}", "games", 1)    # ch=True makes ZADD count changed scores, so 1 means a new personal best.    improved = client.zadd("leaderboard", {player: score}, gt=True, ch=True)    if improved:        client.publish("arcade:events", f"{player} reached {score}")    return f"{player}: {score}" + (" (new best!)" if improved else "")  for player, score in [("ada", 7300), ("grace", 6800), ("ada", 6900), ("grace", 7600)]:    print(submit_score(player, score)) # Mallory spams submissions, hoping a lucky one sticks.for i in range(1, 8):    print(submit_score("mallory", 5000 + i * 100)) print("\n🏆 Top 3")top = client.zrange("leaderboard", 0, 2, desc=True, withscores=True)for i, (name, score) in enumerate(top, start=1):    print(f"{i}. {name} {score:.0f}") time.sleep(0.5)  # let the feed print its last messageslistener.stop()

python level_9.py