【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 主應用程序的子應用程序。