Instantly share code, notes, and snippets.

Embed
What would you like to do?
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()
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