پاسخ تمرینات آموزش ماژول Asyncio – بهبود عملکرد با Gather

import asyncio

async def download_file(name):
    print(f'download {name} started')
    await asyncio.sleep(2)
    return name

async def main():
    res = await asyncio.gather(
        download_file('test1.txt'),
        download_file('test2.txt'),
    )
    print(res)   # ['test1.txt', 'test2.txt']

asyncio.run(main())   # ~2s, not 4s

import asyncio

async def download_file(name):
    print(f'download {name} started')
    await asyncio.sleep(2)
    return name

async def main():
    res = await asyncio.gather(
        download_file('test1.txt'),
        download_file('test2.txt'),
    )
    print(res)   # ['test1.txt', 'test2.txt']

asyncio.run(main())   # ~2s, not 4s

import asyncio

async def download_file(name):
    assert '%' not in name, "Bad file name!"
    await asyncio.sleep(1)
    return name

async def main():
    res = await asyncio.gather(
        download_file('test.txt'),
        download_file('text%.txt'),
        return_exceptions=True,
    )
    print(res)   # ['test.txt', AssertionError('Bad file name!')]

asyncio.run(main())

import asyncio

async def download_file(name):
    assert '%' not in name, "Bad file name!"
    await asyncio.sleep(1)
    return name

async def main():
    res = await asyncio.gather(
        download_file('test.txt'),
        download_file('text%.txt'),
        return_exceptions=True,
    )
    print(res)   # ['test.txt', AssertionError('Bad file name!')]

asyncio.run(main())

import asyncio

async def call_api(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise RuntimeError(f"{name} failed")
    return f"{name} ok"

async def main():
    results = await asyncio.gather(
        call_api("users", 1),
        call_api("orders", 2, fail=True),
        call_api("payments", 1),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            print("error:", r)
        else:
            print("success:", r)

asyncio.run(main())

import asyncio

async def call_api(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise RuntimeError(f"{name} failed")
    return f"{name} ok"

async def main():
    results = await asyncio.gather(
        call_api("users", 1),
        call_api("orders", 2, fail=True),
        call_api("payments", 1),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            print("error:", r)
        else:
            print("success:", r)

asyncio.run(main())