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驱动程序。

以下是包括的方言 –

• 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 的一个子类，被映射到数据库中的一个 书表 。 Books 类中的属性与目标表中的列的数据类型相对应。请注意，id属性与书表中的主键相对应。

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

注意配置类中 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() 操作函数，将样本数据添加到book表中。为了验证，你可以使用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用户界面执行 get_books() 函数的结果。

更新和删除操作由 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')