第五章 FastAPI的依赖注入系统

Author: liaogx

run.py

@@ -9,7 +9,7 @@
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from coronavirus import application
-from tutorial import app03, app04
+from tutorial import app03, app04, app05
 
 app = FastAPI(
     title='FastAPI Tutorial and Coronavirus Tracker API Docs',
@@ -42,6 +42,7 @@
 
 app.include_router(app03, prefix="/chapter03", tags=['第三章 请求参数和验证'])
 app.include_router(app04, prefix="/chapter04", tags=['第四章 响应处理和FastAPI配置'])
+app.include_router(app05, prefix="/chapter05", tags=['第五章 FastAPI的依赖注入系统'])
 app.include_router(application, prefix="/coronavirus", tags=['新冠病毒疫情跟踪器API'])
 
 if __name__ == '__main__':

tutorial/__init__.py

@@ -3,3 +3,5 @@
 # __author__ = '__Jack__'
 
 from .chapter03 import app03
+from .chapter04 import app04
+from .chapter05 import app05

tutorial/chapter05.py

@@ -2,73 +2,132 @@
 # -*- coding:utf-8 -*-
 # __author__ = '__Jack__'
 
-# from typing import List
-#
-# from fastapi import Form, File, UploadFile
-# from starlette.requests import Request
-#
-# from fastapi import APIRouter
-#
-#
-# app02 = APIRouter()
-#
-#
-# @app02.post("/user/")
-# async def login_post(request: Request, username: str = Form(...), password: str = Form(...)):
-#     """
-#     Form表单操作
-#     :param request:
-#     :param username:
-#     :param password:
-#     :return:
-#     """
-#
-#     print(f"username: {username}, password: {password}")
-#
-#     return templates.TemplateResponse("post.html", {"request": request, "username": username, "password": password})
-#
-#
-# @app02.get("/user/")
-# async def login_get(request: Request):
-#     return templates.TemplateResponse("post.html", {"request": request})
-#
-#
-# @app02.get("/file/")
-# async def login_get(request: Request):
-#     return templates.TemplateResponse("file.html", {"request": request})
-#
-#
-# @app02.post("/file/")
-# async def upload_file(request: Request, files_list: List[bytes] = File(...), files_name: List[UploadFile] = File(...)):
-#     """
-#     上传多个文件
-#     :param request:
-#     :param files_list: List[bytes]表示可上传多个文件，bytes表示上传一个文件
-#     :param files_name:
-#     :return:
-#     """
-#
-#     return templates.TemplateResponse("file.html", {
-#         "request": Request,
-#         "file_sizes": [len(file) for file in files_list],  # 用字节长度表示文件大小
-#         "filenames": [file.filename for file in files_name]
-#     })
-#
-#
-# @app02.post("/create_file/")
-# async def create_file(request: Request, file1: bytes = File(...), file2: UploadFile = File(...), notes: str = Form(...)):
-#     """
-#     上传单个文件
-#     :param request:
-#     :param file1: bytes类型可以统计长度，但是没有content_type属性
-#     :param file2: UploadFile类有content_type属性，但是不能len()统计文件长度
-#     :param notes:
-#     :return:
-#     """
-#
-#     return templates.TemplateResponse("file.html", {
-#         "request": Request,
-#         "file_size": len(file1),
-#         "notes": notes,
-#         "file2_content_type": file2.content_type
-#     })
+from typing import Optional
+
+from fastapi import APIRouter
+from fastapi import Depends, HTTPException, Header
+
+app05 = APIRouter()
+
+"""Dependencies 导入、创建和声明依赖"""
+
+
+async def common_parameters(q: Optional[str] = None, page: int = 1, limit: int = 100):
+    return {"q": q, "page": page, "limit": limit}
+
+
+@app05.get("/dependency01")
+async def dependency01(commons: dict = Depends(common_parameters)):
+    return commons
+
+
+@app05.get("/dependency02")
+def dependency02(commons: dict = Depends(common_parameters)):  # 可以在async def中调用def依赖，也可以在def中导入async def依赖
+    return commons
+
+
+"""Classes as Dependencies 类作为依赖项"""
+
+fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
+
+
+class CommonQueryParams:
+    def __init__(self, q: Optional[str] = None, page: int = 1, limit: int = 100):
+        self.q = q
+        self.page = page
+        self.limit = limit
+
+
+@app05.get("/classes_as_dependencies")
+# async def classes_as_dependencies(commons: CommonQueryParams = Depends(CommonQueryParams)):
+# async def classes_as_dependencies(commons: CommonQueryParams = Depends()):
+async def classes_as_dependencies(commons=Depends(CommonQueryParams)):
+    response = {}
+    if commons.q:
+        response.update({"q": commons.q})
+    items = fake_items_db[commons.page: commons.page + commons.limit]
+    response.update({"items": items})
+    return response
+
+
+"""Sub-dependencies 子依赖"""
+
+
+def query(q: Optional[str] = None):
+    return q
+
+
+def sub_query(q: str = Depends(query), last_query: Optional[str] = None):
+    if not q:
+        return last_query
+    return q
+
+
+@app05.get("/sub_dependency")
+async def sub_dependency(final_query: str = Depends(sub_query, use_cache=True)):
+    """use_cache默认是True, 表示当多个依赖有一个共同的子依赖时，每次request请求只会调用子依赖一次，多次调用将从缓存中获取"""
+    return {"sub_dependency": final_query}
+
+
+"""Dependencies in path operation decorators 路径操作装饰器中的多依赖"""
+
+
+async def verify_token(x_token: str = Header(...)):
+    """没有返回值的子依赖"""
+    if x_token != "fake-super-secret-token":
+        raise HTTPException(status_code=400, detail="X-Token header invalid")
+
+
+async def verify_key(x_key: str = Header(...)):
+    """有返回值的子依赖，但是返回值不会被调用"""
+    if x_key != "fake-super-secret-key":
+        raise HTTPException(status_code=400, detail="X-Key header invalid")
+    return x_key
+
+
+@app05.get("/dependency_in_path_operation", dependencies=[Depends(verify_token), Depends(verify_key)])  # 这时候不是在函数参数中调用依赖，而是在路径操作中
+async def dependency_in_path_operation():
+    return [{"user": "user01"}, {"user": "user02"}]
+
+
+"""Global Dependencies 全局依赖"""
+
+# app05 = APIRouter(dependencies=[Depends(verify_token), Depends(verify_key)])
+
+
+"""Dependencies with yield 带yield的依赖"""
+
+
+# 这个需要Python3.7才支持，Python3.6需要pip install async-exit-stack async-generator
+# 以下都是伪代码
+
+async def get_db():
+    db = "db_connection"
+    try:
+        yield db
+    finally:
+        db.endswith("db_close")
+
+
+async def dependency_a():
+    dep_a = "generate_dep_a()"
+    try:
+        yield dep_a
+    finally:
+        dep_a.endswith("db_close")
+
+
+async def dependency_b(dep_a=Depends(dependency_a)):
+    dep_b = "generate_dep_b()"
+    try:
+        yield dep_b
+    finally:
+        dep_b.endswith(dep_a)
+
+
+async def dependency_c(dep_b=Depends(dependency_b)):
+    dep_c = "generate_dep_c()"
+    try:
+        yield dep_c
+    finally:
+        dep_c.endswith(dep_b)