Last active
June 21, 2018 16:46
-
-
Save andymitchhank/ff6f501eecf142e6b611e17e253d346b to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
import signal | |
from threading import Thread | |
import time | |
from decorators import logger | |
@logger | |
async def do_some_work(x): | |
await asyncio.sleep(x) | |
@logger | |
def more_work(x): | |
time.sleep(x) | |
@logger | |
def block_the_main_thread(): | |
loop = asyncio.get_event_loop() | |
loop.run_until_complete(do_some_work(5)) | |
tasks = [asyncio.ensure_future(do_some_work(2)), | |
asyncio.ensure_future(do_some_work(5))] | |
loop.run_until_complete(asyncio.gather(*tasks)) | |
def start_loop(loop): | |
asyncio.set_event_loop(loop) | |
loop.run_forever() | |
@logger | |
def on_a_new_thread(): | |
new_loop = asyncio.new_event_loop() | |
t = Thread(target=start_loop, args=(new_loop,)) | |
t.start() | |
new_loop.call_soon_threadsafe(more_work, 20) | |
asyncio.run_coroutine_threadsafe(do_some_work(5), new_loop) | |
last_future = asyncio.run_coroutine_threadsafe(do_some_work(10), new_loop) | |
# Since we know which coroutine will finish last, | |
# we can stop the loop when it's done | |
last_future.add_done_callback(lambda f: new_loop.stop()) | |
if __name__ == '__main__': | |
block_the_main_thread() | |
on_a_new_thread() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import datetime | |
import inspect | |
now = datetime.datetime.now | |
def format_fn(func, *args, **kwargs): | |
fkwargs = ', '.join(f'{k}={v}' for k, v in kwargs.items()) | |
rv = f'{func.__name__}(' | |
if args: | |
rv += ', '.join(str(a) for a in args) | |
if kwargs: | |
rv += ', '.join(f'{k}={v}' for k, v in kwargs.items()) | |
rv += ')' | |
return rv | |
def logger(func): | |
def inner(*args, **kwargs): | |
fn = format_fn(func, *args, **kwargs) | |
print(f'[{now()}] Entering {fn}') | |
rv = func(*args, **kwargs) | |
print(f'[{now()}] Exiting {fn}') | |
return rv | |
async def inner_async(*args, **kwargs): | |
fn = format_fn(func, *args, **kwargs) | |
print(f'[{now()}] Entering {fn}') | |
rv = await func(*args, **kwargs) | |
print(f'[{now()}] Exiting {fn}') | |
return rv | |
return inner_async if inspect.iscoroutinefunction(func) else inner | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment