FastAPI不要求您使用 SQL（關系）數(shù)據(jù)庫。

但是您可以使用任何您想要的關系數(shù)據(jù)庫。

在這里，我們將看到一個使用SQLAlchemy的示例。

您可以輕松地將其調(diào)整為 SQLAlchemy 支持的任何數(shù)據(jù)庫，例如：

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

在此示例中，我們將使用SQLite，因為它使用單個文件并且 Python 已集成支持。因此，您可以復制此示例并按原樣運行它。

稍后，對于您的生產(chǎn)應用程序，您可能希望使用像PostgreSQL這樣的數(shù)據(jù)庫服務器。

提示

有一個帶有FastAPI和PostgreSQL的官方項目生成器，全部基于Docker，包括前端和更多工具：https : //github.com/tiangolo/full-stack-fastapi-postgresql

筆記

請注意，大部分代碼是SQLAlchemy您將用于任何框架的標準代碼。

該FastAPI具體的代碼是小一如既往。

ORM

FastAPI可與任何數(shù)據(jù)庫和任何樣式的庫配合使用以與數(shù)據(jù)庫通信。

一個常見的模式是使用“ORM”：一個“對象關系映射”庫。

ORM 具有在代碼和數(shù)據(jù)庫表（“關系”）中的對象之間進行轉換（“映射”）的工具。

使用 ORM，您通常會創(chuàng)建一個表示 SQL 數(shù)據(jù)庫中的表的類，該類的每個屬性都表示一個列，具有名稱和類型。

例如，一個類Pet可以代表一個 SQL 表pets。

并且該類的每個實例對象代表數(shù)據(jù)庫中的一行。

例如，一個對象orion_cat（ 的實例Pet）可以有一個屬性orion_cat.type，用于列type。該屬性的值可以是，例如"cat"。

這些 ORM 還具有在表或實體之間建立連接或關系的工具。

這樣，您也可以擁有一個屬性orion_cat.owner，所有者將包含該寵物所有者的數(shù)據(jù)，取自表owner。

所以，orion_cat.owner.name可能是這個寵物主人的名字（來自表中的name列owners）。

它可能具有類似"Arquilian".

當您嘗試從您的寵物對象訪問它時，ORM 將完成所有工作以從相應的表所有者那里獲取信息。

常見的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 文件）都是一個包。

現(xiàn)在讓我們看看每個文件/模塊的作用。

創(chuàng)建 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 創(chuàng)建一個數(shù)據(jù)庫 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 數(shù)據(jù)庫（用 SQLite 數(shù)據(jù)庫打開一個文件）。

該文件將位于文件中的同一目錄中sql_app.db。

這就是為什么最后一部分是./sql_app.db.

如果您使用的是PostgreSQL數(shù)據(jù)庫，則只需取消注釋該行：

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

...并使用您的數(shù)據(jù)庫數(shù)據(jù)和憑據(jù)（相當于 MySQL、MariaDB 或任何其他）對其進行調(diào)整。

提示

如果您想使用不同的數(shù)據(jù)庫，這是必須修改的主線。

創(chuàng)建 SQLAlchemy engine

第一步是創(chuàng)建一個 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()

筆記

論據(jù)：

connect_args={"check_same_thread": False}

...僅用于SQLite. 其他數(shù)據(jù)庫不需要它。

技術細節(jié)

默認情況下，SQLite 將只允許一個線程與其通信，假設每個線程將處理一個獨立的請求。

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

但是在 FastAPI 中，使用普通函數(shù) ( def) 可以針對同一個請求與數(shù)據(jù)庫交互多個線程，因此我們需要讓 SQLite 知道它應該允許使用connect_args={"check_same_thread": False}.

此外，我們將確保每個請求在依賴項中都有自己的數(shù)據(jù)庫連接會話，因此不需要該默認機制。

創(chuàng)建一個SessionLocal班級

SessionLocal該類的每個實例都是一個數(shù)據(jù)庫會話。該類本身還不是數(shù)據(jù)庫會話。

但是一旦我們創(chuàng)建了一個SessionLocal類的實例，這個實例就會成為實際的數(shù)據(jù)庫會話。

