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 Common errors
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 ...'.
Warnings
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__.
Imports
- Service
from mode import Service - Worker
from mode import Worker - BackgroundThreadService
from mode import BackgroundThreadService
Quickstart
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())