aiomisc - 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 making architecture of the asynchronous I/O software.
This is where
aiomisc comes in. It 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.
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
Installation is possible in standard ways, such as PyPI or installation from a git repository directly.
Installing from PyPI:
pip3 install aiomisc
Installing 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.zip
The package contains several extras and you can install additional dependencies if you specify them in this way.
pip3 install "aiomisc[uvloop]"
pip3 install "aiomisc[aiohttp]"
Complete table of extras bellow:
For running aiohttp applications.
For running ASGI applications
use croniter for scheduling tasks
You might using rich for logging
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]"
This section will cover how this library creates and uses the event loop and creates services. For more details see 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 for
3 seconds. It would seem nothing interesting, but the whole point is in
At the first glance the
entrypoint did not do much, just 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 as there are no services
in this example.
Alternatively, you can choose not to use an
entrypoint, just create an
event-loop and set it as a default for current thread:
import asyncio import aiomisc # * Installs uvloop event loop is it's has been installed. # * Creates and set `aiomisc.thread_pool.ThreadPoolExecutor` # as a default executor # * Sets just created event-loop as a current event-loop for this thread. aiomisc.new_event_loop() async def main(): await asyncio.sleep(1) if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(main())
The example above is useful if your code is already using an implicitly created
event loop, you will have to modify less code, just add
aiomisc.new_event_loop() and all calls to
will return the created instance.
However, you can do with one call. Following example closes implicitly created asyncio event loop and install a new one:
import asyncio import aiomisc async def main(): await asyncio.sleep(3) if __name__ == '__main__': loop = aiomisc.new_event_loop() loop.run_until_complete(main())
The main thing that an
entrypoint does is start and gracefully
The service concept within this library means a class derived from
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
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.
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()
entrypoint passes control to the body of the context manager only
after all service instances have started. As mentioned above, a start is
considered to be the completion of the
start method or the setting of
an start event with
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 Reference 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.
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
Installed Poetry as
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
- 1. Tutorial
- 2. Modules
- 2.1. entrypoint
- 2.3. Logging configuration
- 2.4. Services
- 2.4.10. Multiple services
- 2.4.11. Configuration
- 2.4.12. aiohttp service
- 2.4.13. asgi service
- 2.4.14. Memory Tracer
- 2.4.15. Profiler
- 2.4.16. Raven service
- 2.5. Abstract connection pool
- 2.6. Context
- 2.10. Circuit Breaker
- 2.11. cutout
- 2.13. asynchronous file operations
- 2.14. Working with threads
- 2.16. Utilities
- 2.18. Logging configuration
- 2.19. Pytest plugin
- 2.21. Plugins
- 2.22. Statistic counters
- 3. API Reference