SQL（關聯式）資料庫

資訊

這些文件即將更新。 🎉

目前版本假設使用 Pydantic v1 和 SQLAlchemy 2.0 以前的版本。

新的文件將包含 Pydantic v2，並將在 SQLModel（也基於 SQLAlchemy）更新以使用 Pydantic v2 後使用它。

FastAPI 並不要求您使用 SQL（關聯式）資料庫。

但您可以使用任何您想要的關聯式資料庫。

這裡我們將看到一個使用 SQLAlchemy 的範例。

您可以輕鬆地將其調整為任何 SQLAlchemy 支援的資料庫，例如

• PostgreSQL
• MySQL
• SQLite
• Oracle
• Microsoft SQL Server 等等。

在此範例中，我們將使用 SQLite，因為它使用單個檔案，且 Python 具有內建支援。因此，您可以複製此範例並按原樣執行它。

稍後，對於您的正式應用程式，您可能需要使用像 PostgreSQL 這樣的資料庫伺服器。

提示

有一個官方專案產生器，包含 FastAPI 和 PostgreSQL，全部基於 Docker，包括前端和其他工具：https://github.com/tiangolo/full-stack-fastapi-postgresql

注意事項

請注意，大多數程式碼都是您在任何框架中使用的標準 SQLAlchemy 程式碼。

FastAPI 特定的程式碼一如既往地簡潔。

ORM

FastAPI 可與任何資料庫和任何類型的資料庫通訊函式庫搭配使用。

常見的模式是使用「ORM」：「物件關聯對映」函式庫。

ORM 具有在程式碼中的「*物件*」和資料庫表格（「*關聯*」）之間轉換（「*對映*」）的工具。

使用 ORM，您通常會建立一個代表 SQL 資料庫中表格的類別，該類別的每個屬性都代表一個欄位，包含名稱和類型。

例如，Pet 類別可以代表 SQL 表格 pets。

該類別的每個「*實例*」物件都代表資料庫中的一列。

例如，一個物件 orion_cat（Pet 的一個實例）可以有一個屬性 orion_cat.type，用於 type 欄位。該屬性的值可以是，例如 "cat"。

這些 ORM 也具有建立表格或實體之間的連接或關聯的工具。

這樣，您也可以擁有一個屬性 orion_cat.owner，而 owner 將包含從「owners」表格中取得的此寵物擁有者的資料。

所以，orion_cat.owner.name 可能會是這隻寵物的擁有者姓名（來自 owners 表格中的 name 欄位）。

它的值可能像 "Arquilian"。

當您嘗試從寵物物件存取它時，ORM 會執行所有工作，從對應的表格 *owners* 中取得資訊。

常見的 ORM 例如：Django-ORM（Django 框架的一部分）、SQLAlchemy ORM（SQLAlchemy 的一部分，獨立於框架）和 Peewee（獨立於框架）等等。

這裡我們將看到如何使用 SQLAlchemy ORM。

您可以用類似的方式使用任何其他 ORM。

提示

文件中有一個使用 Peewee 的等效文章。

檔案結構

在這些範例中，假設您有一個名為 my_super_project 的目錄，其中包含一個名為 sql_app 的子目錄，其結構如下：

.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

檔案 __init__.py 只是一個空檔案，但它告訴 Python sql_app 及其所有模組（Python 檔案）是一個套件。

現在讓我們看看每個檔案／模組的功能。

安裝 SQLAlchemy

首先，您需要安裝 SQLAlchemy。

確保您建立一個虛擬環境，啟動它，然後安裝它，例如：

$ pip install sqlalchemy

---> 100%

建立 SQLAlchemy 部分

讓我們參考檔案 sql_app/database.py。

導入 SQLAlchemy 部分

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.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)

Base = declarative_base()

為 SQLAlchemy 建立資料庫 URL

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.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)

Base = declarative_base()

在此範例中，我們「連線」到 SQLite 資料庫（開啟包含 SQLite 資料庫的檔案）。

該檔案將位於同一目錄中的 sql_app.db 檔案。

