FastAPI+SQLAlcheym，实现对数据库的增删改查

2023-06-0816:46:35后端程序开发Comments2,492 views1字数 5946阅读模式

FastAPI对数据库的操作通常是采用SQLAlchemy来实现的，这主要是由于后者会以面向对象的方式来操作表，同时SQLAlchemy用统一的接口使得程序与数据库相分离，降低了耦合度。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

在FastAPI的官方文档中，有专门对SQLAlcheym的介绍，只不过该示例中是两个表，且表与表之间有外键约束，为了更清楚让刚入门的同学快速掌握FastAPI对数据库的操作，本文将以一个简单的表来说明如何在FastAPI中实现对数据库的增删改查操作。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

整体思路

在官方文档中，FastAPI采用SQLAlchemy对数据库的操作是采用一种非常清晰的模块划分方式，即要操作一个数据表，通常用五个文件来实现：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

数据库配置文件：databases.py，主要完成对数据库的连接；
数据库中表对应的模型文件：models.py，建立所需要的表对应的数据模型；
数据库中表所对应的架构文件：schemas.py，这里采用pydantic来完成对数据表的基本校验，其实质是建立与表对应的类，它与models.py中类的区别在于：models.py中是与表严格对应的，而schemas则可以根据表模型来定制适合不同场景的类。
增删改查操作文件：crud.py，该文件中主要完成对数据库的各种读写操作。
主文件main.py，这里配置各种路由及get、post等方法。

本例中将以一个图书数据库来实现增删改查操作。接下来一起看看各种文件的设定。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

数据库的连接

在本例中，我们将以sqlite3数据库来实现图书数据库的管理，这里要注意的是，如果采用sqlite3，则配置中的check_same_thread标记要设定为False，这是因为：sqlite3数据库本身并非一个网络数据库，其默认只能在同一线程中使用，如果不设定该标记，则SQLAlchemy会提示错误，具体代码如下所示：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

from sqlalchemy import create_engine # type: ignore
from sqlalchemy.ext.declarative import declarative_base # type: ignore
from sqlalchemy.orm import sessionmaker # type: ignore

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

在上述代码中，通过创建一个SessionLocal变量，将当前数据库的连接保存于其中，这样可在其它文件中将其导入使用。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

Base是SQLAlchemy的总类，与数据库表对应的数据模型均须继承它。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

提示：如果要使用mysql数据库，则连接字符串如下："mysql+pymysql://用户名:密码@主机IP:端口/数据库名?charset=utf8"文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

建立数据表模型

为演示简单，这里的数据表仅包含三个字段：ID、书名、定价。在当前目录下建立文件models.py，代码如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

from sqlalchemy import Column, Integer, String  # type: ignore
from .database import Base


class Books(Base):
    __tablename__ = "books"

    id = Column(Integer, primary_key=True, index=True)
    bookname = Column(String(100), unique=True)
    prices = Column(Integer)

在上述代码中，我们建立了表名为books的类Books，该类包含了之前提到的三个字段，注意，这里的书名价格用的是整数表示。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

在这里，我们用from .database import Base导入了刚才创建的基类，对于某些编辑器会提示这里的导入警告，一个合适的办法是在当前文件夹下新建一个__init__.py空文件，这样语法检查器会认为这是一个包，可以被导入而不会出现警告信息。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

建立架构文件

在之前一篇文章中我们介绍过利用pydantic来检验前端传输数据的功能，这里同样用该包来完成输入或输出数据的类设定，在同一个目录下建立文件schemas.py文件，其中建立一个与模型类完全一致的类，这主要是要利用其id属性来完成更改数据的功能，再建立一个去掉id的类，用来新建数据时使用，代码如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

from typing import Union
from pydantic import BaseModel

class Books(BaseModel): 
    id: Union[int, None]=None   
    bookname : str
    prices : Union[int, None] = None

    class Config:
        orm_mode = True

class BooksBase(BaseModel):
    bookname : str
    prices : Union[int, None] = None

在上述代码中，类Books有一个内部类class Config，这是pydantic中的一个配置，将其中的orm_mode设定为True，即告诉pydantic，这是一个可以直接映射为对象关系模型的类。而BooksBase类则只是对数据进行校验。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

当然，我们可以利用类的继承功能，将上述代码改写如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

