Fastapi 简明教程

FastAPI - SQL Databases

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

我们不会使用 DB-API 合规的数据库驱动程序，而会使用 SQLAlchemy 作为 Python 代码和数据库之间的界面（我们将使用 SQLite 数据库，因为 Python 已内置对它的支持）。SQLAlchemy 是一个流行的 SQL 工具包和 Object Relational Mapper 。

对象关系映射是一种在面向对象编程语言中不同类型系统之间转换数据的编程技术。通常，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() 方法在数据库中创建相应的表。

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

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} 路由使用 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() 函数的结果。

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

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