本次的程式碼與目錄結構可以參考 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
asynccontextmanager
dependency上面提到的問題
async with
是async
的 context managerasync_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
# ...
現在這樣寫就可以正常執行了!
decorator
來注入 AsyncSession
雖然我們現在可以透過 async with get_db() as db_session
直接來取得 AsyncSession
操作 DB
可以不用透過 FastAPI 中的 Depends
來達成
但是我們還是需要在每個 CRUD function 中都寫上
async with get_db() as db_session
並多一個縮排
decorator
for functiondecorator
來將 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 functions
decorator
for 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
UserCrudManager
instance 來操作 CRUD修改完後,只需要在 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
decorator
for functiondecorator
for classUserCrudManager
來 CRUD