Skip to content

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
You can’t perform that action at this time.