一、ORM

FastAPI可與任何數(shù)據(jù)庫和任何樣式的庫配合使用以與數(shù)據(jù)庫通信。
一個常見的模式是使用“ORM”：一個“對象關系映射”庫。

對象-關系映射(Object Relational Mapping,簡稱ORM,對象關系映射)是一種為了解決面向對象與關系數(shù)據(jù)庫存在的互不匹配的現(xiàn)象的技術。 簡單的說，ORM是通過使用描述對象和數(shù)據(jù)庫之間映射的元數(shù)據(jù)，將程序中的對象自動持久化到關系數(shù)據(jù)庫中。本質上就是將數(shù)據(jù)從一種形式轉換到另外一種形式。

ORM 具有在代碼和數(shù)據(jù)庫表（“關系”）中的對象之間進行轉換（“映射”）的工具。
使用 ORM，通常會創(chuàng)建一個表示 SQL 數(shù)據(jù)庫中的表的類，該類的每個屬性都表示一個列，具有名稱和類型。

二、文件結構

假設您有一個名為的目錄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

三、配置調(diào)試環(huán)境launch.json

在開始調(diào)試前，我們需要進行一些配置。首先，我們需要安裝并啟動一個插件——Python調(diào)試器。在VSCode市場中搜索并安裝Python插件。然后，在VSCode的側邊欄中打開項目文件夾。接下來，我們需要創(chuàng)建一個launch.json文件，用于配置調(diào)試器。在VSCode的側邊欄中點擊調(diào)試按鈕，然后選擇“創(chuàng)建一個配置文件”-“Python”。在生成的launch.json文件中，我們可以根據(jù)需要進行一些調(diào)試選項的配置，例如調(diào)試入口文件、環(huán)境變量等。

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python 調(diào)試程序: FastAPI", // 設置調(diào)試配置的名稱。將會在啟動配置的下拉菜單中顯示。
            "type": "debugpy", //指定調(diào)試器的類型為debugpy。是 vs code 用于計算調(diào)試代碼需要用哪個擴展。
            "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

# 創(chuàng)建 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 創(chuàng)建一個數(shù)據(jù)庫 URL
# “連接”到一個 SQLite 數(shù)據(jù)庫（用 SQLite 數(shù)據(jù)庫打開一個文件）。
# 該文件將位于文件中的同一目錄中sql_app.db。
# 當前工作目錄(根目錄)./sql_app.db.

engine = create_engine(  # 創(chuàng)建一個 SQLAlchemy“引擎”
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    # connect_args={"check_same_thread": False}僅用于SQLite. 其他數(shù)據(jù)庫不需要它
    # 技術細節(jié)
    # 默認情況下，SQLite 將只允許一個線程與其通信，假設每個線程將處理一個獨立的請求。
    # 這是為了防止意外地為不同的事物（對于不同的請求）共享相同的連接。
    # 但是在 FastAPI 中，使用普通函數(shù) ( def) 可以針對同一個請求與數(shù)據(jù)庫交互多個線程，
    # 因此我們需要讓 SQLite 知道它應該允許使用connect_args={"check_same_thread": False}.
    # 此外，我們將確保每個請求在依賴項中都有自己的數(shù)據(jù)庫連接會話，因此不需要該默認機制。
   echo=True,  # 參數(shù)表示連接發(fā)出的SQL將被記錄為標準輸出。運行代碼能直接在命令行打印出對應的SQL語句,查看對應的SQL語句，有助于排錯.
)

SessionLocal = sessionmaker(
    autocommit=False, autoflush=False, bind=engine
)  # 創(chuàng)建一個SessionLocal
# SessionLocal該類的每個實例都是一個數(shù)據(jù)庫會話。該類本身還不是數(shù)據(jù)庫會話。
# 但是一旦我們創(chuàng)建了一個SessionLocal類的實例，這個實例就會成為實際的數(shù)據(jù)庫會話。
# 我們命名它SessionLocal以區(qū)別于Session我們從 SQLAlchemy 導入的。
# 要創(chuàng)建SessionLocal類，請使用函數(shù)sessionmaker：

Base = declarative_base()  # 創(chuàng)建一個Base
# 使用declarative_base()返回一個類的函數(shù)。將從這個類繼承來創(chuàng)建每個數(shù)據(jù)庫模型或類（ORM 模型）。

