Store data in Redis with Flask Async

For a future project idea I wanted to test how complicated Flask + Redis in Python async is. Redis-py got official async support with version 4.2.x and Flask has async support since version 2.0. Some additional libraries are needed to run this: asgiref and uvicorn.

First I needed a redis database to test this. I could have started a local docker container with redis, but I want to host this later on so I chose the free plan of the Redis Cloud (30MB are more than enough).

First we need a redis client that can process everything async:

import os
import json
import redis.asyncio as redis

class Store:
    def __init__(self):
        self.r = redis.from_url(os.environ.get("REDIS_URL"))

    async def save(self, name, state):
        return await self.r.set(f"game-{name}", json.dumps(state))

    async def load(self, name):
        game = await self.r.get(f"game-{name}")
        if game:
            return json.loads(game)
        return game

On save the game state, which is a dict, is stored as a json string in redis. If the key is found on load this is reversed, otherwise the "None" is returned.

Now the annotated

import uuid

from asgiref.wsgi import WsgiToAsgi
from flask import Flask, make_response, request

# has the previous snippet
from .store import Store

wsgi_app = Flask(__name__)
store = Store()

async def new():
    name = str(uuid.uuid4()).split("-")[-1]
    user = request.cookies.get("user_id", str(uuid.uuid4()))
    state = {"name": name, "players": [user]}
    await, state)

    resp = make_response(f"new game created: {name}<br/> you are: {user} (cookie set)")
    resp.set_cookie("user_id", user)
    return resp

async def join(name):
    state = await store.load(name)
    if state:
        user = request.cookies.get("user_id", str(uuid.uuid4()))
        if user not in state["players"]:
        await, state)

        resp = make_response(f"{name} found, players: {state['players']}")
        resp.set_cookie("user_id", user)
        return resp
    return "game not found"

async def index():
    return "nothing to see"

app = WsgiToAsgi(wsgi_app)

A lot is happening here:

The wsgi_app is the "normal" Flask app. With WsgiToAsgi this can now run via uvicorn. For example like this: uvicorn main:app --reload. The REDIS_URL should be set before running (for a local docker setup this would be redis://localhost:6379). Without the asgi setup there is no async mainloop for the Redis store and the Flask app to share. This breaks pretty fast, so the whole app has to run in one asgi context.

The /new endpoint created a new "game" and adds the current user as player. The user-id is stored in a cookie and read when already set. To return a response with a cookie in it a response object is created via make_response. The current game name is the last part of a uuid4. In the "real" version I use a name generator that creates nicer names, i.e. "dark-red-deer-42". The /join/<name> endpoint adds the current user as player (if not already in the players list). This is only to demonstrate how endpoints like this would work.

What is missing: no error checking for the redis client! All the should be checked for returned values.

One final note, I run this on with a Procfile like this: web: uvicorn app.main:app --host= --port=8080