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()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.5.2.tar.gz
(9.3 kB
view details)
Built Distribution
aiomisc-0.5.2-py3-none-any.whl
(13.6 kB
view details)
File details
Details for the file aiomisc-0.5.2.tar.gz.
File metadata
- Download URL: aiomisc-0.5.2.tar.gz
- Upload date:
- Size: 9.3 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 | 12f3b42dee1ea56ac4f924b58ad73079e02f31742a68112709f77111dccbde6f |
|
| MD5 | 651df8755eff4e01c3d9801cb4965d78 |
|
| BLAKE2b-256 | a58b33b282508d8cf08c649726be148a0d723ce879f88fa8e3a5169fbbb98765 |
File details
Details for the file aiomisc-0.5.2-py3-none-any.whl.
File metadata
- Download URL: aiomisc-0.5.2-py3-none-any.whl
- Upload date:
- Size: 13.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 | bf0ad35b52bb1b472f0d5ab754b5edac4f4b90a33706e25cd62ae2c1975362fb |
|
| MD5 | b7c9d5bd94111734233b4703e47182bf |
|
| BLAKE2b-256 | 239db7a11f787d500a5c95094ae8b6150df3183a15e39395b5a0abc88cd68fd2 |
