SQL 数据库

FastAPI不需要您使用SQL（关系）数据库。
但是您可以使用所需的任何关系数据库。
这里你可以看到一个使用SQLAlchemy的例子.
您可以轻松地使其适应SQLAlchemy支持的任何数据库，例如:
PostgreSQL
MySQL
SQLite
Oracle
* Microsoft SQL Server等
在此示例中，我们将使用SQLite，因为它使用单个文件并且Python具有集成的支持。因此，您可以复制此示例并按原样运行它。
稍后，对于您的生产应用程序，您可能想要使用PostgreSQL之类的数据库服务器。

:::info 提示
有一个使用FastAPI和PostgreSQL的官方项目生成器，它们均基于Docker，包括前端和更多工具：https://github.com/tiangolo/full-stack-fastapi-postgresql :::
:::tips 注意
大多数代码是您将在任何框架中使用的标准SQLAlchemy代码。 FastAPI专用代码与以往一样小。 :::

ORMs

FastAPI可与任何数据库和任何样式的库一起使用，以与数据库进行通信。
一种常见的模式是使用“ ORM”：“对象关系映射”库。
ORM具有在代码和数据库表（“关系”）中的对象之间进行转换（“映射”）的工具。
使用ORM，通常可以创建一个表示SQL数据库中的表的类，该类的每个属性表示一个具有名称和类型的列。
例如，Pet类可以表示SQL表pets。
该类的每个实例对象都代表数据库中的一行。
例如，对象orioncat（Pet的实例）可以为列类型具有属性orioncat.type。该属性的值可以是例如“cat”。
这些ORM还具有在表或实体之间建立连接或关系的工具。
这样，您还可以拥有一个orion_cat.owner属性，并且所有者将包含该宠物所有者的数据，这些数据取自表所有者。
因此，orion_cat.owner.name可以是该宠物所有者的名称（来自所有者表中的名称列）。
它的值可能类似于“ Arquilian”。
当您尝试从宠物对象访问信息时，ORM会做所有工作以从相应的表所有者那里获取信息。
常见的ORM例如：Django-ORM（Django框架的一部分），SQLAlchemy ORM（SQLAlchemy的一部分，独立于框架）和Peewee（独立于框架），等等。
在这里，我们将看到如何使用SQLAlchemy ORM。
以类似的方式，您可以使用任何其他ORM。

:::info 提示
在文档中有一篇使用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创建数据库url

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

在此示例中，我们将“连接”到SQLite数据库（使用SQLite数据库打开文件）。
该文件将位于文件sql_app.db中的同一目录中。
这就是为什么最后一部分是./sql_app.db。
如果您使用的是PostgreSQL数据库，则只需取消注释以下行：

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

…并使其适应您的数据库数据和凭据（等效于MySQL，MariaDB或其他任何数据库）。

:::tips 提示
如果要使用其他数据库，这是必须修改的主行 :::

创建SQLAlchemyengine

第一步是创建一个SQLAlchemy“引擎”。
稍后我们将在其他地方使用此engine。

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

注意

参数:

connect_args={"check_same_thread": False}

…仅适用于SQLite。其他数据库则不需要

