FastAPI教程 SQL（关系）数据库

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。

所以，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 文件）都是一个包。

现在让我们看看每个文件/模块的作用。

创建 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以区别于Session我们从 SQLAlchemy 导入的。

我们稍后将使用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 也使用术语“模型”来指代不同的东西，数据验证、转换以及文档类和实例。

Base从database（database.py上面的文件）导入。

创建从它继承的类。

这些类是 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 在数据库中为这些模型中的每一个使用的表的名称。

创建模型属性/列

现在创建所有模型（类）属性。

这些属性中的每一个都代表其相应数据库表中的一列。

我们使用ColumnSQLAlchemy 作为默认值。

而我们通过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, 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")

创建关系

现在创建关系。

为此，我们使用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表中的数据库中获取项目并在此处填充它们。

并且在访问中的属性owner时Item，它将包含表中的UserSQLAlchemy 模型users。它将使用owner_id带有外键的属性/列来知道从users表中获取哪条记录。

创建 Pydantic 模型

现在让我们检查文件sql_app/schemas.py。

提示

为了避免 SQLAlchemy模型和 Pydantic模型之间的混淆，我们将使用models.py带有 SQLAlchemy 模型的文件schemas.py和带有 Pydantic 模型的文件。

这些 Pydantic 模型或多或少地定义了一个“模式”（有效的数据形状）。

因此，这将有助于我们在使用两者时避免混淆。

创建初始 Pydantic模型/模式

创建一个ItemBase和UserBasePydantic模型（或者说“模式”）以在创建或读取数据时具有共同的属性。

并创建一个继承自它们的ItemCreateand UserCreate（因此它们将具有相同的属性），以及创建所需的任何其他数据（属性）。

因此，用户password在创建它时也会有一个。

但是为了安全起见，password其他 Pydantic模型中不会出现，例如在读取用户时不会从 API 发送。

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模型使用定义属性=，并将类型作为参数传递给Column，例如：

name = Column(String)

虽然 Pydantic模型使用声明类型:，但新的类型注释语法/类型提示：

name: str

牢记这一点，这样您在使用=和:使用它们时就不会感到困惑。

创建用于读取/返回的Pydantic模型/模式

现在创建将在读取数据时使用的Pydantic模型（模式），当从 API 返回数据时。

例如，在创建项目之前，我们不知道分配给它的 ID 是什么，但是在读取它时（从 API 返回它时）我们已经知道它的 ID。

同样，在读取用户时，我们现在可以声明items将包含属于该用户的项目。

不仅这些项目的ID，但我们在Pydantic中定义的所有数据模型读取项目：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

现在，在Pydantic模型读取，Item并且User，添加一个内部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模型读取数据，即使它不是dict，而是 ORM 模型（或任何其他具有属性的任意对象）。

