91. Event Loop in asyncio

Here are 10 Python code snippets demonstrating how to work with the event loop in Python's asyncio for running asynchronous tasks:

1. Basic Event Loop Execution

Copy

import asyncio

async def greet():
    print("Hello, World!")
    await asyncio.sleep(1)
    print("Goodbye, World!")

asyncio.run(greet())

This creates a coroutine greet() and runs it using asyncio.run().

---

2. Running Multiple Coroutines

Copy

import asyncio

async def task1():
    print("Task 1 starting...")
    await asyncio.sleep(2)
    print("Task 1 done.")

async def task2():
    print("Task 2 starting...")
    await asyncio.sleep(1)
    print("Task 2 done.")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

asyncio.gather() runs multiple coroutines concurrently.

---

3. Using asyncio.create_task

Copy

import asyncio

async def task(name, delay):
    print(f"{name} starting...")
    await asyncio.sleep(delay)
    print(f"{name} done.")

async def main():
    t1 = asyncio.create_task(task("Task 1", 2))
    t2 = asyncio.create_task(task("Task 2", 1))
    await t1
    await t2

asyncio.run(main())

asyncio.create_task() schedules coroutines to run concurrently.

---

4. Running Tasks Sequentially

Copy

import asyncio

async def task(name, delay):
    print(f"{name} starting...")
    await asyncio.sleep(delay)
    print(f"{name} done.")

async def main():
    await task("Task 1", 2)
    await task("Task 2", 1)

asyncio.run(main())

Tasks are executed one after another using await.

---

5. Using asyncio.sleep for Delays

Copy

import asyncio

async def countdown(number):
    while number > 0:
        print(f"Countdown: {number}")
        await asyncio.sleep(1)
        number -= 1
    print("Liftoff!")

asyncio.run(countdown(5))

This demonstrates using asyncio.sleep() to create a delay in an asynchronous function.

---

6. Handling Exceptions in Coroutines

Copy

import asyncio

async def faulty_task():
    raise ValueError("Something went wrong!")

async def main():
    try:
        await faulty_task()
    except ValueError as e:
        print(f"Caught an exception: {e}")

asyncio.run(main())

Exceptions in coroutines can be caught and handled with standard try/except blocks.

---

7. Running Tasks with Timeouts

Copy

import asyncio

async def long_task():
    await asyncio.sleep(5)
    print("Task completed!")

async def main():
    try:
        await asyncio.wait_for(long_task(), timeout=3)
    except asyncio.TimeoutError:
        print("Task timed out!")

asyncio.run(main())

asyncio.wait_for() allows setting a timeout for a coroutine.

---

8. Using asyncio.run_until_complete

Copy

import asyncio

async def say_hello():
    await asyncio.sleep(1)
    print("Hello, asyncio!")

loop = asyncio.get_event_loop()
loop.run_until_complete(say_hello())

run_until_complete() runs the event loop until the coroutine is done. Note: asyncio.run() is preferred in most cases.

---

9. Using asyncio.as_completed

Copy

import asyncio

async def task(name, delay):
    await asyncio.sleep(delay)
    return f"{name} completed after {delay} seconds."

async def main():
    tasks = [task("Task 1", 3), task("Task 2", 1), task("Task 3", 2)]
    for completed in asyncio.as_completed(tasks):
        result = await completed
        print(result)

asyncio.run(main())

asyncio.as_completed() yields results of tasks as they complete.

---

10. Running Background Tasks

Copy

import asyncio

async def background_task():
    while True:
        print("Background task running...")
        await asyncio.sleep(2)

async def main():
    task = asyncio.create_task(background_task())
    await asyncio.sleep(6)  # Main task runs for 6 seconds
    task.cancel()  # Cancel the background task
    try:
        await task
    except asyncio.CancelledError:
        print("Background task cancelled.")

asyncio.run(main())

This demonstrates running a long-running background task that can be canceled.

---

These snippets cover a variety of scenarios for running and managing asynchronous tasks using the asyncio event loop in Python.