FastAPI - SQL 数据库

在上一章中，我们使用了 Python 列表作为内存数据库来使用 FastAPI 执行 CRUD 操作。相反，我们可以使用任何关系数据库（例如 MySQL、Oracle 等）来执行存储、检索、更新和删除操作。

我们不使用符合 **DB-API** 的数据库驱动程序，而是使用 **SQLAlchemy** 作为 Python 代码和数据库之间的接口（我们将使用 SQLite 数据库，因为 Python 对其有内置支持）。SQLAlchemy 是一个流行的 SQL 工具包和 **对象关系映射器**。

对象关系映射 (ORM) 是一种编程技术，用于在面向对象编程语言中转换不兼容类型系统之间的数据。通常，面向对象语言（如 Python）中使用的类型系统包含非标量类型。但是，大多数数据库产品（如 Oracle、MySQL 等）中的数据类型是原始类型，例如整数和字符串。

在 ORM 系统中，每个类都映射到底层数据库中的一个表。ORM 代替您编写繁琐的数据库接口代码，让您可以专注于系统逻辑的编程。

为了使用 SQLAlchemy，我们需要首先使用 PIP 安装程序安装该库。

pip install sqlalchemy

SQLAlchemy 旨在与为特定数据库构建的 DBAPI 实现一起工作。它使用方言系统与各种类型的 DBAPI 实现和数据库进行通信。所有方言都需要安装相应的 DBAPI 驱动程序。

包含以下方言：

Firebird
Microsoft SQL Server
MySQL
Oracle
PostgreSQL
SQLite
Sybase

由于我们将使用 SQLite 数据库，因此我们需要为名为 test.db 的数据库创建一个数据库引擎。从 sqlalchemy 模块导入 **create_engine()** 函数。

from sqlalchemy import create_engine
from sqlalchemy.dialects.sqlite import *
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args = {"check_same_thread": False})

为了与数据库交互，我们需要获取其句柄。会话对象是数据库的句柄。会话类使用 **sessionmaker()** 定义 - 一个可配置的会话工厂方法，它绑定到引擎对象。

from sqlalchemy.orm import sessionmaker, Session
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

接下来，我们需要一个声明性基类，用于在声明性系统中存储类和映射表的目录。

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

**Books**（**Base** 的子类）映射到数据库中的 **book** 表。**Books** 类中的属性对应于目标表中列的数据类型。请注意，id 属性对应于 book 表中的主键。

from sqlalchemy import Column, Integer, String
class Books(Base):
   __tablename__ = 'book'
   id = Column(Integer, primary_key=True, nullable=False)
   title = Column(String(50), unique=True)
   author = Column(String(50))
   publisher = Column(String(50))
   Base.metadata.create_all(bind=engine)

**create_all()** 方法在数据库中创建相应的表。

现在，我们必须声明一个与声明性基类子类（上面定义的 Books 类）对应的 Pydantic 模型。

from typing import List
from pydantic import BaseModel, constr
class Book(BaseModel):
   id: int
   title: str
   author:str
   publisher: str
   class Config:
      orm_mode = True

请注意 config 类中 **orm_mode=True** 的使用，这表示它与 SQLAlchemy 的 ORM 类映射。

其余代码与内存 CRUD 操作类似，不同之处在于操作函数通过 SQLalchemy 接口与数据库交互。FastAPI 应用对象的 POST 操作定义如下：

from fastapi import FastAPI, Depends
app=FastAPI()
def get_db():
   db = session()
   try:
      yield db
   finally:
   db.close()
@app.post('/add_new', response_model=Book)
def add_book(b1: Book, db: Session = Depends(get_db)):
   bk=Books(id=b1.id, title=b1.title, author=b1.author,
publisher=b1.publisher)
   db.add(bk)
   db.commit()
   db.refresh(bk)
   return Books(**b1.dict())

首先建立数据库会话。来自 POST 请求体的数据作为新行添加到 book 表中。执行 **add_book()** 操作函数以将示例数据添加到 books 表中。要进行验证，您可以使用 SQLiteStudio（SQLite 数据库的 GUI 工具）。

定义了两个用于 GET 操作的操作函数，一个用于获取所有记录，另一个用于匹配路径参数的记录。

以下是绑定到 /list 路由的 **get_books()** 函数。执行时，其服务器响应是所有记录的列表。

@app.get('/list', response_model=List[Book])
def get_books(db: Session = Depends(get_db)):
   recs = db.query(Books).all()
   return recs

** /book/{id}** 路由调用 **get_book()** 函数，其中 id 作为路径参数。SQLAlchemy 的查询返回与给定 id 对应的对象。

@app.get('/book/{id}', response_model=Book)
def get_book(id:int, db: Session = Depends(get_db)):
   return db.query(Books).filter(Books.id == id).first()

下图显示了从 Swagger UI 执行 **get_books()** 函数的结果。

更新和删除操作由 **update_book()** 函数（访问 ** /update/{id}** 路由时执行）和访问 ** /delete/{id}** 路由时调用的 **del_book()** 函数执行。

@app.put('/update/{id}', response_model=Book)
def update_book(id:int, book:Book, db: Session = Depends(get_db)):
   b1 = db.query(Books).filter(Books.id == id).first()
   b1.id=book.id
   b1.title=book.title
   b1.author=book.author
   b1.publisher=book.publisher
   db.commit()
   return db.query(Books).filter(Books.id == id).first()
@app.delete('/delete/{id}')
def del_book(id:int, db: Session = Depends(get_db)):
   try:
      db.query(Books).filter(Books.id == id).delete()
      db.commit()
   except Exception as e:
      raise Exception(e)
   return {"delete status": "success"}

如果您打算使用 SQLite 之外的任何其他数据库，则只需相应地更改方言定义。例如，要使用 MySQL 数据库和 **pymysql** 驱动程序，请将引擎对象的语句更改为以下内容：

engine = create_engine('mysql+pymysql://user:password@localhost/test')

打印页面