:::tips 详细教程
默认情况下，假定每个线程将处理一个独立的请求，SQLite将仅允许一个线程与其通信。
这是为了防止为不同的事物（针对不同的请求）意外共享同一连接。
但是在FastAPI中，使用普通函数（def），一个以上的线程可以与同一请求的数据库进行交互，因此我们需要使SQLite知道它应该使用`connectargs = {“checksame_thread”：False}允许它。
另外，我们将确保每个请求都具有依赖关系，从而获得其自己的数据库连接会话，因此不需要该默认机制。 :::

创建一个SessionLocal类

SessionLocal类的每个实例将是一个数据库会话。该类本身还不是数据库会话。
但是，一旦我们创建SessionLocal类的实例，该实例将成为实际的数据库会话。
我们将其命名为SessionLocal，以区别于我们从SQLAlchemy导入的Session。
稍后我们将使用Session（从SQLAlchemy导入的会话）。
要创建SessionLocal类，请使用sessionmaker函数：

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

创建一个Base类

现在，我们将使用函数declarative_base()返回一个类。
稍后，我们将从该类继承以创建每个数据库模型或类（ORM模型）：

Base = declarative_base()

创建一个数据库模型

现在让我们看一下文件sql_app/models.py

从Base类创建SQLAlchemy模型

我们将使用之前创建的这个基类来创建SQLAlchemy模型。

:::info 提示
SQLAlchemy使用术语“model”来指代与数据库交互的这些类和实例。
但是Pydantic也使用术语“模型”来指代不同的东西，即数据验证，转换以及文档类和实例。 :::

从database中导入Base(上面的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这些模型中的每个模型在数据库中使用的表的名称。

创建模型属性和元素

现在创建所有模型（类）属性。
这些属性中的每一个都代表其相应数据库表中的一列。
我们使用SQLAlchemy中的Column作为默认值。
然后，我们传递一个SQLAlchemy类“类型”（如Integer，String和Boolean），该类将数据库中的类型定义为参数。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from .database import Base
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, 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")

创建关系

现在创建关系。
为此，我们使用SQLAlchemy ORM提供的relationship。
这或多或少将成为“魔术”属性，其中将包含与该表相关的其他表中的值。

# 省去了部分代码，上面的代码中已经有了  
from sqlalchemy.orm import relationship
items = relationship("Item", back_populates="owner")
owner = relationship("User", back_populates="items")

当访问User中的属性item时，如my_user.items中一样，它将具有项SQLAlchemy模型的列表（来自items表），该模型的外键指向用户表中的该记录。
当您访问my_user.items时，SQLAlchemy实际上会从items表中的数据库中获取项目，并在此处填充它们。
并且在访问Item中的属性owner时，它将包含来自users表的User SQLAlchemy模型。
它将使用owner_id属性/列及其外键来知道要从users表中获取哪条记录。

创建Pydantic模型

我们来查看文件sql_app/schemas.py.

:::tips 提示
为了避免SQLAlchemy模型和Pydantic模型之间的混淆，我们将使用SQLAlchemy模型的文件models.py和使用Pydantic模型的schemas.py。
这些Pydantic模型或多或少定义了“模式”（有效数据形状）。 因此，这将有助于我们避免在同时使用两者时产生混淆。 :::

创建初始化Pydantic模型/模式

创建一个ItemBase和UserBase Pydantic模型（或称“schemas”）以在创建或读取数据时具有共同的属性。
并创建一个从它们继承的ItemCreate和UserCreate（这样它们将具有相同的属性），以及创建所需的任何其他数据（属性）。
因此，用户在创建密码时还将具有password。
但是为了安全起见，该password不会在其他Pydantic模型中使用，例如，在读取用户时不会从API发送该密码

from typing import List
from pydantic import BaseModel
class ItemBase(BaseModel):
    title: str
    description: 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。
同样，在读取用户时，我们现在可以声明项目将包含属于该用户的项目。
不仅是这些项目的ID，还包括我们在Pydantic模型中定义的用于读取项目的所有数据：Item。

class Item(ItemBase):
    id: int
    owner_id: int
class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

:::info 提示
请注意，在读取用户（从API返回）时将使用的“用户” Pydantic模型不包含密码。 :::

使用Pydantic’s的orm_mod

现在，在用于读取的Pydantic模型，Item和User中，添加一个内部Config类。
Config类用于为Pydantic提供配置。
在Config类中，设置属性orm_mode = True。

class Item(ItemBase):
    id: int
    owner_id: int
    class Config:
        orm_mode = True
class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []
    class Config:
        orm_mode = True

:::info 提示
请注意，它使用=来分配一个值，例如：
orm_mode = True
它不像以前的类型声明那样使用：。
这是在设置配置值，而不是声明类型。 :::

Pydantic的orm_mode将告诉Pydantic模型读取数据，即使它不是dict而是ORM模型（或任何其他具有属性的任意对象）。
这样，而不是仅尝试从dict获取id值，如：

id = data["id"]

它还将尝试从属性获取它，如：

id = data.id

因此，Pydantic模型与ORM兼容，您可以在路径操作中的response_model参数中声明它。
您将能够返回数据库模型，并且它将从中读取数据。

关于ORM mode的详细教程

SQLAlchemy和许多其他默认情况下是“延迟加载”。
例如，这意味着除非您尝试访问将包含该数据的属性，否则它们不会从数据库中获取关系数据。
例如，访问属性items：

current_user.items

将使SQLAlchemy转到items表并获得该用户的项目，但不可以。
如果没有orm_mode，则如果您从路径操作中返回SQLAlchemy模型，则该模型将不包含关系数据。
即使您在Pydantic模型中声明了这些关系。
但是在ORM模式下，由于Pydantic本身会尝试从属性访问其所需的数据（而不是假设一个dict），因此您可以声明要返回的特定数据，并且即使从ORM中也可以获取该数据。 。

CRUD工具

让我们来看一下文件sql_app/crud.py.
在此文件中，我们将具有可重用的功能来与数据库中的数据进行交互。
CRUD来自：创建(Create)，读取(Read)，更新(Update)和删除(Delete)。
…尽管在这个例子中我们只展示了创建和读取

读数据

从sqlalchemy.orm导入Session，这将允许您声明db参数的类型，并在函数中进行更好的类型检查和完成。
导入models（SQLAlchemy模型）和schemas（Pydantic模型/模式）。
创建实用程序功能以： 通过ID和电子邮件读取单个用户
读取多个用户
* 阅读单个item。

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

:::info 提示
通过创建与路径操作功能无关的仅用于与数据库交互（获取用户或项目）的功能，您可以更轻松地在多个部分中重用它们，并为其添加单元测试。 :::

创建数据

现在创建实用程序函数来创建数据。
这些步骤是：
用您的数据创建一个SQLAlchemy模型实例。
add将该实例对象添加到您的数据库会话中。
commit将更改提交到数据库（以便将其保存）。
refresh刷新您的实例（以便它包含数据库中的任何新数据，例如生成的ID）。

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

:::info 提示
User的SQLAlchemy模型包含hashed_password，该密码应包含密码的安全哈希值。
但是由于API客户端提供的是原始密码，因此您需要提取原始密码并在应用程序中生成哈希密码。
然后，将hashed_password参数与要保存的值一起传递。 :::
:::tips 警告
此示例不安全，密码不进行哈希处理。 在现实生活中的应用程序中，您需要对密码进行哈希处理，并且永远不要将其保存为纯文本格式。
有关更多详细信息，请返回到教程中的“安全性”部分。 在这里，我们仅关注数据库的工具和机制。 :::
:::info 提示
我们没有将每个关键字参数传递给Item并从Pydantic模型中读取每个关键字参数，而是使用Pydantic模型的数据通过以下方式生成了dict：
item.dict()
然后将dict的键值对传递为SQLAlchemy Item的关键字参数，带有： Item(**item.dict())
然后，我们将Pydantic模型未提供的额外关键字参数ownerid传递给： Item(**item.dict(),ownerid=user_id）
:::

主要的FastAPI app

现在，在文件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初始化数据库（创建表等）。
而且您还将Alembic用于“迁移”（这是它的主要工作）。
“迁移”是每当您更改SQLAlchemy模型的结构，添加新属性等以在数据库中复制这些更改，添加新列，新表等时所需的一组步骤。
您可以找到一个项目生成-模板中的模板中的FastAPI项目中的Alembic示例。具体在源代码的alembic目录中。

创建依赖

:::info 信息
为此，您需要使用Python 3.7或更高版本，或者在Python 3.6中安装“反向端口”：
$ pip install async-exit-stack async-generator
这是安装async-exit-stack和async-generator

您还可以将替代方法与最后说明的“中间件”一起使用。 :::

现在，使用我们在sql_app/databases.py文件中创建的SessionLocal类创建依赖项。
我们需要每个请求有一个独立的数据库会话/连接（SessionLocal），在所有请求中使用相同的会话，然后在请求完成后将其关闭。
然后将为下一个请求创建一个新会话。
为此，我们将创建一个带有yield的新依赖关系，如之前在带有yield的依赖关系的部分中所述。
我们的依赖关系将创建一个新的SQLAlchemy SessionLocal，它将在单个请求中使用，然后在请求完成后将其关闭。

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

:::info 信息
我们将创建SessionLocal()和处理请求放在try块中。
然后在finally块中将其关闭。
这样，我们确保在请求后数据库会话始终关闭。即使在处理请求时出现异常。
但是您不能从退出代码中引发另一个异常（在yield之后）。 在yield和HTTPException的依赖项中查看更多 :::

然后，在路径操作函数中使用依赖项时，我们使用直接从SQLAlchemy导入的Session类型声明它。
然后，这将为我们提供更好的路径操作功能内的编辑器支持，因为编辑器将知道db参数的类型为Session：

@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

:::tips 详细教程
参数db实际上是SessionLocal类型的，但是此类（使用sessionmaker()创建）是SQLAlchemy Session的“代理”，因此，编辑器实际上并不知道提供了哪些方法。
但是通过将类型声明为Session，编辑器现在可以知道可用的方法（.add()、.query()、.commit()等），并可以提供更好的支持（例如完成）。
类型声明不影响实际对象。 :::

创建你的FastAPI路径操作

现在，最后是标准的FastAPI路径操作代码。

@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并使用该会话。

:::info 提示
请注意，您返回的值是SQLAlchemy模型或SQLAlchemy模型的列表。
但是，由于所有路径操作都有一个使用orm_mode的Pydantic模型/架构的response_model，因此将在其中提取Pydantic模型中声明的数据，并将其返回给客户端，同时进行所有常规过滤和验证。 ::: :::info 提示**
还要注意，有些response_models具有标准的Python类型，例如List[schemas.Item]。
但是，由于该List的内容/参数是带有orm_mode的Pydantic模型，因此可以正常检索数据并将其返回给客户端，而不会出现问题。 :::

关于def和async def对比

在这里，我们在路径操作函数内部和依赖项中使用SQLAlchemy代码，然后它将与外部数据库进行通信。
那可能需要一些“等待”。
但是由于SQLAlchemy不具有直接使用await的兼容性，因此类似于：

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

我们这样来替代:

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

然后，我们应该使用普通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)
    ...

:::info 非常详细的说明
如果您好奇并具有深厚的技术知识，则可以在Async文档中查看有关如何处理async def和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
  from pydantic import BaseModel
  class ItemBase(BaseModel):
    title: str
    description: 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

  检查

  您可以复制此代码并按原样使用。

:::info 信息
实际上，此处显示的代码是测试的一部分。就像这些文档中的大多数代码一样。 :::

您可以使用Uvicorn运行它：
在命令行下uvicorn sql_app.main:app --reload
然后打开浏览器进入[http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs)
您将能够与FastAPI应用程序进行交互，从真实的数据库中读取数据：

直接与数据库交互

如果要独立于FastAPI直接浏览SQLite数据库（文件）以调试其内容，添加表，列，记录，修改数据等，你可以使用DB Browser fro SQLite。
它看起来像这样：

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

中间件的替代数据库会话

如果您不能将依赖关系与yield一起使用-例如，如果您没有使用Python 3.7并且无法安装上面提到的针对Python 3.6的“反向端口”，则可以在“中间件”类似的方式。
“中间件”基本上是始终针对每个请求执行的功能，其中某些代码在端点功能之前执行，而某些代码在端点功能之后执行.

创建一个中间件

我们将添加的中间件（只是一个函数）将为每个请求创建一个新的SQLAlchemy SessionLocal，将其添加到请求中，然后在请求完成后将其关闭。

@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

:::info 信息
我们将创建SessionLocal()和处理请求放在try块中。
然后在finally块中将其关闭。
这样，我们确保在请求后数据库会话始终关闭。即使在处理请求时出现异常。 :::

关于request.state

request.state是每个Request对象的属性。它可以存储附加到请求本身的任意对象，例如本例中的数据库会话。您可以在Starlette的有关“请求”状态的文档中阅读有关此内容的更多信息。
在这种情况下，对于我们来说，这有助于我们确保在所有请求中使用单个数据库会话，然后再关闭（在中间件中）。

与yield或中间件的依赖

在此处添加中间件类似于具有yield的依赖项所做的事情，但有一些区别：
它需要更多代码，并且稍微复杂一些。
中间件必须是异步功能。
如果其中包含必须“等待”网络的代码，则可能会“阻塞”您的应用程序，从而导致性能降低。
尽管在这里，SQLAlchemy的工作方式可能不是很成问题。
但是，如果您向具有大量I/O等待的中间件添加更多代码，则可能会出现问题。
每个请求都运行一个中间件。
因此，将为每个请求创建一个连接。
即使处理该请求的路径操作不需要DB

:::info 提示
当依赖关系足以满足用例时，最好将依赖关系与yield一起使用 :::
:::info 信息
与yield的依赖关系最近已添加到FastAPI。
本教程的先前版本仅包含带有中间件的示例，并且可能有多个使用该中间件进行数据库会话管理的应用程序 :::