我們命名它SessionLocal以區(qū)別于Session我們從 SQLAlchemy 導入的。

我們稍后將使用Session（從 SQLAlchemy 導入的）。

要創(chuàng)建SessionLocal類，請使用函數(shù)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()

創(chuàng)建一個Base班級

現(xiàn)在我們將使用declarative_base()返回一個類的函數(shù)。

稍后我們將從這個類繼承來創(chuàng)建每個數(shù)據(jù)庫模型或類（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()

創(chuàng)建數(shù)據(jù)庫模型

現(xiàn)在讓我們看看文件sql_app/models.py。

從Base類創(chuàng)建 SQLAlchemy 模型

我們將使用Base我們之前創(chuàng)建的這個類來創(chuàng)建 SQLAlchemy 模型。

提示

SQLAlchemy 使用術語“模型”來指代與數(shù)據(jù)庫交互的這些類和實例。

但是 Pydantic 也使用術語“模型”來指代不同的東西，數(shù)據(jù)驗證、轉換以及文檔類和實例。

Base從database（database.py上面的文件）導入。

創(chuàng)建從它繼承的類。

這些類是 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, index=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)
    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 在數(shù)據(jù)庫中為這些模型中的每一個使用的表的名稱。

創(chuàng)建模型屬性/列

現(xiàn)在創(chuàng)建所有模型（類）屬性。

這些屬性中的每一個都代表其相應數(shù)據(jù)庫表中的一列。

我們使用ColumnSQLAlchemy 作為默認值。

而我們通過SQLAlchemy的類“類型”，如Integer，String和Boolean，它定義了數(shù)據(jù)庫的類型，作為參數(shù)。

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

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

創(chuàng)建關系

現(xiàn)在創(chuàng)建關系。

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

這將或多或少成為一個“神奇”屬性，其中包含與此相關的其他表中的值。

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

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

當訪問中的屬性items在User中my_user.items，它將有一個ItemSQLAlchemy 模型列表（來自items表），這些模型具有指向users表中此記錄的外鍵。

當您訪問 時my_user.items，SQLAlchemy 實際上會從items表中的數(shù)據(jù)庫中獲取項目并在此處填充它們。

并且在訪問 中的屬性owner時Item，它將包含表中的UserSQLAlchemy 模型users。它將使用owner_id帶有外鍵的屬性/列來知道從users表中獲取哪條記錄。

創(chuàng)建 Pydantic 模型

現(xiàn)在讓我們檢查文件sql_app/schemas.py。

提示

為了避免 SQLAlchemy模型和 Pydantic模型之間的混淆，我們將使用models.py帶有 SQLAlchemy 模型的文件schemas.py和帶有 Pydantic 模型的文件。

這些 Pydantic 模型或多或少地定義了一個“模式”（有效的數(shù)據(jù)形狀）。

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

創(chuàng)建初始 Pydantic模型/模式

創(chuàng)建一個ItemBase和UserBasePydantic模型（或者說“模式”）以在創(chuàng)建或讀取數(shù)據(jù)時具有共同的屬性。

并創(chuàng)建一個繼承自它們的ItemCreateand UserCreate（因此它們將具有相同的屬性），以及創(chuàng)建所需的任何其他數(shù)據(jù)（屬性）。

因此，用戶password在創(chuàng)建它時也會有一個。

但是為了安全起見，password其他 Pydantic模型中不會出現(xiàn)，例如在讀取用戶時不會從 API 發(fā)送。

from typing import List, Optional

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = 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模型使用 定義屬性=，并將類型作為參數(shù)傳遞給Column，例如：

name = Column(String)

雖然 Pydantic模型使用聲明類型:，但新的類型注釋語法/類型提示：

name: str

牢記這一點，這樣您在使用=和:使用它們時就不會感到困惑。

創(chuàng)建用于讀取/返回的Pydantic模型/模式

現(xiàn)在創(chuàng)建將在讀取數(shù)據(jù)時使用的Pydantic模型（模式），當從 API 返回數(shù)據(jù)時。

例如，在創(chuàng)建項目之前，我們不知道分配給它的 ID 是什么，但是在讀取它時（從 API 返回它時）我們已經(jīng)知道它的 ID。

同樣，在讀取用戶時，我們現(xiàn)在可以聲明items將包含屬于該用戶的項目。

不僅這些項目的ID，但我們在Pydantic中定義的所有數(shù)據(jù)模型讀取項目：Item。

from typing import List, Optional

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = 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

提示

請注意，讀取用戶（從 API 返回）時將使用User的 Pydantic模型不包含password.

使用 Pydantic orm_mode

現(xiàn)在，在Pydantic模型讀取，Item并且User，添加一個內(nèi)部Config類。

此類Config用于向 Pydantic 提供配置。

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

from typing import List, Optional

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = 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

它不:用于之前的類型聲明。

這是設置配置值，而不是聲明類型。

Pydanticorm_mode會告訴 Pydantic模型讀取數(shù)據(jù)，即使它不是dict，而是 ORM 模型（或任何其他具有屬性的任意對象）。

這樣，而不是僅僅嘗試id從 a獲取值dict，如下所示：

id = data["id"]

它還會嘗試從屬性中獲取它，例如：

id = data.id

有了這個，Pydantic模型與 ORM 兼容，您只需response_model在路徑操作的參數(shù)中聲明它。

您將能夠返回一個數(shù)據(jù)庫模型，它會從中讀取數(shù)據(jù)。

ORM模式的技術細節(jié)

SQLAlchemy 和許多其他的默認情況下是“延遲加載”。

這意味著，例如，除非您嘗試訪問包含該數(shù)據(jù)的屬性，否則它們不會從數(shù)據(jù)庫中獲取關系數(shù)據(jù)。

例如，訪問屬性items：

current_user.items

將使 SQLAlchemy 轉到該items表并獲取該用戶的項目，但不是之前。

沒有orm_mode，如果您從路徑操作返回 SQLAlchemy 模型，它將不包括關系數(shù)據(jù)。

即使您在 Pydantic 模型中聲明了這些關系。

但是在 ORM 模式下，由于 Pydantic 本身會嘗試從屬性（而不是假設為dict）訪問它需要的數(shù)據(jù)，您可以聲明要返回的特定數(shù)據(jù)，它甚至可以從 ORM 中獲取它。

CRUD 工具

現(xiàn)在讓我們看看文件sql_app/crud.py。

在這個文件中，我們將有可重用的函數(shù)來與數(shù)據(jù)庫中的數(shù)據(jù)進行交互。

