【FastAPI】--3.进阶教程(二)

【FastAPI】--进阶教程1-CSDN博客

【FastAPI】--基础教程-CSDN博客

1.FastAPI - CORS

2.FastAPI - CRUD 操作

2.1.Create

2.2.Read

2.3.Update

2.4.Delete

3.FastAPI - 使用 GraphQL

4.FastAPI - Websockets

5.FastAPI - 事件处理程序

6.FastAPI - 安装 Flask 应用程序

1.FastAPI - CORS

Cross-Origin Resource Sharing (CORS) 是指在一个客户端浏览器上运行的前端应用程序试图通过 JavaScript 代码与后端通信，而后端与前端位于不同的"来源"的情况。这里的来源是协议、域名和端口号的组合。因此，http://localhost 和 https://localhost 的来源不同。

前端运行在 http://localhost:3000，后端运行在 http://localhost:8000。
浏览器会阻止前端直接访问后端 API，除非后端配置了 CORS。

要明确指定允许的来源，请导入 CORSMiddleware 并将来源列表添加到应用程序的中间件。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = ["http://192.168.211.:8000","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"}

以下是 FastAPI app.add_middleware(CORSMiddleware) 参数 的详细表格说明，涵盖所有核心配置选项及其用途、示例和注意事项：

CORS 中间件参数完整表格

参数	类型	说明
`allow_origins`	`List[str]`	允许访问的源（域名列表）。
`allow_origin_regex`	`str`	用正则表达式匹配允许的源。
`allow_methods`	`List[str]`	允许的 HTTP 方法。
`allow_headers`	`List[str]`	允许的请求头。
`allow_credentials`	`bool`	是否允许跨域携带 Cookie（如 `Authorization: Bearer`）。
`expose_headers`	`List[str]`	允许前端 JavaScript 访问的响应头。
`max_age`	`int`	预检请求（OPTIONS）的缓存时间（秒）。

2.FastAPI - CRUD 操作

CRUD（Create, Read, Update, Delete）是 Web 开发中最基础的操作。FastAPI 实现 CRUD 的完整示例，包含 路由设计、请求验证、数据库交互 和 错误处理。

（创、读、更新、删除）

REST 架构使用 HTTP 动词或方法来对资源进行操作。 POST、GET、PUT和DELETE方法分别执行CREATE、READ、UPDATE和DELETE操作。

2.1.Create

在下面的示例中，我们将使用 Python 列表作为内存数据库并对其执行 CRUD 操作。首先，让我们设置一个 FastAPI 应用程序对象并声明一个名为 Book 的 Pydantic 模型。

from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
data = []
class Book(BaseModel):id: inttitle: strauthor: strpublisher: str#出版社@app.post("/book")
def add_book(book: Book):data.append(book.dict())return dataif __name__ == "__main__":import uvicornuvicorn.run(app='__main__:app', host='127.0.0.1', port=8000, reload=True)

在Swagger UI中，执行几次这个操作函数并添加一些数据。

2.2.Read

要检索列表，定义绑定到 @app.get() 装饰器的操作函数，如下所示 −

@app.get("/list")
def get_books():return data

要检索一本书，其 id 作为路径参数，定义 get() 操作装饰器和 get_book() 函数如下 −

@app.get("/book/{id}")
def get_book(id: int):id = id - 1return data[id]

2.3.Update

接下来，定义修改数据列表中对象的@app.put()装饰器。这个装饰器也有一个 id 字段的路径参数。

在 FastAPI 中，@app.put() 用于定义 HTTP PUT 请求 的路由，通常用于 更新资源。

@app.put("/book/{id}")
def add_book(id: int, book: Book):data[id] = bookreturn data

2.4.Delete

最后，我们定义了@app.delete() 装饰器，用于删除路径参数对应的一个对象。

@app.delete("/book/{id}")
def delete_book(id: int):data.pop(id-1)return data

3.FastAPI - 使用 GraphQL

GraphQL 是一种用于 API 的 查询语言 和 运行时，由 Facebook 于 2012 年开发并在 2015 年开源。它提供了一种更高效、灵活且强类型的方式来与 API 交互，尤其适合现代复杂应用（如 React/Vue 前端、移动端等）。

由于 GraphQL 与 ASGI 兼容，因此可以轻松地与 FastAPI 应用程序集成。 GraphQL 有很多 Python 库。下面列出了其中一些 −

Strawberry

Ariadne

Tartiflette

Graphene

