标签：__ return FastAPI app fastapi import def
FastAPI
1. FastAPI
简介
1. 介绍
FastAPI 是一个用于构建 API 快速（高性能）的 web 框架 使用 Python 3.6+ 并基于标准的 Python 类型提示 开发快捷 性能和NodeJS GO相当 并集成SwaggerUI
2. 特征
快速：可与 NodeJS 和 Go 并肩的极高性能（归功于 Starlette 和 Pydantic） 最快的 Python web 框架之一
高效编码：提高功能开发速度约 200％ 至 300％
更少 bug：减少约 40％ 的人为（开发者）导致错误
智能：极佳的编辑器支持 处处皆可自动补全 减少调试时间
简单：设计的易于使用和学习 阅读文档的时间更短
简短：使代码重复最小化 通过不同的参数声明实现丰富功能
健壮：生产可用级别的代码 还有自动生成的交互式文档
标准化：基于（并完全兼容）API 的相关开放标准：OpenAPI (以前被称为 Swagger) 和 JSON Schema
集成两个交互式文档(docs或者redoc访问)
集成数据校验(Pydantic封装了验证数据类型等API)
集成认证授权方式(HttpBasicAuth,OAuth2, jwt等)
支持外部扩展插件丰富API
兼容Starlette特性(支持websocket,graphql,流响应,后台任务处理等)
3. 参考文档
https://fastapi.tiangolo.com/zh/
快速入门
依赖
pip install fastapi
pip install uvicorn[standard]
实例
uvicorn 01:app --reload --host 0.0.0.0 --port 8888
ps:
uvicorn [fastApi应用所在文件名]:[fastApi应用名] --reload --port [端口] --host [IP]
--reload: 服务变动会自动重启
实际会有三个交互式api路由地址
http://localhost:8888/docs
http://localhost:8888/redoc
http://localhost:8888/openapi.json
import uvicorn
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"Hello": "fastapi"}
if __name__ == '__main__':
# app实例启动(reload=True app实例需要用"启动文件名:fastapi实例名"形式)
uvicorn.run(app="01:app", host="0.0.0.0", port=8888, reload=True)
# 控制台启动
# uvicorn 01:app --host 0.0.0.0 --port 8888 --reload
2. 请求参数
查询字符串
from typing import Union, List
import uvicorn
from fastapi import FastAPI, Query
from typing_extensions import Annotated
app = FastAPI()
@app.get("/req/queryParam")
async def query_param(require: str, q: Union[List[str], None] = Query(default=None, alias="q"), page: int = 1, size: int = Query(default=10)):
"""
查询字符串
:param require: 不赋值则为必选参数 等价于 require:str=Query(default=...) 等价于require:str=Query(default=Required) alias起别名
:param q: 可选参数
:param page: 可选参数携带默认值1
:param size: 可选参数携带默认值10 可通过Query()显式指定该参数为查询字符串参数 可对参数进行更复杂的校验
:return:
"""
return {"queryParam": {"require": require, "q": q, "page": page, "size": size}}
if __name__ == '__main__':
uvicorn.run(app="02:app", host="0.0.0.0", port=8888, reload=True)
路径参数
from enum import Enum
import uvicorn
from fastapi import FastAPI, Path
app = FastAPI()
class DataTypes(str, Enum):
a = "typeA"
b = "typeB"
o = "typeOther"
@app.get("/req/{type_}/{id_}/{path_:path}")
async def path_param(type_: DataTypes, path_: str, id_: int = Path()):
"""
路径参数
:param type_: 路径参数 可指定数据类型 可以是基本数据类型或者具体的实体类 封装了参数校验
:param path_: path修饰的参数即为路径参数
:param id_: 可通过Path()显式指定该参数为路径参数 可对参数进行更复杂的校验
:return:
"""
return {"pathParams": {"id_": id_, "dataType": type_, "path_": path_}}
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
请求体
Json
from typing import Union
import uvicorn
from fastapi import FastAPI, Body
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class Stu(BaseModel):
name: str
age: int
desc: Union[str, None] = None
@app.post("/req/v1/json")
async def json_v1(stu: Stu):
"""
请求体-JSON
:param stu: 等价于Annotated[Stu, Body()] 可指定实体类<需要继承自BaseModel>
:return:
"""
return {"json": {"stu": stu}}
@app.post("/req/v2/json")
async def json_v2(stu: Annotated[Stu, Body()], info: Annotated[str, Body()]):
"""
请求体-JSON
:param stu: 等价于Annotated[Stu, Body()] 可指定实体类<需要继承自BaseModel>
存在多个参数时 默认以{"stu":{..}}方式传参
:param info: 使用Annotated[str, Body()]指定为请求体参数
:return:
"""
return {"json": {"stu": stu, "info": info}}
if __name__ == '__main__':
uvicorn.run(app="02:app", host="0.0.0.0", port=8888, reload=True)
pip install python-multipart
import uvicorn
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/req/form")
async def form_param(name2: str = Form(default=None), age2: int = Form(default=...)):
"""
表单数据
:param name2:
:param age2:
:return:
"""
return {"name": name2, "age": age2}
if __name__ == '__main__':
uvicorn.run(app="02:app", host="0.0.0.0", port=8888, reload=True)
File
pip install python-multipart
from typing import List
import uvicorn
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/req/file")
async def file(f1: bytes = File(default=None), f2: UploadFile = None, f3: List[UploadFile] = None):
"""
文件上传
:param f1: bytes 多用于小文件
:param f2: UploadFile 多用于大文件
提供相关属性
filename文件名|file文件属性|size文件大小|headers相关头信息
提供相关方法
write(data: Union[str, bytes]): 写数据
read(size:int): 读数据
seek(offset:int): 移动到文件offset字节处位置
close(): 关闭文件
:param f3: 多文件上传
:return:
"""
# f1保存
with open("a.jpg", "wb") as fw:
fw.write(f1)
# f2保存
with open(f2.filename, "wb") as fw:
for d in iter(lambda: f2.file.read(1024*1), b''):
fw.write(d)
# f3保存
for f in f3:
with open(f.filename, "wb") as fw:
fw.write(f.file.read())
return {"f1": len(f1), "f2": f2.filename, "f3": len(f3)}
if __name__ == '__main__':
uvicorn.run(app="02:app", host="0.0.0.0", port=8888, reload=True)
参数校验
from typing import Union
import uvicorn
from fastapi import FastAPI, Body, Query, Path, Form
from pydantic import BaseModel, Field
from typing_extensions import Annotated
app = FastAPI()
class Inner(BaseModel):
desc: str = None
class Check(BaseModel):
"""
高级校验BaseModel内部字段使用Field()
"""
name: str = Field(default=None)
inner: Union[Inner, None] = None # 嵌套模型
@app.post("/req/check/{path_}")
async def check_param(query_: Union[str, None] = Query(default=None, min_length=1, max_length=10, pattern="^a", alias="q"),
path_: int = Path(default=..., lt=10, ge=1),
# form_: float = Form(default=None, lt=10.0),
json_: str = Body(default=None),
other_: Union[Check, None] = None,
):
"""
参数校验
PS: 所有支持校验的数据类型https://docs.pydantic.dev/usage/types
Query()|Path()|Form()|Body()| Field() 均可自定义高级参数校验
:param query_: 查询字符串
:param path_: 路径参数
:param form_: 表单数据(json和form两者选其一)
:param json_: JSON数据
:param other_: JSON数据-模型类-实体类
:return:
"""
return {"query_": query_, "path_": path_, "json_": json_, "other_": other_}
if __name__ == '__main__':
uvicorn.run(app="02:app", host="0.0.0.0", port=8888, reload=True)
3. 响应参数
纯文本
from typing import Union, Any
import uvicorn
from fastapi import FastAPI, Response
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, Field
from starlette.requests import Request
from starlette.responses import JSONResponse, FileResponse, HTMLResponse, RedirectResponse, StreamingResponse, \
PlainTextResponse
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
@app.get("/res/text", response_class=PlainTextResponse)
def res_v1():
"""
返回纯文本或字节响应
可通过指定response_class 或者return 对应Response子类
:return:
"""
data = "ok"
return data
if __name__ == '__main__':
uvicorn.run(app="03:app", host="0.0.0.0", port=8888, reload=True)
JSON
from typing import Union, Any
import uvicorn
from fastapi import FastAPI, Response
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, Field
from starlette.requests import Request
from starlette.responses import JSONResponse, FileResponse, HTMLResponse, RedirectResponse, StreamingResponse, \
PlainTextResponse
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
class Res1(BaseModel):
code: int = Field(default=0)
msg: str = Field(default="")
class Res2(Res1):
data: Union[Any, None] = None
@app.get("/res/json",
status_code=200,
response_model=Union[Res1, Res2],
response_model_exclude_none=True,
response_model_exclude_unset=True,
tags=["resV2"], # swaggerUi 文档标签展示
summary="接口名称", # swaggerUi 文档展示
description="接口描述" # swaggerUi 文档展示
)
def res_v2(res: Res2):
"""
dict|list|Pydantic模型<继承BaseModel>|数据库模型<继承BaseModel> 内部class Config属性orm_=True等
默认使用JSONResponse返回响应数据
response_model:指定返回的实体类对象
可以为单个实体类或者多个
Union[a,b]:表示任意一个 以实际传参为准
a: 表示特定实体类
List[a]: 实体类集合
Dict[str,str]: 字典 key:str;val:str
...
response_model_exclude_unset：过滤未设置值的字段
response_model_exclude_none: 过滤为None的字段
:param res:
:return:
"""
content = jsonable_encoder(res) # 优先转换为兼容JSON的数据类型
return JSONResponse(content=content) # 返回JSON数据
# return res # 默认使用JSONResponse返回响应数据
if __name__ == '__main__':
uvicorn.run(app="03:app", host="0.0.0.0", port=8888, reload=True)
XML
from typing import Union, Any
import uvicorn
from fastapi import FastAPI, Response
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, Field
from starlette.requests import Request
from starlette.responses import JSONResponse, FileResponse, HTMLResponse, RedirectResponse, StreamingResponse, \
PlainTextResponse
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
@app.get("/res/xml")
def res_v3():
"""
返回XML
:return:
"""
data = """
<xml><name>fastapi</name></xml>
"""
return Response(content=data, media_type="application/xml")
if __name__ == '__main__':
uvicorn.run(app="03:app", host="0.0.0.0", port=8888, reload=True)
HTML
# 加载静态模板
pip install jinja2
# 加载静态资源
pip install aiofiles
|----static
|----static/a.css
|----a.html
p {
color: green;
font-size: 18px;
}
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>html</title>
<link href="{{ url_for('static', path='a.css') }}" rel="stylesheet">
</head>
<body>
<p>{{name}}</p>
</body>
</html>
from typing import Union, Any
import uvicorn
from fastapi import FastAPI, Response
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, Field
from starlette.requests import Request
from starlette.responses import JSONResponse, FileResponse, HTMLResponse, RedirectResponse, StreamingResponse, \
PlainTextResponse
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
@app.get("/res/v1/html")
def res_v4_1():
"""
返回html
:return:
"""
html = """
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>html</title>
</head>
<body>
<p>fastapi</p>
</body>
</html>
"""
return HTMLResponse(content=html, status_code=200)
app.mount(path="/static", # 访问路由前缀
app=StaticFiles(directory="./static"), # 文件路径
name="static" # url_for 引用
) # 配置加载静态资源
template = Jinja2Templates(directory=".") # 配置加载静态模板
@app.get("/res/v2/html")
def res_v4_2(request: Request, name: str = "fastapi"):
"""
加载静态模板：pip install jinja2
加载静态资源：pip install aiofiles
必须携带Request参数
:return:
"""
return template.TemplateResponse("a.html", {"request": request, "name": name})
if __name__ == '__main__':
uvicorn.run(app="03:app", host="0.0.0.0", port=8888, reload=True)
文件
from typing import Union, Any
import uvicorn
from fastapi import FastAPI, Response
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, Field
from starlette.requests import Request
from starlette.responses import JSONResponse, FileResponse, HTMLResponse, RedirectResponse, StreamingResponse, \
PlainTextResponse
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
@app.get("/res/file")
def res_v5_1():
"""
返回文本
:return:
"""
return FileResponse(path=r"G:\workDoc\png\alice.jpg",
media_type="image/png",
status_code=200,
filename="a.png",
content_disposition_type="attachment")
@app.get("/res/stream")
def res_v5_2():
"""
文件流
:return:
"""
async def data_iter():
with open(r"G:\workDoc\png\alice.jpg", "rb") as f:
yield f.read()
return StreamingResponse(data_iter())
@app.get("/res/redirect")
def res_v6():
"""
临时跳转
:return:
"""
return RedirectResponse("http://localhost:8888/res/stream")
if __name__ == '__main__':
uvicorn.run(app="03:app", host="0.0.0.0", port=8888, reload=True)
跳转
from typing import Union, Any
import uvicorn
from fastapi import FastAPI, Response
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, Field
from starlette.requests import Request
from starlette.responses import JSONResponse, FileResponse, HTMLResponse, RedirectResponse, StreamingResponse, \
PlainTextResponse
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
app = FastAPI()
@app.get("/res/redirect")
def res_v6():
"""
临时跳转
:return:
"""
return RedirectResponse("http://localhost:8888/res/stream")
if __name__ == '__main__':
uvicorn.run(app="03:app", host="0.0.0.0", port=8888, reload=True)
异常响应
"""
异常类
局部默认
全局自定义
重写默认
"""
import uvicorn
from fastapi import FastAPI, HTTPException, Request
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from starlette.responses import JSONResponse
app = FastAPI()
class MyExc(Exception):
"""
"""
def __init__(self, type_):
self.type_ = type_
@app.exception_handler(MyExc)
async def my_exc(req: Request, exc: MyExc):
"""
全局配置：自定义异常类处理
:param req:
:param exc:
:return:
"""
return JSONResponse(status_code=400,
content=jsonable_encoder({"detail": f"非法{exc.type_}"})
)
@app.get("/error")
async def error(type_: int):
"""
:param type_:
:return:
"""
if type_ == 1:
# 抛出默认异常类
raise HTTPException(status_code=400, detail="非法type_")
elif type_ == 2:
# 抛出自定义异常类
raise MyExc(type_=type_)
elif type_ == 3:
# 抛出全局异常
raise ValueError("报错了")
# 参数类型错误抛出RequestValidationError异常
return {"data": type_}
@app.exception_handler(Exception)
async def all_exc(req: Request, exc: Exception):
"""
全局配置：捕获所有异常类
:param req:
:param exc:
:return:
"""
return JSONResponse(status_code=500,
content=jsonable_encoder({"detail1": str(exc), "reqBody1": req.body, "reqMethod1": req.method})
)
@app.exception_handler(RequestValidationError)
async def req_valid(req: Request, exc: RequestValidationError):
return JSONResponse(status_code=400,
content=jsonable_encoder({"detail2": exc.errors(), "reqBody2": exc.body, "reqMethod2": req.method}))
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
4. 依赖注入
"""
依赖注入
简介：
FastAPI 提供了简单易用，但功能强大的依赖注入系统。这个依赖系统设计的简单易用，可以让开发人员轻松地把组件集成至 FastAPI
依赖注入常用于以下场景：
共享业务逻辑（复用相同的代码逻辑）
共享数据库连接
实现安全、验证、角色权限等等
实现：
创建依赖项(函数或者可调用对象<a()>)
声明依赖项(需要Depends()和一个新的参数接收 传给Depends的参数必须为可调用参数)
ps:
如果在同一个路径（/depend1）操作 多次声明了同一个依赖项，例如，多个依赖项共用一个子依赖项，FastAPI 在处理同一请求时，只调用一次该子依赖项。
FastAPI 不会为同一个请求多次调用同一个依赖项，而是把依赖项的返回值进行「缓存」，并把它传递给同一请求中所有需要使用该返回值的「依赖项」。
在高级使用场景中，如果不想使用「缓存」值，而是为需要在同一请求的每一步操作（多次）中都实际调用依赖项，可以把 Depends 的参数 use_cache 的值设置为 False 即Depends(依赖项对象, use_cache=False)
"""
from typing import Union, Dict
import uvicorn
from fastapi import Depends, FastAPI, Header, HTTPException
from typing_extensions import Annotated
async def verify(authKey: str = Header()):
if authKey != "fastapi":
raise HTTPException(status_code=400, detail="authKey非法")
# 依赖项 全局配置
app = FastAPI(dependencies=[Depends(verify)])
# 依赖项-函数
async def common_param(
q: Union[str, None] = None, skip: int = 0, limit: int = 10
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/depend1")
async def depend_v1(commons: dict = Depends(common_param)):
return commons
# 依赖项-类对象
class CommonParam:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 10):
self.q = q
self.skip = skip
self.limit = limit
def to_dict(self):
return {"q": self.q, "skip": self.skip, "limit": self.limit}
@app.get("/depend2")
async def depend_v2(commons: CommonParam = Depends()):
"""
三种方式均可
commons: CommonParam = Depends()
<=>
commons = Depends(CommonParam)
<=>
commons: CommonParam = Depends(CommonParam)
:param commons:
:return:
"""
return commons.to_dict()
# 依赖项-嵌套使用
async def inner(q: Union[str, None] = None):
return q
async def outer(q: str = Depends(inner), skip: int = 0, limit: int = 10):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/depend3")
async def depend_v3(commons: dict = Depends(outer)):
return commons
# 依赖项-作用于路径操作装饰器上
@app.get("/depend3", dependencies=[Depends(outer)])
async def depend_v3():
return outer()
# 依赖项-可调用对象
class CallBack:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 10):
self.q = q
self.skip = skip
self.limit = limit
def __call__(self):
return {"q": self.q, "skip": self.skip, "limit": self.limit}
cb = CallBack()
@app.get("/depend4")
async def depend_v4(commons: Annotated[Dict, Depends(cb)]):
"""必须使用Annotated声明"""
return commons
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
5. 会话和鉴权
Cookie&Session
"""
cookie + session
"""
import uvicorn
from fastapi import FastAPI, Cookie, Request
from starlette.middleware.sessions import SessionMiddleware
from starlette.responses import JSONResponse
from typing_extensions import Annotated
app = FastAPI()
# 使用session必须注册中间件SessionMiddleware
app.add_middleware(SessionMiddleware, secret_key="session", session_cookie="mySessionId")
@app.get("/cookie/set")
async def set_cookie():
response = JSONResponse(content={"cookie-set": "ok"})
response.set_cookie(key="webFrame", value="fastapi", max_age=3600)
return response
@app.get("/cookie/get")
async def get_cookie(req: Request, webFrame: Annotated[str, Cookie()] = None):
return {"cookie-get1": webFrame, "cookie-get2": req.cookies.get("webFrame")}
@app.get("/session/set")
async def set_session(req: Request):
req.session.update({"key": "fastapi"})
return {"session-set": "ok"}
@app.get("/session/get")
async def get_session(req: Request):
return {"session-get": req.session.get("key")}
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
JWT
pip install python-jose
"""
jwt
pip install python-jose
"""
from datetime import datetime, timedelta
import uvicorn
from fastapi import FastAPI, HTTPException, Depends
from jose import jwt, JWTError
from pydantic import BaseModel
from starlette import status
app = FastAPI()
SECRET_KEY = "secret"
ALGORITHM = "HS256"
class User(BaseModel):
name: str
psw: str
exp: int
@app.get("/token/get")
async def get_token(user: User):
data = user.model_dump().copy()
seconds = data.get("exp")
data.update({"exp": datetime.utcnow() + timedelta(seconds=seconds)})
encode_jwt = jwt.encode(claims=data, key=SECRET_KEY, algorithm=ALGORITHM)
return {"token": encode_jwt}
async def verify(name: str, psw: str, token: str):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
p_name = payload.get("name")
p_psw = payload.get("psw")
if name == p_name and psw == p_psw:
return True
else:
return False
except:
return False
@app.get("/token/verify")
async def verify_token(name: str, psw: str, token: str):
auth_exc = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
detail="authFail",
)
if not token:
raise auth_exc
try:
if verify(name, psw, token):
return {"verify": "ok", "token": token}
except JWTError:
raise auth_exc
@app.get("/")
async def index(auth: bool = Depends(verify)):
"""
http://localhost:8888/?name=fastapi&psw=fastapi&token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYW1lIjoiZmFzdGFwaSIsInBzdyI6ImZhc3RhcGkiLCJleHAiOjE2OTYzMzQ1NDZ9.LB-osTQGRtsj4plGJFPKXOZYqCNMs1oDBmWzUlpkB0w
:param auth:
:return:
"""
if auth:
return {"msg": "ok", "code": 0}
else:
return {"msg": "authError", "code": -1}
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
OAuth2
"""
oauth2
"""
from typing import Union
import uvicorn
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
# 模拟底库人员数据
fake_users_db = {
"fastapi": {
"username": "fastapi",
"hashed_password": "secret-fastapi",
"disabled": False,
}
}
app = FastAPI()
def fake_hash_password(password: str):
return "secret-" + password
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="oauth2")
class User(BaseModel):
username: str
disabled: Union[bool, None] = 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)
def fake_decode_token(token):
user = get_user(fake_users_db, token)
return user
# 依赖项1
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
# 依赖项2
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("/oauth2")
async def oauth2(form_data: OAuth2PasswordRequestForm = Depends()):
"""
oauth2鉴权处理
使用到OAuth2PasswordRequestForm依赖项 传参需要为表单数据 详见类OAuth2PasswordRequestForm相关实例属性
也可以不使用OAuth2PasswordRequestForm依赖项处理 详见下oauth2
:param form_data:
:return:
"""
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("/auth")
async def auth(name: str, current_user: User = Depends(get_current_active_user)):
if name != current_user.username:
raise HTTPException(status_code=400, detail="invalidUser")
return current_user
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
HTTP Basic Auth
import secrets
import uvicorn
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from typing_extensions import Annotated
app = FastAPI()
security = HTTPBasic()
def get_current_username(
credentials: Annotated[HTTPBasicCredentials, Depends(security)]
):
"""
HTTPBasicCredentials 继承自BaseModel 存在username 和passoword属性
:param credentials:
:return:
"""
current_username_bytes = credentials.username.encode("utf8")
correct_username_bytes = b"fastapi"
# 使用secrets.compare_digest比对
is_correct_username = secrets.compare_digest(
current_username_bytes, correct_username_bytes
)
current_password_bytes = credentials.password.encode("utf8")
correct_password_bytes = b"fastapi"
is_correct_password = secrets.compare_digest(
current_password_bytes, correct_password_bytes
)
if not (is_correct_username and is_correct_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Basic"},
)
return credentials.username
@app.get("/auth")
def read_current_user(username: Annotated[str, Depends(get_current_username)]):
return {"username": username}
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
6. 中间件
"""
中间件-(请求拦截器)
TrustedHostMiddleware|HTTPSRedirectMiddleware|SessionMiddleware|CORSMiddleware等
PS:
如果使用了 yield 关键字的依赖项 依赖中的退出代码将在执行中间件后执行
如果有任何后台任务(稍后记录) 它们将在执行中间件后运行
"""
import time
import uvicorn
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import Message
app = FastAPI()
# 中间件-CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost",
"http://localhost:8888"
],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 自定义中间件
@app.middleware("costTime")
async def req_handler(req: Request, api_func):
st = time.time()
print("costTimeMiddleWare执行开始...")
response = await api_func(req)
et = round((time.time()-st), 8)
print(f"{api_func.__name__}耗时:{et*1000}ms")
print("costTimeMiddleWare执行结束...")
return response
# 重写默认的中间件类
class MyMiddleWare(BaseHTTPMiddleware):
# 重写dispatch方法(必须为dispatch)
async def dispatch(self, request, api_func):
print("MyMiddleWare执行开始...")
response = await api_func(request)
print("MyMiddleWare执行结束...")
return response
app.add_middleware(MyMiddleWare)
# 自定义中间件类
class MyClassMiddleWare:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] == "http":
req = Request(scope)
print(">>>", type(scope), scope)
print(">>>", type(receive), receive)
print(">>>", type(send), send)
print(f"method:{req.method};url:{req.url};")
return await self.app(scope, receive, send)
async def msg_handler(msg: Message) -> None:
# 服务启动初始化执行
print(">>>", type(msg), msg)
await send(msg)
await self.app(scope, receive, msg_handler)
app.add_middleware(MyClassMiddleWare)
@app.get("/")
async def main():
return {"message": "Hello fastapi"}
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
7. 扩展
多 WSGI应用启动
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.wsgi import WSGIMiddleware
from flask import Flask, jsonify
# 创建Flask实例
flask_app = Flask(__name__)
# 创建FastAPI实例
app = FastAPI()
# 创建子FastAPI实例
sub_api = FastAPI()
@flask_app.route("/")
def flask_main():
return jsonify({"msg": "Hello Flask"})
@app.get("/v2")
def read_main():
return {"msg": "Hello FastAPI"}
@sub_api.get("/")
def sub_main():
return {"msg": "Hello SubFastAPI"}
app.mount("/v1", WSGIMiddleware(flask_app)) # 添加Flask应用(非ASGIApp需要使用WSGIMiddleware来包装)
app.mount("/v3", sub_api) # 添加子FastAPI应用
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
后台任务
"""
后台任务BackgroundTasks
请求结束后处理的任务
"""
import asyncio
from typing import Union
import uvicorn
from fastapi import BackgroundTasks, Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
# 后台任务函数
async def task_run(num):
"""
任务函数-可同步或异步
:param num:
:return:
"""
await asyncio.sleep(3)
print(f"{num}-后台耗时任务执行结束...")
# 依赖项
def on_depend(background_tasks: BackgroundTasks, q: Union[str, None] = None):
if q:
background_tasks.add_task(task_run, q)
return q
@app.get("/task1")
async def send_notification(
num: str, background_tasks: BackgroundTasks
):
"""直接添加后台任务"""
background_tasks.add_task(task_run, num)
return {"task": "done"}
@app.get("/task2")
async def send_notification(background_tasks: BackgroundTasks, q: Annotated[str, Depends(on_depend)]
):
"""
使用依赖项
:param background_tasks:
:param q:
:return:
"""
background_tasks.add_task(task_run, q)
return {"task": "done"}
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
路由组
|--app
| |--__init__.py
| |--main.py # 启动文件
| |--dependencies.py # 依赖项
| |--routers
| | |--__init__.py
| | |--user.py # 用户子模块
# dependencies.py
from fastapi import Header, HTTPException
async def verify(auth: str = Header()):
if auth != "fastapi":
raise HTTPException(status_code=400, detail="authKey非法")
# user.py
from fastapi import APIRouter, Request, Depends
# from dependencies import verify
# router = APIRouter(
# prefix="/user",
# tags=["user"],
# dependencies=[Depends(verify)]
# )
router = APIRouter()
@router.get("/")
async def user(req: Request):
return {"msg": "user", "url": req.url}
# main.py
import uvicorn
from fastapi import FastAPI, Request, Depends
from dependencies import verify
from routers import users
app = FastAPI(dependencies=[Depends(verify)])
app.include_router(users.router,
prefix="/user",
tags=["user"],
dependencies=[],
)
@app.get("/")
async def root(req: Request):
return {"msg": "root", "url": req.url}
if __name__ == '__main__':
uvicorn.run(app=app, host="0.0.0.0", port=8888)
文档
标签：__,
return,
FastAPI,
app,
fastapi,
import,
def
From： https://www.cnblogs.com/fsh19991001/p/17741765.html