在現代網頁應用程式開發中,高效能的後端 API 設計至關重要。本文將比較 Flask 和 FastAPI 兩種 Python 網頁框架在實作後端 API 的差異,並探討如何利用 SQLAlchemy 和 Alchemical 簡化資料函式庫操作。兩種框架都採用 SQLAlchemy 與資料函式庫互動,但 FastAPI 更進一步利用了非同步特性提升效能。Flask 版本使用 db.session.execute() 執行查詢,並透過 to_dict() 方法將模型物件序列化為字典,最後以列表推導式組合資料。FastAPI 則使用 async with db.Session() 建立非同步資料函式庫會話,並以 await session.stream() 非同步執行查詢,再以非同步列表推導式處理結果,充分展現非同步操作的優勢。兩種框架的路由定義方式也有所不同,FastAPI 使用 APIRouter 提供更簡潔的路由定義方式,而 Flask 則使用 Blueprint。
使用JSON處理伺服器回應資料
當伺服器傳回包含特定內容的字串時,可以將其解碼為方便的格式進行處理。以在瀏覽器中執行的JavaScript為例,JSON.parse()函式能夠將JSON格式的資料轉換為根據JavaScript物件、陣列和基本資料型別的結構。
仔細觀察上述範例的結構,您會發現這種特定的表示方式不僅包含了產品資訊,還包含了與其相關的一些關係,例如製造商和原產國,這些資訊以JSON表示形式嵌入到父實體中。
建立遞迴包含關係的序列化系統
建立一個能夠輕鬆生成遞迴包含關係的回應的序列化系統其實並不困難。基本想法是為每個模型類別新增一個to_dict()方法,該方法傳回物件的字典版本,可以序列化為JSON格式。
以下以Product模型的to_dict()方法實作為例:
class Product(Model):
# ...
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'manufacturer': self.manufacturer.to_dict(),
'year': self.year,
'cpu': self.cpu,
'countries': [country.to_dict() for country in self.countries],
}
內容解密:
to_dict()方法將Product物件轉換為字典格式,以便進行JSON序列化。self.manufacturer.to_dict()遞迴地將製造商物件轉換為字典格式,並嵌入到產品字典中。[country.to_dict() for country in self.countries]同樣遞迴地處理原產國的列表,將每個國家物件轉換為字典格式。- 這種設計確保了序列化邏輯集中在單一位置,便於維護。
對於大型和複雜的模型,使用專門的序列化函式庫(如Marshmallow)可能是一個更好的選擇。
Alchemical套件簡化SQLAlchemy的使用
在使用SQLAlchemy時,每個專案都需要建立幾個關鍵物件:
- 一個
Engine例項。 - 一個具有明確索引和約束命名規範的
MetaData例項。 - 一個模型宣告基礎類別。
- 一個與引擎相關聯的
Session基礎類別。
使用Alchemical簡化組態
Alchemical套件(由本文作者建立和維護)試圖透過將上述所有專案封裝到一個Alchemical例項中來簡化SQLAlchemy的使用。您可以使用pip安裝Alchemical:
(venv) $ pip install alchemical
考慮以下範例,它提供了與db.py模組相同的功能:
import os
from dotenv import load_dotenv
from alchemical import Alchemical
load_dotenv()
db = Alchemical(os.environ['DATABASE_URL'])
內容解密:
Alchemical類別封裝了SQLAlchemy所需的多個關鍵元件,使得初始化變得更加簡單。db物件包含了所有必要的SQLAlchemy專案,可以透過其屬性或方法存取。db.get_engine()用於取得內部管理的引擎例項。db.metadata提供了對MetaData例項的存取。db.create_all()和db.drop_all()方法分別用於建立和銷毀資料函式庫表格。db.Model是宣告基礎類別,用於定義模型。db.Session是會話基礎類別,用於建立會話。db.begin()上下文管理器簡化了會話和事務的管理。
Alchemical的額外功能
- 支援多資料函式庫連線。
- 簡化了Alembic資料函式庫遷移的支援。
- 非同步支援(當從
alchemical.aio模組匯入時)。 - 與Flask網頁框架的整合(當從
alchemical.flask模組匯入時)。
範例網頁應用程式
本文的GitHub倉函式庫中包含了一個完整的小型網頁應用程式範例,該程式展示了一個RetroFun訂單表格,支援高效的分頁、排序、搜尋和資訊提示。
表格前端實作
範例應用程式中的表格使用了grid.js函式庫。該表格組態為「伺服器端」模式,這意味著它透過向伺服器發出請求來取得資料。這是組態表格最有效的方式,因為只請求需要顯示的資料。
請求引數與JSON回應結構
當表格需要顯示新資料時,會向伺服器的/api/orders端點發出請求,攜帶以下查詢字串引數:
start: 需要顯示的第一個元素的根據1的索引。length: 表格需要的元素數量。sort: 以逗號分隔的排序列表,每個專案以+或-開頭,表示升序或降序,後面跟隨欄位名稱。search: 使用者在搜尋框中輸入的搜尋字串,如果沒有搜尋請求,則為空字串。
伺服器的/api/orders端點需要接受這些查詢字串引數,根據它們執行資料函式庫查詢,並使用以下JSON結構傳回請求的專案:
{
"data": [
{ ... order ... },
{ ... order ... },
...
],
}
內容解密:
/api/orders端點負責處理來自前端的請求,執行相應的資料函式庫查詢,並以JSON格式傳回結果。- 請求引數用於控制傳回資料的分頁、排序和篩選。
- JSON回應結構中的
data欄位包含了符合請求條件的訂單資料。
資料函式庫查詢與API端點實作
在建立一個支援分頁、搜尋及排序功能的訂單管理系統後端時,資料函式庫查詢邏輯與API端點的設計是至關重要的部分。本文將探討如何使用SQLAlchemy實作相關的資料函式庫查詢,以及如何在Flask框架中定義API端點以滿足前端的需求。
資料函式庫查詢邏輯
資料函式庫查詢是後端邏輯的核心部分,主要涉及兩個主要的查詢:計算符合搜尋條件的訂單總數以及擷取特定分頁的訂單資料。
計算訂單總數
計算訂單總數的查詢相對簡單。當沒有搜尋條件時,直接計算訂單表的總筆數;若有搜尋條件,則需要進行多表的join操作,以支援對客戶名稱及產品名稱的搜尋。
def total_orders(search):
if not search:
return sa.select(sa.func.count(Order.id))
return (
sa.select(sa.func.count(sa.distinct(Order.id)))
.join(Order.customer)
.join(Order.order_items)
.join(OrderItem.product)
.where(
sa.or_(
Customer.name.ilike(f'%{search}%'),
Product.name.ilike(f'%{search}%'),
)
)
)
擷取分頁訂單資料
擷取分頁訂單資料的查詢則複雜許多,需要支援搜尋、排序及分頁功能。
def paginated_orders(start, length, sort, search):
total = sa.func.sum(OrderItem.quantity * OrderItem.unit_price).label(None)
q = (
sa.select(Order, total)
.join(Order.customer)
.join(Order.order_items)
.join(OrderItem.product)
.group_by(Order)
.distinct()
)
# 新增搜尋條件
if search:
q = q.where(
sa.or_(
Customer.name.ilike(f'%{search}%'),
Product.name.ilike(f'%{search}%'),
)
)
# 新增排序邏輯
if sort:
order = []
for s in sort.split(','):
direction = s[0]
name = s[1:]
if name == 'customer':
column = Customer.name
elif name == 'total':
column = total
else:
column = getattr(Order, name)
if direction == '-':
column = column.desc()
order.append(column)
if not order:
order = [Order.timestamp.desc()]
q = q.order_by(*order)
# 新增分頁邏輯
q = q.offset(start).limit(length)
return q
API端點實作
在Flask框架中,API端點透過Blueprint進行定義。以下展示了兩個主要的端點:根URL(/)及訂單API(/api/orders)。
from flask import Blueprint, render_template, request
from .models import db
from . import queries
bp = Blueprint('routes', __name__)
@bp.route('/')
def index():
return render_template('index.html')
@bp.route('/api/orders')
def get_orders():
start = request.args.get('start')
length = request.args.get('length')
sort = request.args.get('sort')
search = request.args.get('search')
data_query = queries.paginated_orders(start, length, sort, search)
total_query = queries.total_orders(search)
orders = db.session.execute(data_query)
data = [{**o[0].to_dict(), 'total': o[1]} for o in orders]
return {
'data': data,
'total': db.session.scalar(total_query),
}
內容解密:
total_orders函式的作用:此函式根據是否存在搜尋字串,傳回計算訂單總數的SQLAlchemy查詢。若無搜尋條件,直接計算訂單總數;若有,則進行多表join以支援對客戶名稱及產品名稱的搜尋。paginated_orders函式的實作邏輯:此函式構建了一個支援搜尋、排序及分頁的查詢。首先,定義了一個根據多表join的基礎查詢;接著,根據搜尋字串新增where條件;然後,根據排序引數動態構建排序邏輯;最後,加上分頁限制。- Flask端點的定義:使用Blueprint定義了兩個端點。根URL傳回包含前端JavaScript程式碼的HTML頁面;
/api/orders端點根據請求引數生成並執行資料函式庫查詢,傳回符合前端要求的JSON回應。
重點整理:
- 資料函式庫查詢邏輯分為計算訂單總數及擷取分頁資料兩部分。
total_orders函式處理計算符合搜尋條件的訂單總數。paginated_orders函式處理分頁、搜尋及排序邏輯。- Flask中的API端點透過Blueprint定義,包含根URL及
/api/orders。 /api/orders端點負責執行資料函式庫查詢並傳回JSON格式的回應。
FastAPI 與 Flask 後端實作比較
本章節將探討如何使用 FastAPI 和 Flask 實作後端服務,並對兩者的實作細節進行比較。以下程式碼片段需要仔細研究,以全面理解其工作原理:
資料查詢與處理
在資料查詢方面,兩種框架都使用了 SQLAlchemy 來與資料函式庫互動。以下為 Flask 版本的資料處理範例:
orders = db.session.execute(data_query)
data = [{**o[0].to_dict(), 'total': o[1]} for o in orders]
內容解密:
db.session.execute(data_query):執行資料查詢並將結果儲存在orders變數中。o[0].to_dict():將查詢結果中的Order模型序列化為字典格式。'total': o[1]:將查詢結果中的第二個值(總計)加入到字典中。- 使用列表推導式將所有結果轉換為包含
total屬性的字典列表。
FastAPI 路由實作
FastAPI 的路由定義在 router.py 模組中。以下為相關程式碼:
from fastapi import APIRouter
from fastapi.responses import FileResponse
from .models import db
from . import queries
router = APIRouter()
@router.get('/')
async def index():
return FileResponse('retrofun/html/index.html')
@router.get('/api/orders')
async def get_orders(start: int, length: int, sort: str = '', search: str = ''):
data_query = queries.paginated_orders(start, length, sort, search)
total_query = queries.total_orders(search)
async with db.Session() as session:
orders = await session.stream(data_query)
data = [{**o[0].to_dict(), 'total': o[1]} async for o in orders]
return {
'data': data,
'total': await session.scalar(total_query),
}
內容解密:
@router.get('/'):定義根 URL 的處理函式,傳回前端 HTML 檔案。@router.get('/api/orders'):定義/api/orders端點的處理函式,支援分頁、排序和搜尋功能。async with db.Session() as session:使用非同步方式建立資料函式庫會話。await session.stream(data_query):非同步執行資料查詢並傳回結果流。async for o in orders:使用非同步列表推導式處理查詢結果。return {'data': data, 'total': await session.scalar(total_query)}:傳回包含資料和總計的 JSON 回應。
Flask 後端實作
Flask 後端的實作主要涉及應用程式的初始化和路由定義。以下是專案結構和關鍵程式碼片段:
專案結構
- main.py # 應用程式入口點
- config.py # Flask 組態變數
- retrofun # 應用程式邏輯的 Python 包
- __init__.py # 包初始化(載入環境變數)
- app.py # 應用程式工廠函式
- models.py # Alchemical 資料函式庫例項和模型定義
- queries.py # 支援訂單表格分頁的資料函式庫查詢
- routes.py # 應用程式的 Flask Blueprint
- templates # Flask 範本目錄
- index.html # 應用程式的 HTML 頁面
- migrations # Alembic 資料函式庫遷移目錄
- alembic.ini # Alembic 組態檔案
- .flaskenv # Flask 特定的環境變數
- .env.template # 環境變數範本
- requirements.txt # 應用程式依賴項
資料函式庫初始化
在 models.py 中,使用 Alchemical 初始化資料函式庫:
from alchemical.flask import Alchemical
db = Alchemical()
內容解密:
- 從
alchemical.flask包匯入Alchemical類別。 - 建立
db物件作為 Alchemical 資料函式庫例項。
應用程式工廠函式
在 app.py 中,使用工廠函式模式初始化 Flask 應用程式例項:
from flask import Flask
from .models import db
def create_app():
app = Flask(__name__)
db.init_app(app)
return app
內容解密:
- 建立 Flask 應用程式例項。
- 使用
db.init_app(app)初始化 Alchemical 資料函式庫擴充功能。