ورود
ثبت نام

پاسخ تمرینات آموزش ماژول Asyncio – کنسل کردن Coroutine

Please login to bookmark Close


پاسخ تمرین ۱

متد cancel همان task را کنسل می‌کند. وقتی پس از آن دوباره منتظر همان task بمانیم، پایتون خطای asyncio.CancelledError را صادر می‌کند که باید آن را بگیریم تا کنسل‌شدن را درست مدیریت کنیم.

پاسخ تمرین ۲

asyncio.wait_for هنگام timeout، task را کنسل می‌کند و خطای asyncio.TimeoutError پرتاب می‌کند (نه CancelledError). پس باید except asyncio.TimeoutError بنویسیم.

پاسخ تمرین ۳

به‌جای حلقه‌ی دستی، از asyncio.wait_for با یک timeout استفاده می‌کنیم؛ این یک فراخوانی خودش منتظر task می‌ماند و اگر در مهلت تمام نشود، خودکار کنسلش می‌کند.

پاسخ تمرین ۴

shield یک coroutine را از کنسل‌شدن مستقیم محافظت می‌کند. اما اگر با وجود shield مهلت تمام شود، به‌جای CancelledError خطای TimeoutError می‌گیریم و خودِ task در پس‌زمینه همچنان به اجرا ادامه می‌دهد (متوقف نمی‌شود).

پاسخ تمرین ۵

چون shield فقط جلوی رسیدنِ کنسل به task را می‌گیرد، نه خودِ task را متوقف می‌کند؛ آن task هنوز در پس‌زمینه در حال اجراست. پس اگر نتیجه‌اش را می‌خواهیم (یا فقط می‌خواهیم بگذاریم درست تمام شود)، باید بعداً دوباره منتظرش بمانیم، نه اینکه بدون رسیدگی رهایش کنیم.

پاسخ تمرین ۶

import asyncio

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

async def manage():
    t1 = asyncio.create_task(download_file('test1.txt'))
    seconds = 0
    while not t1.done():
        await asyncio.sleep(1)
        seconds += 1
        if seconds == 5:
            t1.cancel()
    try:
        await t1
    except asyncio.CancelledError:
        print('download cancelled!')

asyncio.run(manage())

پاسخ تمرین ۷

import asyncio

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

async def manage():
    t1 = asyncio.create_task(download_file('test.txt'))
    try:
        await asyncio.wait_for(t1, timeout=5)
    except asyncio.TimeoutError:
        print('download cancelled (timed out)!')

asyncio.run(manage())

پاسخ تمرین ۸

import asyncio

async def call_api():
    await asyncio.sleep(10)
    return {"status": "ok"}

async def get_data_with_timeout():
    try:
        return await asyncio.wait_for(call_api(), timeout=3)
    except asyncio.TimeoutError:
        print("API too slow, using a default value")
        return {"status": "unknown"}

asyncio.run(get_data_with_timeout())

راه‌حل جایگزین (پایتون ۳.۱۱+): استفاده از async with asyncio.timeout(3):.

Please login to bookmark Close
نظرات

دیدگاهتان را بنویسید

فهرست مطالب

سرفصل دوره

بستن

تمرین

این قسمت تمرین ندارد!

بستن

پاسخ تمرین ها

هنوز برای تمرین‌های این قسمت پاسخی ثبت نشده است!

بستن

اشتراک گذاری

بستن

چرا بهتره از فیلترشکن استفاده کنید؟

من همه ویدئو ها و پادکست های کُدباز رو توی یوتیوب و ساندکلود و پلتفرم هایی آپلود می‌کنم که اغلب فیلتر هستند.

اغلب آموزش‌ها ویدئو و پادکست دارند. پس اگر می‌خواهید از محتوای سایت بیشترین استفاده رو ببرید نیاز به فیلتر شکن دارید.

توجه داشته باشید که برای خرید از فروشگاه بهتره فیلتر شکن رو خاموش کنید.

بستن

تنظیمات

انتخاب زبان
تغییر تم
بستن