CRUD來源于：? reate，- [R EAD，ù PDATE，和d elete。

...雖然在這個例子中我們只是創(chuàng)建和閱讀。

讀取數(shù)據(jù)

Session從導入sqlalchemy.orm，這將允許您聲明db參數(shù)的類型，并在您的函數(shù)中進行更好的類型檢查和完成。

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

創(chuàng)建實用函數(shù)以：

• 通過 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

提示

通過創(chuàng)建獨立于您的路徑操作函數(shù)的僅專用于與數(shù)據(jù)庫交互（獲取用戶或項目）的函數(shù)，您可以更輕松地在多個部分中重用它們，并為它們添加單元測試。

創(chuàng)建數(shù)據(jù)

現(xiàn)在創(chuàng)建實用函數(shù)來創(chuàng)建數(shù)據(jù)。

步驟是：

• 使用您的數(shù)據(jù)創(chuàng)建 SQLAlchemy 模型實例。
• add 該實例對象到您的數(shù)據(jù)庫會話。
• commit 對數(shù)據(jù)庫的更改（以便保存）。
• refresh 您的實例（以便它包含來自數(shù)據(jù)庫的任何新數(shù)據(jù)，例如生成的 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 模型User包含一個hashed_password應該包含密碼的安全散列版本。

但是由于 API 客戶端提供的是原始密碼，因此您需要將其提取并在您的應用程序中生成散列密碼。

然后傳遞hashed_password帶有要保存的值的參數(shù)。

警告

這個例子不安全，密碼沒有散列。

在現(xiàn)實生活中的應用程序中，您需要對密碼進行哈希處理，并且永遠不要以明文形式保存它們。

有關更多詳細信息，請返回教程中的安全部分。

在這里，我們只關注數(shù)據(jù)庫的工具和機制。

提示

我們沒有將每個關鍵字參數(shù)傳遞給ItemPydantic模型并從中讀取每個參數(shù)，而是dict使用 Pydantic模型的數(shù)據(jù)生成一個：

item.dict()

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

Item(**item.dict())

然后我們傳遞owner_idPydantic模型未提供的額外關鍵字參數(shù)，使用：

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

主要FastAPI應用程序

現(xiàn)在在文件中sql_app/main.py讓我們集成并使用我們之前創(chuàng)建的所有其他部分。

創(chuàng)建數(shù)據(jù)庫表

以一種非常簡單的方式創(chuàng)建數(shù)據(jù)庫表：

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初始化您的數(shù)據(jù)庫（創(chuàng)建表等）。

而且您還將使用 Alembic 進行“遷移”（這是它的主要工作）。

“遷移”是每當您更改 SQLAlchemy 模型的結構、添加新屬性等以在數(shù)據(jù)庫中復制這些更改、添加新列、新表等時所需的一組步驟。

您可以在Project Generation-Template的模板中找到 FastAPI 項目中的 Alembic 示例。具體地，在所述alembic源代碼中的目錄。

創(chuàng)建依賴

現(xiàn)在使用SessionLocal我們在sql_app/databases.py文件中創(chuàng)建的類來創(chuàng)建依賴項。

我們需要SessionLocal每個請求有一個獨立的數(shù)據(jù)庫會話/連接（），在所有請求中使用同一個會話，然后在請求完成后關閉它。

然后將為下一個請求創(chuàng)建一個新會話。

為此，我們將創(chuàng)建一個新的依賴關系yield，正如之前關于依賴關系yield的部分所解釋的那樣。

我們的依賴將創(chuàng)建一個新的 SQLAlchemy SessionLocal，它將在單個請求中使用，然后在請求完成后關閉它。

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()請求的創(chuàng)建和處理放在一個try塊中。

然后我們在finally塊中關閉它。

這樣我們可以確保在請求之后數(shù)據(jù)庫會話總是關閉。即使在處理請求時出現(xiàn)異常。

但是您不能從退出代碼（之后yield）引發(fā)另一個異常。在依賴項中查看更多信息和yieldHTTPException

然后，在路徑操作函數(shù)中使用依賴項時，我們使用Session直接從 SQLAlchemy 導入的類型聲明它。

這將在路徑操作函數(shù)中為我們提供更好的編輯器支持，因為編輯器將知道db參數(shù)的類型Session：

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

技術細節(jié)

參數(shù)db實際上是 type SessionLocal，但是這個類（創(chuàng)建于sessionmaker()）是 SQLAlchemy 的“代理” Session，因此，編輯器并不真正知道提供了哪些方法。

但是，作為申報類型Session，編輯器現(xiàn)在可以知道可用的方法（.add()，.query()，.commit()等），并能提供更好的支持（如完成）。類型聲明不影響實際對象。

創(chuàng)建您的FastAPI 路徑操作

現(xiàn)在，最后，這是標準的FastAPI 路徑操作代碼。

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

我們在依賴項中的每個請求之前創(chuàng)建數(shù)據(jù)庫會話yield，然后關閉它。

然后我們可以在路徑操作函數(shù)中創(chuàng)建所需的依賴項，直接獲取該會話。

這樣，我們就可以crud.get_user直接從路徑操作函數(shù)內(nèi)部調(diào)用并使用該會話。

提示

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

但是由于所有路徑操作都response_model使用 Pydantic模型/模式orm_mode，因此 Pydantic 模型中聲明的數(shù)據(jù)將從它們中提取并返回給客戶端，并進行所有正常的過濾和驗證。

提示

還要注意，有response_models標準的 Python 類型，如List[schemas.Item].

但是作為內(nèi)容/的該參數(shù)List是一個Pydantic模型與orm_mode，該數(shù)據(jù)將被檢索并返回到客戶端為常，沒有任何問題。

關于defvsasync def

在這里，我們在路徑操作函數(shù)和依賴項中使用 SQLAlchemy 代碼，反過來，它將與外部數(shù)據(jù)庫進行通信。

這可能需要一些“等待”。

但是由于 Sqlalchemy 不具有await直接使用的兼容性，就像使用以下內(nèi)容一樣：

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

...而我們正在使用：

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

然后我們應該聲明路徑操作函數(shù)和不帶 的依賴項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)
    ...

