FastAPI 作為高效能的 Python Web 框架，搭配 SQLAlchemy ORM 能夠簡化資料函式庫操作流程。本文除了介紹 SQLAlchemy ORM 的基本用法外，也探討 SQLAlchemy Core 的應用，讓開發者能更彈性地控制 SQL 查詢。此外，文章也涵蓋了使用 PyMongo 操作 MongoDB 的方法，提供更全面的資料函式庫互動方案。透過整合這些技術，開發者可以更有效率地建構資料函式庫驅動的 Web 應用，並提升整體效能。文章中也包含了實務上的程式碼範例，讓讀者更容易理解並應用到實際專案中。
使用SQLAlchemy與FastAPI進行資料函式庫操作
在現代Web開發中，資料函式庫的操作是不可或缺的一部分。FastAPI作為一個高效能的Web框架，支援與SQLAlchemy這類別ORM（Object-Relational Mapping）工具的整合，能夠大幅簡化資料函式庫的操作流程。本章將探討如何結合FastAPI與SQLAlchemy進行資料函式庫的CRUD（Create, Read, Update, Delete）操作。
SQLAlchemy簡介
SQLAlchemy是一個Python SQL工具包和ORM系統，它為應用程式開發人員提供了SQL的全部功能和靈活性。透過SQLAlchemy，開發者可以使用Python程式碼來操作資料函式庫，而無需直接撰寫SQL陳述式。
FastAPI與SQLAlchemy的整合
要將FastAPI與SQLAlchemy結合，首先需要安裝必要的套件。可以使用pip來安裝fastapi、uvicorn（用於執行FastAPI應用程式）以及sqlalchemy。
pip install fastapi uvicorn sqlalchemy
接著，定義資料函式庫連線和模型。以下是一個基本的例子，展示如何定義一個名為Books的資料表：
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 定義資料函式庫連線URL
SQLALCHEMY_DATABASE_URL = "sqlite:///./mydata.sqlite3"
# 建立資料函式庫引擎
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
# 建立SessionLocal類別
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 定義Base類別，用於宣告模型
Base = declarative_base()
class Books(Base):
__tablename__ = "booklist"
id = Column(Integer, primary_key=True)
title = Column(String)
author = Column(String)
price = Column(Integer)
publisher = Column(String)
# 建立資料表
Base.metadata.create_all(bind=engine)
內容解密：
- 資料函式庫連線URL：定義了應用程式如何連線到資料函式庫。在這個例子中，使用的是SQLite資料函式庫。
create_engine函式：根據提供的URL建立一個資料函式庫引擎例項。SessionLocal類別：用於管理資料函式庫會話（session）。declarative_base函式：傳回一個Base類別，用於定義模型。Books模型：定義了名為booklist的資料表結構，包括欄位名稱和型別。
CRUD操作實作
接下來，將展示如何使用FastAPI和SQLAlchemy進行CRUD操作。
1. 新增（Create）
from fastapi import FastAPI, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
app = FastAPI()
class Book(BaseModel):
title: str
author: str
price: int
publisher: str
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/books", response_model=Book)
def add_book(b1: Book, db: Session = Depends(get_db)):
bkORM = Books(**b1.dict())
db.add(bkORM)
db.commit()
db.refresh(bkORM)
return bkORM
內容解密：
add_book函式：處理新增書籍的請求。它接收一個Book物件，將其轉換為Books模型例項，並儲存到資料函式庫中。get_db函式：提供一個資料函式庫會話（session）給路徑操作函式使用。
2. 讀取（Read）
@app.get("/books", response_model=List[Book])
def get_books(db: Session = Depends(get_db)):
recs = db.query(Books).all()
return recs
@app.get("/books/{id}", response_model=Book)
def get_book(id: int, db: Session = Depends(get_db)):
return db.query(Books).filter(Books.id == id).first()
內容解密：
get_books函式：傳回所有書籍的列表。get_book函式：根據提供的ID傳回單一書籍的詳細資訊。
3. 更新（Update）
@app.put("/books/{id}", response_model=Book)
def update_book(id: int, price: int = Body(), db: Session = Depends(get_db)):
bkORM = db.query(Books).filter(Books.id == id).first()
bkORM.price = price
db.commit()
return bkORM
內容解密：
update_book函式：更新指定ID的書籍價格。
4. 刪除（Delete）
@app.delete("/books/{id}")
def del_book(id: int, db: Session = Depends(get_db)):
try:
db.query(Books).filter(Books.id == id).delete()
db.commit()
except Exception as e:
raise Exception(e)
return "book deleted successfully"
內容解密：
del_book函式：刪除指定ID的書籍。
使用SQLAlchemy Core進行非同步操作
SQLAlchemy Core提供了另一種方式來與資料函式庫互動，支援非同步操作。首先，需要安裝databases套件：
pip install databases
然後，可以使用以下程式碼建立非同步的資料函式庫連線和操作：
import databases
import sqlalchemy
DATABASE_URL = "sqlite:///./mydata.sqlite3"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
booklist = sqlalchemy.Table(
"booklist",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("title", sqlalchemy.String),
sqlalchemy.Column("author", sqlalchemy.String),
sqlalchemy.Column("price", sqlalchemy.Integer),
sqlalchemy.Column("publisher", sqlalchemy.String),
)
# 非同步連線資料函式庫
async def connection():
await database.connect()
# 使用SQLAlchemy Core進行CRUD操作
@app.post("/books")
async def add_book(b1: Book):
query = booklist.insert().values(title=b1.title, author=b1.author, price=b1.price, publisher=b1.publisher)
last_record_id = await database.execute(query)
return {**b1.dict(), "id": last_record_id}
內容解密：
- 非同步連線：使用
databases套件建立非同步的資料函式庫連線。 - SQLAlchemy Core：直接使用SQLAlchemy Core來定義資料表和進行CRUD操作。
使用SQLAlchemy Core進行資料函式庫操作
在前面的章節中，我們已經瞭解瞭如何使用SQLAlchemy Core來與資料函式庫進行互動。本文將探討如何使用SQLAlchemy Core來執行常見的資料函式庫操作，包括新增、查詢、更新和刪除。
Table類別的方法
SQLAlchemy Core提供了一個Table類別，用於代表資料函式庫中的表格。這個類別提供了多種方法來執行資料函式庫操作。
查詢方法
Table類別提供了多種查詢方法，包括select()、fetch_all()和fetch_one()。
query = table.select().where(condition)
rows = db.fetch_all(query)
row = db.fetch_one(query)
在上述程式碼中，table.select().where(condition)用於生成一個查詢物件，db.fetch_all(query)用於執行查詢並傳回所有結果，db.fetch_one(query)用於執行查詢並傳回第一個結果。
更新方法
update()方法用於更新資料函式庫中的資料。
query = table.update().where(condition).values(field1=value1, ...)
db.execute(query)
在上述程式碼中，table.update().where(condition).values(field1=value1, ...)用於生成一個更新查詢物件，db.execute(query)用於執行更新查詢。
刪除方法
delete()方法用於刪除資料函式庫中的資料。
query = table.delete().where(condition)
db.execute(query)
在上述程式碼中，table.delete().where(condition)用於生成一個刪除查詢物件，db.execute(query)用於執行刪除查詢。
FastAPI路徑操作
在FastAPI中，我們可以使用SQLAlchemy Core來執行資料函式庫操作。下面的範例展示瞭如何使用SQLAlchemy Core來執行新增、查詢、更新和刪除操作。
新增操作
@app.post("/books", response_model=Book)
async def add_book(b1: Book, db=Depends(get_db)):
query = booklist.insert().values(id=b1.id, title=b1.title, author=b1.author, price=b1.price, publisher=b1.publisher)
await db.execute(query)
return "Book added successfully"
查詢操作
@app.get("/books", response_model=List[Book])
async def get_books(db=Depends(get_db)):
query = booklist.select()
return await db.fetch_all(query)
@app.get("/books/{id}")
async def get_book(id: int, db=Depends(get_db)):
query = booklist.select().where(booklist.c.id == id)
return await db.fetch_one(query)
更新操作
@app.put("/books/{id}")
async def update_book(id: int, new_price: int = Body(), db=Depends(get_db)):
query = booklist.update().where(booklist.c.id == id).values(price=new_price)
await db.execute(query)
return "Book updated successfully"
刪除操作
@app.delete("/books/{id}")
async def del_book(id: int, db=Depends(get_db)):
query = booklist.delete().where(booklist.c.id == id)
await db.execute(query)
return "Book deleted successfully"
使用PyMongo進行MongoDB操作
MongoDB是一種檔案導向的NoSQL資料函式庫。PyMongo是MongoDB的官方Python驅動程式。
安裝PyMongo
pip3 install pymongo
連線MongoDB
from pymongo import MongoClient
client = MongoClient("mongodb://localhost:27017/")
建立資料函式庫和集合
db = client["mydatabase"]
collection = db["mycollection"]
在上述程式碼中，client["mydatabase"]用於建立一個名為mydatabase的資料函式庫，db["mycollection"]用於建立一個名為mycollection的集合。
詳細解析FastAPI與資料函式庫互動的核心技術
為什麼選擇SQLAlchemy Core？
在眾多的ORM（Object-Relational Mapping）工具中，SQLAlchemy Core憑藉其靈活性、效能和豐富的功能成為了Python開發者的首選。它不僅支援多種資料函式庫系統，還提供了核心表達語言，使得開發者能夠以程式化的方式構建複雜的SQL查詢。
SQLAlchemy Core與FastAPI的完美結合
FastAPI作為一個現代化的Web框架，其非同步處理能力和自動化檔案生成等特性使其成為構建高效API的理想選擇。當與SQLAlchemy Core結合使用時，開發者可以充分利用兩者的優勢，實作高效、靈活的資料函式庫操作。
非同步資料函式庫操作的實作
在FastAPI中，非同步路徑操作函式可以與SQLAlchemy Core無縫結合，透過使用await關鍵字等待非同步資料函式庫操作的完成。這種模式不僅提高了應用程式的效能，還使得程式碼更加簡潔易讀。
PyMongo與MongoDB的互動
對於需要使用NoSQL資料函式庫的場景，PyMongo提供了與MongoDB互動的直接方式。透過PyMongo，開發者可以輕鬆地進行檔案的增刪改查操作，並且能夠充分利用MongoDB的靈活性和可擴充套件性。
最佳實踐與效能最佳化
在使用SQLAlchemy Core和PyMongo時，遵循最佳實踐對於保證應用程式的效能和可維護性至關重要。這包括合理使用索引、最佳化查詢邏輯、避免不必要的資料函式庫操作等。
隨著資料函式庫技術的不斷發展和Web應用需求的日益複雜，掌握如何有效地將FastAPI與各種資料函式庫技術結合使用，將成為開發者的重要技能。未來，我們可以預見更多高效、靈活的資料函式庫互動技術將被開發出來，以滿足不斷變化的應用需求。
總之，本章節透過介紹SQLAlchemy Core和PyMongo的使用，為讀者提供了一個全面瞭解如何在FastAPI中進行資料函式庫操作的機會。無論是對於初學者還是有經驗的開發者，這些知識都將有助於構建更高效、更可靠的Web應用程式。