這就是為什麼最後一部分是 ./sql_app.db。

如果您使用的是 PostgreSQL 資料庫，您只需取消註釋這一行：

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

... 並使用您的資料庫資料和憑證進行調整（MySQL、MariaDB 或其他資料庫也一樣）。

提示

如果您想使用不同的資料庫，這是您必須修改的主要行。

建立 SQLAlchemy engine

第一步是建立一個 SQLAlchemy「引擎」。

我們稍後將在其他地方使用這個 engine。

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.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)

Base = declarative_base()

注意事項

參數

connect_args={"check_same_thread": False}

... 僅適用於 SQLite。其他資料庫不需要它。

「技術細節」

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

這是為了防止意外地將相同的連線用於不同的請求。

但是在 FastAPI 中，使用普通函數 (def) 時，多個執行緒可能會針對同一個請求與資料庫互動，因此我們需要讓 SQLite 知道它應該允許使用 connect_args={"check_same_thread": False}。

此外，我們將確保每個請求在其依賴項中獲得自己的資料庫連線工作階段，因此不需要該預設機制。

建立 SessionLocal 類別

SessionLocal 類別的每個實例都將是一個資料庫工作階段。該類別本身還不是資料庫工作階段。

但一旦我們建立了 SessionLocal 類別的實例，這個實例將會是實際的資料庫連線階段。

我們將它命名為 SessionLocal，以區別於我們從 SQLAlchemy 導入的 Session。

我們稍後會使用 Session（從 SQLAlchemy 導入的）。

要建立 SessionLocal 類別，請使用 sessionmaker 函式。

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.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)

Base = declarative_base()

建立一個 Base 類別

現在我們將使用 declarative_base() 函式，它會返回一個類別。

稍後我們將繼承這個類別來建立每個資料庫模型或類別（ORM 模型）。

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.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)

Base = declarative_base()

建立資料庫模型

現在讓我們看看 sql_app/models.py 檔案。

從 Base 類別建立 SQLAlchemy 模型

我們將使用之前建立的這個 Base 類別來建立 SQLAlchemy 模型。

提示

SQLAlchemy 使用術語「**模型**」來指稱這些與資料庫互動的類別和實例。

但 Pydantic 也使用術語「**模型**」來指稱不同的東西，即資料驗證、轉換和文件類別及實例。

從 database（上面的 database.py 檔案）導入 Base。

建立繼承它的類別。

這些類別是 SQLAlchemy 模型。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=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)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

__tablename__ 屬性告訴 SQLAlchemy 在資料庫中用於這些模型中每個模型的表格名稱。

建立模型屬性／欄位

現在建立所有模型（類別）屬性。

這些屬性中的每一個都代表其對應資料庫表格中的一個欄位。

我們使用 SQLAlchemy 的 Column 作為預設值。

我們傳遞一個 SQLAlchemy 類別「類型」，例如 Integer、String 和 Boolean，它定義資料庫中的類型，作為參數。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=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)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

建立關聯

現在建立關聯。

為此，我們使用 SQLAlchemy ORM 提供的 relationship。

這將或多或少成為一個「神奇」的屬性，它將包含來自與此表格相關的其他表格的值。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=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)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

當在 User 中存取 items 屬性時，例如在 my_user.items 中，它將包含一個 Item SQLAlchemy 模型列表（來自 items 表格），這些模型具有一個指向 users 表格中此紀錄的外鍵。

當您存取 my_user.items 時，SQLAlchemy 實際上會從資料庫的 items 表格中提取項目並在此處填入它們。

當在 Item 中存取 owner 屬性時，它將包含來自 users 表格的 User SQLAlchemy 模型。它將使用帶有外鍵的 owner_id 屬性／欄位來知道要從 users 表格中獲取哪個紀錄。

建立 Pydantic 模型

現在讓我們檢查 sql_app/schemas.py 檔案。

提示

為了避免混淆 SQLAlchemy 的*模型*和 Pydantic 的*模型*，我們將使用包含 SQLAlchemy 模型的 models.py 檔案和包含 Pydantic 模型的 schemas.py 檔案。