信息

如果您需要異步連接到關系數(shù)據(jù)庫，請參閱異步 SQL（關系）數(shù)據(jù)庫。

非常技術性的細節(jié)

如果您很好奇并且擁有深厚的技術知識，您可以在Async文檔中查看有關如何處理此async defvs的非常技術性的細節(jié)。def

遷移

因為我們直接使用 SQLAlchemy 并且我們不需要任何插件來使用FastAPI，所以我們可以直接將數(shù)據(jù)庫遷移與Alembic集成。

由于與 SQLAlchemy 和 SQLAlchemy 模型相關的代碼存在于單獨的獨立文件中，您甚至可以使用 Alembic 執(zhí)行遷移，而無需安裝 FastAPI、Pydantic 或其他任何東西。

同樣，您將能夠在與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, index=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)
    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:

from typing import List, Optional

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = 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:

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

信息

：Uvicorn 在 http://127.0.0.1:8000 上運行（按 CTRL+C 退出）

然后，您可以在http://127.0.0.1:8000/docs 上打開瀏覽器。

您將能夠與您的FastAPI應用程序交互，從真實數(shù)據(jù)庫中讀取數(shù)據(jù)：

直接與數(shù)據(jù)庫交互

如果您想獨立于 FastAPI 直接探索 SQLite 數(shù)據(jù)庫（文件），以調(diào)試其內(nèi)容，添加表、列、記錄、修改數(shù)據(jù)等，您可以使用DB Browser for SQLite。

它看起來像這樣：

您還可以使用在線 SQLite 瀏覽器，如SQLite Viewer或ExtendsClass。

使用中間件的替代數(shù)據(jù)庫會話

如果您不能使用依賴項yield——例如，如果您沒有使用Python 3.7并且無法安裝上面提到的Python 3.6的“backports” ——您可以在類似的“中間件”中設置會話道路。

“中間件”基本上是一個始終為每個請求執(zhí)行的函數(shù)，其中一些代碼在端點函數(shù)之前執(zhí)行，一些代碼在端點函數(shù)之后執(zhí)行。

創(chuàng)建中間件

我們將添加的中間件（只是一個函數(shù)）將為SessionLocal每個請求創(chuàng)建一個新的 SQLAlchemy ，將其添加到請求中，然后在請求完成后關閉它。

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()請求的創(chuàng)建和處理放在一個try塊中。

然后我們在finally塊中關閉它。

這樣我們可以確保在請求之后數(shù)據(jù)庫會話總是關閉。即使在處理請求時出現(xiàn)異常。

關于 request.state

request.state是每個Request對象的屬性。它用于存儲附加到請求本身的任意對象，例如本例中的數(shù)據(jù)庫會話。您可以在Starlette 關于Requeststate的文檔中閱讀更多相關信息。

在這種情況下，它幫助我們確保通過所有請求使用單個數(shù)據(jù)庫會話，然后關閉（在中間件中）。

與yield或 中間件的依賴關系

在這里添加一個中間件類似于依賴 with 的yield作用，但有一些區(qū)別：

• 它需要更多的代碼，而且有點復雜。
• 中間件必須是一個async函數(shù)。如果其中有必須“等待”網(wǎng)絡的代碼，它可能會在那里“阻塞”您的應用程序并稍微降低性能。雖然這里的工作方式可能不是很成問題SQLAlchemy。但是，如果您將更多代碼添加到有大量I/O等待的中間件，則可能會出現(xiàn)問題。
• 每個請求都會運行一個中間件。因此，將為每個請求創(chuàng)建一個連接。即使處理該請求的路徑操作不需要數(shù)據(jù)庫。

提示

yield當它們足以滿足用例時，最好使用依賴項。

信息

yield最近向FastAPI添加了依賴項。

本教程的先前版本只有帶有中間件的示例，并且可能有幾個應用程序使用中間件進行數(shù)據(jù)庫會話管理。