在本章節中，我們將使用 SQLite 作為我們的目標資料庫進行使用。在前幾章有介紹了怎麼創建一個 FastAPI 的 app 以及檔案架構，這次我們使用一體式架構建置我們這次的系統。

.
├─src
    __init__.py
		models.py
		database.py
    main.py

首先我們初始化完一個專案後並將專案結構整形為以上結構，並在 src 中新增一個database.py檔案，database.py 該檔案負責處理與資料庫連接的設定等。

資料庫連接設定

# src/utils.py

import uuid

def generate_uuid() -> str:
    """generate uuid string

    Returns
    -------
    str
        uuid string
    """
    return str(uuid.uuid4())

# src/database.py

from sqlalchemy import create_engine, String
from sqlalchemy.orm import sessionmaker, DeclarativeBase, mapped_column
from sqlalchemy import DateTime
from sqlalchemy.sql import func
from src.utils import generate_uuid

SQLALCHEMY_DATABASE_URL = "sqlite:///./example.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase):
    id = mapped_column(String, primary_key=True, default=generate_uuid)
    created_at = mapped_column(DateTime(timezone=True), server_default=func.now())
    updated_at = mapped_column(
        DateTime(timezone=True), onupdate=func.now(), server_default=func.now()
    )

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

首先我們根據官方文件中的範例程式來進行調整。

定義資料庫連線URL

SQLALCHEMY_DATABASE_URL = "sqlite:///./example.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

不同的資料庫有不同的連線URL格式，並且也需要安裝不同的套件。

建立資料庫引擎

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

建立一個引擎物件，可以讓我們透過這個物件與資料庫建立連線。

connect_args={"check_same_thread": False}

本次使用 SQLite，故此參數須設定為 False，以下是官方文件說明 :

預設情況下，SQLite 只允許一個線程與其通訊，假設每個線程會處理獨立的請求。

這是為了防止不同的事物（對於不同的請求）意外共享相同的連接。

但在 FastAPI 中，使用普通函數（def）可能會有多個線程與同一個請求的資料庫進行交互，因此我們需要通知 SQLite，應允許這種情況，可以使用 connect_args={"check_same_thread": False}。

此外，我們將確保每個請求都在相依性（dependency）中獲得自己的資料庫連接會話，因此不需要使用預設的機制。

實體化一個 sessionmaker 物件

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

顧名思義該物件幫助我們建立與DB之間的Session

定義資料表基底

class Base(DeclarativeBase):
    id = mapped_column(String, primary_key=True, default=generate_uuid)
    created_at = mapped_column(DateTime(timezone=True), server_default=func.now())
    updated_at = mapped_column(
        DateTime(timezone=True), onupdate=func.now(), server_default=func.now()
    )

藉由繼承 DeclarativeBase 這個類別物件，可以讓我們使用聲明的方式來定義與資料庫中的資料表對應。在此定義 Base model 時，可以先將每張表都會使用到的欄位先定義好，之後便可複用到各個表。

• id : UUID 主鍵
• created_at : 紀錄該筆資料建立時間
• updated_at: 紀錄該筆資料更動時間

獲取 Session 方法

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

我們會在路由中藉由依賴注入的方式使用此方法得到一個 db 的 Session，並在請求結束時將其關閉。

定義資料表

我們在 src 中新增 models.py 檔，此檔案定義資料表對應。

# src/models.py

from sqlalchemy import ForeignKey, Integer, String, Float
from sqlalchemy.orm import mapped_column, relationship

from src.database import Base

class Customer(Base):
    __tablename__ = "customers"

    customer_name = mapped_column(String(30))
    orders = relationship("Order", back_populates="customer", lazy="joined")

class Order(Base):
    __tablename__ = "orders"

    customer_id = mapped_column(String, ForeignKey("customers.id"), index=True)
    item_id = mapped_column(String, ForeignKey("items.id"), index=True)
    quantity = mapped_column(Integer, default=1)
    customer = relationship("Customer", back_populates="orders")
    item = relationship("Item", lazy="joined")

class Item(Base):
    __tablename__ = "items"

    item_name = mapped_column(String(30))
    price = mapped_column(Float, default=0.0)
    quantity = mapped_column(Integer, default=0)

我們透過宣告類別並繼承剛剛在 database.py 中的 Base model來定義資料表物件。在這邊我們定義三張表，分別儲存顧客資訊、訂單資訊、商品資訊。

顧客資訊

