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
  • Configuration from environment
• run() shortcut
• Services
  • TCPServer
  • UDPServer
  • TLSServer
  • TCPClient
  • TLSClient
  • RobustTCPClient
  • RobustTLSClient
  • PeriodicService
  • CronService
  • Multiple services
  • Configuration
  • aiohttp service
  • asgi service
  • Memory Tracer
  • Profiler
  • Raven service
  • SDWatchdogService
  • ProcessService
  • RespawningProcessService
• Abstract connection pool
• Context
• timeout decorator
• Async backoff
• asyncretry
• Circuit Breaker
• cutout
• Aggregate decorator
• asynchronous file operations
• Working with threads
  • contextvars support
  • @threaded
  • @threaded_separate
  • Threaded iterator decorator
  • aiomisc.IteratorWrapper
  • aiomisc.IteratorWrapperSeparate
  • aiomisc.ThreadPoolExecutor
  • aiomisc.sync_wait_coroutine
• ProcessPoolExecutor
• Utilities
  • Select
  • cancel_tasks
  • awaitable
  • Bind socket
  • Recurring callback
  • Periodic callback
  • Cron callback
• Worker Pool
  • Example
• Logging configuration
  • Color
  • JSON
  • JournalD
  • Rich
  • Buffered log handler
• Pytest plugin
  • Basic usage
  • pytest markers
  • Passing default context
  • Testing services
  • Event loop policy overriding
  • Thread pool overriding
  • entrypoint arguments
  • aiohttp test client
  • TCPProxy
• Signal
  • pre_start
  • post_start
  • pre_stop
  • post_stop
• 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