# 226. Python's AnyIO Library

#### **🔹 1. Basic AnyIO Example**

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

```python
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`**

```python
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`**

```python
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.

```python
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.

```python
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**

```python
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**

```python
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**

```python
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.

```python
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`**

```python
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.

***
