閱讀(4.2k) 書簽贊(0) 我要糾錯(cuò)

FastAPI教程 SQL（關(guān)系）數(shù)據(jù)庫

2022-07-19 10:06 更新

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

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

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

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

PostgreSQL
MySQL
SQLite
Oracle
Microsoft SQL Server 等

在此示例中，我們將使用SQLite，因?yàn)樗褂脝蝹€(gè)文件并且 Python 已集成支持。因此，您可以復(fù)制此示例并按原樣運(yùn)行它。

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

提示

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

筆記

請注意，大部分代碼是SQLAlchemy您將用于任何框架的標(biāo)準(zhǔn)代碼。

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

ORM

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

一個(gè)常見的模式是使用“ORM”：一個(gè)“對象關(guān)系映射”庫。

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

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

例如，一個(gè)類Pet可以代表一個(gè) SQL 表pets。

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

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

這些 ORM 還具有在表或?qū)嶓w之間建立連接或關(guān)系的工具。

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

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

它可能具有類似"Arquilian".

當(dāng)您嘗試從您的寵物對象訪問它時(shí)，ORM 將完成所有工作以從相應(yīng)的表所有者那里獲取信息。

常見的ORM有例如：Django-ORM（Django框架的一部分）、SQLAlchemy ORM（SQLAlchemy的一部分，獨(dú)立于框架）和Peewee（獨(dú)立于框架）等。

在這里，我們將看到如何使用SQLAlchemy ORM。

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

提示

文檔中有一篇使用 Peewee 的等效文章。

文件結(jié)構(gòu)

對于這些示例，假設(shè)您有一個(gè)名為的目錄my_super_project，其中包含一個(gè)名為的子目錄sql_app，其結(jié)構(gòu)如下：

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

該文件__init__.py只是一個(gè)空文件，但它告訴 Python，sql_app它的所有模塊（Python 文件）都是一個(gè)包。

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

創(chuàng)建 SQLAlchemy 部件

讓我們參考文件sql_app/database.py。

導(dǎo)入 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)建一個(gè)數(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()

在這個(gè)例子中，我們“連接”到一個(gè) SQLite 數(shù)據(jù)庫（用 SQLite 數(shù)據(jù)庫打開一個(gè)文件）。

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

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

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

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

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

提示

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

創(chuàng)建 SQLAlchemy engine

第一步是創(chuàng)建一個(gè) 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ù)庫不需要它。

技術(shù)細(xì)節(jié)

默認(rèn)情況下，SQLite 將只允許一個(gè)線程與其通信，假設(shè)每個(gè)線程將處理一個(gè)獨(dú)立的請求。

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

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

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

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

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

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

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

我們稍后將使用Session（從 SQLAlchemy 導(dǎo)入的）。

要創(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)建一個(gè)Base班級

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

稍后我們將從這個(gè)類繼承來創(chuàng)建每個(gè)數(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)建的這個(gè)類來創(chuàng)建 SQLAlchemy 模型。

提示

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

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

Base從database（database.py上面的文件）導(dǎo)入。

創(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ù)庫中為這些模型中的每一個(gè)使用的表的名稱。

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

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

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

我們使用ColumnSQLAlchemy 作為默認(rèn)值。

而我們通過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)建關(guān)系

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

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

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

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")

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

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

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

創(chuàng)建 Pydantic 模型

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

提示

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

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

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

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

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

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

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

但是為了安全起見，password其他 Pydantic模型中不會出現(xiàn)，例如在讀取用戶時(shí)不會從 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 風(fēng)格和 Pydantic 風(fēng)格

請注意，SQLAlchemy模型使用定義屬性=，并將類型作為參數(shù)傳遞給Column，例如：

name = Column(String)

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

name: str

牢記這一點(diǎn)，這樣您在使用=和:使用它們時(shí)就不會感到困惑。

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

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

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

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

不僅這些項(xiàng)目的ID，但我們在Pydantic中定義的所有數(shù)據(jù)模型讀取項(xiàng)目：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 返回）時(shí)將使用User的 Pydantic模型不包含password.

使用 Pydantic orm_mode

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

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

在Config類中，設(shè)置屬性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

提示

