日韩性视频-久久久蜜桃-www中文字幕-在线中文字幕av-亚洲欧美一区二区三区四区-撸久久-香蕉视频一区-久久无码精品丰满人妻-国产高潮av-激情福利社-日韩av网址大全-国产精品久久999-日本五十路在线-性欧美在线-久久99精品波多结衣一区-男女午夜免费视频-黑人极品ⅴideos精品欧美棵-人人妻人人澡人人爽精品欧美一区-日韩一区在线看-欧美a级在线免费观看

歡迎訪問 生活随笔!

生活随笔

當前位置: 首頁 > 编程资源 > 编程问答 >内容正文

编程问答

fastapi 安全性 / APIRouter / BackgroundTasks / 元数据 / 测试调试

發布時間:2024/7/5 编程问答 33 豆豆
生活随笔 收集整理的這篇文章主要介紹了 fastapi 安全性 / APIRouter / BackgroundTasks / 元数据 / 测试调试 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

文章目錄

    • 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. 調試

learn from https://fastapi.tiangolo.com/zh/tutorial/security/first-steps/

1. 例子

# 安全性 main.py from fastapi import FastAPI, Depends from fastapi.security import OAuth2PasswordBearerapp = 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 BaseModelapp = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # oauth2_scheme(some, parameters) 是課調用的,可以被Depends使用class User(BaseModel):username: stremail: Optional[str] = Nonefull_name: Optional[str] = Nonedisabled: Optional[bool] = Nonedef 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 BaseModelfake_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" + passwordapp = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # oauth2_scheme(some, parameters) 是課調用的,可以被Depends使用class User(BaseModel):username: stremail: Optional[str] = Nonefull_name: Optional[str] = Nonedisabled: Optional[bool] = Noneclass UserInDB(User):hashed_password: strdef get_user(db, username: str):if username in db:user_dict = db[username]return UserInDB(**user_dict) # 過濾,獲取hasded_passworddef 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 userasync 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, HTTPExceptionasync 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 routerapp = 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.routertags=["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.routertags=["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 APIRouterrouter = 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, FastAPIapp = 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 Optionalapp = 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 Optionaldescription = """ goodApp API helps you do awesome stuff. 🚀🚀🚀## ItemsYou can **read items**.## UsersYou 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")


  • 如果想完全禁用 OpenAPI 模式,可以將其設置為 openapi_url=None,這樣也會禁用使用它的文檔用戶界面

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 TestClientapp = 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 定義 clientresponse = client.get("/")assert response.status_code == 200assert 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 itemmain13.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:10d:\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10: DeprecationWarning: "@coroutine" decorator is deprecated since Python 3.8, use "async def" insteaddef 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 FastAPIapp = FastAPI()@app.get("/") def root():a = "a"b = "b" + areturn {"hello world": b}if __name__ == "__main__":uvicorn.run(app, host="0.0.0.0", port=8000)

然后調用 python main14.py

總結

以上是生活随笔為你收集整理的fastapi 安全性 / APIRouter / BackgroundTasks / 元数据 / 测试调试的全部內容,希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯,歡迎將生活随笔推薦給好友。