# Python 異步與同步操作整合教學
本文件介紹如何在 Python 異步程式設計中處理同步與非阻塞操作，並在 FastAPI 中實作範例。重點包含：
* 同步函式與異步函式的區別
* 何謂非阻塞操作
* 如何判斷函數是否可等待
* 如何將同步函式轉換為非阻塞操作（改寫或包裝）
* FastAPI 中的應用範例

## 同步 vs. 異步函式
- 同步函式
直接執行會阻塞後續程式碼，例如：

import time

def blocking_task():
    print("開始阻塞等待")
    time.sleep(3)  # 阻塞操作
    print("等待結束")

- 異步函式
使用 async def 定義，並用 await 進行非阻塞等待：

import asyncio

async def async_task():
    print("開始等待")
    await asyncio.sleep(3)  # 非阻塞等待
    print("等待結束")

## 何謂非阻塞操作
非阻塞操作：在等待 I/O 或耗時操作時，事件迴圈仍能處理其他任務，避免整個程式「卡住」。
例子：
* 使用 asyncio.sleep()：在等待時不阻塞其他任務。
* 使用 aiohttp：進行網路請求時不阻塞主事件迴圈。

## 如何判斷一個函數是否可等待
利用 inspect 模組檢查函式或物件：
- 判斷是否為協程函式

import inspect

async def async_func():
    return 123

def sync_func():
    return 456

print(inspect.iscoroutinefunction(async_func))  # True
print(inspect.iscoroutinefunction(sync_func))   # False

- 判斷返回物件是否可等待

import inspect
coro = async_func()
print(inspect.isawaitable(coro))  # True

- 自定義可等待物件：實現 __await__() 方法的物件也是可等待的。

## 將同步函式改成異步
有兩種常見做法
- 改寫為異步函式
- 使用包裝（asyncio.to_thread）

### 改寫為異步函式
如果函式內部有非阻塞版本操作，可以直接改寫：

import asyncio

async def async_task():
    print("開始等待")
    await asyncio.sleep(3)
    print("等待結束")

### 使用包裝（asyncio.to_thread）

import asyncio
import time

def blocking_task():
    print("開始阻塞等待")
    time.sleep(3)
    print("等待結束")

async def async_wrapper():
    await asyncio.to_thread(blocking_task)

# 執行異步包裝的函式
asyncio.run(async_wrapper())

## FastAPI 中的應用範例
- 使用 asyncio.to_thread 處理同步操作
在 FastAPI 中，由於底層操作（例如與 MinIO 的連線）為同步，需使用 asyncio.to_thread 來避免阻塞事件迴圈。

假設你有一個同步的 MinioClient 類別，其部分方法如 insert、read 等都是同步實作，範例如下：

from fastapi import FastAPI, Depends
import asyncio
import time

app = FastAPI()

# 假設這是你的同步 MinioClient 類別（部分示例）
class MinioClient:
    def insert(self, data: dict) -> dict:
        # 模擬同步上傳操作
        time.sleep(3)
        return {"status": "uploaded", "data": data}
    
    def read(self, identifier: str) -> str:
        time.sleep(1)
        return f"data for {identifier}"
    
    def close(self) -> None:
        print("MinioClient connection pool closed")

# 全局單例：在應用啟動時初始化，應用關閉時釋放資源
minio_client = None

@app.on_event("startup")
async def startup_event():
    global minio_client
    minio_client = MinioClient()

@app.on_event("shutdown")
async def shutdown_event():
    if minio_client:
        minio_client.close()

def get_minio_client():
    return minio_client

# 非阻塞上傳 API，使用 asyncio.to_thread 包裝同步方法
@app.post("/upload")
async def upload_object(data: dict, client: MinioClient = Depends(get_minio_client)):
    result = await asyncio.to_thread(client.insert, data)
    return result

# 非阻塞讀取 API
@app.get("/read/{identifier}")
async def read_object(identifier: str, client: MinioClient = Depends(get_minio_client)):
    result = await asyncio.to_thread(client.read, identifier)
    return {"result": result}

## 連線池的生命週期管理
不希望每次 API 請求後就關閉連線池，可以利用 FastAPI 的啟動與關閉事件來控制連線池的建立與釋放，確保連線池只在應用啟動時初始化，在應用終止時才關閉。
較舊版本

@app.on_event("startup")
async def startup_event():
    global minio_client
    minio_client = MinioClient()  # 單例，整個應用生命週期共用

@app.on_event("shutdown")
async def shutdown_event():
    if minio_client:
        minio_client.close()  # 只有在應用關閉時釋放連線池

新版本

async def lifespan(app: FastAPI):
    # 启动前的操作
    global minio_client
    minio_client = MinioClient()
    await startup_instance(app)
    yield
    # 关闭前的操作
    minio_client.close()



app = FastAPI(lifespan=lifespan)

# 結論

- 非阻塞操作 能讓你的應用在等待 I/O 時依然處理其他任務，提高效能。
- 若底層函式是同步的，可以使用 asyncio.to_thread 在 FastAPI 中包裝，使其非阻塞。
- 透過 FastAPI 的啟動與關閉事件，可以管理全局單例資源（如連線池）的生命週期，避免每次請求都關閉連線。

這些技巧可以讓你在設計高效的異步 API 時，更靈活地整合同步和異步操作。