請注意，它正在分配一個(gè)值=，例如：

orm_mode = True

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

這是設(shè)置配置值，而不是聲明類型。

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

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

id = data["id"]

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

id = data.id

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

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

ORM模式的技術(shù)細(xì)節(jié)

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

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

例如，訪問屬性items：

current_user.items

將使 SQLAlchemy 轉(zhuǎn)到該items表并獲取該用戶的項(xiàng)目，但不是之前。

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

即使您在 Pydantic 模型中聲明了這些關(guān)系。

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

CRUD 工具

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

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

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

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

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

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

導(dǎo)入models（SQLAlchemy 模型）和schemas（Pydantic模型/模式）。

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

通過 ID 和電子郵件讀取單個(gè)用戶。
讀取多個(gè)用戶。
閱讀多個(gè)項(xiàng)目。

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)建獨(dú)立于您的路徑操作函數(shù)的僅專用于與數(shù)據(jù)庫交互（獲取用戶或項(xiàng)目）的函數(shù)，您可以更輕松地在多個(gè)部分中重用它們，并為它們添加單元測試。

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

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

步驟是：

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

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

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

警告

這個(gè)例子不安全，密碼沒有散列。

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

有關(guān)更多詳細(xì)信息，請返回教程中的安全部分。

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

提示

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

item.dict()

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

Item(**item.dict())

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

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

主要FastAPI應(yīng)用程序

現(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 進(jìn)行“遷移”（這是它的主要工作）。

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

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

創(chuàng)建依賴

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

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

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

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

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

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)建和處理放在一個(gè)try塊中。

然后我們在finally塊中關(guān)閉它。

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

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

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

這將在路徑操作函數(shù)中為我們提供更好的編輯器支持，因?yàn)榫庉嬈鲗⒅纃b參數(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

技術(shù)細(xì)節(jié)

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

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

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

現(xiàn)在，最后，這是標(biāo)準(zhǔ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

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

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

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

提示

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

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

提示

還要注意，有response_models標(biāo)準(zhǔn)的 Python 類型，如List[schemas.Item].

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

關(guān)于defvsasync def

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

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

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

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

...而我們正在使用：

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

然后我們應(yīng)該聲明路徑操作函數(shù)和不帶的依賴項(xiàng)async def，只用一個(gè)普通的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)
    ...

信息

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

非常技術(shù)性的細(xì)節(jié)

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

遷移

因?yàn)槲覀冎苯邮褂?SQLAlchemy 并且我們不需要任何插件來使用FastAPI，所以我們可以直接將數(shù)據(jù)庫遷移與Alembic集成。

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

同樣，您將能夠在與FastAPI無關(guān)的代碼的其他部分中使用相同的 SQLAlchemy 模型和實(shí)用程序。

例如，在帶有Celery、RQ或ARQ的后臺任務(wù)工作者中。

查看所有文件

請記住，您應(yīng)該有一個(gè)名為的目錄my_super_project，其中包含一個(gè)名為sql_app.

sql_app 應(yīng)該有以下文件：

sql_app/__init__.py: 是一個(gè)空文件。
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

核實(shí)

您可以復(fù)制此代碼并按原樣使用。

信息

事實(shí)上，此處顯示的代碼是測試的一部分。作為這些文檔中的大部分代碼。

然后你可以用 Uvicorn 運(yùn)行它：

uvicorn sql_app.main:app --reload

信息

：Uvicorn 在 http://127.0.0.1:8000 上運(yùn)行（按 CTRL+C 退出）

重啟?

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

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

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

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

它看起來像這樣：

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

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

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

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

創(chuàng)建中間件

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

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)建和處理放在一個(gè)try塊中。

然后我們在finally塊中關(guān)閉它。

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

關(guān)于 request.state

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

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

與yield或中間件的依賴關(guān)系

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

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

提示

yield當(dāng)它們足以滿足用例時(shí)，最好使用依賴項(xiàng)。

信息

yield最近向FastAPI添加了依賴項(xiàng)。

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

以上內(nèi)容是否對您有幫助：

← FastAPI教程 CORS（跨域資源共享）

FastAPI教程更大的應(yīng)用 - 多個(gè)文件 →

寫筆記

我要補(bǔ)充