面试题：Python异步IO在复杂场景下的应用

Question

假设有一个复杂的系统，需要同时处理来自多个不同数据源（如不同的API接口，数据库查询等，每个操作都需要一定延迟模拟真实场景）的异步请求，并且不同数据源返回的数据之间存在依赖关系（例如，数据源B的请求参数需要根据数据源A的返回结果来确定）。请使用Python的asyncio库设计并实现一个解决方案，确保数据的正确获取和处理，同时考虑如何优雅地处理可能出现的异常情况，如某个数据源请求超时或返回错误数据。

Answer 1

import asyncio
import aiohttp
import async_timeout


async def fetch_data_from_api_a():
    async with aiohttp.ClientSession() as session:
        try:
            async with async_timeout.timeout(10):
                async with session.get('url_of_api_a') as response:
                    if response.status == 200:
                        return await response.json()
                    else:
                        raise Exception(f"API A returned non - 200 status: {response.status}")
        except asyncio.TimeoutError:
            raise Exception("API A request timed out")


async def fetch_data_from_api_b(data_from_a):
    params = {'param': data_from_a['key']}  # 根据数据源A的返回结果确定数据源B的请求参数
    async with aiohttp.ClientSession() as session:
        try:
            async with async_timeout.timeout(10):
                async with session.get('url_of_api_b', params=params) as response:
                    if response.status == 200:
                        return await response.json()
                    else:
                        raise Exception(f"API B returned non - 200 status: {response.status}")
        except asyncio.TimeoutError:
            raise Exception("API B request timed out")


async def main():
    try:
        data_a = await fetch_data_from_api_a()
        data_b = await fetch_data_from_api_b(data_a)
        print(f"Data from API A: {data_a}")
        print(f"Data from API B: {data_b}")
    except Exception as e:
        print(f"An error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

1. fetch_data_from_api_a 函数：
   • 使用 aiohttp 库发起异步HTTP请求到API A。
   • async_timeout.timeout 用于设置请求超时时间为10秒。
   • 如果响应状态码为200，返回JSON格式的数据，否则抛出异常。
2. fetch_data_from_api_b 函数：
   • 根据数据源A返回的数据构建请求参数 params。
   • 同样使用 aiohttp 库发起异步HTTP请求到API B，设置超时时间为10秒。
   • 根据响应状态码处理数据或抛出异常。
3. main 函数：
   • 依次调用 fetch_data_from_api_a 和 fetch_data_from_api_b 来获取数据。
   • 使用 try - except 块捕获可能出现的异常并打印错误信息。
4. asyncio.run(main())：
   • 运行 main 函数，启动异步任务。

这样可以确保在处理多个异步数据源请求时，数据获取和处理的正确性，并优雅地处理超时和错误情况。如果数据源是数据库查询，可以使用支持异步操作的数据库驱动（如 asyncpg 对于PostgreSQL）来代替 aiohttp 的HTTP请求部分。