文章目录
1. 例子
2. 获取当前用户
3. 使用密码和 Bearer 的简单 OAuth2
4. 使用（哈希）密码和 JWT Bearer 令牌的 OAuth2
5. 多个应用文件
5.1 APIRouter
6. BackgroundTasks
7. 元数据
7.1 标题、描述和版本
7.2 openapi_tags 标签元数据
7.3 OpenAPI URL
7.4 文档 URLs
8. 测试
9. 调试
1. 例子
# 安全性 main.py from fastapi import FastAPI, Depends from fastapi.security import OAuth2PasswordBearer app = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # oauth2_scheme(some, parameters) 是课调用的，可以被Depends使用 @app.get("/items/") async def read_items(token: str = Depends(oauth2_scheme)): return {"token": token}
运行 uvicorn main:app --reload
打开 http://127.0.0.1:8000/docs#/default/read_items_items__get
2. 获取当前用户
# 安全性 from fastapi import FastAPI, Depends from typing import Optional from fastapi.security import OAuth2PasswordBearer from pydantic import BaseModel app = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # oauth2_scheme(some, parameters) 是课调用的，可以被Depends使用 class User(BaseModel): username: str email: Optional[str] = None full_name: Optional[str] = None disabled: Optional[bool] = None def fake_decode_token(token): return User(username=token+"fakedecoded", email="abc.mail", full_name="michael") async def get_current_user(token: str = Depends(oauth2_scheme)): return fake_decode_token(token) @app.get("/users/me/") async def read_users_me(current_user: User = Depends(get_current_user)): return current_user
3. 使用密码和 Bearer 的简单 OAuth2
- OAuth2 规定在使用「password 流程」时，客户端/用户必须将
username和password字段作为表单数据发送 OAuth2PasswordRequestForm
# 安全性 from fastapi import FastAPI, Depends, HTTPException, status from typing import Optional from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from pydantic import BaseModel fake_users_db = { "johndoe": { "username": "johndoe", "full_name": "John Doe", "email": "johndoe@example.com", "hashed_password": "fakehashedsecret", "disabled": False, }, "alice": { "username": "alice", "full_name": "Alice Wonderson", "email": "alice@example.com", "hashed_password": "fakehashedsecret2", "disabled": True, }, } def fake_hash_password(password: str): return "fakehashed" + password app = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # oauth2_scheme(some, parameters) 是课调用的，可以被Depends使用 class User(BaseModel): username: str email: Optional[str] = None full_name: Optional[str] = None disabled: Optional[bool] = None class UserInDB(User): hashed_password: str def get_user(db, username: str): if username in db: user_dict = db[username] return UserInDB(**user_dict) # 过滤，获取hasded_password def fake_decode_token(token): return get_user(fake_users_db, token) async def get_current_user(token: str = Depends(oauth2_scheme)): user = fake_decode_token(token) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials", headers={"WWW-Authenticate": "Bearer"}, ) return user async def get_current_active_user(current_user: User = Depends(get_current_user)): if current_user.disabled: raise HTTPException(status_code=400, detail="Inactive user") return current_user @app.post("/token") async def login(form_data: OAuth2PasswordRequestForm = Depends()): user_dict = fake_users_db.get(form_data.username) if not user_dict: raise HTTPException(status_code=400, detail="Incorrect username or password") user = UserInDB(**user_dict) hashed_password = fake_hash_password(form_data.password) if not hashed_password == user.hashed_password: raise HTTPException(status_code=400, detail="Incorrect username or password") return {"access_token": user.username, "token_type": "bearer"} @app.get("/users/me") async def read_users_me(current_user: User = Depends(get_current_active_user)): return current_user
4. 使用（哈希）密码和 JWT Bearer 令牌的 OAuth2
JWT 表示 「JSON Web Tokens」。它是一个将 JSON 对象编码为密集且没有空格的长字符串的标准
安装 python-jose 以在 Python 中生成和校验 JWT 令牌
pip install python-jose[cryptography]
PassLib 是一个用于处理哈希密码的很棒的 Python 包, 推荐的算法是 「Bcrypt」
pip install passlib[bcrypt]
参考：https://fastapi.tiangolo.com/zh/tutorial/security/oauth2-jwt/
5. 多个应用文件
__init__.py 可以使得目录下的包可以被其他目录导入，该文件可以为空
5.1 APIRouter
# dependencies.py # 我们了解到我们将需要一些在应用程序的好几个地方所使用的依赖项。 # 因此，我们将它们放在它们自己的 dependencies 模块 from fastapi import Header, HTTPException async def get_token_header(x_token: str = Header(...)): if x_token != "fake-super-secret-token": raise HTTPException(status_code=400, detail="X-Token header invalid") async def get_query_token(token: str): if token != "jessica": raise HTTPException(status_code=400, detail="No Jessica token provided")
# main.py 你的应用程序中将所有内容联结在一起的主文件 # 你的大部分逻辑现在都存在于其自己的特定模块中 # 因此主文件的内容将非常简单 from fastapi import Depends, FastAPI from dependencies import get_query_token, get_token_header from internal import admin from routers import items, users # from .routers.items import router # 以下两行名称重叠，注意避免 # from .routers.users import router app = FastAPI(dependencies=[Depends(get_query_token)]) app.include_router(users.router) app.include_router(items.router) # users.router 包含了 app/routers/users.py 文件中的 APIRouter # items.router 包含了 app/routers/items.py 文件中的 APIRouter app.include_router( admin.router, prefix="/admin", # 添加路径前缀，而不必修改admin.router tags=["admin"], dependencies=[Depends(get_token_header)], responses={418: {"description": "I'm a teapot"}}, # 但这只会影响我们应用中的 APIRouter， # 而不会影响使用admin.router的任何其他代码 ) # app.include_router()，可以将每个 APIRouter 添加到主 FastAPI 应用程序中 # 多次使用不同的 prefix 包含同一个路由器 app.include_router( admin.router, prefix="/admin_test", # 添加路径前缀，而不必修改admin.router tags=["admin"], dependencies=[Depends(get_token_header)], responses={418: {"description": "I'm a teapot, diff "}}, # 但这只会影响我们应用中的 APIRouter， # 而不会影响使用admin.router的任何其他代码 ) # 也可以在另一个 APIRouter 中包含一个 APIRouter # router.include_router(other_router) @app.get("/") async def root(): return {"message": "Hello Bigger Applications!"}
# internal/admin.py from fastapi import APIRouter router = APIRouter() @router.post("/") async def update_admin(): return {"message": "Admin getting schwifty"}
# routers/items.py # 此模块中的所有路径操作都有相同的： # 路径 prefix：/items # tags：（仅有一个 items 标签） # 额外的 responses # dependencies：它们都需要我们创建的 X-Token 依赖项 from fastapi import APIRouter, Depends, HTTPException from dependencies import get_token_header router = APIRouter( prefix="/items", # 前缀不能以 / 作为结尾 tags=["items"], dependencies=[Depends(get_token_header)], responses={404: {"description": "Not found"}}, # 这些参数将应用于此路由器中包含的所有路径操作 ) fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}} @router.get("/") async def read_items(): return fake_items_db @router.get("/{item_id}") async def read_item(item_id: str): if item_id not in fake_items_db: raise HTTPException(status_code=404, detail="Item not found") return {"name": fake_items_db[item_id]["name"], "item_id": item_id} @router.put( "/{item_id}", tags=["custom"], responses={403: {"description": "Operation forbidden"}}, ) # 这个路径操作将包含标签的组合：["items"，"custom"] # 也会有两个响应，一个用于 404，一个用于 403 async def update_item(item_id: str): if item_id != "plumbus": raise HTTPException( status_code=403, detail="You can only update the item: plumbus" ) return {"item_id": item_id, "name": "The great Plumbus"}
# routers/users.py # 将与用户相关的路径操作与其他代码分开，以使其井井有条 from fastapi import APIRouter router = APIRouter() # 你可以将 APIRouter 视为一个「迷你 FastAPI」类 # 在此示例中，该变量被命名为 router，但你可以根据你的想法自由命名 @router.get("/users/", tags=["users"]) async def read_users(): return [{"username": "Rick"}, {"username": "Morty"}] @router.get("/users/me", tags=["users"]) async def read_user_me(): return {"username": "fakecurrentuser"} @router.get("/users/{username}", tags=["users"]) async def read_user(username: str): return {"username": username}
6. BackgroundTasks
background_tasks.add_task(func, param, keywordparam=value)- add_task 参数： 函数名， 顺序参数， 关键字参数
from fastapi import BackgroundTasks, FastAPI app = FastAPI() def write_notification(email: str, message=""): with open("log.txt", mode="w") as email_file: content = f"notification for {email}: {message}" email_file.write(content) # 任务函数，可以是普通函数 或 async def函数 @app.post("/send-notification/{email}") async def send_notification(email: str, background_tasks: BackgroundTasks): background_tasks.add_task(write_notification, email, message="hello, sending email to me") # add_task 参数： 函数名， 顺序参数， 关键字参数 return {"message": "Notification sent in the background"}
- 依赖注入
from fastapi import BackgroundTasks, FastAPI, Depends from typing import Optional app = FastAPI() def write_log(message: str): with open("log.txt", mode="a") as log: log.write(message) def get_query(background_tasks: BackgroundTasks, q: Optional[str] = None): if q: message = f"found query: {q}\n" background_tasks.add_task(write_log, message) return q @app.post("/send-notification/{email}") async def send_notification(email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)): message = f"message to {email}\n" background_tasks.add_task(write_log, message) return {"message": "Message sent"}
7. 元数据
7.1 标题、描述和版本
from fastapi import BackgroundTasks, FastAPI, Depends from typing import Optional description = """ goodApp API helps you do awesome stuff. 🚀🚀🚀 ## Items You can **read items**. ## Users You will be able to: * **Create users** (_not implemented_). * **Read users** (_not implemented_). """ app = FastAPI( title="good_App", description=description, version="0.0.100", terms_of_service="http://example.com/terms/", contact={ "name": "Deadpoolio the Amazing", "url": "http://x-force.example.com/contact/", "email": "dp@x-force.example.com", }, license_info={ "name": "Apache 2.0", "url": "https://www.apache.org/licenses/LICENSE-2.0.html", }, )
7.2 openapi_tags 标签元数据
from fastapi import FastAPI tags_metadata = [ { "name": "users", # 必须的，跟相应的 tags 参数有相同的名 "description": "Operations with users. The **login** logic is also here.", }, { "name": "items", "description": "Manage items. So _fancy_ they have their own docs.", "externalDocs": { "description": "Items external docs", "url": "https://fastapi.tiangolo.com/", # url 必须的 }, }, ] app = FastAPI(openapi_tags=tags_metadata) @app.get("/users/", tags=["users"]) async def get_users(): return [{"name": "Harry"}, {"name": "Ron"}] @app.get("/items/", tags=["items"]) async def get_items(): return [{"name": "wand"}, {"name": "flying broom"}]
7.3 OpenAPI URL
- 添加
openapi_url参数
app = FastAPI(openapi_tags=tags_metadata, openapi_url="/api/v100/michael.json")
7.4 文档 URLs
你可以配置两个文档用户界面，包括：
Swagger UI：服务于 /docs
可以使用参数 docs_url 设置它的 URL。
可以通过设置 docs_url=None 禁用它。
ReDoc：服务于 /redoc
可以使用参数 redoc_url 设置它的 URL。
可以通过设置 redoc_url=None 禁用它。
如 app = FastAPI(docs_url="/mydocs", redoc_url=None)
8. 测试
from fastapi import FastAPI from fastapi.testclient import TestClient app = FastAPI() @app.get("/") async def read_main(): return {"msg": "Hello World"} client = TestClient(app) def test_read_main(): # 测试函数是普通 def， 这样你可以用 pytest 来执行 # 否则的话，需要使用 @pytest.mark.anyio装饰函数 # 且 使用 from httpx import AsyncClient 定义 client response = client.get("/") assert response.status_code == 200 assert response.json() == {"msg": "Hello World"}
如何测试
pip install pytest
(pt19) D:\gitcode\Python_learning\fastapi\9> pytest main13.py =================================================================== test session starts ==================================================================== platform win32 -- Python 3.8.10, pytest-6.2.5, py-1.10.0, pluggy-1.0.0 rootdir: D:\gitcode\Python_learning\fastapi\9 collected 1 item main13.py . [100%] ===================================================================== warnings summary ===================================================================== ..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10 ..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10 ..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10 ..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10 ..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10 d:\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10: DeprecationWarning: "@coroutine" decorator is deprecated since Python 3.8, use "async def" instead def run(*args, loop=None, executor=None, **kwargs): -- Docs: https://docs.pytest.org/en/stable/warnings.html ============================================================== 1 passed, 5 warnings in 0.45s ===============================================================
9. 调试
# main14.py import uvicorn from fastapi import FastAPI app = FastAPI() @app.get("/") def root(): a = "a" b = "b" + a return {"hello world": b} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)
然后调用 python main14.py
