aiomisc - miscellaneous utils for asyncio¶
Miscellaneous utils for asyncio.
Installation¶
Installing from PyPI:
pip3 install aiomisc
With uvloop:
pip3 install "aiomisc[uvloop]"
With aiohttp:
pip3 install "aiomisc[aiohttp]"
Installing from github.com:
pip3 install git+https://github.com/aiokitchen/aiomisc.git
pip3 install \
https://github.com/aiokitchen/aiomisc/archive/refs/heads/master.zip
Quick Start¶
Async entrypoint with logging and useful arguments:
import argparse
import asyncio
import logging
from aiomisc import entrypoint
from aiomisc.log import LogLevel, LogFormat
parser = argparse.ArgumentParser()
parser.add_argument(
"-L", "--log-level", help="Log level",
default=LogLevel.default(),
choices=LogLevel.choices(),
)
parser.add_argument(
"--log-format", help="Log format",
default=LogFormat.default(),
choices=LogFormat.choices(),
)
parser.add_argument(
"-D", "--debug", action='store_true',
help="Run loop and application in debug mode"
)
parser.add_argument(
"--pool-size", help="Thread pool size",
default=4, type=int,
)
log = logging.getLogger(__name__)
async def main():
log.info('Starting')
await asyncio.sleep(3)
log.info('Exiting')
if __name__ == '__main__':
arg = parser.parse_args()
with entrypoint(
log_level=arg.log_level,
log_format=arg.log_format
) as loop:
loop.run_until_complete(main())
Install event loop on program start:
import asyncio
import aiomisc
# Installing uvloop event loop
# and set `aiomisc.thread_pool.ThreadPoolExecutor`
# as default executor
aiomisc.new_event_loop()
async def main():
await asyncio.sleep(1)
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
Close current 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())
Table Of Contents¶
- entrypoint
run()
shortcut- Services
- Abstract connection pool
- Context
- timeout decorator
- Async backoff
- asyncretry
- Circuit Breaker
- cutout
- Aggregate decorator
- asynchronous file operations
- Working with threads
- ProcessPoolExecutor
- Utilities
- Worker Pool
- Logging configuration
- Pytest plugin
- Signal
- Plugins
- Statistic counters
- API Reference
Versioning¶
This software follows Semantic Versioning
How to develop?¶
Should be installed:
virtualenv
GNU Make as make
Python 3.5+ as python3
For setting up developer environment just type
make develop