fastapi
FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架,使用 Python 3.6+ 并基于标准的 Python 类型提示。
关键特性:
- 快速:可与 NodeJS 和 Go 比肩的极高性能(归功于 Starlette 和 Pydantic)。最快的 Python web 框架之一。
- 高效编码:提高功能开发速度约 200% 至 300%。*
- 更少 bug:减少约 40% 的人为(开发者)导致错误。*
- 智能:极佳的编辑器支持。处处皆可自动补全,减少调试时间。
- 简单:设计的易于使用和学习,阅读文档的时间更短。
- 简短:使代码重复最小化。通过不同的参数声明实现丰富功能。bug 更少。
- 健壮:生产可用级别的代码。还有自动生成的交互式文档。
- 标准化:基于(并完全兼容)API 的相关开放标准:OpenAPI (以前被称为 Swagger) 和JSON Schema
文档: https://fastapi.tiangolo.com
源码: https://github.com/tiangolo/fastapi
第一步
- 安装fastapi
pip install fastapi
- 安装
uvicorn
来作为服务器:
pip install "uvicorn[standard]"
最简单fastapi程序
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
uvicorn main:app --reload
自带文档
http://127.0.0.1:8000/docs
http://127.0.0.1:8000/redoc
路径参数
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
浏览器访问 http://127.0.0.1:8000/items/3
响应:
{"item_id":3}
浏览器访问http://127.0.0.1:8000/items/foo
响应
{
"detail": [
{
"loc": [
"path",
"item_id"
],
"msg": "value is not a valid integer",
"type": "type_error.integer"
}
]
}
因为路径参数 item_id
传入的值为 "foo"
,它不是一个 int
。
查询参数
声明不属于路径参数的其他函数参数时,它们将被自动解释为"查询字符串"参数
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
return {"skip":skip}
浏览器访问http://127.0.0.1:8000/items/?skip=0&limit=10
响应:
{"skip":0,"limit":10}
浏览器访问127.0.0.1:8000/items?skip=0&limit=10.1
响应:
{"detail":
[
{
"loc": ["query", "limit"],
"msg": "value is not a valid integer",
"type": "type_error.integer"
}
]
}
请求参数
from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
tax: Union[float, None] = None
app = FastAPI()
@app.post("/items")
async def create_item(item: Item):
return item
请求http://127.0.0.1:8000/items 请求类型post 参数:{ "name":"yyc", "price":99.0, "tax":18.8 }
响应:
{
"name": "yyc",
"price": 99.0,
"tax": 18.8
}
请求http://127.0.0.1:8000/items 请求类型post 参数:{ "name":"yyc" }
响应:
{
"detail": [
{
"loc": [
"body",
"price"
],
"msg": "field required",
"type": "value_error.missing"
}
]
}
多个参数
from typing import Union
from fastapi import FastAPI, Body
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
class User(BaseModel):
username: str
full_name: Union[str, None] = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, user: User, item: Item, importance: int = Body(), q: Union[str, None] = None):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance, "q": q}
return results
嵌套字段
from typing import Set, Union
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Image(BaseModel):
url: str
name: str
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
tags: Set[str] = set()
image: Union[Image, None] = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
results = {"item_id": item_id, "item": item}
return results
请求体 + 路径参数
from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
app = FastAPI()
@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item):
return {"item_id": item_id, **item.dict()}
请求: http://127.0.0.1:8000/items/1 post请求
请求参数:
{ "name":"yyc","price":99.0, "description": "这个nb","tax":18.8 }
响应:
{
"item_id": 1,
"name": "yyc",
"description": "这个人nb",
"price": 99.0,
"tax": 18.8
}
请求体 + 路径参数 + 查询参数
from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
app = FastAPI()
@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: Union[str, None] = None):
result = {"item_id": item_id, **item.dict()}
if q:
result.update({"q": q})
return result
请求http://127.0.0.1:8000/items/1?q=yyc put请求
请求参数:
{ "name":"yyc", "price":99.0, "description": "这个人nb", "tax":18.8 }
响应:
{
"item_id": 1,
"name": "yyc",
"description": "这个人nb",
"price": 99.0,
"tax": 18.8,
"q": "yyc"
}
文件上传
pip install python-multipart
案例:
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/files/")
async def create_file(file: bytes = File()):
return {"file_size": len(file)}
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
return {"filename": file.filename}
from typing import Union
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/files/")
async def create_file(file: Union[bytes, None] = File(default=None)):
if not file:
return {"message": "No file sent"}
else:
return {"file_size": len(file)}
@app.post("/uploadfile/")
async def create_upload_file(file: Union[UploadFile, None] = None):
if not file:
return {"message": "No upload file sent"}
else:
return {"filename": file.filename}
多文件上传
3.8版本即以上
from typing import List
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse
app = FastAPI()
@app.post("/files/")
async def create_files(files: List[bytes] = File()):
return {"file_sizes": [len(file) for file in files]}
@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile]):
return {"filenames": [file.filename for file in files]}
@app.get("/")
async def main():
content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
"""
return HTMLResponse(content=content)
3.9版本即以上
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse
app = FastAPI()
@app.post("/files/")
async def create_files(files: list[bytes] = File()):
return {"file_sizes": [len(file) for file in files]}
@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):
return {"filenames": [file.filename for file in files]}
@app.get("/")
async def main():
content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
"""
return HTMLResponse(content=content)
参数校验
查询参数
from typing import Union
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
# q 的值为str类型或为none 默认值为none 最小长度3,最大长度50 支持正则匹配
async def read_items(
q: Union[str, None] = Query(
default=None, min_length=3, max_length=50, regex="^fixedquery$"
)
):
results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
if q:
results.update({"q": q})
return results
路径参数
from fastapi import FastAPI, Path, Query
app = FastAPI()
@app.get("/items/{item_id}")
async def read_items(
*,
item_id: int = Path(title="The ID of the item to get", ge=0, le=1000),
q: str,
size: float = Query(gt=0, lt=10.5)
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
return results
gt
:大于ge
:大于等于lt
:小于le
:小于等于
Cookie 参数
from typing import Union
from fastapi import Cookie, FastAPI
app = FastAPI()
@app.get("/items/")
async def read_items(ads_id: Union[str, None] = Cookie(default=None)):
return {"ads_id": ads_id}
设置cookie
from fastapi import FastAPI, Response
app = FastAPI()
@app.post("/cookie-and-object/")
def create_cookie(response: Response):
response.set_cookie(key="fakesession", value="fake-cookie-session-value")
return {"message": "Come to the dark side, we have cookies"}
header参数
from typing import Union
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items/")
async def read_items(user_agent: Union[str, None] = Header(default=None)):
return {"User-Agent": user_agent}
响应
{"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.54"}
from typing import List, Union
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items/")
async def read_items(x_token: Union[List[str], None] = Header(default=None)):
return {"X-Token values": x_token}
header:
X-Token: foo
X-Token: bar
响应:
{
"X-Token values": [
"bar",
"foo"
]
}
request
request.base_url # 获取baseurl
request.url # 获取请求路径
request.url.path # 获取path
request.headers # 获取请求头
响应模型
你可以在任意的路径操作中使用 response_model
参数来声明用于响应的模型:
from typing import List, Union
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
tags: List[str] = []
@app.post("/items/", response_model=Item)
# 规定返回响应的类型必须是Item类型
async def create_item(item: Item):
return item
当响应参数中 有多余参数时不返回
from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
app = FastAPI()
class UserIn(BaseModel):
username: str
password: str
email: EmailStr
full_name: Union[str, None] = None
class UserOut(BaseModel):
username: str
email: EmailStr
full_name: Union[str, None] = None
@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
return user
请求http://127.0.0.1:8000/user/ post 请求
请求参数
{
"username":"yyc",
"password":"123456",
"email":"[email protected]",
"full_name":"杨爸爸"
}
响应:
{
"username": "yyc",
"email": "[email protected]",
"full_name": "杨爸爸"
}
状态码
from fastapi import FastAPI, status
app = FastAPI()
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
return {"name": name}
from fastapi import FastAPI
app = FastAPI()
@app.post("/items/", status_code=201)
async def create_item(name: str):
return {"name": name}
表单数据
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
return {"username": username}
异常处理
from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo": "The Foo Wrestlers"}
@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
if item_id not in items:
raise HTTPException(
status_code=404,
detail="Item not found",
headers={"X-Error": "There goes my error"},
)
return {"item": items[item_id]}
自定义异常处理
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class UnicornException(Exception):
def __init__(self, name: str):
self.name = name
app = FastAPI()
@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
return JSONResponse(
status_code=418,
content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
)
@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
if name == "yolo":
raise UnicornException(name=name)
return {"unicorn_name": name}
请求 /unicorns/yolo
时,路径操作会触发 UnicornException
。
但该异常将会被 unicorn_exception_handler
处理。
接收到的错误信息清晰明了,HTTP 状态码为 418
,JSON 内容如下:
{"message": "Oops! yolo did something. There goes a rainbow..."}
覆盖请求验证异常
请求中包含无效数据时,FastAPI 内部会触发 RequestValidationError
。
该异常也内置了默认异常处理器。
覆盖默认异常处理器时需要导入 RequestValidationError
,并用 @app.excption_handler(RequestValidationError)
装饰异常处理器。
这样,异常处理器就可以接收 Request
与异常。
from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
app = FastAPI()
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
return PlainTextResponse(str(exc.detail), status_code=exc.status_code)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
return PlainTextResponse(str(exc), status_code=400)
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 3:
raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
return {"item_id": item_id}
访问 /items/foo
,可以看到以下内容替换了默认 JSON 错误信息:
{
"detail": [
{
"loc": [
"path",
"item_id"
],
"msg": "value is not a valid integer",
"type": "type_error.integer"
}
]
}
以下是文本格式的错误信息:
1 validation error for Request
path -> item_id
value is not a valid integer (type=type_error.integer)
更多异常处理:
https://fastapi.tiangolo.com/zh/tutorial/handling-errors/
标签
from typing import Set, Union
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
tags: Set[str] = set()
@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):
return item
@app.get("/items/", tags=["items"])
async def read_items():
return [{"name": "Foo", "price": 42}]
@app.get("/users/", tags=["users"])
async def read_users():
return [{"username": "johndoe"}]
路由参数
summary 和 description 参数
from typing import Set, Union
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
tags: Set[str] = set()
@app.post(
"/items/",
response_model=Item,
summary="Create an item",
description="Create an item with all the information, name, description, price, tax and a set of unique tags",
)
async def create_item(item: Item):
return item
json转换
在某些情况下,您可能需要将数据类型(如Pydantic模型)转换为与JSON兼容的数据类型(如dict
、list
等)。
比如,如果您需要将其存储在数据库中。
对于这种要求, FastAPI提供了jsonable_encoder()
函数。
3.6及以上
from datetime import datetime
from typing import Union
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
fake_db = {}
class Item(BaseModel):
title: str
timestamp: datetime
description: Union[str, None] = None
app = FastAPI()
@app.put("/items/{id}")
def update_item(id: str, item: Item):
json_compatible_item_data = jsonable_encoder(item)
fake_db[id] = json_compatible_item_data
3.10及以上
from datetime import datetime
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
fake_db = {}
class Item(BaseModel):
title: str
timestamp: datetime
description: str | None = None
app = FastAPI()
@app.put("/items/{id}")
def update_item(id: str, item: Item):
json_compatible_item_data = jsonable_encoder(item)
fake_db[id] = json_compatible_item_data
中间件
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request) # 获取响应
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
cors跨域
allow_origins
- 一个允许跨域请求的源列表。例如['https://example.org', 'https://www.example.org']
。你可以使用['*']
允许任何源。allow_origin_regex
- 一个正则表达式字符串,匹配的源允许跨域请求。例如'https://.*\.example\.org'
。allow_methods
- 一个允许跨域请求的 HTTP 方法列表。默认为['GET']
。你可以使用['*']
来允许所有标准方法。allow_headers
- 一个允许跨域请求的 HTTP 请求头列表。默认为[]
。你可以使用['*']
允许所有的请求头。Accept
、Accept-Language
、Content-Language
以及Content-Type
请求头总是允许 CORS 请求。allow_credentials
- 指示跨域请求支持 cookies。默认是False
。另外,允许凭证时allow_origins
不能设定为['*']
,必须指定源。expose_headers
- 指示可以被浏览器访问的响应头。默认为[]
。max_age
- 设定浏览器缓存 CORS 响应的最长时间,单位是秒。默认为600
。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = [
"http://localhost.tiangolo.com",
"https://localhost.tiangolo.com",
"http://localhost",
"http://localhost:8080",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
async def main():
return {"message": "Hello World"}
websocket
pip install websockets
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<form action="" onsubmit="sendMessage(event)">
<input type="text" id="messageText" autocomplete="off"/>
<button>Send</button>
</form>
<ul id='messages'>
</ul>
<script>
var ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
};
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
from typing import Union
from fastapi import (
Cookie,
Depends,
FastAPI,
Query,
WebSocket,
WebSocketException,
status,
)
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<form action="" onsubmit="sendMessage(event)">
<label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label>
<label>Token: <input type="text" id="token" autocomplete="off" value="some-key-token"/></label>
<button onclick="connect(event)">Connect</button>
<hr>
<label>Message: <input type="text" id="messageText" autocomplete="off"/></label>
<button>Send</button>
</form>
<ul id='messages'>
</ul>
<script>
var ws = null;
function connect(event) {
var itemId = document.getElementById("itemId")
var token = document.getElementById("token")
ws = new WebSocket("ws://localhost:8000/items/" + itemId.value + "/ws?token=" + token.value);
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
};
event.preventDefault()
}
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
async def get_cookie_or_token(
websocket: WebSocket,
session: Union[str, None] = Cookie(default=None),
token: Union[str, None] = Query(default=None),
):
if session is None and token is None:
raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
return session or token
@app.websocket("/items/{item_id}/ws")
async def websocket_endpoint(
websocket: WebSocket,
item_id: str,
q: Union[int, None] = None,
cookie_or_token: str = Depends(get_cookie_or_token),
):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(
f"Session cookie or query token value is: {cookie_or_token}"
)
if q is not None:
await websocket.send_text(f"Query parameter q is: {q}")
await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")
标签:None,FastAPI,app,item,str,fastapi,import
From: https://www.cnblogs.com/yangyucai/p/17015818.html