{"id":329,"date":"2024-01-28T10:58:15","date_gmt":"2024-01-28T10:58:15","guid":{"rendered":"https:\/\/learnpython.elegantwallp.com\/?p=329"},"modified":"2024-01-28T10:58:16","modified_gmt":"2024-01-28T10:58:16","slug":"python-asyncio-wait_for","status":"publish","type":"post","link":"https:\/\/learnpython.elegantwallp.com\/2024\/01\/28\/python-asyncio-wait_for\/","title":{"rendered":"Python asyncio.wait_for()"},"content":{"rendered":"\n

Summary<\/strong>: in this tutorial, you\u2019ll learn how to use the asyncio.wait_for()<\/code> function to wait for a coroutine to complete with a timeout.<\/p>\n\n\n\n

Introduction to the Python asyncio.wait_for() function<\/h2>\n\n\n\n

In the previous tutorial, you learned how to cancel a task that is in progress by using the\u00a0cancel()<\/code>\u00a0method of the\u00a0Task<\/code>\u00a0object.<\/p>\n\n\n\n

To wait for a task to complete with a timeout, you can use the asyncio.wait_for()<\/code> function. The asyncio.wait_for()<\/code> function waits for a single task to be completed with a timeout.<\/p>\n\n\n\n

When a timeout occurs, the\u00a0asyncio.wait_for()<\/code>\u00a0function cancels the task and raises the\u00a0TimeoutError<\/code>\u00a0exception. Otherwise, it returns the result of the task. For example:<\/p>\n\n\n\n

import asyncio from asyncio.exceptions import TimeoutError async def call_api(message, result=1000, delay=3): print(message) await asyncio.sleep(delay) return result async def main(): task = asyncio.create_task( call_api('Calling API...', result=2000, delay=5) ) MAX_TIMEOUT = 3 try: await asyncio.wait_for(task, timeout=MAX_TIMEOUT) except TimeoutError: print('The task was cancelled due to a timeout') asyncio.run(main()) <\/code>Code language: JavaScript (javascript)<\/small><\/code><\/pre>\n\n\n\n
Output:<\/p>\n\n\n\n
Calling API... The task was cancelled due to a timeout<\/code><\/code><\/pre>\n\n\n\n
How it works<\/p>\n\n\n\n
First, define a\u00a0call_api()<\/code>\u00a0coroutine that takes 3 seconds to complete by default:<\/p>\n\n\n\n
async def call_api(message, result=1000, delay=3): print(message) await asyncio.sleep(delay) return result<\/code>Code language: JavaScript (javascript)<\/small><\/code><\/pre>\n\n\n\n
Second, create a task that wraps the call_api coroutine and takes 5 seconds to complete:<\/p>\n\n\n\n
task = asyncio.create_task( call_api('Calling API...', result=2000, delay=5) )<\/code>Code language: JavaScript (javascript)<\/small><\/code><\/pre>\n\n\n\n
Third, use the\u00a0asyncio.wait_for()<\/code>\u00a0function to wait for the task to complete with a timeout of 3 seconds. Since the task takes 5 seconds to complete, a timeout will occur and a\u00a0TimeoutError<\/code>\u00a0will be raised:<\/p>\n\n\n\n
MAX_TIMEOUT = 3 try: await asyncio.wait_for(task, timeout=MAX_TIMEOUT) except TimeoutError: print('The task was cancelled due to a timeout')<\/code>Code language: PHP (php)<\/small><\/code><\/pre>\n\n\n\n
Shielding a task from cancellation<\/h2>\n\n\n\n
Sometimes, you may want to inform users that a task is taking longer than expected after a certain amount of time but not cancel the task when a timeout is exceeded.<\/p>\n\n\n\n
To do that, you can wrap the task with the\u00a0asyncio.shield()<\/code>\u00a0function. The\u00a0asyncio.shield()<\/code>\u00a0prevents the cancellation of a task. For example:<\/p>\n\n\n\n
import asyncio from asyncio.exceptions import TimeoutError async def call_api(message, result=1000, delay=3): print(message) await asyncio.sleep(delay) return result async def main(): task = asyncio.create_task( call_api('Calling API...', result=2000, delay=5) ) MAX_TIMEOUT = 3 try: await asyncio.wait_for(asyncio.shield(task), timeout=MAX_TIMEOUT) except TimeoutError: print('The task took more than expected and will complete soon.') result = await task print(result) asyncio.run(main()) <\/code>Code language: JavaScript (javascript)<\/small><\/code><\/pre>\n\n\n\n
Output:<\/p>\n\n\n\n
Calling API... The task took more than expected and will complete soon. 2000<\/code><\/code><\/pre>\n\n\n\n
In this example, the task takes 5 seconds to complete. Once the timeout is 3 seconds, the TimeoutEror<\/code> exception is raised. However, the task is not canceled due to the asyncio.shield()<\/code> function.<\/p>\n\n\n\n
In the exception handling section, we await<\/code> for the task to be completed and print out the result.<\/p>\n","protected":false},"excerpt":{"rendered":"
Summary: in this tutorial, you\u2019ll learn how to use the asyncio.wait_for() function to wait for a coroutine to complete with a timeout. Introduction to the Python asyncio.wait_for() function In the previous tutorial, you learned how to cancel a task that is in progress by using the\u00a0cancel()\u00a0method of the\u00a0Task\u00a0object. To wait for a task to complete with a […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[41],"tags":[],"class_list":["post-329","post","type-post","status-publish","format-standard","hentry","category-1-python-concurrency"],"_links":{"self":[{"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/posts\/329","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/comments?post=329"}],"version-history":[{"count":1,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/posts\/329\/revisions"}],"predecessor-version":[{"id":330,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/posts\/329\/revisions\/330"}],"wp:attachment":[{"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/media?parent=329"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/categories?post=329"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/tags?post=329"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}