# 导入Strawberry库（用于构建GraphQL服务）
import strawberry# 定义一个Book类型（GraphQL对象类型）
@strawberry.type
class Book:title: str      # 书名（字符串类型）author: str     # 作者（字符串类型）price: int      # 价格（整数类型）# 定义查询入口点
@strawberry.type
class Query:# 定义一个查询字段book，返回Book类型@strawberry.fielddef book(self) -> Book:# 返回一个固定的Book对象实例return Book(title="Computer Fundamentals", author="Sinha", price=300)# 导入FastAPI（用于创建Web服务）
from fastapi import FastAPI
# 导入Strawberry的ASGI适配器
from strawberry.asgi import GraphQL# 创建GraphQL Schema（模式），指定查询入口为Query类
schema = strawberry.Schema(query=Query)# 创建GraphQL应用实例
graphql_app = GraphQL(schema)# 创建FastAPI应用实例
app = FastAPI()# 添加HTTP路由（处理GraphQL查询）
app.add_route("/book", graphql_app)
# 添加WebSocket路由（处理GraphQL订阅）
app.add_websocket_route("/book", graphql_app)# 主程序入口
if __name__ == "__main__":import uvicorn# 使用uvicorn运行FastAPI应用uvicorn.run(app='__main__:app',  # 指定应用对象host='127.0.0.1',    # 监听本地地址port=8000,           # 使用8000端口reload=True          # 开发模式：代码修改后自动重启)

http://127.0.0.1:8000/book

4.FastAPI - Websockets

WebSocket 是客户端和服务器之间的持久连接，用于在两者之间提供双向、全双工 通信。通过单个 TCP/IP 套接字连接在 HTTP 上进行通信。它可以看作是 HTTP 的升级，而不是一个协议本身。

HTTP 的局限性之一是它是一种严格的半双工或单向协议。另一方面，使用 WebSockets，我们可以发送基于消息的数据，与 UDP 类似，但具有 TCP 的可靠性。 WebSocket 使用 HTTP 作为初始传输机制，但在收到 HTTP 响应后保持 TCP 连接。同一个连接对象，它可以用于客户端和服务器之间的双向通信。因此，可以使用 WebSocket API 构建实时应用程序。

FastAPI 通过 FastAPI 模块中的 WebSocket 类支持 WebSockets。以下示例演示了 WebSocket 在 FastAPI 应用程序中的功能。

1.首先我们有一个index() 函数来渲染一个模板(socket.html)。它绑定到"/"路由。 HTML 文件 socket.html 位于"templates"文件夹中。

main.py

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
templates = Jinja2Templates(directory="templates")#模板目录
from fastapi.staticfiles import StaticFiles
app = FastAPI()
app.mount("/static", StaticFiles(directory="./templates/static"), name="static")#静态文件目录
@app.get("/", response_class=HTMLResponse)
async def index(request: Request):return templates.TemplateResponse("socket.html", {"request": request})

socket.html

<!DOCTYPE html>
<html><head><title>Chat</title><script src="{{ url_for('static', path='ws.js') }}"></script></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></body>
</html>

在 socket.html 中，调用了在表单提交时执行的 JavaScript 函数。因此，为了服务 JavaScript，首先安装"static"文件夹。 JavaScript 文件 ws.js 位于"static"文件夹中。

ws.js

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()
}

加载 JavaScript 代码时，它会创建一个监听"ws://localhost:8000/ws"的 websocket。 sendMessage() 函数将输入消息定向到 WebSocket URL。

此路由调用应用程序代码中的 websocket_endpoint() 函数。传入的连接请求被接受，传入的消息在客户端浏览器上回显。将以下代码添加到 main.py。

from fastapi import 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}")

整个路由结构：

5.FastAPI - 事件处理程序

事件处理程序是当某个确定的事件发生时要执行的函数。在 FastAPI 中，标识了两个这样的事件 −

startup
shutdown。

FastAPI 的应用程序对象具有 on_event() 装饰器，它使用其中一个事件作为参数。当相应的事件发生时，使用此装饰器注册的函数将被触发。

这是启动和关闭事件处理程序的简单示例。当应用程序启动时，开始时间会在控制台日志中回显。同样，当服务器按ctrl+c停止时，也会显示关机时间。

from fastapi import FastAPI
import datetime
app = FastAPI()
@app.on_event("startup")
async def startup_event():print('Server started :', datetime.datetime.now())
@app.on_event("shutdown")
async def shutdown_event():print('server Shutdown :', datetime.datetime.now())if __name__ == '__main__':import uvicornuvicorn.run(app='__main__:app', host='127.0.0.1', port=8000, reload=True)

6.FastAPI - 安装 Flask 应用程序

用 Flask 或 Django 框架编写的 WSGI 应用程序可以包装在 WSGIMiddleware 中，并将其安装在 FastAPI 应用程序上以使其符合 ASGI 标准。

from flask import Flask
import uvicorn
from fastapi.middleware.wsgi import WSGIMiddleware
from fastapi import FastAPIflask_app = Flask(__name__)
@flask_app.route("/")
def index_flask():return "Hello World from Flask!"app = FastAPI()
@app.get("/")
def index():return {"message": "Hello World from FastAPI!"}app.mount("/flask", WSGIMiddleware(flask_app))if __name__ == '__main__':uvicorn.run(app='__main__:app', host='127.0.0.1', port=8000, reload=True)

使用 mount() 方法将 flask 应用程序挂载为 FastAPI 主应用程序的子应用程序。