第八章 中间件、CORS、后台任务、测试用例

Author: liaogx

run.py

@@ -2,14 +2,18 @@
 # -*- coding:utf-8 -*-
 # __author__ = '__Jack__'
 
+import time
+
 import uvicorn
-from fastapi import FastAPI
-from fastapi.exceptions import RequestValidationError
-from fastapi.responses import PlainTextResponse
-from starlette.exceptions import HTTPException as StarletteHTTPException
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
 
 from coronavirus import application
-from tutorial import app03, app04, app05, app06
+from tutorial import app03, app04, app05, app06, app08
+
+# from fastapi.exceptions import RequestValidationError
+# from fastapi.responses import PlainTextResponse
+# from starlette.exceptions import HTTPException as StarletteHTTPException
 
 app = FastAPI(
     title='FastAPI Tutorial and Coronavirus Tracker API Docs',
@@ -40,11 +44,34 @@
 #     return PlainTextResponse(str(exc), status_code=400)
 
 
-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(app06, prefix="/chapter06", tags=['第六章 安全、认证和授权'])
-app.include_router(application, prefix="/coronavirus", tags=['新冠病毒疫情跟踪器API'])
+@app.middleware('http')
+async def add_process_time_header(request: Request, call_next):  # call_next将接收request请求最为参数
+    start_time = time.time()
+    response = await call_next(request)
+    process_time = time.time() - start_time
+    response.headers['X-Process-Time'] = str(process_time)  # 添加自定义的以“X-”开头的请求头
+    return response
+
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[
+        "http://127.0.0.1.tiangolo.com",
+        "https://127.0.0.1.tiangolo.com",
+        "http://127.0.0.1",
+        "http://127.0.0.1:8080",
+    ],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+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(app06, prefix='/chapter06', tags=['第六章 安全、认证和授权'])
+app.include_router(app08, prefix='/chapter08', tags=['第八章 中间件、CORS、后台任务、测试用例'])
+app.include_router(application, prefix='/coronavirus', tags=['新冠病毒疫情跟踪器API'])
 
 if __name__ == '__main__':
-    uvicorn.run("run:app", host="0.0.0.0", port=8000, reload=True, debug=True, workers=1)
+    uvicorn.run('run:app', host='0.0.0.0', port=8000, reload=True, debug=True, workers=1)

tutorial/__init__.py

@@ -6,3 +6,4 @@
 from .chapter04 import app04
 from .chapter05 import app05
 from .chapter06 import app06
+from .chapter08 import app08

tutorial/chapter08.py

@@ -0,0 +1,47 @@
+#!/usr/bin/python3
+# -*- coding:utf-8 -*-
+# __author__ = '__Jack__'
+
+from typing import Optional
+
+from fastapi import APIRouter, BackgroundTasks, Depends
+
+app08 = APIRouter()
+
+"""【见run.py】Middleware 中间件"""
+
+# 注：带yield的依赖的退出部分的代码 和 后台任务 会在中间件之后运行
+
+"""【见run.py】CORS (Cross-Origin Resource Sharing) 跨源资源共享"""
+
+# 域的概念：协议+域名+端口
+
+"""Background Tasks 后台任务"""
+
+
+def bg_task(framework: str):
+    with open("README.md", mode="a") as f:
+        f.write(f"## {framework} 框架精讲")
+
+
+@app08.post("/background_tasks")
+async def run_bg_task(framework: str, background_tasks: BackgroundTasks):
+    """
+    :param framework: 被调用的后台任务函数的参数
+    :param background_tasks: FastAPI.BackgroundTasks
+    :return:
+    """
+    background_tasks.add_task(bg_task, framework)
+    return {"message": "任务已在后台运行"}
+
+
+def continue_write_readme(background_tasks: BackgroundTasks, q: Optional[str] = None):
+    if q:
+        background_tasks.add_task(bg_task, "\n> 整体的介绍 FastAPI，快速上手开发，结合 API 交互文档逐个讲解核心模块的使用\n")
+    return q
+
+
+@app08.post("/dependency/background_tasks")
+async def dependency_run_bg_task(q: str = Depends(continue_write_readme)):
+    if q:
+        return {"message": "README.md更新成功"}

tutorial/test_chapter08.py

@@ -0,0 +1,29 @@
+#!/usr/bin/python3
+# -*- coding:utf-8 -*-
+# __author__ = '__Jack__'
+
+from fastapi.testclient import TestClient
+
+from run import app
+
+"""Testing 测试用例"""
+
+client = TestClient(app)  # 先pip install pytest
+
+
+def test_run_bg_task():  # 函数名用“test_”开头是 pytest 的规范。注意不是async def
+    response = client.post(url="/chapter08/background_tasks?framework=FastAPI")
+    assert response.status_code == 200
+    assert response.json() == {"message": "任务已在后台运行"}
+
+
+def test_dependency_run_bg_task():
+    response = client.post(url="/chapter08/dependency/background_tasks")
+    assert response.status_code == 200
+    assert response.json() is None
+
+
+def test_dependency_run_bg_task_q():
+    response = client.post(url="/chapter08/dependency/background_tasks?q=1")
+    assert response.status_code == 200
+    assert response.json() == {"message": "README.md更新成功"}