昨天我們快速地展式怎麼建立空的資料庫，但也大幅簡化了不少東西，今天再來把這部份補充說明一下。

Schema

先前有介紹過，FastAPI 其中一個優點是，它整合了 pydantic，因此，我們可以透過 pydantic 建立資料庫的 schema (也就是 pydantic model)，幫助我們在後續操作資料庫之前，可以先檢查型別是否正確。

pydantic model 在 [Day 08] API 的 Response 中有介紹過

讓我們再看一次昨天的 models.py，回憶一下我們的資料庫格式。

# models.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String

from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

接下來看一下官網的範例 (有簡化過)

# schemas.py
from typing import Union
from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int


class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

可以看到，User 和 Item 都有拆成三個部份，都是一個 Base、一個 Create，和一個完整的。這樣設計的好處是，之後在建立或是回傳等不同使用情境時，都有合適的 schema 可以用。

稍後我們再把 schema 補到 API 設定上。

資料庫連線

在昨天的範例中，我們連一支 API 都沒有，僅僅只是呼叫 DB 的連線 (session) 而已。實際在開發時，資料庫連線本身就是一個值得討論的議題。到底什麼時候該連？什麼時候該斷？才能不浪費時間反覆連線，又不占用資源，或甚至同時間多筆資料操作導致資料丟失。

這邊先講結論：

在 API 收到 Request 時建立連線，回傳 Response 後中斷連線

首先，我們要確保不論發生什麼事 (成功或失敗)，最終資料庫都要被中斷連線，避免一直佔住資源，最終導致整個後端掛掉。這邊我們可以使用 python 的 try... except... finally... (其實只要 try 和 finally 就夠了) 來幫我們做到這件事。

另外，先前也有快速提到 FastAPI 的 Dependency Injection，我們可以在 API 的函數的參數內添加 Depends()，讓 Depends 內的函數的 output 給 API 使用。這邊就可以讓函數 output 出資料庫的連線 (session)，就可以讓資料庫的連線時機是在 API 收到 Request 後才開始。

此外，再透過 yield (而非 return) 的方式，讓我們可以在 API 回傳 Response 後繼續處理 session，也就是關閉連線。

這就是為什麼昨天的 get_db() 的寫法會使用 try... finally... 和 yield 的原因了。

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

因此，比較正確用法會類似這樣 (擷取部份的 main.py 範例)

@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

這邊我們定義了 response_model 方便我們過濾資料，並用 Depends() 搭配 get_db() 讓我們可以在 API 的開始與結束進行資料庫的連線與中斷連線。

API 函數內的 crud 則是負責操作資料庫的函數 (還沒介紹到)

其他連線方式？

FastAPI 還有介紹一個退一步的做法，那就是在 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

但並不建議這樣做，除非 python 版本太舊沒有 yield，或是什麼其他特別原因，不然還是建議要用上面的作法。

在 API 多次連線和中斷連線？

聽起來或許很奇怪，但如果一個 API 需要進行較複雜的操作 (通常會再包成一個或多個函數) 後，才會操作資料庫，那可能會有人 (例如我QQ) 會把資料庫操作放到更底層的函數才 import 進來，如果要照上面範例的做法，等於就要不停地把 session 往內層的函數傳遞，程式碼看起來就很亂。當 API 使用量小的時候或許不太會有問題，但當使用量增大時，可能會遇到下面這個問題：

1. 速度較慢
   這個應該很好理解，連線與中斷連線本身較需要花費時間
2. 資料丟失
   發生機率不高，但如果同時有多筆交易資料 (至少需要操作資料庫兩次，分別是增加和減少)，那可能就會覆蓋掉某些資料 (詳細可以去查查 Database transaction 相關資料)

因此，還是盡量避免在一個 API 內多次連線和中斷連線。

重新連線

來聊聊一個實際上工作上遇到的資料庫連線問題。

那時候使用的資料庫 MariaDB，前後端建立起來之後測試都沒問題，結果後來發現放到隔天 (伺服器不關機)，資料庫就連不上了，進而導致 API 發生錯誤。但只要重新啟動後端，就又都正常。一開始以為是記憶體滿了之類的問題，但偏偏其他與資料庫無關的 API 都很正常。找了幾天 (因為要放一陣子才能重現問題 XD)，最終才知道是因為 MariaDB (和 MySQL) 在連線一段時間後 (預設是 8 小時)，會自動中斷連線。

這邊指的「中斷」不是指關閉 session，而是在 database.py 內的 create_engine() 那邊的中斷。

因此，在 create_engine() 除了資料庫的連線 URL 之外，還需要加上 pool_recyle，讓 SQLAlchemy 自動重新連線。

engine = create_engine('mysql+mysqldb://...', pool_recycle=3600)

或是也可以直接換一個資料庫啦 (咦？)

這部份在 SQLAlchemy 也有說明

重點回顧

今天我們把上次跳過的 schema 和 資料庫連線 的部份補上了，明天終於可以開始操作資料庫了~

祝大家中秋節快樂~