一、ORM

FastAPI可与任何数据库和任何样式的库配合使用以与数据库通信。
一个常见的模式是使用“ORM”：一个“对象关系映射”库。

对象-关系映射(Object Relational Mapping,简称ORM,对象关系映射)是一种为了解决面向对象与关系数据库存在的互不匹配的现象的技术。 简单的说，ORM是通过使用描述对象和数据库之间映射的元数据，将程序中的对象自动持久化到关系数据库中。本质上就是将数据从一种形式转换到另外一种形式。

ORM 具有在代码和数据库表（“关系”）中的对象之间进行转换（“映射”）的工具。
使用 ORM，通常会创建一个表示 SQL 数据库中的表的类，该类的每个属性都表示一个列，具有名称和类型。

二、文件结构

假设您有一个名为的目录my_super_project，其中包含一个名为的子目录sql_app，其结构如下：

my_super_project
└── .vscode
    ├── launch.json
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

三、配置调试环境launch.json

在开始调试前，我们需要进行一些配置。首先，我们需要安装并启动一个插件——Python调试器。在VSCode市场中搜索并安装Python插件。然后，在VSCode的侧边栏中打开项目文件夹。接下来，我们需要创建一个launch.json文件，用于配置调试器。在VSCode的侧边栏中点击调试按钮，然后选择“创建一个配置文件”-“Python”。在生成的launch.json文件中，我们可以根据需要进行一些调试选项的配置，例如调试入口文件、环境变量等。

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python 调试程序: FastAPI", // 设置调试配置的名称。将会在启动配置的下拉菜单中显示。
            "type": "debugpy", //指定调试器的类型为debugpy。是 vs code 用于计算调试代码需要用哪个扩展。
            "request": "launch",
            "module": "uvicorn", 
            "args": [ 
                "main:app",
                "--reload"
            ],
            "cwd": "${workspaceRoot}/sql_app/", 
            "jinja": true
        }
    ]
}

四、程序文件代码

1、sql_app/__init__.py

__init__.py只是一个空文件，但它告诉 Python，sql_app它的所有模块（Python 文件）都是一个包。

2.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 创建一个数据库 URL
# “连接”到一个 SQLite 数据库（用 SQLite 数据库打开一个文件）。
# 该文件将位于文件中的同一目录中sql_app.db。
# 当前工作目录(根目录)./sql_app.db.

engine = create_engine(  # 创建一个 SQLAlchemy“引擎”
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    # connect_args={"check_same_thread": False}仅用于SQLite. 其他数据库不需要它
    # 技术细节
    # 默认情况下，SQLite 将只允许一个线程与其通信，假设每个线程将处理一个独立的请求。
    # 这是为了防止意外地为不同的事物（对于不同的请求）共享相同的连接。
    # 但是在 FastAPI 中，使用普通函数 ( def) 可以针对同一个请求与数据库交互多个线程，
    # 因此我们需要让 SQLite 知道它应该允许使用connect_args={"check_same_thread": False}.
    # 此外，我们将确保每个请求在依赖项中都有自己的数据库连接会话，因此不需要该默认机制。
   echo=True,  # 参数表示连接发出的SQL将被记录为标准输出。运行代码能直接在命令行打印出对应的SQL语句,查看对应的SQL语句，有助于排错.
)

SessionLocal = sessionmaker(
    autocommit=False, autoflush=False, bind=engine
)  # 创建一个SessionLocal
# SessionLocal该类的每个实例都是一个数据库会话。该类本身还不是数据库会话。
# 但是一旦我们创建了一个SessionLocal类的实例，这个实例就会成为实际的数据库会话。
# 我们命名它SessionLocal以区别于Session我们从 SQLAlchemy 导入的。
# 要创建SessionLocal类，请使用函数sessionmaker：

Base = declarative_base()  # 创建一个Base
# 使用declarative_base()返回一个类的函数。将从这个类继承来创建每个数据库模型或类（ORM 模型）。

3.sql_app/models.py:

# 创建数据库模型

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

from database import Base #从database.py文件中导入Base类

# 从Base类创建 SQLAlchemy 模型
"""
我们将使用Base我们之前创建的这个类来创建 SQLAlchemy 模型。
提示
SQLAlchemy 使用术语“模型”来指代与数据库交互的这些类和实例。
但是 Pydantic 也使用术语“模型”来指代不同的东西，数据验证、转换以及文档类和实例。
Base从database（database.py上面的文件）导入。
创建从它继承的类。
这些类是 SQLAlchemy 模型。
"""


class User(Base):
    __tablename__ = "users"  # 该__tablename__属性告诉 SQLAlchemy 在数据库中为这些模型中的每一个使用的表的名称。

    # 创建模型属性/列
    # 现在创建所有模型（类）属性。
    # 这些属性中的每一个都代表其相应数据库表中的一列。
    # 我们使用ColumnSQLAlchemy 作为默认值。
    # 而我们通过SQLAlchemy的类“类型”，如Integer，String和Boolean，它定义了数据库的类型，作为参数。

    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")  # 创建关系