3.sql_app/models.py:

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

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

from database import Base #從database.py文件中導入Base類

# 從Base類創(chuàng)建 SQLAlchemy 模型
"""
我們將使用Base我們之前創(chuàng)建的這個類來創(chuàng)建 SQLAlchemy 模型。
提示
SQLAlchemy 使用術語“模型”來指代與數(shù)據(jù)庫交互的這些類和實例。
但是 Pydantic 也使用術語“模型”來指代不同的東西，數(shù)據(jù)驗證、轉換以及文檔類和實例。
Base從database（database.py上面的文件）導入。
創(chuàng)建從它繼承的類。
這些類是 SQLAlchemy 模型。
"""


class User(Base):
    __tablename__ = "users"  # 該__tablename__屬性告訴 SQLAlchemy 在數(shù)據(jù)庫中為這些模型中的每一個使用的表的名稱。

    # 創(chuàng)建模型屬性/列
    # 現(xiàn)在創(chuàng)建所有模型（類）屬性。
    # 這些屬性中的每一個都代表其相應數(shù)據(jù)庫表中的一列。
    # 我們使用ColumnSQLAlchemy 作為默認值。
    # 而我們通過SQLAlchemy的類“類型”，如Integer，String和Boolean，它定義了數(shù)據(jù)庫的類型，作為參數(shù)。

    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")  # 創(chuàng)建關系


# 使用relationship方法由 SQLAlchemy ORM 提供的。
# 這將或多或少成為一個“神奇”屬性，其中包含與此相關的其他表中的值。
# 當訪問中的屬性items在User中my_user.items，它將有一個Item 、SQLAlchemy 模型列表（來自items表），這些模型具有指向users表中此記錄的外鍵。
# 當您訪問 時my_user.items，SQLAlchemy 實際上會從items表中的數(shù)據(jù)庫中獲取項目并在此處填充它們。
# 并且在訪問 中的屬性owner時Item，它將包含表中的UserSQLAlchemy 模型users。它將使用owner_id帶有外鍵的屬性/列來知道從users表中獲取哪條記錄。


class Item(Base):
    __tablename__ = "items"  # 該__tablename__屬性告訴 SQLAlchemy 在數(shù)據(jù)庫中為這些模型中的每一個使用的表的名稱。

    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:

# 創(chuàng)建 Pydantic 模型
# 提示
# 為了避免 SQLAlchemy模型和 Pydantic模型之間的混淆，我們將使用models.py帶有 SQLAlchemy 模型的文件schemas.py和帶有 Pydantic 模型的文件。
# 這些 Pydantic 模型或多或少地定義了一個“模式”（有效的數(shù)據(jù)形狀）。
# 因此，這將有助于我們在使用兩者時避免混淆。

from typing import List, Optional

from pydantic import BaseModel

# SQLAlchemy 風格和 Pydantic 風格
# 請注意，SQLAlchemy模型使用 定義屬性=，并將類型作為參數(shù)傳遞給Column，例如：
# name = Column(String)
# Pydantic模型使用聲明類型 : ，但新的類型注釋語法/類型提示：
# name: str
# 牢記這一點，這樣在使用=和:使用它們時就不會感到困惑。

# 創(chuàng)建初始 Pydantic模型/模式
# 創(chuàng)建一個ItemBase和UserBase的Pydantic模型（或者說“模式”）以在創(chuàng)建或讀取數(shù)據(jù)時具有共同的屬性。
# 并創(chuàng)建一個繼承自它們的ItemCreate和UserCreate（因此它們將具有相同的屬性），以及創(chuàng)建所需的任何其他數(shù)據(jù)（屬性）。
# 因此，用戶password在創(chuàng)建它時也會有一個。
# 但是為了安全起見，password其他 Pydantic模型中不會出現(xiàn)，例如在讀取用戶時不會從 API 發(fā)送。


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 工具
# 可重用的函數(shù)來與數(shù)據(jù)庫中的數(shù)據(jù)進行交互


# 導入schemas（Pydantic模型/模式），
# Session從導入sqlalchemy.orm，這將允許聲明db參數(shù)的類型，并在您的函數(shù)中進行更好的類型檢查和完成。
from sqlalchemy.orm import Session

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


