Development Status
- 5 - Production/Stable
Framework
- AsyncIO
- Pytest
- aiohttp
Intended Audience
- Developers
- Education
- Information Technology
- System Administrators
Natural Language
- English
- Russian
Operating System
- MacOS
- Microsoft :: Windows
- POSIX :: Linux
- POSIX
Programming Language
- Python :: 3
- Python :: 3.11
- Python :: 3.12
- Python :: 3.13
- Python :: 3.14
- Python :: Implementation :: CPython
- Python
Topic
- Internet :: WWW/HTTP
- Internet
- Software Development :: Libraries :: Python Modules
- Software Development :: Libraries
- Software Development
- System :: Archiving :: Compression
- System :: Logging
- System
- Utilities
Typing
- Typed
Miscellaneous utils for asyncio.
As a programmer, you are no stranger to the challenges that come with building and maintaining software applications. One area that can be particularly difficult is designing the architecture of software that uses asynchronous I/O.
This is where aiomisc comes in. aiomisc is a Python library that provides a collection of utility functions and classes for working with asynchronous I/O in a more intuitive and efficient way. It is built on top of the asyncio library and is designed to make it easier for developers to write asynchronous code that is both reliable and scalable.
With aiomisc, you can take advantage of powerful features like worker pools, connection pools, circuit breaker pattern, and retry mechanisms such as asyncbackoff and asyncretry to make your asyncio code more robust and easier to maintain. In this documentation, we’ll take a closer look at what aiomisc has to offer and how it can help you streamline your asyncio service development.
Why use aiomisc?
Problem: Production asyncio applications require significant boilerplate for logging, graceful shutdown, thread pools, and error handling.
Solution: aiomisc handles infrastructure so you can focus on business logic.
Plain asyncio |
With aiomisc |
|---|---|
Manual signal handling |
Built into entrypoint |
Manual logging setup |
Single parameter |
Manual thread pool |
Automatic + @threaded |
Try/finally cleanup |
Service stop() method |
Installation
Installation is possible in standard ways, such as PyPI or installation from a git repository directly.
Installing from PyPI:
pip3 install aiomiscInstalling from github.com:
# Using git tool
pip3 install git+https://github.com/aiokitchen/aiomisc.git
# Alternative way using http
pip3 install \
https://github.com/aiokitchen/aiomisc/archive/refs/heads/master.zipThe package contains several extras and you can install additional dependencies if you specify them in this way.
With uvloop:
pip3 install "aiomisc[uvloop]"With aiohttp:
pip3 install "aiomisc[aiohttp]"Complete table of extras below:
example |
description |
|---|---|
pip install aiomisc[aiohttp] |
For running aiohttp applications. |
pip install aiomisc[asgi] |
For running ASGI applications |
pip install aiomisc[carbon] |
|
pip install aiomisc[cron] |
use croniter for scheduling tasks |
pip install aiomisc[raven] |
|
pip install aiomisc[rich] |
Use rich for logging |
pip install aiomisc[uvicorn] |
|
pip install aiomisc[uvloop] |
use uvloop as a default event loop |
You can combine extras values by separating them with commas, for example:
pip3 install "aiomisc[aiohttp,cron,rich,uvloop]"Quick Start
This section will cover how this library creates and uses the event loop and creates services. Of course, you can’t write about everything here, but you can read about a lot in the Tutorial section, and you can always refer to the Modules and API reference sections for help.
Event-loop and entrypoint
Let’s look at this simple example first:
import asyncio
import logging
import aiomisc
log = logging.getLogger(__name__)
async def main():
log.info('Starting')
await asyncio.sleep(3)
log.info('Exiting')
if __name__ == '__main__':
with aiomisc.entrypoint(log_level="info", log_format="color") as loop:
loop.run_until_complete(main())This code declares an asynchronous main() function that exits after 3 seconds. It would seem nothing interesting, but the whole point is in the entrypoint.
What does the entrypoint do, it would seem not so much, it creates an event-loop and transfers control to the user. However, under the hood, the logger is configured in a separate thread, a pool of threads is created, services are started, but more on that later and there are no services in this example.
Alternatively, you can use the standard asyncio.Runner:
import asyncio
async def main():
await asyncio.sleep(1)
if __name__ == '__main__':
with asyncio.Runner() as runner:
runner.run(main())Services
The main thing that an entrypoint does is start and gracefully stop services.
The service concept within this library means a class derived from the aiosmic.Service class and implementing the async def start(self) -> None: method and optionally the async def stop(self, exc: Optional[ Exception]) -> None method.
The concept of stopping a service is not necessarily is pressing Ctrl+C keys by user, it’s actually just exiting the entrypoint context manager.
The example below shows what your service might look like:
from aiomisc import entrypoint, Service
class MyService(Service):
async def start(self):
do_something_when_start()
async def stop(self, exc):
do_graceful_shutdown()
with entrypoint(MyService()) as loop:
loop.run_forever()The entry point can start as many instances of the service as it likes, and all of them will start concurrently.
There is also a way if the start method is a payload for a service, and then there is no need to implement the stop method, since the running task with the start function will be canceled at the stop stage. But in this case, you will have to notify the entrypoint that the initialization of the service instance is complete and it can continue.
Like this:
import asyncio
from threading import Event
from aiomisc import entrypoint, Service
event = Event()
class MyService(Service):
async def start(self):
# Send signal to entrypoint for continue running
self.start_event.set()
await asyncio.sleep(3600)
with entrypoint(MyService()) as loop:
assert event.is_set()The whole power of this library is in the set of already implemented or abstract services. Such as: AIOHTTPService, ASGIService, TCPServer, UDPServer, TCPClient, PeriodicService, CronService and so on.
Unfortunately in this section it is not possible to pay more attention to this, please pay attention to the Tutorial section section, there are more examples and explanations, and of cource you always can find out an answer on the /api/index or in the source code. The authors have tried to make the source code as clear and simple as possible, so feel free to explore it.
Versioning
This software follows Semantic Versioning
Summary: it’s given a version number MAJOR.MINOR.PATCH, increment the:
MAJOR version when you make incompatible API changes
MINOR version when you add functionality in a backwards compatible manner
PATCH version when you make backwards compatible bug fixes
Additional labels for pre-release and build metadata are available as extensions to the MAJOR.MINOR.PATCH format.
In this case, the package version is assigned automatically with poem-plugins, it using on the tag in the repository as a major and minor and the counter, which takes the number of commits between tag to the head of branch.
How to develop?
This project, like most open source projects, is developed by enthusiasts, you can join the development, submit issues, or send your merge requests.
In order to start developing in this repository, you need to do the following things.
Should be installed:
Python 3.7+ as python3
Installed Poetry as poetry
For setting up developer environment just execute:
# installing all dependencies poetry install # setting up pre-commit hooks poetry run pre-commit install # adding poem-plugins to the poetry poetry self add poem-plugins
