FastAPI - SQL 数据库

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

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

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

在 ORM 系统中，每个类都映射到底层数据库中的一个表。ORM 无需您自己编写繁琐的数据库接口代码，而是会为您处理这些问题，同时您可以专注于对系统逻辑进行编程。

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

pip install sqlalchemy

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

以下是包含的方言 -

  • 火鸟

  • 微软SQL服务器

  • MySQL

  • 甲骨文

  • PostgreSQL

  • SQLite

  • 赛贝斯

由于我们将使用 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})

为了与数据库交互，我们需要获取它的句柄。会话对象是数据库的句柄。Session 类是使用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的子类，映射到数据库中的图书表。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

注意配置类中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 请求正文中的数据将作为新行添加到图书表中。执行add_book()操作函数将样本数据添加到books表中。要进行验证，您可以使用 SQLiteStudio，这是一个用于 SQLite 数据库的 GUI 工具。

FastAPI SQL 数据库

定义了两种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}路由使用 id 作为路径参数调用get_book()函数。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()函数的结果。

FastAPI SQL 数据库

更新和删除操作由update_book()函数（在访问/update/{id}路由时执行）和del_book()函数执行，当路由/delete/{id}作为 URL 给出时调用。

@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')