class Customer(Base):
    __tablename__ = "customers"

    customer_name = mapped_column(String(30))
    orders = relationship("Order", back_populates="customer", lazy="joined")

• customer_name : 儲存30個字元的顧客名稱。
• orders : 顧客擁有的訂單。

relationship 為與關聯資料的對應，在此 oders 會對應到 orders 表中與該顧客有關聯的資料，會在拿取資料時一起發出 join 查詢。

訂單資訊

class Order(Base):
    __tablename__ = "orders"

    customer_id = mapped_column(String, ForeignKey("customers.id"), index=True)
    item_id = mapped_column(String, ForeignKey("customers.id"), index=True)
    quantity = mapped_column(Integer, default=1)
    customer = relationship("Customer", back_populates="orders")
    item = relationship("Item", lazy="joined")

• customer_id : 外來鍵，紀錄顧客的關聯性
• item_id : 外來鍵，紀錄商品的關聯性
• quantity : 本次訂單買了多少商品
• customer : 同時查詢顧客資料
• item : 同時查詢商品資料

商品資訊

class Item(Base):
    __tablename__ = "items"

    item_name = mapped_column(String(30))
    price = mapped_column(Float, default=0.0)
    quantity = mapped_column(Integer, default=0)

• item_name : 商品名稱，最多30個字
• price : 價錢，預設為 0.0
• quantity : 目前庫存數量，預設 0

製作遷移檔

我們定義完資料表關聯物件後，我們便可以使用 alembic 製作遷移檔，以將變更同步到目標資料庫中。

安裝 alembic

pip install alembic

使用 pip 直接安裝即可。

初始化 alembic

(Fast) PS D:\workspace\python\FastAPI\fastapi_hello> alembic init alembic
Creating directory D:\workspace\python\FastAPI\fastapi_hello\alembic ...  done
Creating directory D:\workspace\python\FastAPI\fastapi_hello\alembic\versions ...  done
Generating D:\workspace\python\FastAPI\fastapi_hello\alembic.ini ...  done
Generating D:\workspace\python\FastAPI\fastapi_hello\alembic\env.py ...  done
Generating D:\workspace\python\FastAPI\fastapi_hello\alembic\README ...  done
Generating D:\workspace\python\FastAPI\fastapi_hello\alembic\script.py.mako ...  done
Please edit configuration/connection/logging settings in 'D:\\workspace\\python\\FastAPI\\fastapi_hello\\alembic.ini' before proceeding.

使用 alembic init 即可創建出 alembic 的資料夾，其中便包含了其遷移所用的程式碼。

設定 alembic

# alembic.ini

sqlalchemy.url = sqlite:///./example.db

首先到 alembic.ini 中將資料庫連接字串設定成與database.py中相同

# alembic/env.py

from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

from src.database import Base
from src.models import Customer, Item, Order

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = Base.metadata

在此處將 target_metadata 更改為 database.py 中的Base model 的 metadata，以讓 alembic 可以正確抓取到資料表。

此外，需要將你要與資料庫關聯的模型引用進來，alembic 才會抓取的到與資料庫裡的差別，以建立遷移檔。

製作遷移檔

(Fast) PS D:\workspace\python\FastAPI\fastapi_hello> alembic revision --autogenerate -m "init"
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'customers'
INFO  [alembic.autogenerate.compare] Detected added table 'items'
INFO  [alembic.autogenerate.compare] Detected added table 'orders'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_orders_customer_id' on '['customer_id']'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_orders_item_id' on '['item_id']'
Generating D:\workspace\python\FastAPI\fastapi_hello\alembic\versions\e96ee114f582_init.py ...  done

使用 alembic revision --autogenerate -m "" 指令建立遷移檔

• —autogenerate : 從設定好的SQLalchemy模型與DB比較後自動產生遷移檔
• -m : comment

同步變更

(Fast) PS D:\workspace\python\FastAPI\fastapi_hello> alembic upgrade head
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 8b24c4cdc30b, init

使用 alembic upgrade head 直接將版本更新到最新的遷移檔版本即可。

如果有問題也可以用 alembic downgrade 來進行降版。

小結

今天我們知道怎麼設定資料庫連線，定義好我們的 database model，以及透過這些 model 來產生遷移檔，到這裡我們已經將資料庫處理完了，下一章節我們就可以來針對資料表創建 CRUD 接口。

參考資料

https://fastapi.tiangolo.com/zh/tutorial/sql-databases