from typing import Union
from pydantic import BaseModel

class BooksBase(BaseModel):     
    bookname : str
    prices : Union[int, None] = None

class Books(BooksBase): 
    id: Union[int, None]=None
    class Config:
        orm_mode = True

增删改查

对数据表的操作单独用一个文件来收集，在同一个目录下建立文件crud.py，其中的增删改查代码如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

from datetime import date
from sqlalchemy.orm import Session
from . import models, schemas 

# 根据书名查询，支持模糊查询
def get_books_by_name(db: Session, bookname: str):
    return db.query(models.Books).filter(models.Books.bookname.like(f"%{bookname}%")).all()

# 根据书的`ID`删除
def delete_book_by_Id(db:Session, bookId:int):
    db_book = db.query(models.Books).filter(models.Books.id == bookId).one_or_none()
    if db_book is None:
        return None
    db.delete(db_book)
    db.commit()
    return True

# 增加书籍信息
def create_book(db:Session, book:schemas.BooksBase):
    curBook = models.Books(
        bookname = book.bookname,
        prices = book.prices,        
    )
    db.add(curBook)
    db.commit()
    db.refresh(curBook)
    return curBook

# 根据书的`ID`修改书籍信息
def update_book_by_id(db:Session, bookId:int, book:schemas.BooksBase):
    db_book = db.query(models.Books).filter(models.Books.id == bookId).one_or_none()
    if db_book is None:
        return None

    # Update model class variable from requested fields 
    for var, value in vars(book).items():
        setattr(db_book, var, value) if value else None
    db.commit()
    db.refresh(db_book)
    return db_book

值得一提的是，上述修改书籍的代码中，我们利用了Python中的vars这个函数，它可以将一个对象的所有属性以字典的方式列举。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

整合路由

最后，我们在该目录下的main.py文件中设定各种不同的路由，调用上述crud.py中的函数来完成对数据表的增删改查，代码如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session #type: ignore
from . import crud, schemas, models
from .database import SessionLocal, engine

# 根据模板文件创建对应的数据表
models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# 设定数据库连接
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 查询书籍
@app.get("/books/", response_model=list[schemas.Books])
def get_hospital_nums(bookname:str, db:Session=Depends(get_db)):
    db_books = crud.get_books_by_name(db, bookname=bookname)    
    if not db_books:
        raise HTTPException(status_code=400, detail="当前书籍名称未查询到相匹配的书籍。")
    return db_books

# 删除书籍
@app.post("/deleteBook/{bookid}")
def delete_book(bookid:int, db:Session=Depends(get_db)):
    return crud.delete_book_by_Id(db, bookId=bookid)

# 修改书籍
@app.post("/updateBook/{bookid}", response_model=schemas.Books)
def update_book(bookid:int, book:schemas.BooksBase, db:Session=Depends(get_db)):
    return crud.update_book_by_id(db, bookId=bookid, book=book)

# 新增书籍
@app.post("/books/", response_model=schemas.Books)
def create_book(book: schemas.BooksBase, db: Session = Depends(get_db)):
    db_book = crud.get_books_by_name(db, bookname=book.bookname)
    if db_book:
        raise HTTPException(status_code=400, detail="该书籍已经存在。")
    return crud.create_book(db=db, book=book)

至此，五个文件已经全部完成，接下来我们要测试一下这些功能。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

在导入文件时，一定要注意：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

上述文件中，我们是用from . import XXX的方式导入当前文件夹下的其它文件，在这种情况下，如果要成功运行，必须将目录切换至backend的上一级来运行才可以。如果省略.，直接用import XXX来导入当前文件夹下的其它文件，则必须在backend这个当前目录下运行服务器uvicorn的命令。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

我们采用在backend文件夹的上一级运行服务器的方式。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

测试增删改查功能

下图展示了我们创建的五个文件组织形式：
![[2022-09-29_15-59.png]]文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

将目录切换至backend的上一级，在命令窗口运行下列命令：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html

uvicorn backend.main:app --reload --port 8001

然后你会发现在backend的上一级目录中生成了一个数据库文件sql_app.db，此时如果用DB Browser for SQLite这个软件来查看该数据库文件时，会发现我们定义的表已经在其中创建了，如下图所示：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/bc/46000.html