Plugging into the event-loop when using ASGIApp and long-lived connections

[acnebs]

Hi, I'm trying to use python-socketio with FastAPI, which I do like this:

app: ASGIApp = ASGIApp(socketio_server=sio, other_asgi_app=fast_app)

And at the moment I'm just running that app object with uvicorn.

Anyway, I need to create a redis connection to use in my socketio server handlers for some coordination and such. I'm already using Redis for my python-socketio Manager, so it made sense to use redis for this too.

When I was using the WSGI version of python-socketio, I was actually able to hijack python-socketio's existing connection to redis (sio.server.manager.redis) However, AsyncRedisManager is not storing the connection in that way.

I do not want to create a new connection to redis in each of my handlers, as that seems very inefficient. But I cannot figure out how to have a connection that can be shared between my handlers in async-land.

[miguelgrinberg]

The aioredis manager uses two connections, pub and sub. This is not a technique that I would recommend, but if you want to steal the redis connection, can't you just use one of those?

A better approach would be for you to subclass the AsyncServer instance, or even better your FastAPI instance, and then create an app-wide redis connection in your specialized subclass.

[acnebs]

Thanks for the tip, I'll look into that!

Another idea I had was to store a connection as an attr of a socketio.AsyncNamespace class, by connecting to redis in the on_connect handler. But for some reason, awaiting an aioredis.create_connection_pool call in that event handler is a syntax error. Can't really figure out why though – I'm a bit new to async in python.

Do you know why this doesn't work?

[miguelgrinberg]

Was the connect handler an async function or a standard function? You can only use await on async def functions.

[acnebs]

Yep, it was async. Here is the relevant code:

REDIS_URL = "redis://localhost"

class V1Socket(AsyncNamespace):
    async def on_connect(self, sid, environ):
        self.redis = await create_redis_pool(config.REDIS_URL)

Python is flagging a syntax error at the end of the self.redis line and I am very confused.

EDIT: Scratch that, as you can see from the pasted code, there was some sort of invisible character at the end of my line, which wasn't showing up in my editor (or the linter, or the traceback). No idea how it got there, but there ya go.

Regardless, does the AsyncNamespace route seem like a reasonable way to handle my issue? Any gotchas that I'm overlooking?

[miguelgrinberg]

The on_connect() handler will be invoked every time a client connects, you need to have the connection created once only. Better to do something like this:

class V1Socket(AsyncNamespace):
    @classmethod
    async def create(self):
        self.redis = await create_redis_pool(config.REDIS_URL)

namespace = await V1Socket.create()

[acnebs]

Thanks! With that solution, I was still having issues with my setup because I had no place to await the V1Socket.create() method.

I ended up adding a get_redis method to my AsyncNamespace sub-class like this:

class V1Socket(AsyncNamespace):
    async def get_redis(self) -> RedisConnection:
        if not hasattr(self, "_redis"):
            self._redis = await create_redis_pool(config.REDIS_URL)
        return self._redis

and then just awaiting it in my handlers.

async def on_disconnect(self, sid):
    redis = await self.get_redis()

That is working for me for now, so I'll close this. But obviously if anyone has a better solution, feel free to post it here.