Asyncio module failing to create task

Question

My Source Code:

import asyncio

async def mycoro(number):
    print(f'Starting {number}')
    await asyncio.sleep(1)
    print(f'Finishing {number}')
    return str(number)

c = mycoro(3)
task = asyncio.create_task(c)
loop = asyncio.get_event_loop()
loop.run_until_complete(task)
loop.close()

Error:

RuntimeError: no running event loop
sys:1: RuntimeWarning: coroutine 'mycoro' was never awaited

I was watching a tutorial and according to my code it was never awaited when I did and it clearly does in the video I was watching.

Answer 1

Simply run the coroutine directly without creating a task for it:

import asyncio

async def mycoro(number):
    print(f'Starting {number}')
    await asyncio.sleep(1)
    print(f'Finishing {number}')
    return str(number)

c = mycoro(3)
loop = asyncio.get_event_loop()
loop.run_until_complete(c)
loop.close()

The purpose of asyncio.create_task is to create an additional task from inside a running task. Since it directly starts the new task, it must be used inside a running event loop – hence the error when using it outside.

Use loop.create_task(c) if a task must be created from outside a task.

---

In more recent version of asyncio, use asyncio.run to avoid having to handle the event loop explicitly:

c = mycoro(3)
asyncio.run(c)

In general, use asyncio.create_task only to increase concurrency. Avoid using it when another task would block immediately.

# bad task usage: concurrency stays the same due to blocking
async def bad_task():
    task = asyncio.create_task(mycoro(0))
    await task

# no task usage: concurrency stays the same due to stacking
async def no_task():
     await mycoro(0)

# good task usage: concurrency is increased via multiple tasks
async def good_task():
    tasks = [asyncio.create_task(mycoro(i)) for i in range(3)]
    print('Starting done, sleeping now...')
    await asyncio.sleep(1.5)
    await asyncio.gather(*tasks)  # ensure subtasks finished

Answer 2

Change the line

task = asyncio.Task(c)