225. Python’s Asynchronous Generators

🔹 1. Basic Async Generator Example

Copy

import asyncio

async def count_up_to(max):
    count = 1
    while count <= max:
        yield count
        count += 1
        await asyncio.sleep(1)  # Simulate an async operation

async def main():
    async for number in count_up_to(5):
        print(number)

asyncio.run(main())

✅ Fix: This example shows an async generator that yields numbers while performing asynchronous operations (simulated with await asyncio.sleep).

---

🔹 2. Async Generator with External Data (Simulated I/O Operation)

Copy

import asyncio

async def fetch_data_from_database():
    for i in range(3):
        await asyncio.sleep(1)  # Simulate a database call
        yield f"Data {i}"

async def main():
    async for data in fetch_data_from_database():
        print(data)

asyncio.run(main())

✅ Fix: An async generator is used to fetch data asynchronously, simulating database calls with await asyncio.sleep.

---

🔹 3. Async Generator with Error Handling

Copy

import asyncio

async def process_data(data_list):
    for item in data_list:
        if item == "error":
            raise ValueError("Error occurred while processing data.")
        await asyncio.sleep(1)
        yield f"Processed {item}"

async def main():
    try:
        async for result in process_data(["data1", "data2", "error", "data3"]):
            print(result)
    except ValueError as e:
        print(f"Handled error: {e}")

asyncio.run(main())

✅ Fix: An async generator with proper error handling using try-except to catch errors while processing asynchronous data.

---

🔹 4. Async Generator with a Custom Stop Condition

Copy

import asyncio

async def generate_numbers(limit):
    i = 0
    while True:
        if i >= limit:
            break
        yield i
        i += 1
        await asyncio.sleep(0.5)

async def main():
    async for number in generate_numbers(5):
        print(number)

asyncio.run(main())

✅ Fix: The stop condition inside the generator (break when i >= limit) ensures that it stops after yielding the specified number of values.

---

🔹 5. Using Async Generators for Streaming Data

Copy

import asyncio

async def stream_data():
    for i in range(5):
        await asyncio.sleep(1)  # Simulate I/O-bound task
        yield f"Streaming data {i}"

async def main():
    async for data in stream_data():
        print(data)

asyncio.run(main())

✅ Fix: Streaming data with an async generator, simulating data being fetched or generated over time, without blocking.

---

🔹 6. Async Generator with async for and External Data Source (File I/O)

Copy

import asyncio

async def read_file_lines(file_path):
    with open(file_path, 'r') as file:
        for line in file:
            await asyncio.sleep(0.5)  # Simulate reading delay
            yield line.strip()

async def main():
    async for line in read_file_lines('sample.txt'):
        print(line)

asyncio.run(main())

✅ Fix: An async generator that reads lines from a file asynchronously. Each line is yielded after a simulated delay.

---

🔹 7. Async Generator with Multiple Consumers

Copy

import asyncio

async def generate_numbers():
    for i in range(1, 6):
        await asyncio.sleep(1)
        yield i

async def consumer(name, number_generator):
    async for number in number_generator:
        print(f"{name} consumed: {number}")

async def main():
    number_generator = generate_numbers()
    consumer1 = asyncio.create_task(consumer("Consumer 1", number_generator))
    consumer2 = asyncio.create_task(consumer("Consumer 2", number_generator))

    await asyncio.gather(consumer1, consumer2)

asyncio.run(main())

✅ Fix: Two consumers are consuming data from a single async generator, ensuring that the generator is asynchronously yielding values to both consumers.

---

🔹 8. Async Generator with a Time Limit

Copy

import asyncio

async def time_limited_generator(max_time):
    start_time = asyncio.get_event_loop().time()
    while asyncio.get_event_loop().time() - start_time < max_time:
        yield "Data"
        await asyncio.sleep(0.5)

async def main():
    async for data in time_limited_generator(2):
        print(data)

asyncio.run(main())

✅ Fix: This async generator continues to yield until the time limit is reached, demonstrating a controlled streaming process.

---

🔹 9. Async Generator with a Custom Async Iterator

Copy

import asyncio

class AsyncRange:
    def __init__(self, start, end):
        self.current = start
        self.end = end

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current >= self.end:
            raise StopAsyncIteration
        await asyncio.sleep(1)  # Simulate a delay
        self.current += 1
        return self.current - 1

async def main():
    async for number in AsyncRange(1, 5):
        print(number)

asyncio.run(main())

✅ Fix: A custom asynchronous iterator is used to implement an async range generator with __aiter__ and __anext__.

---

🔹 10. Async Generator with a Complex Workflow (Fetching & Processing)

Copy

import asyncio

async def fetch_data():
    # Simulate fetching data from a network or database
    await asyncio.sleep(1)
    return [1, 2, 3, 4, 5]

async def process_data(data):
    for item in data:
        await asyncio.sleep(0.5)  # Simulate processing time
        yield f"Processed {item}"

async def main():
    data = await fetch_data()
    async for result in process_data(data):
        print(result)

asyncio.run(main())

✅ Fix: This example combines fetching data asynchronously and then processing it using an async generator, which is ideal for workflows where data fetching and processing are interdependent.

---

🚀 Why Use Async Generators?

FeatureBenefit

Efficient Memory Use

✅ Async generators yield data one piece at a time, preventing memory overload.

Non-blocking I/O

✅ They work well with I/O-bound operations, like file I/O, network requests, or database queries.

Improved Performance

✅ Async generators allow for concurrent tasks without blocking the program.

Lazy Evaluation

✅ They delay evaluation until the data is needed, improving program responsiveness.

---

🚀 When to Use Async Generators?

ScenarioUse Async Generators?

Streaming Data from a Remote API

✅ Yes

Processing Large Data Files

✅ Yes

Concurrent Task Execution

✅ Yes

Asynchronous Iteration Over Network or Disk I/O

✅ Yes

---