這些 Pydantic 模型或多或少定義了一個「schema」（一個有效的資料形狀）。

因此，這將有助於我們在同時使用兩者時避免混淆。

建立初始 Pydantic *模型*／schemas

建立 ItemBase 和 UserBase Pydantic *模型*（或者我們可以說「schemas」）以便在建立或讀取資料時具有共同的屬性。

並建立繼承自它們的 ItemCreate 和 UserCreate（因此它們將具有相同的屬性），以及建立所需的任何額外數據（屬性）。

因此，使用者在建立時也會有 password（密碼）。

但為了安全起見，password（密碼）不會出現在其他 Pydantic *模型* 中，例如，在從 API 讀取使用者時，它不會被發送。

Python 3.10+Python 3.9+Python 3.8+

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


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

    class Config:
        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

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 Config:
        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

from typing import List, 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 Config:
        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

SQLAlchemy 風格和 Pydantic 風格

請注意，SQLAlchemy *模型* 使用 = 定義屬性，並將類型作為參數傳遞給 Column，如下所示

name = Column(String)

而 Pydantic *模型* 使用 : 宣告類型，這是新的類型註釋語法/類型提示

name: str

請記住這些，這樣在使用 = 和 : 時才不會混淆。

建立用於讀取/返回的 Pydantic *模型*/結構

現在建立將在讀取數據時，從 API 返回數據時使用的 Pydantic *模型*（結構）。

例如，在建立項目之前，我們不知道將分配給它的 ID 是什麼，但在讀取它時（從 API 返回它時），我們將已經知道它的 ID。

同樣地，在讀取使用者時，我們現在可以宣告 items（項目）將包含屬於此使用者的項目。

不僅是這些項目的 ID，還包括我們在用於讀取項目的 Pydantic *模型* Item 中定義的所有數據。

Python 3.10+Python 3.9+Python 3.8+

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


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

    class Config:
        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

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 Config:
        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

from typing import List, 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 Config:
        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

提示

請注意，User，即在讀取使用者時（從 API 返回它時）將使用的 Pydantic *模型*，不包含 password（密碼）。

使用 Pydantic 的 orm_mode

現在，在用於讀取的 Pydantic *模型* Item 和 User 中，新增一個內部 Config 類別。

這個 Config 類別用於提供 Pydantic 的配置。

在 Config 類別中，設定屬性 orm_mode = True。

Python 3.10+Python 3.9+Python 3.8+

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


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

    class Config:
        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

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 Config:
        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

from typing import List, 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 Config:
        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

提示

請注意，它是使用 = 分配值，如下所示

orm_mode = True

它不像之前的類型宣告那樣使用 :。

這是設定配置值，而不是宣告類型。

Pydantic 的 orm_mode 將告知 Pydantic *模型* 即使數據不是 dict（字典），而是 ORM 模型（或任何其他具有屬性的任意物件）也要讀取數據。

這樣，它就不會僅嘗試從 dict（字典）中獲取 id 值，如下所示

id = data["id"]

它也會嘗試從屬性中獲取它，如下所示

id = data.id

這樣，Pydantic *模型* 就與 ORM 相容，您只需在 *路徑操作* 中的 response_model 參數中宣告它即可。

您將能夠返回數據庫模型，它將從中讀取數據。

關於 ORM 模式的技術細節

SQLAlchemy 和許多其他預設都是「延遲載入」的。

這意味著，例如，除非您嘗試存取包含該數據的屬性，否則它們不會從數據庫中提取關係的數據。

例如，存取屬性 items（項目）

current_user.items

會使 SQLAlchemy 前往 items（項目）表並獲取此使用者的項目，但不會在此之前執行。

如果沒有 orm_mode，如果您從 *路徑操作* 中返回 SQLAlchemy 模型，則它不會包含關係數據。

即使您在 Pydantic 模型中宣告了這些關係也是如此。

但在 ORM 模式下，由於 Pydantic 本身會嘗試從屬性（而不是假設一個 dict）中訪問所需的數據，您可以聲明想要返回的特定數據，它就能夠去獲取它，即使是從 ORM 中獲取。

