本次的程式碼與目錄結構可以參考 FastAPI Tutorial : Day16 branch
在昨天(Day15)我們完成初步非同步存取 DB 的操作
也透過 Depends 將 AsyncSession 注入到 CRUD function 中
我們今天會注重於要如何再次優化我們 CRUD 的架構
並如何把 AsyncSession 的 Dependency Injection 做的更好
目前的寫法都是將 CRUD 分寫成 function 如： get_users 、 get_user_id_by_email 、 create_user 等
但是我們也可以將所有得 CRUD functions 都包裝在 class 中
crud/users.py
class UserCrud:
def __init__(self,db_session:AsyncSession):
self.db_session = db_session
async def get_users(self,db_session:AsyncSession, keyword:str=None,last:int=0,limit:int=50):
stmt = select(UserModel.name,UserModel.id,UserModel.email,UserModel.avatar)
if keyword:
stmt = stmt.where(UserModel.name.like(f"%{keyword}%"))
stmt = stmt.offset(last).limit(limit)
result = await db_session.execute(stmt)
users = result.all()
return users
# ... CRUD functions
async def get_user_crud():
async with get_db() as db_session:
yield UserCrud(db_session)這樣的好處是：我們不需要為每個 CRUD function 都注入 AsyncSession
可以直接透過 Depends(get_user_crud) 來取得 UserCrud class 來操作
api/users.py
from crud.users import UserCrud , get_user_crud_manager
# ...
db_depends = Depends(get_user_crud) # <--- 修改
@router.get("/users",
response_model=List[UserSchema.UserRead],
response_description="Get list of user",
)
async def get_users(page_parms:dict= Depends(pagination_parms), userCrud:UserCrud=db_depends):
users = await userCrud.get_users(**page_parms)
return users在使用 Depends 與 yield 時，常見的錯誤如下：
我們可能會想：「如果要在 api/users.py 中使用 Depends(get_user_crud) 來取得 UserCrud class 來操作」
為什麼不直接在 crud/users.py 中使用 Depends(get_db) 來取得 AsyncSession 來操作呢？
crud/users.py
class UserCrud:
def __init__(self):
self.db_session = Depends(get_db) # <--- 修改
async def get_users(self,db_session:AsyncSession, keyword:str=None,last:int=0,limit:int=50):
stmt = select(UserModel.name,UserModel.id,UserModel.email,UserModel.avatar)
if keyword:
stmt = stmt.where(UserModel.name.like(f"%{keyword}%"))
stmt = stmt.offset(last).limit(limit)
result = await db_session.execute(stmt)
users = result.all()
return users
# ... CRUD functions直接這樣寫的話，會報錯如下：
寫著 AttributeError: 'Depends' object has no attribute 'execute' ：Depends 沒有 execute 的屬性
但依照我們的邏輯，self.db_session 應該是 AsyncSession 才對啊（？
這是因為在 FastAPI 中，Depends 必須要寫在 API endpoint 的 handle funtion 中 !
也就是說，我們只能在 api/users.py 中使用 Depends(get_user_crud) 來取得 UserCrud class 來操作
或是透過 Depends(get_db) 來注入 AsyncSession 到 CRUD function 中
那可能又會想說：「為什麼一定要透過 Depends 來取得 AsyncSession 呢？
應該也可以直接在 crud/users.py 中使用 async with get_db() as db_session 來取得 AsyncSession 才對啊（？
crud/users.py
# ...
async def get_users(self,keyword:str=None,last:int=0,limit:int=50):
async with get_db() as db_session: # <--- 新增
stmt = select(UserModel.name,UserModel.id,UserModel.email,UserModel.avatar)
if keyword:
stmt = stmt.where(UserModel.name.like(f"%{keyword}%"))
stmt = stmt.offset(last).limit(limit)
# result = await self.db_session.execute(stmt)
result = await db_session.execute(stmt)
users = result.all()
return users這樣寫的話，會報錯如下：
寫著 TypeError: 'async_generator' object does not support the asynchronous context manager protocol
查看 FastAPI Doc 中，與 Depends 和 yield 相關的部分
上面提到 @contextlib.asynccontextmanager
稍微了解後，我們可以知道 async with 是 async 的 context manager
async_generator 不支援 async 的 context manager
上面提到的問題
async with是async的 context manager
async_generator不支援async的 context manager
我們可以透過 @contextlib.asynccontextmanager 來將 async_generator 轉換成 async 的 context manager
所以應該要為 get_db 加上 @contextlib.asynccontextmanager
database/generic.py
from contextlib import asynccontextmanager
@asynccontextmanager # <--- 新增
async def get_db():
async with SessionLocal() as db:
async with db.begin():
yield db
# ...那上面直接使用 async with get_db() as db_session 來取得 AsyncSession 操作 DB 的方式就可以了 !
crud/users.py
# ...
async def get_users(self,keyword:str=None,last:int=0,limit:int=50):
async with get_db() as db_session:
stmt = select(UserModel.name,UserModel.id,UserModel.email,UserModel.avatar)
if keyword:
stmt = stmt.where(UserModel.name.like(f"%{keyword}%"))
stmt = stmt.offset(last).limit(limit)
# result = await self.db_session.execute(stmt)
result = await db_session.execute(stmt)
users = result.all()
return users
# ...現在這樣寫就可以正常執行了！
雖然我們現在可以透過 async with get_db() as db_session 直接來取得 AsyncSession 操作 DB
可以不用透過 FastAPI 中的 Depends 來達成
但是我們還是需要在每個 CRUD function 中都寫上
async with get_db() as db_session
並多一個縮排
decoratorfor function 所以我們可以透過decorator來將async with get_db() as db_session注入到 CRUD function 中
database/generic.py
# ...
# decorator dependency for getting db session
def db_session_decorator(func):
# print("in db_context_decorator")
async def wrapper(*args, **kwargs):
async with get_db() as db_session:
kwargs["db_session"] = db_session
result = await func(*args, **kwargs)
return result
# print("out db_context_decorator")
return wrapper在 wrapper function 中，我們為接下來要掛上 @db_session_decorator decorator 的 func 中
db_session 參數設為 AsyncSession
所以在原本的 CRUD function 中，我們只需要要加上 db_session 參數和 @db_session_decorator
就可以注入 AsyncSession 使用了
crud/users.py
# ...
from database.generic import db_session_decorator # <--- 新增
# ...
class UserCrudManager: # <--- 修改
@db_session_decorator # <--- 新增
async def get_users(self,db_session:AsyncSession, keyword:str=None,last:int=0,limit:int=50):
stmt = select(UserModel.name,UserModel.id,UserModel.email,UserModel.avatar)
# ...
return users
@db_session_decorator # <--- 新增
async def get_user_by_id(self,user_id: int):
# ...
# ... CRUD functionsdecoratorfor class
但是這樣的寫法，我們還是需要在每個 CRUD function 中都寫上 db_session 參數和 @db_session_decorator
所以我們可以透過為 UserCrudManager 加上 decorator
來為所有 methods 都加上 @db_session_decorator
這邊我們透過 setattr 來為 cls 中的每個 methods 都加上 db_session_decorator
database/generic.py
# ...
def crud_class_decorator(cls):
# print("in db_class_decorator")
for name, method in cls.__dict__.items():
if callable(method):
setattr(cls, name, db_session_decorator(method))
# print("out db_class_decorator")
return cls我們只需要為 UserCrudManager 加上 @crud_class_decorator
並為每個 CRUD methods 加上 db_session 參數
crud/users.py
# ...
@crud_class_decorator # <--- 新增
class UserCrudManager:
# 為每個 CRUD methods 加上 db_session 參數
async def get_users(self,db_session:AsyncSession, keyword:str=None,last:int=0,limit:int=50): # <--- 修改
stmt = select(UserModel.name,UserModel.id,UserModel.email,UserModel.avatar)
# ...
return users
async def get_user_by_id(self,db_session:AsyncSession,user_id: int): # <--- 修改
# ...
# ... CRUD functions修改完後，只需要在 api/users.py 建立一個 UserCrudManager 的 instance
就可以透過 instanceName.action 來操作 CRUD 了
api/users.py
# ...
# from crud import users as UserCrud # <--- 刪除
from crud.users import UserCrudManager
UserCrud = UserCrudManager()
# 這邊我們同樣民名為 UserCrud ，但是實際上是 UserCrudManager 的 instance
# ...
@router.get("/users",
response_model=List[UserSchema.UserRead],
response_description="Get list of user",
)
async def get_users(page_parms:dict= Depends(pagination_parms), userCrud:UserCrud=db_depends):
users = await UserCrud.get_users(**page_parms)
return users
# ... API endpoints再跑一次昨天的 benchmark 來比較一下
ab -n 50000 -c 32 http://127.0.0.1:8001/sync/api/users
- 昨天的
- 今天的
昨天以 Depends 注入 AsyncSession 總時間約 105 秒
今天透過 asynccontextmanager 與 decorator 注入 AsyncSession 總時間約 98 秒
可以看到效能上有稍微提升
從 sync CRUD 修改成 async CRUD 要更新的 code 也比較少
- 常見
Depends與yield的錯誤 - 透過
@contextlib.asynccontextmanager來將async_generator轉換成async的 context manager - 透過
decorator來注入AsyncSession
decoratorfor functiondecoratorfor class
- 使用
UserCrudManagerinstance 來操作 CRUD