Using SocketIO with TaskIQ tasks

[kevinhikaruevans]

I have a FastAPI application and I'm slowly integrating TaskIQ with it. It's pretty simple but I thought I'd share how I'm using SocketIO with TaskIQ, in case any one else has a similar use-case.

I'm using SocketIO's namespace classes with fastapi_socketio (a simple wrapper for sio). In my FastAPI app, I'm using AsyncRedisManager:

from fastapi_socketio import SocketManager
from socketio import AsyncRedisManager
...
redis_manager = AsyncRedisManager(url=Config.REDIS_URL)
socket_manager = SocketManager(app=app, client_manager=redis_manager)

from sio.sensors import namespace as sensor_namespace
from sio.clients import namespace as client_namespace

socket_manager._sio.register_namespace(sensor_namespace) # or socket_manager.register_namespace() for later versions
socket_manager._sio.register_namespace(client_namespace)

And in my TaskIQ tasks, I'm doing something similar:

from socketio import AsyncRedisManager, AsyncServer
...
redis_manager = AsyncRedisManager(url=Config.REDIS_URL)
sio = AsyncServer(client_manager=redis_manager)

from sio.sensors import namespace as sensor_namespace
from sio.clients import namespace as client_namespace

sio.register_namespace(client_namespace)
sio.register_namespace(sensor_namespace)

Then, from my async tasks, I'm able to call:

await client_namespace.emit('test')
# or
await sio.emit('test', 'some-data')

Basically, we're using Redis to pubsub messages between the TaskIQ worker and the FastAPI process.