Mode

raw JSON →
4.4.0 verified Fri May 01 auth: no python

Mode is an async service-based programming library built on top of asyncio. It provides Service, Worker, and BackgroundThreadService classes for composing asynchronous services. Current version is 4.4.0, requires Python ~=3.6. Release cadence is irregular.

pip install mode
error RuntimeError: Task <Task pending ...> got Future <Future pending ...> attached to a different loop
cause Mixing services created in different event loops, often due to threading or incorrect loop handling.
fix
Ensure all Mode services are created and started within the same event loop. Use 'asyncio.get_event_loop()' consistently or 'asyncio.run()' for top-level.
error TypeError: 'async_generator' object is not iterable
cause Using yield in Service.run() which is a coroutine, not an async generator.
fix
Change run() to use async/await pattern: 'async def run(self): await asyncio.sleep(...)' instead of 'while True: yield await ...'.
deprecated The 'Worker' class is deprecated in favor of 'Service' for custom services. Use Service as base for new code.
fix Replace 'class MyWorker(Worker)' with 'class MyService(Service)' and adjust run() accordingly.
breaking In version 4.0.0, the 'run()' method signature changed: it no longer takes an 'async' generator approach but must be a coroutine. Old code using 'yield' in run() breaks.
fix Change run() to be an async def that uses await asyncio.sleep() or other async patterns instead of yield.
gotcha Service.start() must be awaited; common mistake is to call start() without await, leading to unstarted services.
fix Always use 'await service.start()' inside an async context.
gotcha Avoid overriding '__init__' in subclasses unless you call super().__init__() properly. Missing super() can break internal state.
fix If you override __init__, always call 'super().__init__(**kwargs)' at the end of your __init__.

Simple example of creating and running a Mode Service.

import asyncio
from mode import Service

class MyService(Service):
    async def run(self) -> None:
        while not self.should_stop:
            await asyncio.sleep(1)
            print('Service running')

async def main():
    service = MyService()
    await service.start()
    await asyncio.sleep(5)
    await service.stop()

if __name__ == '__main__':
    asyncio.run(main())