ORM
- FastAPI 可與任何數據庫和任何樣式的庫配合使用並和數據庫通信
- object-relational mapping 對象關系映射
- ORM 具有在代碼和數據庫表（關系）中的對象之間進行轉換（映射）的工具
- 使用 ORM，通常會創建一個表示 SQL 數據表的類，該類的每個屬性都表示一個列，具有名稱和類型
小栗子
- Pet 類可以表示 SQL 表 pets
- 並且 Pet 類的每個實例對象代表數據庫中的一行數據
- 例如，對象 orion_cat（Pet 的一個實例）可以具有屬性 orion_cat.type，用於列類型，屬性的值可以是：貓
項目架構
. └── sql_app ├── __init__.py ├── curd.py ├── database.py ├── main.py ├── models.py └── schemas.py
前提
需要先安裝 sqlalchemy
pip install sqlalchemy
使用 sqlite
- 后面的栗子，暫時跟着官網，先使用 sqlite 數據庫來演示
- 后面有時候再通過 Mysql 來寫多一篇文章
database.py 代碼
# 1、導入 sqlalchemy 部分的包 from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 2、聲明 database url SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" # SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db" # 3、創建 sqlalchemy 引擎 engine = create_engine( url=SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} ) # 4、創建一個 database 會話 session = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 5、返回一個 ORM Model Base = declarative_base()
聲明 database 連接 url
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" # SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
第一行是 slite 連接 url
其他數據庫連接 url 的寫法
# sqlite-pysqlite 庫 sqlite+pysqlite:///file_path # mysql-mysqldb 庫 mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname> # mysql-pymysql 庫 mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] # mysql-mysqlconnector 庫 mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname> # oracle-cx_Oracle 庫 oracle+cx_oracle://user:pass@hostname:port[/dbname][?service_name=<service>[&key=value&key=value...]] # postgresql-pypostgresql 庫 postgresql+pypostgresql://user:password@host:port/dbname[?key=value&key=value...] # SQL Server-PyODBC 庫 mssql+pyodbc://<username>:<password>@<dsnname>
創建一個數據庫引擎
engine = create_engine( url=SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} )
- {"check_same_thread": False} 僅適用於 SQlite，其他數據庫不需要用到
- 默認情況下，SQLite 將只允許一個線程與其通信，假設每個線程只處理一個獨立的請求
- 這是為了防止被不同的事物（對於不同的請求）共享相同的連接
- 但是在 FastAPI 中，使用普通函數 (def) 可以針對同一請求與數據庫的多個線程進行交互，因此需要讓 SQLite 知道它應該允許使用多線程
- 需要確保每個請求在依賴項中都有自己的數據庫連接會話，因此不需要設置為同一個線程
創建一個數據庫會話
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
- SessionLocal 類的每個實例都是一個數據庫會話
- 但 sessionmaker 本身還不是數據庫會話
- 但是一旦創建了 SessionLocal 類的實例，這個實例就會成為實際的數據庫會話
- 將其命名為 SessionLocal ，方便區分從 SQLAlchemy 導入的 Session
- 稍后將使用 Session（從 SQLAlchemy 導入的那個）
創建一個 ORM 模型基類
Base = declarative_base()
后面會通過繼承這個 Base 類，來創建每個數據庫 Model，也稱為 ORM Model
models.py 代碼
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String from sqlalchemy.orm import relationship from .database import Base class User(Base): # 1、表名 __tablename__ = "users" # 2、類屬性，每一個都代表數據表中的一列 # Column 就是列的意思 # Integer、String、Boolean 就是數據表中，列的類型 id = Column(Integer, primary_key=True, index=True, autoincrement=True) email = Column(String, unique=True, index=True) hashed_password = Column(String) is_active = Column(Boolean, default=True) items = relationship("Item", back_populates="owner") class Item(Base): __tablename__ = "items" id = Column(Integer, primary_key=True, index=True, autoincrement=True) title = Column(String, index=True) description = Column(String, index=True) owner_id = Column(Integer, ForeignKey("users.id")) owner = relationship("User", back_populates="items")
注意：有 autoincrement 就不要用 default 了哈
Column
列，一個屬性代表數據表中的一列
常用參數
| 參數 | 作用 |
| primary_key | 如果設為 True ,這列就是表的主鍵 |
| unique | 如果設為 True ,這列不允許出現重復的值 |
| index | 如果設為 True ,為這列創建索引,提升查詢效率 |
| nullable |
|
| default | 為這列定義默認值 |
| autoincrement |
如果設為 True ,這列自增 |
String、Integer、Boolean
代表數據表中每一列的數據類型
schemas.py 代碼
背景
為了避免混淆 SQLAlchemy 模型和 Pydantic 模型之間，將使用文件 models.py 編寫 SQLAlchemy 模型和文件 schemas.py 編寫 Pydantic 模型
實際代碼
from typing import List, Optional from pydantic import BaseModel # Item 的基類,表示創建和查詢 Item 時共有的屬性 class ItemBase(BaseModel): title: str description: Optional[str] = None # 創建 Item 時的 Model class ItemCreate(ItemBase): pass # 查詢 Item 時的 Model class Item(ItemBase): id: int owner_id: int # 向 Pydantic 提供配置 class Config: # orm_mode 會告訴 Pydantic 模型讀取數據，即使它不是字典，而是 ORM 模型（或任何其他具有屬性的任意對象） orm_mode = True class UserBase(BaseModel): email: str class UserCreate(UserBase): password: str class User(UserBase): id: int is_active: bool items: List[Item] = [] class Config: orm_mode = True
ItemBase、UserBase
基類，聲明在創建或讀取數據時共有的屬性
ItemCreate、UserCreate
創建數據時使用的 Model
Item、User
讀取數據時使用的 Model
orm_mode
class Config: orm_mode = True
- 這是一個 Pydantic 配置項
- orm_mode 會告訴 Pydantic 模型讀取數據，即使它不是字典，而是 ORM 模型（或任何其他具有屬性的任意對象）
# 正常情況 id = data["id"] # 還會嘗試從對象獲取屬性 id = data.id
設置了 orm_mode，Pydantic 模型與 ORM 就兼容了，只需在路徑操作的 response_model 參數中聲明它即可
orm_mode 的技術細節
- SQLAlchemy 默認情況下 lazy loading 懶加載，即需要獲取數據時，才會主動從數據庫中獲取對應的數據
- 比如獲取屬性 current_user.items ，SQLAlchemy 會從 items 表中獲取該用戶的 item 數據，但在這之前不會主動獲取
如果沒有 orm_mode
- 從路徑操作中返回一個 SQLAlchemy 模型，它將不會包括關系數據（比如 user 中有 item，則不會返回 item，后面再講實際的栗子）
- 在 orm_mode 下，Pydantic 會嘗試從屬性訪問它要的數據，可以聲明要返回的特定數據，它甚至可以從 ORM 中獲取它
curd.py 代碼
作用
- 主要用來編寫與數據庫交互的函數，增刪改查，方便整個項目不同地方都能進行復用
- 並且給這些函數添加專屬的單元測試
實際代碼
代碼只實現了查詢和創建
- 根據 id 查詢 user
- 根據 email 查詢 user
- 查詢所有 user
- 創建 user
- 查詢所有 item
- 創建 item
from sqlalchemy.orm import Session from .models import User, Item from .schemas import UserCreate, ItemCreate # 根據 id 獲取 user def get_user(db: Session, user_id: int): return db.query(User).filter(User.id == user_id).first() # 根據 email 獲取 user def get_user_by_email(db: Session, email: str): return db.query(User).filter(User.email == email).first() # 獲取所有 user def get_users(db: Session, size: int = 0, limit: int = 100): return db.query(User).offset(size).limit(limit).all() # 創建 user，user 類型是 Pydantic Model def create_user(db: Session, user: UserCreate): fake_hashed_password = user.password + "superpolo" # 1、使用傳進來的數據創建 SQLAlchemy Model 實例對象 db_user = User(email=user.email, hashed_password=fake_hashed_password) # 2、將實例對象添加到數據庫會話 Session 中 db.add(db_user) # 3、將更改提交到數據庫 db.commit() # 4、刷新實例，方便它包含來自數據庫的任何新數據，比如生成的 ID db.refresh(db_user) return db_user # 獲取所有 item def get_items(db: Session, size: int = 0, limit: int = 100): return db.query(Item).offset(size).limit(limit).all() # 創建 item，item 類型是 Pydantic Model def create_item(db: Session, item: ItemCreate, user_id: int): db_item = Item(**item.dict(), owner_id=user_id) db.add(db_item) db.commit() db.refresh(db_item) return db_item
create_user、create_item
函數內的操作步驟如下
# 1、使用傳進來的數據創建 SQLAlchemy Model 實例對象 db_user = User(email=user.email, hashed_password=fake_hashed_password) # 2、將實例對象添加到數據庫會話 Session 中 db.add(db_user) # 3、將更改提交到數據庫 db.commit() # 4、刷新實例，方便它包含來自數據庫的任何新數據，比如生成的 ID db.refresh(db_user)
main.py 代碼
from typing import List import uvicorn from fastapi import Depends, FastAPI, HTTPException, status, Path, Query, Body from sqlalchemy.orm import Session from models import Base from schemas import User, UserCreate, ItemCreate, Item from database import SessionLocal, engine import curd Base.metadata.create_all(bind=engine) app = FastAPI() # 依賴項,獲取數據庫會話對象 def get_db(): db = SessionLocal() try: yield db finally: db.close() # 創建用戶 @app.post("/users", response_model=User) async def create_user(user: UserCreate, db: Session = Depends(get_db)): # 1、先查詢用戶是否有存在 db_user = curd.get_user_by_email(db, user.email) if db_user: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="user 已存在") res_user = curd.create_user(db, user) return res_user # 根據 user_id 獲取用戶 @app.get("/user_id/{user_id}", response_model=User) async def get_user(user_id: int = Path(...), db: Session = Depends(get_db)): return curd.get_user(db, user_id) # 根據 email 獲取用戶 @app.get("/user_email/{email}", response_model=User) async def get_user_by_email(email: str = Path(...), db: Session = Depends(get_db)): return curd.get_user_by_email(db, email) # 獲取所有用戶 @app.get("/users_all/", response_model=List[User]) async def get_users(skip: int = Query(0), limit: int = Query(100), db: Session = Depends(get_db)): return curd.get_users(db, skip, limit) # 創建 item @app.post("/users/{user_id}/items", response_model=Item) async def get_user_item(user_id: int = Path(...), item: ItemCreate = Body(...), db: Session = Depends(get_db)): return curd.create_user_item(db, item, user_id) # 獲取所有 item @app.get("/items/", response_model=List[Item]) async def get_items(skip: int = Query(0), limit: int = Query(100), db: Session = Depends(get_db)): return curd.get_items(db, skip, limit) if __name__ == "__main__": uvicorn.run(app="main:app", host="127.0.0.1", port=8080, reload=True, debug=True)
依賴項
def get_db(): db = SessionLocal() try: yield db finally: db.close()
- 每個請求都有一個獨立的數據庫會話（SessionLocal）
- 在請求完成后會自動關閉它
- 然后下一個請求來的時候，會創建一個新會話
聲明依賴項
async def create_user(user: UserCreate, db: Session = Depends(get_db))
- SessionLocal 是 sessionmaker() 創建的，是 SQLAlchemy Session 的代理
- 通過聲明 db: Session ，IDE 就可以提供智能代碼提示啦
使用中間件 middleware 代替依賴項聲明數據庫會話
# 中間件 @app.middleware("http") async def db_session_middleware(request: Request, call_next): # 默認響應 response = Response("Internal server error", status_code=500) try: request.state.db = SessionLocal() response = await call_next(request) finally: # 關閉數據庫會話 request.state.db.close() return response # 依賴項,獲取數據庫會話對象 def get_db(request: Request): return request.state.db
request.state
- request.state 是每個 Request 對象的一個屬性
- 它用於存儲附加到請求本身的任意對象，例如本例中的數據庫會話 db
- 也就是說，我不叫 db，叫 sqlite_db 也可以，只是一個屬性名
使用中間件 middleware 和使用 yield 的依賴項的區別
- 中間件需要更多的代碼，而且稍微有點復雜
- 中間件必須是一個 async 函數，而且需要有 await 的代碼，可能會阻塞程序並稍稍降低性能
- 每個請求運行的時候都會先運行中間件，所以會為每個請求都創建一個數據庫連接，即使某個請求的路徑操作函數並不需要和數據庫交互
建議
- 創建數據庫連接對象最好還是用帶有 yield 的依賴項來完成
- 在其他使用場景也是，能滿足需求的前提下，最好用帶有 yield 的依賴項來完成