آموزش ماژول asyncio – کنسل کردن coroutine

import asyncio

async def download_file(file_name):
    print(f'dowload {file_name} started')
    await asyncio.sleep(6)
    print(f'dowload {file_name} finished')

async def manage_downloads():
    t1 = asyncio.create_task(download_file('test1.txt'))

    seconds = 0
    while not t1.done():
        print('download still running')
        await asyncio.sleep(1)
        seconds += 1
        if seconds == 5:
            t1.cancel()

    try:
        await t1
    except asyncio.CancelledError:
        print('download cancelled!')

asyncio.run(manage_downloads())

import asyncio

async def download_file(file_name):
    print(f'dowload {file_name} started')
    await asyncio.sleep(6)
    print(f'dowload {file_name} finished')

async def manage_downloads():
    t1 = asyncio.create_task(download_file('test.txt'))

    try:
        await asyncio.wait_for(t1, timeout=5)
    except asyncio.CancelledError:
        print('download cancelled!')

asyncio.run(manage_downloads())

import asyncio

async def download_file(file_name):
    print(f'dowload {file_name} started')
    await asyncio.sleep(6)
    print(f'dowload {file_name} finished')

async def manage_downloads():
    t1 = asyncio.create_task(download_file('test.txt'))

    try:
        await asyncio.wait_for(asyncio.shield(t1), timeout=5)
    except asyncio.TimeoutError:
        print('downloading takes too long but we ar working on it')
        await t1

asyncio.run(manage_downloads())