建站
咨詢
售后
建站咨詢
新聞中心
FastAPI不要求您使用 SQL（關(guān)系）數(shù)據(jù)庫。
但是您可以使用任何您想要的關(guān)系數(shù)據(jù)庫。
在這里，我們將看到一個使用SQLAlchemy的示例。
您可以輕松地將其調(diào)整為 SQLAlchemy 支持的任何數(shù)據(jù)庫，例如：
- PostgreSQL
- MySQL
- SQLite
- Oracle
- Microsoft SQL Server 等
在此示例中，我們將使用SQLite，因為它使用單個文件并且 Python 已集成支持。因此，您可以復(fù)制此示例并按原樣運行它。
稍后，對于您的生產(chǎn)應(yīng)用程序，您可能希望使用像PostgreSQL這樣的數(shù)據(jù)庫服務(wù)器。
提示
有一個帶有FastAPI和PostgreSQL的官方項目生成器，全部基于Docker，包括前端和更多工具：https : //github.com/tiangolo/full-stack-fastapi-postgresql
筆記
請注意，大部分代碼是SQLAlchemy您將用于任何框架的標準代碼。
該FastAPI具體的代碼是小一如既往。
ORM
FastAPI可與任何數(shù)據(jù)庫和任何樣式的庫配合使用以與數(shù)據(jù)庫通信。
一個常見的模式是使用“ORM”：一個“對象關(guān)系映射”庫。
ORM 具有在代碼和數(shù)據(jù)庫表（“關(guān)系”）中的對象之間進行轉(zhuǎn)換（“映射”）的工具。
使用 ORM，您通常會創(chuàng)建一個表示 SQL 數(shù)據(jù)庫中的表的類，該類的每個屬性都表示一個列，具有名稱和類型。
例如，一個類Pet可以代表一個 SQL 表pets。
并且該類的每個實例對象代表數(shù)據(jù)庫中的一行。
例如，一個對象orion_cat（ 的實例Pet）可以有一個屬性orion_cat.type，用于列type。該屬性的值可以是，例如"cat"。
這些 ORM 還具有在表或?qū)嶓w之間建立連接或關(guān)系的工具。
這樣，您也可以擁有一個屬性orion_cat.owner，所有者將包含該寵物所有者的數(shù)據(jù)，取自表owner。
所以，orion_cat.owner.name可能是這個寵物主人的名字（來自表中的name列owners）。
它可能具有類似"Arquilian".
當您嘗試從您的寵物對象訪問它時，ORM 將完成所有工作以從相應(yīng)的表所有者那里獲取信息。
常見的ORM有例如：Django-ORM（Django框架的一部分）、SQLAlchemy ORM（SQLAlchemy的一部分，獨立于框架）和Peewee（獨立于框架）等。
在這里，我們將看到如何使用SQLAlchemy ORM。
以類似的方式，您可以使用任何其他 ORM。
提示
文檔中有一篇使用 Peewee 的等效文章。
文件結(jié)構(gòu)
對于這些示例，假設(shè)您有一個名為的目錄my_super_project，其中包含一個名為的子目錄sql_app，其結(jié)構(gòu)如下：
.
└── sql_app
├── __init__.py
├── crud.py
├── database.py
├── main.py
├── models.py
└── schemas.py
該文件__init__.py只是一個空文件，但它告訴 Python，sql_app它的所有模塊（Python 文件）都是一個包。
現(xiàn)在讓我們看看每個文件/模塊的作用。
創(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)建一個數(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()
在這個例子中，我們“連接”到一個 SQLite 數(shù)據(jù)庫（用 SQLite 數(shù)據(jù)庫打開一個文件）。
該文件將位于文件中的同一目錄中sql_app.db。
這就是為什么最后一部分是./sql_app.db.
如果您使用的是PostgreSQL數(shù)據(jù)庫，則只需取消注釋該行：
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
...并使用您的數(shù)據(jù)庫數(shù)據(jù)和憑據(jù)（相當于 MySQL、MariaDB 或任何其他）對其進行調(diào)整。
提示
如果您想使用不同的數(shù)據(jù)庫，這是必須修改的主線。
創(chuàng)建 SQLAlchemy engine
第一步是創(chuàng)建一個 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ù)細節(jié)
默認情況下，SQLite 將只允許一個線程與其通信，假設(shè)每個線程將處理一個獨立的請求。
這是為了防止意外地為不同的事物（對于不同的請求）共享相同的連接。
但是在 FastAPI 中，使用普通函數(shù) ( def) 可以針對同一個請求與數(shù)據(jù)庫交互多個線程，因此我們需要讓 SQLite 知道它應(yīng)該允許使用connect_args={"check_same_thread": False}.
此外，我們將確保每個請求在依賴項中都有自己的數(shù)據(jù)庫連接會話，因此不需要該默認機制。
創(chuàng)建一個SessionLocal班級
SessionLocal該類的每個實例都是一個數(shù)據(jù)庫會話。該類本身還不是數(shù)據(jù)庫會話。
但是一旦我們創(chuàng)建了一個SessionLocal類的實例，這個實例就會成為實際的數(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)建一個Base班級
現(xiàn)在我們將使用declarative_base()返回一個類的函數(shù)。
稍后我們將從這個類繼承來創(chuàng)建每個數(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)建的這個類來創(chuàng)建 SQLAlchemy 模型。
提示
SQLAlchemy 使用術(shù)語“模型”來指代與數(shù)據(jù)庫交互的這些類和實例。
但是 Pydantic 也使用術(shù)語“模型”來指代不同的東西，數(shù)據(jù)驗證、轉(zhuǎn)換以及文檔類和實例。
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ù)庫中為這些模型中的每一個使用的表的名稱。
創(chuàng)建模型屬性/列
現(xiàn)在創(chuàng)建所有模型（類）屬性。
這些屬性中的每一個都代表其相應(yīng)數(shù)據(jù)庫表中的一列。
我們使用ColumnSQLAlchemy 作為默認值。
而我們通過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 提供的。
這將或多或少成為一個“神奇”屬性，其中包含與此相關(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")
當訪問中的屬性items在User中my_user.items，它將有一個ItemSQLAlchemy 模型列表（來自items表），這些模型具有指向users表中此記錄的外鍵。
當您訪問 時my_user.items，SQLAlchemy 實際上會從items表中的數(shù)據(jù)庫中獲取項目并在此處填充它們。
并且在訪問 中的屬性owner時Item，它將包含表中的UserSQLAlchemy 模型users。它將使用owner_id帶有外鍵的屬性/列來知道從users表中獲取哪條記錄。
創(chuàng)建 Pydantic 模型
現(xiàn)在讓我們檢查文件sql_app/schemas.py。
提示
為了避免 SQLAlchemy模型和 Pydantic模型之間的混淆，我們將使用models.py帶有 SQLAlchemy 模型的文件schemas.py和帶有 Pydantic 模型的文件。
這些 Pydantic 模型或多或少地定義了一個“模式”（有效的數(shù)據(jù)形狀）。
因此，這將有助于我們在使用兩者時避免混淆。
創(chuàng)建初始 Pydantic模型/模式
創(chuàng)建一個ItemBase和UserBasePydantic模型（或者說“模式”）以在創(chuàng)建或讀取數(shù)據(jù)時具有共同的屬性。
并創(chuàng)建一個繼承自它們的ItemCreateand UserCreate（因此它們將具有相同的屬性），以及創(chuàng)建所需的任何其他數(shù)據(jù)（屬性）。
因此，用戶password在創(chuàng)建它時也會有一個。
但是為了安全起見，password其他 Pydantic模型中不會出現(xiàn)，例如在讀取用戶時不會從 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
牢記這一點，這樣您在使用=和:使用它們時就不會感到困惑。
創(chuàng)建用于讀取/返回的Pydantic模型/模式
現(xiàn)在創(chuàng)建將在讀取數(shù)據(jù)時使用的Pydantic模型（模式），當從 API 返回數(shù)據(jù)時。
例如，在創(chuàng)建項目之前，我們不知道分配給它的 ID 是什么，但是在讀取它時（從 API 返回它時）我們已經(jīng)知道它的 ID。
同樣，在讀取用戶時，我們現(xiàn)在可以聲明items將包含屬于該用戶的項目。
不僅這些項目的ID，但我們在Pydantic中定義的所有數(shù)據(jù)模型讀取項目：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
現(xiàn)在，在Pydantic模型讀取，Item并且User，添加一個內(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
提示
請注意，它正在分配一個值=，例如：
orm_mode = True
它不:用于之前的類型聲明。
這是設(shè)置配置值，而不是聲明類型。
Pydanticorm_mode會告訴 Pydantic模型讀取數(shù)據(jù)，即使它不是dict，而是 ORM 模型（或任何其他具有屬性的任意對象）。
這樣，而不是僅僅嘗試id從 a獲取值dict，如下所示：
id = data["id"]
它還會嘗試從屬性中獲取它，例如：
id = data.id
有了這個，Pydantic模型與 ORM 兼容，您只需response_model在路徑操作的參數(shù)中聲明它。
您將能夠返回一個數(shù)據(jù)庫模型，它會從中讀取數(shù)據(jù)。
ORM模式的技術(shù)細節(jié)
SQLAlchemy 和許多其他的默認情況下是“延遲加載”。
這意味著，例如，除非您嘗試訪問包含該數(shù)據(jù)的屬性，否則它們不會從數(shù)據(jù)庫中獲取關(guān)系數(shù)據(jù)。
例如，訪問屬性items：
current_user.items
將使 SQLAlchemy 轉(zhuǎn)到該items表并獲取該用戶的項目，但不是之前。
沒有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。
在這個文件中，我們將有可重用的函數(shù)來與數(shù)據(jù)庫中的數(shù)據(jù)進行交互。
CRUD來源于：? reate，- [R EAD，ù PDATE，和d elete。
...雖然在這個例子中我們只是創(chuàng)建和閱讀。
讀取數(shù)據(jù)
Session從導(dǎo)入sqlalchemy.orm，這將允許您聲明db參數(shù)的類型，并在您的函數(shù)中進行更好的類型檢查和完成。
導(dǎo)入models（SQLAlchemy 模型）和schemas（Pydantic模型/模式）。
創(chuàng)建實用函數(shù)以：
- 通過 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
提示
通過創(chuàng)建獨立于您的路徑操作函數(shù)的僅專用于與數(shù)據(jù)庫交互（獲取用戶或項目）的函數(shù)，您可以更輕松地在多個部分中重用它們，并為它們添加單元測試。
創(chuàng)建數(shù)據(jù)
現(xiàn)在創(chuàng)建實用函數(shù)來創(chuàng)建數(shù)據(jù)。
步驟是：
- 使用您的數(shù)據(jù)創(chuàng)建 SQLAlchemy 模型實例。
- add 該實例對象到您的數(shù)據(jù)庫會話。
- commit 對數(shù)據(jù)庫的更改（以便保存）。
- refresh 您的實例（以便它包含來自數(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包含一個hashed_password應(yīng)該包含密碼的安全散列版本。
但是由于 API 客戶端提供的是原始密碼，因此您需要將其提取并在您的應(yīng)用程序中生成散列密碼。
然后傳遞hashed_password帶有要保存的值的參數(shù)。
警告
這個例子不安全，密碼沒有散列。
在現(xiàn)實生活中的應(yīng)用程序中，您需要對密碼進行哈希處理，并且永遠不要以明文形式保存它們。
有關(guān)更多詳細信息，請返回教程中的安全部分。
在這里，我們只關(guān)注數(shù)據(jù)庫的工具和機制。
提示
我們沒有將每個關(guān)鍵字參數(shù)傳遞給ItemPydantic模型并從中讀取每個參數(shù)，而是dict使用 Pydantic模型的數(shù)據(jù)生成一個：
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 進行“遷移”（這是它的主要工作）。
“遷移”是每當您更改 SQLAlchemy 模型的結(jié)構(gòu)、添加新屬性等以在數(shù)據(jù)庫中復(fù)制這些更改、添加新列、新表等時所需的一組步驟。
您可以在Project Generation-Template的模板中找到 FastAPI 項目中的 Alembic 示例。具體地，在所述alembic源代碼中的目錄。
創(chuàng)建依賴
現(xiàn)在使用SessionLocal我們在sql_app/databases.py文件中創(chuàng)建的類來創(chuàng)建依賴項。
我們需要SessionLocal每個請求有一個獨立的數(shù)據(jù)庫會話/連接（），在所有請求中使用同一個會話，然后在請求完成后關(guān)閉它。
然后將為下一個請求創(chuàng)建一個新會話。
為此，我們將創(chuàng)建一個新的依賴關(guān)系yield，正如之前關(guān)于依賴關(guān)系yield的部分所解釋的那樣。
我們的依賴將創(chuàng)建一個新的 SQLAlchemy SessionLocal，它將在單個請求中使用，然后在請求完成后關(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)建和處理放在一個try塊中。
然后我們在finally塊中關(guān)閉它。
這樣我們可以確保在請求之后數(shù)據(jù)庫會話總是關(guān)閉。即使在處理請求時出現(xiàn)異常。
但是您不能從退出代碼（之后yield）引發(fā)另一個異常。在依賴項中查看更多信息和yieldHTTPException
然后，在路徑操作函數(shù)中使用依賴項時，我們使用Session直接從 SQLAlchemy 導(dǎo)入的類型聲明它。
這將在路徑操作函數(shù)中為我們提供更好的編輯器支持，因為編輯器將知道db參數(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ù)細節(jié)
參數(shù)db實際上是 type SessionLocal，但是這個類（創(chuàng)建于sessionmaker()）是 SQLAlchemy 的“代理” Session，因此，編輯器并不真正知道提供了哪些方法。
但是，作為申報類型Session，編輯器現(xiàn)在可以知道可用的方法（.add()，.query()，.commit()等），并能提供更好的支持（如完成）。類型聲明不影響實際對象。
創(chuàng)建您的FastAPI 路徑操作
現(xià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
我們在依賴項中的每個請求之前創(chuàng)建數(shù)據(jù)庫會話yield，然后關(guān)閉它。
然后我們可以在路徑操作函數(shù)中創(chuàng)建所需的依賴項，直接獲取該會話。
這樣，我們就可以crud.get_user直接從路徑操作函數(shù)內(nèi)部調(diào)用并使用該會話。
提示
請注意，您返回的值是 SQLAlchemy 模型或 SQLAlchemy 模型列表。
但是由于所有路徑操作都response_model使用 Pydantic模型/模式orm_mode，因此 Pydantic 模型中聲明的數(shù)據(jù)將從它們中提取并返回給客戶端，并進行所有正常的過濾和驗證。
提示
還要注意，有response_models標準的 Python 類型，如List[schemas.Item].
但是作為內(nèi)容/的該參數(shù)List是一個Pydantic模型與orm_mode，該數(shù)據(jù)將被檢索并返回到客戶端為常，沒有任何問題。
關(guān)于defvsasync def
在這里，我們在路徑操作函數(shù)和依賴項中使用 SQLAlchemy 代碼，反過來，它將與外部數(shù)據(jù)庫進行通信。
這可能需要一些“等待”。
但是由于 Sqlalchemy 不具有await直接使用的兼容性，就像使用以下內(nèi)容一樣：
user = await db.query(User).first()
...而我們正在使用：
user = db.query(User).first()
然后我們應(yīng)該聲明路徑操作函數(shù)和不帶 的依賴項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)
...
信息
如果您需要異步連接到關(guān)系數(shù)據(jù)庫，請參閱異步 SQL（關(guān)系）數(shù)據(jù)庫。
非常技術(shù)性的細節(jié)
如果您很好奇并且擁有深厚的技術(shù)知識，您可以在Async文檔中查看有關(guān)如何處理此async defvs的非常技術(shù)性的細節(jié)。def
遷移
因為我們直接使用 SQLAlchemy 并且我們不需要任何插件來使用FastAPI，所以我們可以直接將數(shù)據(jù)庫遷移與Alembic集成。
由于與 SQLAlchemy 和 SQLAlchemy 模型相關(guān)的代碼存在于單獨的獨立文件中，您甚至可以使用 Alembic 執(zhí)行遷移，而無需安裝 FastAPI、Pydantic 或其他任何東西。
同樣，您將能夠在與FastAPI無關(guān)的代碼的其他部分中使用相同的 SQLAlchemy 模型和實用程序。
例如，在帶有Celery、RQ或ARQ的后臺任務(wù)工作者中。
查看所有文件
請記住，您應(yīng)該有一個名為的目錄my_super_project，其中包含一個名為sql_app.
sql_app 應(yīng)該有以下文件：
- 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
核實
您可以復(fù)制此代碼并按原樣使用。
信息
事實上，此處顯示的代碼是測試的一部分。作為這些文檔中的大部分代碼。
然后你可以用 Uvicorn 運行它：
uvicorn sql_app.main:app --reload
信息
：Uvicorn 在 http://127.0.0.1:8000 上運行（按 CTRL+C 退出）
重啟?
然后，您可以在http://127.0.0.1:8000/docs 上打開瀏覽器。
您將能夠與您的FastAPI應(yīng)用程序交互，從真實數(shù)據(jù)庫中讀取數(shù)據(jù)：
直接與數(shù)據(jù)庫交互
如果您想獨立于 FastAPI 直接探索 SQLite 數(shù)據(jù)庫（文件），以調(diào)試其內(nèi)容，添加表、列、記錄、修改數(shù)據(jù)等，您可以使用DB Browser for SQLite。
它看起來像這樣：
您還可以使用在線 SQLite 瀏覽器，如SQLite Viewer或ExtendsClass。
使用中間件的替代數(shù)據(jù)庫會話
如果您不能使用依賴項yield——例如，如果您沒有使用Python 3.7并且無法安裝上面提到的Python 3.6的“backports” ——您可以在類似的“中間件”中設(shè)置會話道路。
“中間件”基本上是一個始終為每個請求執(zhí)行的函數(shù)，其中一些代碼在端點函數(shù)之前執(zhí)行，一些代碼在端點函數(shù)之后執(zhí)行。
創(chuàng)建中間件
我們將添加的中間件（只是一個函數(shù)）將為SessionLocal每個請求創(chuàng)建一個新的 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)建和處理放在一個try塊中。
然后我們在finally塊中關(guān)閉它。
這樣我們可以確保在請求之后數(shù)據(jù)庫會話總是關(guān)閉。即使在處理請求時出現(xiàn)異常。
關(guān)于 request.state
request.state是每個Request對象的屬性。它用于存儲附加到請求本身的任意對象，例如本例中的數(shù)據(jù)庫會話。您可以在Starlette 關(guān)于Requeststate的文檔中閱讀更多相關(guān)信息。
在這種情況下，它幫助我們確保通過所有請求使用單個數(shù)據(jù)庫會話，然后關(guān)閉（在中間件中）。
與yield或 中間件的依賴關(guān)系
在這里添加一個中間件類似于依賴 with 的yield作用，但有一些區(qū)別：
- 它需要更多的代碼，而且有點復(fù)雜。
- 中間件必須是一個async函數(shù)。如果其中有必須“等待”網(wǎng)絡(luò)的代碼，它可能會在那里“阻塞”您的應(yīng)用程序并稍微降低性能。雖然這里的工作方式可能不是很成問題SQLAlchemy。但是，如果您將更多代碼添加到有大量I/O等待的中間件，則可能會出現(xiàn)問題。
- 每個請求都會運行一個中間件。因此，將為每個請求創(chuàng)建一個連接。即使處理該請求的路徑操作不需要數(shù)據(jù)庫。
提示
yield當它們足以滿足用例時，最好使用依賴項。
信息
yield最近向FastAPI添加了依賴項。
本教程的先前版本只有帶有中間件的示例，并且可能有幾個應(yīng)用程序使用中間件進行數(shù)據(jù)庫會話管理。
本文題目：創(chuàng)新互聯(lián)FastAPI教程：FastAPI教程SQL（關(guān)系）數(shù)據(jù)庫
本文URL：http://fisionsoft.com.cn/article/cdjoodi.html