这样，而不是仅仅尝试id从 a获取值dict，如下所示：

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来源于：Ç reate，- [R EAD，ù PDATE，和d elete。

…虽然在这个例子中我们只是创建和阅读。

读取数据

Session从导入sqlalchemy.orm，这将允许您声明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

提示

的 SQLAlchemy 模型User包含一个hashed_password应该包含密码的安全散列版本。

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

然后传递hashed_password带有要保存的值的参数。

警告

这个例子不安全，密码没有散列。

在现实生活中的应用程序中，您需要对密码进行哈希处理，并且永远不要以明文形式保存它们。

有关更多详细信息，请返回教程中的安全部分。

在这里，我们只关注数据库的工具和机制。

提示

我们没有将每个关键字参数传递给ItemPydantic模型并从中读取每个参数，而是dict使用 Pydantic模型的数据生成一个：

item.dict()

然后我们将dict的键值对作为关键字参数传递给 SQLAlchemy Item，使用：

Item(**item.dict())

然后我们传递owner_idPydantic模型未提供的额外关键字参数，使用：

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

主要FastAPI应用程序

现在在文件中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

蒸馏笔记

通常，您可能会使用Alembic初始化您的数据库（创建表等）。

而且您还将使用 Alembic 进行“迁移”（这是它的主要工作）。

“迁移”是每当您更改 SQLAlchemy 模型的结构、添加新属性等以在数据库中复制这些更改、添加新列、新表等时所需的一组步骤。

您可以在Project Generation-Template的模板中找到 FastAPI 项目中的 Alembic 示例。具体地，在所述alembic源代码中的目录。

创建依赖

现在使用SessionLocal我们在sql_app/databases.py文件中创建的类来创建依赖项。

我们需要SessionLocal每个请求有一个独立的数据库会话/连接（），在所有请求中使用同一个会话，然后在请求完成后关闭它。

然后将为下一个请求创建一个新会话。

为此，我们将创建一个新的依赖关系yield，正如之前关于依赖关系yield的部分所解释的那样。

我们的依赖将创建一个新的 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()请求的创建和处理放在一个try块中。

然后我们在finally块中关闭它。

这样我们可以确保在请求之后数据库会话总是关闭。即使在处理请求时出现异常。

但是您不能从退出代码（之后yield）引发另一个异常。在依赖项中查看更多信息和 yieldHTTPException

然后，在路径操作函数中使用依赖项时，我们使用Session直接从 SQLAlchemy 导入的类型声明它。

这将在路径操作函数中为我们提供更好的编辑器支持，因为编辑器将知道db参数的类型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

技术细节

参数db实际上是 type SessionLocal，但是这个类（创建于sessionmaker()）是 SQLAlchemy 的“代理” Session，因此，编辑器并不真正知道提供了哪些方法。

但是，作为申报类型Session，编辑器现在可以知道可用的方法（.add()，.query()，.commit()等），并能提供更好的支持（如完成）。类型声明不影响实际对象。

创建您的FastAPI 路径操作

现在，最后，这是标准的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

我们在依赖项中的每个请求之前创建数据库会话yield，然后关闭它。

然后我们可以在路径操作函数中创建所需的依赖项，直接获取该会话。

这样，我们就可以crud.get_user直接从路径操作函数内部调用并使用该会话。

提示

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

但是由于所有路径操作都response_model使用 Pydantic模型/模式orm_mode，因此 Pydantic 模型中声明的数据将从它们中提取并返回给客户端，并进行所有正常的过滤和验证。

提示

还要注意，有response_models标准的 Python 类型，如List[schemas.Item].

但是作为内容/的该参数List是一个Pydantic模型与orm_mode，该数据将被检索并返回到客户端为常，没有任何问题。

关于defvsasync 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文档中查看有关如何处理此async defvs的非常技术性的细节。def

迁移

因为我们直接使用 SQLAlchemy 并且我们不需要任何插件来使用FastAPI，所以我们可以直接将数据库迁移与Alembic集成。

由于与 SQLAlchemy 和 SQLAlchemy 模型相关的代码存在于单独的独立文件中，您甚至可以使用 Alembic 执行迁移，而无需安装 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应用程序交互，从真实数据库中读取数据：

直接与数据库交互

如果您想独立于 FastAPI 直接探索 SQLite 数据库（文件），以调试其内容，添加表、列、记录、修改数据等，您可以使用DB Browser for SQLite。

它看起来像这样：

您还可以使用在线 SQLite 浏览器，如SQLite Viewer或ExtendsClass。

使用中间件的替代数据库会话

如果您不能使用依赖项yield——例如，如果您没有使用Python 3.7并且无法安装上面提到的Python 3.6的“backports” ——您可以在类似的“中间件”中设置会话道路。

“中间件”基本上是一个始终为每个请求执行的函数，其中一些代码在端点函数之前执行，一些代码在端点函数之后执行。

创建中间件

我们将添加的中间件（只是一个函数）将为SessionLocal每个请求创建一个新的 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

信息