.github
.husky
assets
docs
adb.md
adobe-lightroom.md
adobe-photoshop.md
adobe-xd.md
android-studio.md
ansible.md
apt.md
ascii-code.md
aspect-ratio.md
awk.md
bash.md
blender.md
c.md
cargo.md
chmod.md
cmake.md
cmd.md
cocoapods.md
colors-named.md
conan.md
conda.md
cpp.md
cron.md
cs.md
css.md
curl.md
dart.md
django.md
docker.md
dockerfile.md
ejs.md
electron.md
emacs.md
emmet.md
emoji.md
es6.md
expressjs.md
fastapi.md
feds.md
ffmpeg.md
find.md
firefox.md
flask.md
flutter.md
gdb.md
git.md
github-actions.md
github-copilot.md
github.md
gitlab-ci.md
gitlab.md
gmail.md
golang.md
google-chrome.md
graphql.md
grep.md
homebrew.md
html-char.md
html.md
htmx.md
htop.md
http-status-code.md
ini.md
intelli-j-idea.md
iptables.md
iso-639-1.md
java.md
javascript.md
jest.md
jq.md
jquery.md
json.md
julia.md
justfile.md
koajs.md
kotlin.md
kubernetes.md
laravel.md
latex.md
lerna.md
lessjs.md
linux-command.md
lsof.md
lua.md
make.md
markdown.md
matlab.md
mime.md
mitmproxy.md
mongodb.md
mysql.md
nestjs.md
netcat.md
netstat.md
nextjs.md
nginx.md
npm.md
openssl.md
oracle.md
package.json.md
pacman.md
php.md
pm2.md
postgres.md
ps.md
pycharm.md
python.md
pytorch.md
quickreference.md
react-native.md
react.md
reactrouter.md
redis.md
regex.md
resolutions.md
ruby.md
rust.md
sass.md
scala.md
screen.md
sed.md
semver.md
sketch.md
ssh.md
styled-components.md
stylus.md
sublime-text.md
subversion.md
swift.md
swiftui.md
symbol-code.md
sysdig.md
systemd.md
tailwindcss.md
tar.md
taskset.md
tmux.md
toml.md
typescript.md
vim.md
vscode.md
vue.md
vue2.md
webstorm.md
xcode.md
xpath.md
yaml.md
yarn.md
yum.md
zip.md
.dockerignore
.editorconfig
.gitattributes
.gitignore
.lintstagedrc
.markdownlint.json
.npmrc
.prettierignore
.prettierrc
.refsrc.json
CONTRIBUTING.md
Dockerfile
LICENSE
README.md
package.json
renovate.json
12 KiB
12 KiB
FastAPI 备忘清单
FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架,使用 Python 3.6+ 并基于标准的 Python 类型提示。Python: 3.9.5
FastAPI: 0.103.1
入门
最小程序
下面代码会直接启动http服务,也可以使用 uvicorn main:app --reload
from fastapi import FastAPI
import uvicorn
app = FastAPI()
# http://127.0.0.1:8000/
@app.get("/")
async def root():
return {"message": "Hello World"}
if __name__ == '__main__':
uvicorn.run(app='main:app', reload=True)
路径参数
最基本的路径参数
# http://127.0.0.1:8000/items/1
@app.get("/items/{item_id}")
async def read_item(item_id):
return {"item_id": item_id} # item_id自定义
多个路径参数
# http://127.0.0.1:8000/items/1/2
@app.get("/items/{item_id}/{user_id}")
async def read_item(item_id, user_id):
return {"item_id": item_id, "user_id": user_id}
有类型的路径参数
# http://127.0.0.1:8000/items/1
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
文件路径参数
# http://127.0.0.1:8000/file//home/my/my.txt
@app.get("/file/{file_path:path}")
async def read_item(file_path):
return {"file_path": file_path}
查询参数
带默认值的查询参数
# http://127.0.0.1:8000/items/?skip=0&limit=2
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
return fake_items_db[skip: skip + limit]
可选查询参数
# http://127.0.0.1:8000/items/1?q=admin
from typing import Union
@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Union[str, None] = None):
if q:
return {"item_id": item_id, "q": q}
return {"item_id": item_id}
多路径多查询参数
# http://127.0.0.1:8000/users/1/items/2
# or
# http://127.0.0.1:8000/users/1/items/2?q=query&short=true
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
user_id: int, item_id: str, q: Union[str, None] = None, short: bool = False
):
item = {"item_id": item_id, "owner_id": user_id}
if q:
item.update({"q": q})
if not short:
item.update(
{"description": "This is an amazing item that has a long description"}
)
return item
必需查询参数
# http://127.0.0.1:8000/items/123?needy=yes
@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):
item = {"item_id": item_id, "needy": needy}
return item
请求体
from pydantic import BaseModel
from typing import Union
class Item(BaseModel):
name: str = '小明'
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
@app.post("/items/")
async def create_item(item: Item):
print(item.name)
return item
调用
curl -X 'POST' \
'http://127.0.0.1:8000/items/' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"name": "小明",
"description": "string",
"price": 0,
"tax": 0
}'
查询参数和字符串校验
from fastapi import Query
@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, max_length=50)):
results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
if q:
results.update({"q": q})
return results
参数列表
参数 | 含义 | 类型 |
---|---|---|
default | 默认值 | 任意类型或... |
max_length | 最大长度 | int |
min_length | 最小长度 | int |
pattern | 正则匹配 | string |
alias | 别名参数 | string |
deprecated | 准备弃用参数 | bool |
多个相同的查询参数
# http://127.0.0.1:8000/items/?q=foo&q=bar
@app.get("/items/")
async def read_items(q: Union[List[str], None] = Query(default=None)):
query_items = {"q": q}
return query_items
路径参数和数值校验
Path用法基本和Query相同,参考:FastAPI官方文档
导入 Path
from fastapi import FastAPI, Path, Query
from typing_extensions import Annotated
声明元数据
@app.get("/items/{item_id}")
async def read_items(
item_id: Annotated[int, Path(title="The ID of the item to get")],
q: Annotated[Union[str, None], Query(alias="item-query")] = None,
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
return results
参数列表
参数 | 含义 | 类型 |
---|---|---|
... | 和Query具有相同参数 | ... |
ge | 大于等于 | int float |
gt | 大于 | int float |
le | 小于等于 | int float |
le | 小于等于 | int float |
title | api文档的标题 | string |
其他参数
都具有Query的参数,max_length、min_length等
Cookie参数
from fastapi import Cookie
@app.get("/items/")
async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None):
return {"ads_id": ads_id}
Header 参数
from fastapi import Header
@app.get("/items/")
async def read_items(user_agent: Annotated[Union[str, None], Header()] = None,items_id: Annotated[Union[int, None], Header(ge=1)] = None):
return {"User-Agent": user_agent, "items_id": items_id}
表单数据
接收的不是 JSON,而是表单字段时,要使用 Form。
安装
pip install python-multipart
HTML
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
</head>
<body>
<form method="post" action="http://127.0.0.1:8000/login">
<span>账号:</span><input type="text" name="username">
<br>
<span>密码:</span><input type="password" name="password">
<br>
<input type="submit" value="登录">
</form>
</body>
</html>
FastAPI
from fastapi import FastAPI, Form
import uvicorn
app = FastAPI()
@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
return {"username": username}
if __name__ == '__main__':
uvicorn.run(app='main:app', reload=True)
文件上传
from fastapi import FastAPI, UploadFile
from fastapi.responses import HTMLResponse
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
print(file.file.read().decode())
return {"filenames": file.filename, "type": str(type(file.file))}
@app.get("/")
async def main():
content = """<body>
<form action="/uploadfile/" enctype="multipart/form-data" method="post">
<input name="file" type="file" multiple>
<input type="submit">
</form>
</body>"""
return HTMLResponse(content=content)
UploadFile属性
属性名 | 含义 | 返回 |
---|---|---|
filename | 文件名 | 上传的文件名 |
content_type | 内容类型 | MIME 类型 |
file | 文件 | SpooledTemporaryFile具有read,write方法 |
UploadFile async方法
方法名 | 含义 |
---|---|
write(data) | 把 data 写入文件 |
read(size) | 按指定数量的字节读取文件内容 |
seek(offset) | 移动至文件 offset (int )字节处的位置 |
close() | 关闭文件 |
依赖项
依赖项使用场景
- 共享业务逻辑(复用相同的代码逻辑)
- 共享数据库连接
- 实现安全、验证、角色权限
- 等……
创建依赖项
from typing import Union
from fastapi import Depends, FastAPI
app = FastAPI()
# read_items和read_users方法依赖common_parameters
# 白话就是read_items和read_users都需要q,skip,limit查询参数
async def common_parameters(
q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commons
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
return commons
类作为依赖项
from typing import Union
from fastapi import Depends, FastAPI
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
# read_itemsx接收一个commons参数,类型是CommonQueryParams
# CommonQueryParams接收三个参数,这三个参数是调用api的时候传
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response
还可以简写
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()): # 这里的Depends没有传参,FastAPI会自动使用CommonQueryParams
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response
子依赖项
from typing import Union
from fastapi import Cookie, Depends, FastAPI
app = FastAPI()
def query_extractor(q: Union[str, None] = None):
return q
def query_or_cookie_extractor(
q: str = Depends(query_extractor),
last_query: Union[str, None] = Cookie(default=None),
):
if not q:
return last_query
return q
# read_query函数依赖query_or_cookie_extractor函数
# query_or_cookie_extractor函数又依赖query_extractor函数
# 就是说依赖项可以依赖其他依赖项,只要你不晕,可以无数次套娃
@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
return {"q_or_cookie": query_or_default}
不使用缓存
# 使用use_cache = False参数不使用缓存数据
# 不使用use_cache = False,value和value1是一样的
def result_value():
value = randint(1, 99)
return value
def get_value(value: int = Depends(result_value, use_cache=False), value1: int = Depends(result_value, use_cache=False)):
return value, value1
@app.get('/value/')
async def needy_dependency(value: tuple = Depends(get_value)):
return {"value": value}
全局依赖项
from fastapi import Depends, FastAPI, Header, HTTPException
async def verify_token(x_token: str = Header()):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: str = Header()):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
# 全局依赖项很有用,后面的安全性就可以使用全局依赖项
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
@app.get("/items/")
async def read_items():
return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
@app.get("/users/")
async def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]
安全性
待更新
参考
- Python 备忘清单 (jaywcjlove.github.io)
- FastAPI 官方文档 (fastapi.tiangolo.com)