A short note on how to use Starlette together with Python-SocketIO. For a DIY project I want to run a single thread process to control a wake-up light (among other things).


Starlette is basically an asynchronous version on Flask, which are both web frameworks. Personally I use Flask quite often and like the philosophy and resulting code.

Starlette seems to take inspiration from Flask and improves on speed. Additionaly there are some nice quality-of-life improvements (Websocktes, GraphQL). The asynchrounous part gives must easier in-process background tasks, for example sending an e-mail.

In their own words:

Starlette is a lightweight ASGI framework/toolkit, which is ideal for building high performance asyncio services.

SocketIO (and Python-SocketIO)

Socket.IO is a library on top of websockets

Python-SocketIO is a Python library that enables you to work with socket.io in Python. The name is pretty descriptive:

This projects implements Socket.IO clients and servers that can run standalone or integrated with a variety of Python web frameworks.

The problem

I wanted the following things:

  • Starlette with a REST API;
  • SocketIO (through Python-SocketIO) for realtime streams;
  • A background task that periodially streams the current information in the system.

Additionally I wanted to have it in a single thread, and no heavy extra libraries. Such as distributed task queues (Celery) which then require other moving parts.

Part of the reason is that I have a single state in memory of some hardware components. It being a single-threaded async application makes it very easy to argue about updating the state and reading the state.

I ran into some issues when wanting to combine Starlette+Python-SocketIO+background tasks. Starlette runs on Uvicorn and the current way how Python-SocketIO hooks into the async loop was not working anymore.

After some trial and error I got it working:

import logging
import asyncio
import uvicorn
from uvicorn.loops.uvloop import uvloop_setup
from starlette.applications import Starlette
from starlette.responses import JSONResponse
import socketio

# Set some basic logging
    format="%(asctime)-15s %(levelname)-8s %(message)s"

# Create a basic app
sio = socketio.AsyncServer(async_mode='asgi')
star_app = Starlette(debug=True)
app = socketio.ASGIApp(sio, star_app)

async def homepage(request):
    return JSONResponse({'hello': 'world'})

async def connect(sid, environ):
    logging.info(f"connect {sid}")

async def message(sid, data):
    logging.info(f"message {data}")
    # await device.set(data)

async def disconnect(sid):
    logging.info(f'disconnect {sid}')

# Set up the event loop
async def start_background_tasks():
    while True:
        logging.info(f"Background tasks that ticks every 10s.")
        await sio.sleep(10.0)

async def start_uvicorn():
    uvicorn.run(app, host='', port=8000)

async def main(loop):
    bg_task = loop.create_task(start_background_tasks())
    uv_task = loop.create_task(start_uvicorn())
    await asyncio.wait([bg_task, uv_task])

if __name__ == '__main__':
    loop = asyncio.get_event_loop()

Perhaps there are other people running into the same issue and I can save them some time.