CRUD 工具

現在讓我們看看檔案 sql_app/crud.py。

在這個檔案中，我們將有一些可重複使用的函式來與資料庫中的數據互動。

**CRUD** 來自：**C**reate（新增）、**R**ead（讀取）、**U**pdate（更新）和 **D**elete（刪除）。

...雖然在本例中，我們只進行新增和讀取。

讀取數據

從 sqlalchemy.orm 導入 Session，這將允許您聲明 db 參數的類型，並在您的函式中獲得更好的類型檢查和自動完成。

導入 models（SQLAlchemy 模型）和 schemas（Pydantic *模型* / 模式）。

建立工具函式來

• 透過 ID 和電子郵件讀取單個使用者。
• 讀取多個使用者。
• 讀取多個項目。

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

提示

透過建立專門用於與資料庫互動（獲取使用者或項目）的函式，而不依賴您的 *路徑操作函式*，您可以更輕鬆地在多個部分重複使用它們，並且還可以為它們添加單元測試。

新增數據

現在建立工具函式來新增數據。

步驟如下

• 使用您的數據建立一個 SQLAlchemy 模型*實例*。
• 將該實例物件add到您的資料庫會話中。
• 將更改commit到資料庫（以便儲存它們）。
• refresh您的實例（以便它包含來自資料庫的任何新數據，例如產生的 ID）。

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

資訊

在 Pydantic v1 中，該方法稱為 .dict()，在 Pydantic v2 中已棄用（但仍支援），並重新命名為 .model_dump()。

此處的範例使用 .dict() 以與 Pydantic v1 相容，但如果您可以使用 Pydantic v2，則應改用 .model_dump()。

提示

User 的 SQLAlchemy 模型包含一個 hashed_password，其中應包含密碼的安全雜湊版本。

但是由於 API 用戶端提供的是原始密碼，您需要提取它並在您的應用程式中產生雜湊密碼。

然後使用該值傳遞 hashed_password 參數以進行儲存。

警告

此範例不安全，密碼未經過雜湊處理。

在實際應用程式中，您需要對密碼進行雜湊處理，並且永遠不要以純文字形式儲存它們。

如需更多詳細資訊，請返回教學中的安全性章節。

這裡我們只關注資料庫的工具和機制。

提示

我們不是將每個關鍵字參數傳遞給 Item 並從 Pydantic *模型*中讀取每個參數，而是使用以下方法產生一個包含 Pydantic *模型*數據的 dict：

item.dict()

然後，我們將 dict 的鍵值對作為關鍵字參數傳遞給 SQLAlchemy Item，使用：

Item(**item.dict())

然後，我們傳遞 Pydantic *模型*未提供的額外關鍵字參數 owner_id，使用：

Item(**item.dict(), owner_id=user_id)

主要的 FastAPI 應用程式

現在在檔案 sql_app/main.py 中，讓我們整合並使用我們之前建立的所有其他部分。

建立資料庫表格

以非常簡化的方式建立資料庫表格

Python 3.9+Python 3.8+

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

Alembic 備註

通常，您可能會使用 Alembic 初始化您的資料庫（建立表格等）。

你也可以使用 Alembic 進行「遷移」（這是它的主要工作）。

「遷移」是指每當你更改 SQLAlchemy 模型的結構、新增屬性等等時所需的一系列步驟，以便在資料庫中複製這些更改，例如新增欄位、新增表格等等。

你可以在 全端 FastAPI 模板 的 FastAPI 專案中找到 Alembic 的範例。 確切位置在 原始碼中的 alembic 目錄。

建立依賴項

現在使用我們在 sql_app/database.py 檔案中建立的 SessionLocal 類別來建立一個依賴項。

我們需要為每個請求建立一個獨立的資料庫連線/工作階段 (SessionLocal)，在整個請求中使用同一個工作階段，然後在請求完成後關閉它。

然後下一個請求將會建立一個新的工作階段。

