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 流程」时,客户端/用户必须usernamepassword 字段作为表单数据发送
  • 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

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/471555.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

java json path_Java使用JSONPath解析JSON完整内容详解

JsonPath是一种简单的方法来提取给定JSON文档的部分内容。 JsonPath有许多编程语言,如Javascript,Python和PHP,Java。JsonPath提供的json解析非常强大,它提供了类似正则表达式的语法,基本上可以满足所有你想要获得的js…

wndows程序设计之书籍知识与代码摘录-封装一个类似printf的messagebox

//----------------------------------------- //本程序展示了如何实现MessageBoxPrintf函数 //本函数能像printf那样格式化输出 //摘录自&#xff1a;《windows程序设计第五版》 //----------------------------------------- #include<windows.h> #include<tchar.h&…

LeetCode 2050. 并行课程 III(拓扑排序)

文章目录1. 题目2. 解题1. 题目 给你一个整数 n &#xff0c;表示有 n 节课&#xff0c;课程编号从 1 到 n 。 同时给你一个二维整数数组 relations &#xff0c;其中 relations[j] [prevCoursej, nextCoursej] &#xff0c;表示课程 prevCoursej 必须在课程 nextCoursej 之前…

改不改,这是一个问题

概率称号&#xff1a; 给你三盒。里面放的东西。您指定&#xff0c;然后打开了另外两个。发现是空的&#xff0c;要求&#xff1a;你想改变你的选择。 答案是&#xff0c;它应该改变网上评选&#xff0c;这样的概率较大。因此&#xff0c;我要求&#xff0c;空的情况下。选择的…

LeetCode 1521. 找到最接近目标值的函数值(位运算)

文章目录1. 题目2. 解题1. 题目 Winston 构造了一个如上所示的函数 func 。他有一个整数数组 arr 和一个整数 target &#xff0c;他想找到让 |func(arr, l, r) - target| 最小的 l 和 r 。 请你返回 |func(arr, l, r) - target| 的最小值。 请注意&#xff0c; func 的输入参…

LeetCode 2053. 数组中第 K 个独一无二的字符串(哈希)

文章目录1. 题目2. 解题1. 题目 独一无二的字符串 指的是在一个数组中只出现过 一次 的字符串。 给你一个字符串数组 arr 和一个整数 k &#xff0c;请你返回 arr 中第 k 个 独一无二的字符串 。 如果 少于 k 个独一无二的字符串&#xff0c;那么返回 空字符串 “” 。 注意…

java 该改变request url_如何在Java中使用servlet过滤器来更改传入的servlet请求URL?...

如何使用servlet过滤器来更改传入的servlet请求URLweb.xml至web.xml&#xff1f;更新&#xff1a;根据BalusC的以下步骤&#xff0c;我想出了以下代码&#xff1a;public class UrlRewriteFilter implements Filter {Overridepublic void init(FilterConfig config) throws Ser…

LeetCode 2055. 蜡烛之间的盘子(前缀和)

文章目录1. 题目2. 解题1. 题目 给你一个长桌子&#xff0c;桌子上盘子和蜡烛排成一列。 给你一个下标从 0 开始的字符串 s &#xff0c;它只包含字符 * 和 | &#xff0c;其中 * 表示一个 盘子 &#xff0c;| 表示一支 蜡烛 。 同时给你一个下标从 0 开始的二维整数数组 que…

Java可靠性测试fit_Java ParagraphVectors.fit方法代码示例

import org.deeplearning4j.models.paragraphvectors.ParagraphVectors; //导入方法依赖的package包/类public static void main(String[] args) throws Exception {ClassPathResource srcFile new ClassPathResource("/raw_sentences.txt");File file srcFile.get…

flashbackup闪回数据库

某时间删除了一个表 SQL> delete test1;50000 rows deleted.1、关闭数据库&#xff0c;启动到mount状态&#xff08;abort数据库能快速关闭&#xff0c;但是可能会丢数据&#xff09; SQL> shutdown abort; ORACLE instance shut down. SQL> startup mount;2、基于时间…

