Python 非同步程式設計實戰

Python 的 asyncio 模組讓我們能夠撰寫高效的非同步程式碼，特別適合 I/O 密集型任務。

基礎語法

import asyncio

async def fetch_data(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

async def main():
    data = await fetch_data('https://api.example.com/data')
    print(data)

asyncio.run(main())

並行執行

async def main():
    results = await asyncio.gather(
        fetch_data('https://api.example.com/users'),
        fetch_data('https://api.example.com/posts'),
        fetch_data('https://api.example.com/comments'),
    )
    return results

Task 與 Future

async def main():
    task = asyncio.create_task(fetch_data('https://api.example.com/data'))
    result = await task

FastAPI 整合

from fastapi import FastAPI
app = FastAPI()

@app.get('/users/{user_id}')
async def get_user(user_id: int):
    user = await fetch_user(user_id)
    return user