aiomisc - miscellaneous utils for asyncio

Navigation

Verified details

These details have been verified by PyPI
Maintainers
Avatar for mosquito from gravatar.com mosquito Avatar for robot-mosquito from gravatar.com robot-mosquito

Unverified details

These details have not been verified by PyPI
Meta
Classifiers
Report project as malware

Project description

Coveralls Travis CI Latest Version https://img.shields.io/pypi/wheel/aiomisc.svg https://img.shields.io/pypi/pyversions/aiomisc.svg https://img.shields.io/pypi/l/aiomisc.svg

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())

entrypoint

Running and graceful shutdown multiple services in one process.

import asyncio
from aiomisc.entrypoint import entrypoint
from aiomisc.service import Service, TCPServer, UDPServer


class LoggingService(Service):
    async def start(self):
        while True:
            print('Hello from service', self.name)
            await asyncio.sleep(1)


class EchoServer(TCPServer):
    async def handle_client(self, reader: asyncio.StreamReader,
                            writer: asyncio.StreamWriter):
        while True:
            writer.write(await reader.readline())


class UDPPrinter(UDPServer):
    async def handle_datagram(self, data: bytes, addr):
        print(addr, '->', data)


services = (
    LoggingService(name='#1'),
    EchoServer(address='::1', port=8901),
    UDPPrinter(address='::1', port=3000),
)


with entrypoint(*services) as loop:
    loop.run_forever()

Service for aiohttp

Installed aiohttp required.

import aiohttp.web
from aiomisc.entrypoint import entrypoint
from aiomisc.service.aiohttp import AIOHTTPService


async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return aiohttp.web.Response(text=text)


class REST(AIOHTTPService):
    async def create_application(self):
        app = aiohttp.web.Application()

        app.add_routes([
            aiohttp.web.get('/', handle),
            aiohttp.web.get('/{name}', handle)
        ])

        return app


service = REST(address='127.0.0.1', port=8080)


with entrypoint(service) as loop:
    loop.run_forever()

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

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page