92. Coroutines in Python

Here are 10 Python code snippets to demonstrate the use of coroutines in Python with async and await:

1. Basic Coroutine Example

Copy

import asyncio

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

asyncio.run(greet())

Defines a coroutine using async and await, and executes it with asyncio.run.

---

2. Chaining Coroutines

Copy

import asyncio

async def step1():
    print("Step 1 completed.")
    await asyncio.sleep(1)

async def step2():
    print("Step 2 completed.")
    await asyncio.sleep(1)

async def main():
    await step1()
    await step2()

asyncio.run(main())

Coroutines can be chained together using await.

---

3. Returning Values from Coroutines

Copy

import asyncio

async def square(x):
    await asyncio.sleep(1)
    return x * x

async def main():
    result = await square(4)
    print(f"Result: {result}")

asyncio.run(main())

A coroutine can return a value after an await.

---

4. Running Coroutines Concurrently

Copy

import asyncio

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

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

asyncio.run(main())

Multiple coroutines are executed concurrently with asyncio.gather.

---

5. Using async with Loops

Copy

import asyncio

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

asyncio.run(countdown(5))

Coroutines can be used inside loops with await.

---

6. Combining Coroutines and Exception Handling

Copy

import asyncio

async def risky_task():
    await asyncio.sleep(1)
    raise ValueError("Something went wrong!")

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

asyncio.run(main())

Coroutines can include try/except blocks for exception handling.

---

7. Coroutine with Timeout

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=2)
    except asyncio.TimeoutError:
        print("Task timed out!")

asyncio.run(main())

asyncio.wait_for can enforce a timeout for a coroutine.

---

8. Creating a Coroutine Dynamically

Copy

import asyncio

async def dynamic_task(name):
    print(f"Task {name} is running.")
    await asyncio.sleep(2)
    print(f"Task {name} is done.")

tasks = [dynamic_task(i) for i in range(3)]

asyncio.run(asyncio.gather(*tasks))

Dynamically creates coroutines in a list and runs them concurrently.

---

9. Using await in a Coroutine Chain

Copy

import asyncio

async def step1():
    print("Step 1 completed.")
    await asyncio.sleep(1)
    return "Data from Step 1"

async def step2(data):
    print(f"Step 2 received: {data}")
    await asyncio.sleep(1)

async def main():
    data = await step1()
    await step2(data)

asyncio.run(main())

Data is passed between coroutines using await.

---

10. Coroutine with Infinite Loop

Copy

import asyncio

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

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

asyncio.run(main())

An infinite coroutine can be created and later canceled.

---