策略模式java 用例_java策略模式简单用例

运用java策略模式一个小程序/****/package Strategy;import java.util.Arrays;/*** author HuangRong* Funtion 定义策略*/public interface ComputablesStra {public abstract void printForm();}/****/package Strategy;/*** author HuangRong* Function 为策略模式中的上下文…

LeetCode 2057. 值相等的最小索引

文章目录1. 题目2. 解题1. 题目 给你一个下标从 0 开始的整数数组 nums &#xff0c;返回 nums 中满足 i mod 10 nums[i] 的最小下标 i &#xff1b;如果不存在这样的下标&#xff0c;返回 -1 。 x mod y 表示 x 除以 y 的 余数 。 示例 1&#xff1a; 输入&#xff1a;num…

.net String.Format数字格式化输出

内容转载自&#xff1a;http://www.cnblogs.com/lqb/archive/2008/08/04/1259498.html 前面内容这个做的总结的很全&#xff0c;今后有新增的我继续往后补充。请留意我增加的绿色字体&#xff0c;有惊喜哟&#xff01;-:) int a 12345678; //格式为sring输出 Label1.Text str…

java a 2_Java A* 算法(2)

if(Arrays.binarySearch(isPass,map[fy][fx]) 0) { Node node2 new Node(fx, fy); node2.parent node; node2.G node.G 10; //采用manhattan启发算法 两点中的直角 距离 node2.H Math.abs((ex-fxey-fy)if(Arrays.binarySearch(isPass,map[fy][fx]) > 0)  {Node node…

LeetCode 2058. 找出临界点之间的最小和最大距离(链表)

文章目录1. 题目2. 解题1. 题目 链表中的 临界点 定义为一个 局部极大值点 或 局部极小值点 。 如果当前节点的值 严格大于 前一个节点和后一个节点&#xff0c;那么这个节点就是一个 局部极大值点 。 如果当前节点的值 严格小于 前一个节点和后一个节点&#xff0c;那么这个…

LeetCode 2059. 转化数字的最小运算数(BFS)

文章目录1. 题目2. 解题1. 题目 给你一个下标从 0 开始的整数数组 nums &#xff0c;该数组由 互不相同 的数字组成。另给你两个整数 start 和 goal 。 整数 x 的值最开始设为 start &#xff0c;你打算执行一些运算使 x 转化为 goal 。你可以对数字 x 重复执行下述运算&…

java8 supplyasync_java – 为什么CompletableFuture.supplyAsync成功随...

我是Java 8中lambda和异步代码的新手.我不断得到一些奇怪的结果……我有以下代码&#xff1a;import java.util.concurrent.CompletableFuture;public class Program {public static void main(String[] args) {for (int i 0; i < 100; i) {String test "Test_"…

又是一年国庆假期最后一天

时间过的真的很快啊&#xff0c;美好的日子已经快过完了&#xff0c;在长假的最后一天&#xff0c;有很多的朋友可能都会抱怨&#xff0c;不想上班了&#xff0c;好想在好好的玩玩。开开心心的过完了假期的前六天&#xff0c;当突然被告知明天就是最后一天&#xff0c;你会有什…

LeetCode 2062. 统计字符串中的元音子字符串

文章目录1. 题目2. 解题1. 题目 子字符串 是字符串中的一个连续&#xff08;非空&#xff09;的字符序列。 元音子字符串 是 仅 由元音&#xff08;a、e、i、o 和 u&#xff09;组成的一个子字符串&#xff0c;且必须包含 全部五种 元音。 给你一个字符串 word &#xff0c;…

1006实验一实验报告

实验一报告 实验一、DOS命令解释程序的编写 13物联网工程 张学玲 201306104140 一、 实验目的 &#xff08;1&#xff09;认识DOS&#xff1b; &#xff08;2&#xff09;掌握命令解释程序的原理&#xff1b; &#xff08;3&#xff09;掌握简单的DOS调用方法&#xff1b; &a…