226. Python's AnyIO Library

🔹 1. Basic AnyIO Example

This example demonstrates how to run code that works across multiple async frameworks.

import anyio

async def say_hello():
    print("Hello from AnyIO!")

anyio.run(say_hello)

✅ Fix: AnyIO will automatically choose the right async framework (like asyncio, Trio, etc.) to run the function.

---

🔹 2. Running Code on asyncio

import anyio

async def fetch_data():
    print("Fetching data...")

async def main():
    await fetch_data()

anyio.run(main)

✅ Fix: AnyIO will use asyncio if no other framework is specified. This is equivalent to using asyncio.run() directly.

---

🔹 3. Running Code on Trio

import anyio

async def process_data():
    print("Processing data...")

async def main():
    await process_data()

anyio.run(main, backend="trio")

✅ Fix: Here, Trio is explicitly set as the backend, and AnyIO runs the async code using Trio's event loop.

---

🔹 4. Using AnyIO to Run Different Frameworks

You can easily switch between frameworks by changing the backend argument.

import anyio

async def run_task():
    print("Running task...")

async def main():
    await run_task()

# Running with AsyncIO
anyio.run(main, backend="asyncio")

# Running with Trio
anyio.run(main, backend="trio")

✅ Fix: This allows you to run the same code under different backends for testing or deployment.

---

🔹 5. Async Context Manager Example

AnyIO provides support for async context managers.

import anyio

async def task():
    async with anyio.create_task_group() as task_group:
        task_group.start_soon(say_hello)
        task_group.start_soon(say_hello)

async def say_hello():
    print("Hello from Task Group!")

anyio.run(task)

✅ Fix: The context manager is useful when dealing with multiple concurrent tasks.

---

🔹 6. Running Code with Timeouts

import anyio

async def long_running_task():
    await anyio.sleep(2)
    print("Long-running task completed")

async def main():
    async with anyio.move_on_after(1):
        await long_running_task()

anyio.run(main)

✅ Fix: move_on_after() allows for adding timeouts to your async code, and AnyIO handles timeouts in a cross-framework manner.

---

🔹 7. Using AnyIO with Parallelism

import anyio

async def fetch_data(i):
    print(f"Fetching data {i}...")
    await anyio.sleep(1)

async def main():
    async with anyio.create_task_group() as task_group:
        for i in range(5):
            task_group.start_soon(fetch_data, i)

anyio.run(main)

✅ Fix: Task parallelism using AnyIO's create_task_group() and start_soon() makes it easier to run multiple async tasks concurrently.

---

🔹 8. AnyIO with Resources and Cleanup

import anyio

class Resource:
    def __init__(self, name):
        self.name = name

    async def close(self):
        print(f"Cleaning up resource {self.name}")

async def process_resource(resource):
    print(f"Processing {resource.name}...")

async def main():
    async with anyio.open_resource(Resource("Resource 1")) as resource:
        await process_resource(resource)

anyio.run(main)

✅ Fix: This demonstrates resource management in AnyIO, ensuring that resources are properly cleaned up after use.

---

🔹 9. Handling Exceptions in AnyIO

AnyIO provides built-in support for handling exceptions across multiple backends.

import anyio

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

async def main():
    try:
        await risky_task()
    except Exception as e:
        print(f"Error caught: {e}")

anyio.run(main)

✅ Fix: Exception handling works seamlessly across frameworks, allowing consistent behavior across different environments.

---

🔹 10. Using AnyIO's run_in_worker_thread

import anyio

def blocking_task():
    import time
    time.sleep(2)
    return "Blocking task done"

async def main():
    result = await anyio.to_thread(blocking_task)
    print(result)

anyio.run(main)

✅ Fix: Use to_thread() to run blocking code (like CPU-bound tasks) in a separate thread without blocking the event loop.

---

🚀 Benefits of Using AnyIO

1. Cross-Framework Compatibility – Supports multiple async backends (asyncio, Trio, Curio) without changing code.

2. Task Grouping – Simplifies managing multiple tasks concurrently through task groups.

3. Graceful Shutdown and Resource Cleanup – Automatic handling of cleanup, timeouts, and resource management.

4. Error Handling – Consistent exception handling across different async frameworks.

5. Thread Pool Integration – Easily run blocking I/O-bound code in a worker thread with to_thread.

---

When to Use AnyIO:

• Framework Agnosticism – When you want to write code that is flexible enough to switch between async frameworks without heavy modifications.

• Testing Across Multiple Backends – When you need to test or develop for different async frameworks.

• Concurrency Management – When you want to manage multiple async tasks or perform I/O-bound operations efficiently.

---