FastAPI -- 第三弹(自定义响应、中间件、代理、WebSockets)
路径操作的高级配置
OpenAPI 的 operationId
from fastapi import FastAPI
app = FastAPI()
# 通过 operation_id 参数设置
@app.get("/items/", operation_id="some_specific_id_you_define")
async def read_items():
return [{"item_id": "Foo"}]
使用 路径操作函数 的函数名作为 operationId
from fastapi import FastAPI
from fastapi.routing import APIRoute
app = FastAPI()
@app.get("/items/")
async def read_items():
return [{"item_id": "Foo"}]
def use_route_names_as_operation_ids(app: FastAPI) -> None:
"""
Simplify operation IDs so that generated API clients have simpler function
names.
Should be called only after all routes have been added.
"""
for route in app.routes:
if isinstance(route, APIRoute):
# 个人觉得这种操作挺不错，可以在一个地方统一处理，更有利于建立规范 且方便管理
route.operation_id = route.name # in this case, 'read_items'
use_route_names_as_operation_ids(app)
从 OpenAPI 中排除
from fastapi import FastAPI
app = FastAPI()
# 使用 include_in_schema 参数并将其设置为 False，
# openapi 文档中就看不到这个接口的信息
#
# 个人认为，可以作为是否完成的开关使用，默认设置 False，当接口完成开发后设置为 True
@app.get("/items/", include_in_schema=False)
async def read_items():
return [{"item_id": "Foo"}]
docstring 的高级描述
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")
async def create_item(item: Item):
"""
Create an item with all the information:
- **name**: each item must have a name
- **description**: a long description
- **price**: required
- **tax**: if the item doesn't have tax, you can omit this
- **tags**: a set of unique tag strings for this item
> \\f ： 换页 的转义字符 1 <br/>
> 比较有意思的是，这里可以按照 Markdown 的语法写文档注释
> Swagger UI 只支持一些简单的 Markdown 语法
> 比如现在的 `>` Swagger UI就不支持，但是 ReDoc 支持
> 但是比较遗憾的是 这两者对 Markdown 的语法支持的不够完善
但是比较遗憾的是 这两者都不支持换行
但是比较遗憾的是 这两者都不支持换行
但是比较遗憾的是 这两者都不支持换行
`\n` 和 `<br/>` 两者的换行还不一样
对于 `>` 中的换行 要使用 `<br/>`, 不然会作为两段 引用，
只能说对于 Markdown语法的支持两者都有提升空间啊
\f
\\f ： 换页 的转义字符 2
换页的内容，不会再 上面提到的两者中 进行展示
比较有意思的是，这里可以按照 Markdown 的语法写文档注释
然后在文档中就会按照 Markdown 的样式进行展示
:param item: User input.
"""
return item
自定义响应
使用 ORJSONResponse
如果你需要压榨性能，你可以安装并使用
orjson并将响应设置为ORJSONResponse
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
app = FastAPI()
# 使用 ORJSONResponse 代替 JSONResponse
@app.get("/items/", response_class=ORJSONResponse)
async def read_items():
return ORJSONResponse([{"item_id": "Foo"}])
StreamingResponse
如果您有类似文件的对象（例如，由 open() 返回的对象），则可以在 StreamingResponse 中将其返回
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
some_file_path = "large-video-file.mp4"
app = FastAPI()
@app.get("/")
def main():
def iterfile(): # (1)
with open(some_file_path, mode="rb") as file_like: # (2)
yield from file_like # (3)
# 使用 StreamingResponse 返回数据
return StreamingResponse(iterfile(), media_type="video/mp4")
FileResponse
异步传输文件作为响应。
from fastapi import FastAPI
from fastapi.responses import FileResponse
some_file_path = "large-video-file.mp4"
app = FastAPI()
@app.get("/")
async def main():
return FileResponse(some_file_path)
响应 Cookies
使用 Response 参数
app = FastAPI()
@app.post("/cookie-and-object/")
def create_cookie(response: Response):
# 定义一个 response: Response 的参数
# 调用 set_cookie
response.set_cookie(key="fakesession", value="fake-cookie-session-value")
return {"message": "Come to the dark side, we have cookies"}
直接响应 Response
from fastapi import FastAPI
from fastapi.responses import JSONResponse
app = FastAPI()
@app.post("/cookie/")
def create_cookie():
content = {"message": "Come to the dark side, we have cookies"}
# 创建一个 Response 或其子类 对象
response = JSONResponse(content=content)
# 调用 set_cookie
response.set_cookie(key="fakesession", value="fake-cookie-session-value")
return response
响应头
和 响应Cookies 类似也有两种方式
使用 Response 参数
from fastapi import FastAPI, Response
app = FastAPI()
@app.get("/headers-and-object/")
def get_headers(response: Response):
# 声明一个 Response 类的形参
# 像字典添加 键值对 一样，往 headers中添加键值对
response.headers["X-Cat-Dog"] = "alone in the world"
return {"message": "Hello World"}
直接返回 Response
from fastapi import FastAPI
from fastapi.responses import JSONResponse
app = FastAPI()
@app.get("/headers/")
def get_headers():
content = {"message": "Hello World"}
# 将要返回的响应头信息先存到字典中 header_dict
header_dict = {"X-Cat-Dog": "alone in the world", "Content-Language": "en-US"}
# 通过 kv 形式设置 headers， headers=header_dict
return JSONResponse(content=content, headers=header_dict )
直接使用请求
from fastapi import FastAPI, Request
app = FastAPI()
@app.get("/items/{item_id}")
def read_root(item_id: str, request: Request):
# 声明一个形参 request: Request
# 可以从 request 中得到请求信息
client_host = request.client.host
return {"client_host": client_host, "item_id": item_id}
添加 ASGI 中间件
from fastapi import FastAPI
from unicorn import UnicornMiddleware
app = FastAPI()
# 通过 FastAPI 的实例 app 调用 add_middleware 方法添加中间件
app.add_middleware(UnicornMiddleware, some_config="rainbow")
使用代理
在 FastAPI 应用里设置 root_path
from fastapi import FastAPI, Request
# 将 "/api/v1" 赋值给 root_path
app = FastAPI(root_path="/api/v1")
@app.get("/app")
def read_main(request: Request):
return {"message": "Hello World", "root_path": request.scope.get("root_path")}
附加的服务器（多环境部署时）
from fastapi import FastAPI, Request
# 可以通过 servers，将多个环境的地址以列表的形式赋值给 servers
app = FastAPI(
servers=[
{"url": "https://stag.example.com", "description": "Staging environment"},
{"url": "https://prod.example.com", "description": "Production environment"},
],
root_path="/api/v1",
# 将 root_path 从 servers 剔除，root_path_in_servers=False
# 默认 servers 会包含 root_path， root_path_in_servers=True
# root_path_in_servers=False,
)
@app.get("/app")
def read_main(request: Request):
return {"message": "Hello World", "root_path": request.scope.get("root_path")}
WebSockets
安装 WebSockets
pip install websockets
创建 websocket
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="" οnsubmit="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 创建一个 websocket
@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}")
事件：启动 - 关闭
startup 事件
from fastapi import FastAPI
app = FastAPI()
items = {}
# 通过 app.on_event("startup") ”声明“一个函数为启动函数
# 可以做一些预处理工作，比如 往数据库中写入一些初始数据
@app.on_event("startup")
async def startup_event():
items["foo"] = {"name": "Fighters"}
items["bar"] = {"name": "Tenders"}
@app.get("/items/{item_id}")
async def read_items(item_id: str):
return items[item_id]
shutdown 事件
from fastapi import FastAPI
app = FastAPI()
# 通过 app.on_event("shutdown") ”声明“一个函数为关闭函数
@app.on_event("shutdown")
def shutdown_event():
with open("log.txt", mode="a") as log:
log.write("Application shutdown")
@app.get("/items/")
async def read_items():
return [{"name": "Foo"}]
到此结 DragonFangQy 2024.07.18
原文地址：https://blog.csdn.net/f704084109/article/details/140417071
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！