# 使用relationship方法由 SQLAlchemy ORM 提供的。
# 这将或多或少成为一个“神奇”属性，其中包含与此相关的其他表中的值。
# 当访问中的属性items在User中my_user.items，它将有一个Item 、SQLAlchemy 模型列表（来自items表），这些模型具有指向users表中此记录的外键。
# 当您访问 时my_user.items，SQLAlchemy 实际上会从items表中的数据库中获取项目并在此处填充它们。
# 并且在访问 中的属性owner时Item，它将包含表中的UserSQLAlchemy 模型users。它将使用owner_id带有外键的属性/列来知道从users表中获取哪条记录。


class Item(Base):
    __tablename__ = "items"  # 该__tablename__属性告诉 SQLAlchemy 在数据库中为这些模型中的每一个使用的表的名称。

    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"))
    #ForeignKey外键，将无连接的多张表，转换为具有丰富重叠关系的集合。
    #上面的ForeignKey表示，Item.owner_id 列中的值应该被约束到users.id列中的那些值，即其主键。
    owner = relationship("User", back_populates="items")
    #两张表都增加relationship之后，relationship就可以智能的决定，该怎么连接到对方。
    #在Item这一边，Item.owner_id 属性指向User的（一个）实例，在User这一边，User.items 属性指向一组Item的实例。

4.sql_app/schemas.py:

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

from typing import List, Optional

from pydantic import BaseModel

# SQLAlchemy 风格和 Pydantic 风格
# 请注意，SQLAlchemy模型使用 定义属性=，并将类型作为参数传递给Column，例如：
# name = Column(String)
# Pydantic模型使用声明类型 : ，但新的类型注释语法/类型提示：
# name: str
# 牢记这一点，这样在使用=和:使用它们时就不会感到困惑。

# 创建初始 Pydantic模型/模式
# 创建一个ItemBase和UserBase的Pydantic模型（或者说“模式”）以在创建或读取数据时具有共同的属性。
# 并创建一个继承自它们的ItemCreate和UserCreate（因此它们将具有相同的属性），以及创建所需的任何其他数据（属性）。
# 因此，用户password在创建它时也会有一个。
# 但是为了安全起见，password其他 Pydantic模型中不会出现，例如在读取用户时不会从 API 发送。


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

5.sql_app/crud.py:

# CRUD 工具
# 可重用的函数来与数据库中的数据进行交互


# 导入schemas（Pydantic模型/模式），
# Session从导入sqlalchemy.orm，这将允许声明db参数的类型，并在您的函数中进行更好的类型检查和完成。
from sqlalchemy.orm import Session

# 导入models（SQLAlchemy 模型）
import models, schemas


def get_user(db: Session, user_id: int):  # 创建实用函数读取数据，通过 ID 读取单个用户。
    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"
    # SQLAlchemy 模型User包含一个hashed_password应该包含密码的安全散列版本。
    # 但是由于 API 客户端提供的是原始密码，因此要将其提取并在应用程序中生成散列密码。
    # 然后传递hashed_password带有要保存的值的参数。
    # 警告
    # 这个例子不安全，密码没有散列。
    # 在现实生活中的应用程序中，需要对密码进行哈希处理，并且永远不要以明文形式保存它们。
    db_user = models.User(
        email=user.email, hashed_password=fake_hashed_password
    )  # 使用数据创建 SQLAlchemy 模型实例。
    db.add(db_user)  # add 该实例对象到数据库会话。
    db.commit()  # commit 对数据库的更改（以便保存）。
    db.refresh(
        db_user
    )  # refresh实例（以便它包含来自数据库的任何新数据，例如生成的 ID）。
    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.model_dump(), owner_id=user_id
    )  # 使用dict的键值对作为关键字参数传递给 SQLAlchemy模型实例Item。
    # 传递owner_id作为Pydantic模型未提供的额外关键字参数。
    # model_dump()函数代替dict()函数。
    db.add(db_item)  # add 该实例对象到数据库会话。
    db.commit()  # commit 对数据库的更改（以便保存）。
    db.refresh(
        db_item
    )  # refresh实例（以便它包含来自数据库的任何新数据，例如生成的 ID）。
    return db_item

6.sql_app/main.py

# 主要FastAPI应用程序
# main.py集成并使用创建的所有其他部分。

from typing import List

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

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

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

app = FastAPI()


# 创建依赖（创建一个数据库会话），使用sql_app/databases.py文件中创建的SessionLocal类来创建依赖项。
def get_db():
    db = (
        SessionLocal()
    )  # SessionLocal每个请求有一个独立的数据库会话/连接，在所有请求中使用同一个会话，然后在请求完成后关闭它。
    try:  # SessionLocal()请求的创建和处理放在一个try块中。
        yield db  # 创建一个新的依赖关系yield。
    finally:  # finally块中关闭SessionLocal()请求，
        db.close()  # 确保单个请求完成后数据库会话总是关闭。


# 所有路径操作都 response_model 使用 Pydantic模型/模式orm_mode，
# 因此 Pydantic 模型中声明的数据将从它们中提取并返回给客户端，并进行所有正常的过滤和验证。
@app.post("/users/", response_model=schemas.User)  # 创建标准的FastAPI 路径操作代码。
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])
# response_models标准的 Python 类型，如List[schemas.Item].
# 但是作为内容/的该参数List是一个Pydantic模型与orm_mode，该数据将被检索并返回到客户端为常，没有任何问题。
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