def get_user(db: Session, user_id: int):  # 創(chuàng)建實用函數(shù)讀取數(shù)據(jù)，通過 ID 讀取單個用戶。
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(
    db: Session, email: str
):  # 創(chuàng)建實用函數(shù)讀取數(shù)據(jù)，通過電子郵件讀取單個用戶。
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(
    db: Session, skip: int = 0, limit: int = 100
):  ##創(chuàng)建實用函讀取數(shù)據(jù)，讀取多個用戶。
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):  # 創(chuàng)建實用函數(shù)來創(chuàng)建數(shù)據(jù)。
    fake_hashed_password = user.password + "notreallyhashed"
    # SQLAlchemy 模型User包含一個hashed_password應該包含密碼的安全散列版本。
    # 但是由于 API 客戶端提供的是原始密碼，因此要將其提取并在應用程序中生成散列密碼。
    # 然后傳遞hashed_password帶有要保存的值的參數(shù)。
    # 警告
    # 這個例子不安全，密碼沒有散列。
    # 在現(xiàn)實生活中的應用程序中，需要對密碼進行哈希處理，并且永遠不要以明文形式保存它們。
    db_user = models.User(
        email=user.email, hashed_password=fake_hashed_password
    )  # 使用數(shù)據(jù)創(chuàng)建 SQLAlchemy 模型實例。
    db.add(db_user)  # add 該實例對象到數(shù)據(jù)庫會話。
    db.commit()  # commit 對數(shù)據(jù)庫的更改（以便保存）。
    db.refresh(
        db_user
    )  # refresh實例（以便它包含來自數(shù)據(jù)庫的任何新數(shù)據(jù)，例如生成的 ID）。
    return db_user


def get_items(
    db: Session, skip: int = 0, limit: int = 100
):  # 創(chuàng)建實用函數(shù)，閱讀多個項目。
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(
    db: Session, item: schemas.ItemCreate, user_id: int
):  # 創(chuàng)建實用函數(shù)來創(chuàng)建數(shù)據(jù)。
    db_item = models.Item(
        **item.model_dump(), owner_id=user_id
    )  # 使用dict的鍵值對作為關鍵字參數(shù)傳遞給 SQLAlchemy模型實例Item。
    # 傳遞owner_id作為Pydantic模型未提供的額外關鍵字參數(shù)。
    # model_dump()函數(shù)代替dict()函數(shù)。
    db.add(db_item)  # add 該實例對象到數(shù)據(jù)庫會話。
    db.commit()  # commit 對數(shù)據(jù)庫的更改（以便保存）。
    db.refresh(
        db_item
    )  # refresh實例（以便它包含來自數(shù)據(jù)庫的任何新數(shù)據(jù)，例如生成的 ID）。
    return db_item

6.sql_app/main.py

# 主要FastAPI應用程序
# main.py集成并使用創(chuàng)建的所有其他部分。

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


# 創(chuàng)建依賴（創(chuàng)建一個數(shù)據(jù)庫會話），使用sql_app/databases.py文件中創(chuàng)建的SessionLocal類來創(chuàng)建依賴項。
def get_db():
    db = (
        SessionLocal()
    )  # SessionLocal每個請求有一個獨立的數(shù)據(jù)庫會話/連接，在所有請求中使用同一個會話，然后在請求完成后關閉它。
    try:  # SessionLocal()請求的創(chuàng)建和處理放在一個try塊中。
        yield db  # 創(chuàng)建一個新的依賴關系yield。
    finally:  # finally塊中關閉SessionLocal()請求，
        db.close()  # 確保單個請求完成后數(shù)據(jù)庫會話總是關閉。


# 所有路徑操作都 response_model 使用 Pydantic模型/模式orm_mode，
# 因此 Pydantic 模型中聲明的數(shù)據(jù)將從它們中提取并返回給客戶端，并進行所有正常的過濾和驗證。
@app.post("/users/", response_model=schemas.User)  # 創(chuàng)建標準的FastAPI 路徑操作代碼。
def create_user(
    user: schemas.UserCreate, db: Session = Depends(get_db)
):  # 在路徑操作函數(shù)中創(chuàng)建所需的依賴項，直接獲取該數(shù)據(jù)庫會話。
    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].
# 但是作為內(nèi)容/的該參數(shù)List是一個Pydantic模型與orm_mode，該數(shù)據(jù)將被檢索并返回到客戶端為常，沒有任何問題。
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