Skip to content

Instantly share code, notes, and snippets.

@pdonorio
Created April 17, 2020 21:16
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save pdonorio/c00b1e975b62a4ce7f3b540aec0b8e1f to your computer and use it in GitHub Desktop.
Save pdonorio/c00b1e975b62a4ce7f3b540aec0b8e1f to your computer and use it in GitHub Desktop.
Asyncio quick test

asyncio

Quick test/tutorial on concurrent tasks, based on asyncio.

"""
https://docs.python.org/3/library/asyncio-task.html
https://www.youtube.com/watch?v=Xbl7XjFYsN4
"""
import time
import random
import asyncio
#######################
#  WHEN USING A NOTEBOOK
# NOTE: clash of running event loop, because jupyter already runs one
# https://markhneedham.com/blog/2019/05/10/jupyter-runtimeerror-this-event-loop-is-already-running/
# !pip3 install nest_asyncio
# import nest_asyncio
# nest_asyncio.apply()
#######################
async def say_after(what):
delay = random.randint(1, 10)
print(f'{what} - sleeping: {delay}')
await asyncio.sleep(delay)
print(what)
async def main():
tasks = []
for word in ['hello', 'world', 'again']:
tasks.append(asyncio.create_task(say_after(word)))
print(f"started at {time.strftime('%X')}")
for task in tasks:
await task
print(f"finished at {time.strftime('%X')}")
# notebook simplified version:
# await main()
asyncio.run(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment