aiomisc - miscellaneous utils for asyncio
Project description
Miscellaneous utils for asyncio.
Installation
Installing from pypi:
pip3 install aiomiscInstalling from github.com:
pip3 install git+https://github.com/mosquito/aiomisc.gitQuick 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
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
aiomisc-0.7.0.tar.gz
(10.9 kB
view details)
Built Distribution
aiomisc-0.7.0-py3-none-any.whl
(17.6 kB
view details)
File details
Details for the file aiomisc-0.7.0.tar.gz.
File metadata
- Download URL: aiomisc-0.7.0.tar.gz
- Upload date:
- Size: 10.9 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/1.11.0 pkginfo/1.4.2 requests/2.18.4 setuptools/33.1.1 requests-toolbelt/0.8.0 tqdm/4.20.0 CPython/3.6.5
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | c6d64b5aba8eac70b3dc9a3a652bff74bd381b501043cb5f35f3a9808c79140f |
|
| MD5 | 00e0b1679b8584722398878386feb044 |
|
| BLAKE2b-256 | 07e5f0fa35a69a5faf88ac614a74e2df07aa66c33a69830689111907f7eba7bb |
File details
Details for the file aiomisc-0.7.0-py3-none-any.whl.
File metadata
- Download URL: aiomisc-0.7.0-py3-none-any.whl
- Upload date:
- Size: 17.6 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/1.11.0 pkginfo/1.4.2 requests/2.18.4 setuptools/33.1.1 requests-toolbelt/0.8.0 tqdm/4.20.0 CPython/3.6.5
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | 9f671b94a7fce2e7833c20d061dc7bda9129343ec2fa4b06b1a96f31eb469e4e |
|
| MD5 | db9774646d23405cd5b4ebac125b62c7 |
|
| BLAKE2b-256 | b19044c2ecc867836942b3d044fcfc1ce46508dc10f3bed1a1031642687d56b9 |