為此，我們將使用 yield 建立一個新的依賴項，如先前在關於 使用 yield 的依賴項 的章節中所述。

我們的依賴項將建立一個新的 SQLAlchemy SessionLocal，它將用於單個請求，然後在請求完成後關閉它。

Python 3.9+Python 3.8+

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

資訊

我們將 SessionLocal() 的建立和請求的處理放在 try 區塊中。

然後我們在 finally 區塊中關閉它。

透過這種方式，我們確保資料庫工作階段在請求後總是關閉。 即使在處理請求時發生例外狀況。

但是你不能從退出程式碼（在 yield 之後）引發另一個例外狀況。 在 使用 yield 的依賴項和 HTTPException 中查看更多資訊。

然後，在*路徑操作函式*中使用依賴項時，我們使用直接從 SQLAlchemy 導入的 Session 類型來宣告它。

這將在*路徑操作函式*內提供更好的編輯器支援，因為編輯器將知道 db 參數的類型是 Session。

Python 3.9+Python 3.8+

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

「技術細節」

參數 db 實際上是 SessionLocal 類型，但這個類別（使用 sessionmaker() 建立）是 SQLAlchemy Session 的「代理」，因此編輯器並不知道它提供了哪些方法。

但是透過將類型宣告為 Session，編輯器現在可以知道可用的方法（.add()、.query()、.commit() 等等），並且可以提供更好的支援（例如自動完成）。 類型宣告不會影響實際的物件。

建立你的 FastAPI *路徑操作*

現在，終於到了標準的 FastAPI *路徑操作*程式碼。

Python 3.9+Python 3.8+

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

我們在使用 yield 的依賴項中，在每個請求之前建立資料庫工作階段，然後在之後關閉它。

然後我們可以在*路徑操作函式*中建立所需的依賴項，以直接取得該工作階段。

這樣，我們就可以直接從*路徑操作函式*內部呼叫 crud.get_user 並使用該工作階段。

提示

請注意，你返回的值是 SQLAlchemy 模型或 SQLAlchemy 模型的列表。

但是由於所有*路徑操作*都有一個使用 orm_mode 的 Pydantic *模型*/schema 的 response_model，因此在你的 Pydantic 模型中宣告的資料將會從中提取並返回給客戶端，並進行所有正常的篩選和驗證。

提示

另請注意，有一些 response_models 具有標準 Python 類型，例如 List[schemas.Item]。

但由於該 List 的內容/參數是一個帶有 orm_mode 的 Pydantic *模型*，數據將會被正常檢索並返回給客戶端，不會出現問題。

關於 def 與 async def

這裡我們在 *路徑操作函式* 和依賴項中使用 SQLAlchemy 程式碼，而它將會與外部資料庫進行通訊。

這可能會需要一些「等待」時間。

但由於 SQLAlchemy 不相容直接使用 await，就像以下這樣

user = await db.query(User).first()

...而我們使用的是

user = db.query(User).first()

那麼我們應該宣告 *路徑操作函式* 和依賴項時不使用 async def，而只使用普通的 def，如下所示

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    ...

資訊

如果您需要非同步連接到關聯式資料庫，請參閱 非同步 SQL (關聯式) 資料庫。

「非常技術性的細節」

如果您感到好奇且具備深厚的技術知識，您可以查看 非同步 文件中關於如何處理 async def 與 def 的非常技術性的細節。

遷移

因為我們直接使用 SQLAlchemy，而且不需要任何外掛程式讓它與 FastAPI 協作，我們可以直接將資料庫遷移與 Alembic 整合。

由於與 SQLAlchemy 和 SQLAlchemy 模型相關的程式碼位於獨立的檔案中，您甚至可以在不安裝 FastAPI、Pydantic 或任何其他套件的情況下使用 Alembic 執行遷移。

同樣地，您可以在程式碼中其他與 FastAPI 無關的部分使用相同的 SQLAlchemy 模型和工具。

例如，在使用 Celery、RQ 或 ARQ 的背景工作程序中。

檢閱所有檔案

