72. Python's asyncio Event Loop

Here are 10 Python code snippets demonstrating how to use Python's asyncio module for writing asynchronous code, focusing on IO-bound tasks.

---

1. Basic Asyncio Event Loop

Creating a simple asynchronous function and running it using the event loop.

import asyncio

async def say_hello():
    print("Hello, world!")

# Running the event loop
asyncio.run(say_hello())

---

2. Asynchronous Sleep with asyncio.sleep

Using asyncio.sleep to simulate a non-blocking delay.

import asyncio

async def delayed_greeting():
    print("Starting greeting...")
    await asyncio.sleep(2)
    print("Hello after 2 seconds!")

asyncio.run(delayed_greeting())

---

3. Multiple Asynchronous Tasks

Running multiple asynchronous tasks concurrently.

import asyncio

async def task1():
    print("Task 1 starts")
    await asyncio.sleep(1)
    print("Task 1 ends")

async def task2():
    print("Task 2 starts")
    await asyncio.sleep(2)
    print("Task 2 ends")

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

asyncio.run(main())

---

4. Asynchronous IO with aiohttp for HTTP Requests

Performing asynchronous HTTP requests using aiohttp.

import aiohttp
import asyncio

async def fetch_data(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    data = await fetch_data('https://jsonplaceholder.typicode.com/posts')
    print(data[:100])  # Printing only the first 100 characters of the response

asyncio.run(main())

---

5. Handling Asynchronous Exceptions

Handling exceptions in asynchronous functions.

import asyncio

async def might_fail():
    print("This might fail...")
    await asyncio.sleep(1)
    raise ValueError("Something went wrong!")

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

asyncio.run(main())

---

6. Asyncio with Timeouts

Using asyncio.wait_for to apply a timeout to an asynchronous task.

import asyncio

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

async def main():
    try:
        await asyncio.wait_for(long_running_task(), timeout=2)
    except asyncio.TimeoutError:
        print("The task took too long!")

asyncio.run(main())

---

7. Asyncio Queue for Task Synchronization

Using asyncio.Queue for managing tasks in a producer-consumer model.

import asyncio

async def producer(queue):
    for i in range(5):
        await asyncio.sleep(1)
        await queue.put(i)
        print(f"Produced {i}")

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Consumed {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(producer(queue), consumer(queue))

asyncio.run(main())

---

8. Asynchronous File I/O with aiofiles

Performing non-blocking file I/O using aiofiles.

import aiofiles
import asyncio

async def write_to_file():
    async with aiofiles.open('example.txt', mode='w') as file:
        await file.write("This is an example of async file I/O.")

async def main():
    await write_to_file()

asyncio.run(main())

---

9. Asynchronous Database Queries with aiomysql

Performing asynchronous MySQL queries using aiomysql.

import asyncio
import aiomysql

async def fetch_data():
    async with aiomysql.create_pool(user='user', db='test', host='127.0.0.1', port=3306) as pool:
        async with pool.acquire() as conn:
            async with conn.cursor() as cursor:
                await cursor.execute('SELECT * FROM users')
                result = await cursor.fetchall()
                print(result)

asyncio.run(fetch_data())

---

10. Asyncio with Custom Event Loop

Running a custom event loop manually instead of using asyncio.run.

import asyncio

async def custom_loop_task():
    print("Task started in custom loop!")
    await asyncio.sleep(2)
    print("Task finished!")

loop = asyncio.get_event_loop()
loop.run_until_complete(custom_loop_task())
loop.close()

---

These snippets showcase various ways to handle IO-bound tasks using Python's asyncio module. They cover simple asynchronous function execution, concurrent tasks, timeouts, exception handling, and interaction with external systems like HTTP requests, file I/O, and databases.