aiomisc 0.4.6

aiomisc - miscellaneous utils for asyncio

Miscellaneous utils for asyncio.

Installation

Installing from pypi:

pip3 install aiomisc

Installing from github.com:

pip3 install git+https://github.com/mosquito/aiomisc.git

Quick Start

Install event loop on the program starts.

import asyncio
from aiomisc.utils import new_event_loop


# Installing uvloop event loop
# and set `aiomisc.thread_pool.ThreadPoolExecutor`
# as default executor
new_event_loop()


async def main():
    await asyncio.sleep(3)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Close current event loop and install the new one:

import asyncio
from aiomisc.utils import new_event_loop


async def main():
    await asyncio.sleep(3)


if __name__ == '__main__':
    loop = new_event_loop()
    loop.run_until_complete(main())

threaded decorator

Wraps blocking function and run it on the thread pool.

import asyncio
import time
from aiomisc.utils import new_event_loop
from aiomisc.thread_pool import threaded


@threaded
def blocking_function():
    time.sleep(1)


async def main():
    # Running in parallel
    await asyncio.gather(
        blocking_function(),
        blocking_function(),
    )


if __name__ == '__main__':
    loop = new_event_loop()
    loop.run_until_complete(main())

Fast ThreadPoolExecutor

This is the simple thread pool implementation.

Installation as a default thread pool:

import asyncio
from aiomisc.thread_pool import ThreadPoolExecutor

loop = asyncio.get_event_loop()
thread_pool = ThreadPoolExecutor(4, loop=loop)
loop.set_default_executor(thread_pool)

Bind socket

from aiomisc.utils import bind_socket

# IPv4 socket
sock = bind_socket(address="127.0.0.1", port=1234)

# IPv6 socket (on Linux IPv4 socket will be bind too)
sock = bind_socket(address="::1", port=1234)

Periodic callback

Runs coroutine function periodically

import asyncio
import time
from aiomisc.utils import new_event_loop
from aiomisc.periodic import PeriodicCallback


async def periodic_function():
    print("Hello")


if __name__ == '__main__':
    loop = new_event_loop()

    periodic = PeriodicCallback(periodic_function)

    # Call it each second
    periodic.start(1)

    loop.run_forever()

Logging configuration

Setting up colorized logs:

import logging
from aiomisc.log import basic_config


# Configure logging
basic_config(level=logging.INFO, buffered=False, log_format='color')

Setting up json logs:

import logging
from aiomisc.log import basic_config


# Configure logging
basic_config(level=logging.INFO, buffered=False, log_format='json')

Buffered log handler

Parameter buffered=True enables memory buffer which flushing logs in thread.

import logging
from aiomisc.log import basic_config
from aiomisc.periodic import PeriodicCallback
from aiomisc.utils import new_event_loop


# Configure logging globally
basic_config(level=logging.INFO, buffered=False, log_format='json')

async def write_log(loop):
    logging.info("Hello %f", loop.time())

if __name__ == '__main__':
    loop = new_event_loop()

    # Configure
    basic_config(
        level=logging.INFO,
        buffered=True,
        log_format='color',
        flush_interval=2
    )

    periodic = PeriodicCallback(write_log, loop)
    periodic.start(0.3)

    loop.run_forever()

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