請記住，您應該有一個名為 my_super_project 的目錄，其中包含一個名為 sql_app 的子目錄。

sql_app 應該包含以下檔案

• sql_app/__init__.py：是一個空檔案。

• sql_app/database.py:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.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)

Base = declarative_base()

• sql_app/models.py:

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=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)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

• sql_app/schemas.py:

Python 3.10+Python 3.9+Python 3.8+

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


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

    class Config:
        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

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 Config:
        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

from typing import List, 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 Config:
        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

• sql_app/crud.py:

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

• sql_app/main.py:

Python 3.9+Python 3.8+

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

檢查一下

您可以複製此程式碼並按原樣使用。

資訊

實際上，這裡顯示的程式碼是測試的一部分。如同這些文件中大部分的程式碼一樣。

然後您可以使用 Uvicorn 執行它

$ uvicorn sql_app.main:app --reload

<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

然後，您可以在瀏覽器中開啟 http://127.0.0.1:8000/docs。

您將能夠與您的 FastAPI 應用程式互動，從真實的資料庫中讀取數據

直接與資料庫互動

如果您想直接探索 SQLite 資料庫（檔案），而不透過 FastAPI，以便除錯其內容、新增表格、欄位、記錄、修改數據等，您可以使用 DB Browser for SQLite。

它看起來像這樣

您也可以使用線上 SQLite 瀏覽器，例如 SQLite Viewer 或 ExtendsClass。

使用中介軟體的替代資料庫連線方法

如果您無法使用帶有 yield 的依賴注入 -- 例如，如果您沒有使用 Python 3.7 且無法為 Python 3.6 安裝上述提到的「backports」套件 -- 您可以用類似的方式在「中介軟體」中設定連線。

「中介軟體」基本上是一個在每個請求都會執行的函式，其中一些程式碼在端點函式之前執行，而一些程式碼在端點函式之後執行。

建立中介軟體

我們將新增的中介軟體（只是一個函式）會為每個請求建立一個新的 SQLAlchemy SessionLocal，將其新增至請求，然後在請求完成後關閉它。

Python 3.9+Python 3.8+

from fastapi import Depends, FastAPI, HTTPException, Request, Response
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


@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


# Dependency
def get_db(request: Request):
    return request.state.db


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

from typing import List

from fastapi import Depends, FastAPI, HTTPException, Request, Response
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


@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


# Dependency
def get_db(request: Request):
    return request.state.db


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@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

資訊

我們將 SessionLocal() 的建立和請求的處理放在 try 區塊中。

然後我們在 finally 區塊中關閉它。

透過這種方式，我們確保資料庫工作階段在請求後總是關閉。 即使在處理請求時發生例外狀況。

關於 request.state

request.state 是每個 Request 物件的屬性。它用於儲存附加到請求本身的任意物件，例如本例中的資料庫連線。您可以在Starlette 的關於 Request 狀態的說明文件中了解更多資訊。

在本例中，它幫助我們確保在整個請求中使用單個資料庫連線，然後在之後（在中介軟體中）關閉它。

使用 yield 的依賴注入或中介軟體

在此新增中介軟體與使用 yield 的依賴注入類似，但有一些差異

• 它需要更多程式碼，並且稍微複雜一些。
• 中介軟體必須是一個 async 函式。
  • 如果其中的程式碼必須「等待」網路，它可能會「阻塞」您的應用程式並稍微降低效能。
  • 雖然以 SQLAlchemy 的運作方式來看，這裡可能沒有什麼問題。
  • 但是，如果您在中介軟體中新增了更多具有大量 I/O 等待的程式碼，那麼它可能會造成問題。
• 中介軟體會針對每個請求執行。
  • 因此，將為每個請求建立一個連線。
  • 即使處理該請求的路徑操作不需要資料庫也是如此。

提示

當使用 yield 的依賴注入足以應付使用案例時，最好使用它。

資訊

使用 yield 的依賴注入是最近才新增到 FastAPI 的。

本教學的先前版本僅提供中介軟體的範例，並且可能有幾個應用程式使用中介軟體進行資料庫連線管理。