当前位置：极客教程 > fastapi > Fastapi 教程 > FastAPI SQL数据库

FastAPI SQL数据库

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
Python

SQLAlchemy被设计为与为特定数据库建立的DBAPI实现一起运行。它使用方言系统来与各种类型的DBAPI实现和数据库进行通信。所有的方言都需要安装一个合适的DBAPI驱动程序。

以下是包括的方言 –

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

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

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

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

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

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)
Python

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
Python

注意配置类中 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())
Python

首先建立一个数据库会话。来自POST请求体的数据被作为新的行添加到book表中。执行 add_book() 操作函数，将样本数据添加到book表中。为了验证，你可以使用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
Python

/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()
Python

下图显示了从Swagger用户界面执行 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"}
Python

如果你打算使用任何其他数据库来代替SQLite，你只需要相应地改变方言定义。例如，要使用MySQL数据库和 pymysql 驱动，请将引擎对象的声明改为以下内容

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

Python教程

Python 教程

Python 教程
Tkinter 教程

Tkinter 教程
Pandas 教程

Pandas 教程
NumPy 教程

NumPy 教程
Flask 教程

Flask 教程
Django 教程

Django 教程
PySpark 教程

PySpark 教程
wxPython 教程

wxPython 教程
SymPy 教程

SymPy 教程
Seaborn 教程

Seaborn 教程
SciPy 教程

SciPy 教程
RxPY 教程

RxPY 教程
Pycharm 教程

Pycharm 教程
Pygame 教程

Pygame 教程
PyGTK 教程

PyGTK 教程
PyQt 教程

PyQt 教程
PyQt5 教程

PyQt5 教程
PyTorch 教程

PyTorch 教程
Matplotlib 教程

Matplotlib 教程
Web2py 教程

Web2py 教程
BeautifulSoup 教程

BeautifulSoup 教程

Java教程

Java 教程

Java 教程

Web教程

HTML 教程

HTML 教程
CSS 教程

CSS 教程
CSS3 教程

CSS3 教程
jQuery 教程

jQuery 教程
Ajax 教程

Ajax 教程
AngularJS 教程

AngularJS 教程
TypeScript 教程

TypeScript 教程
WordPress 教程

WordPress 教程
Laravel 教程

Laravel 教程
Next.js 教程

Next.js 教程
PhantomJS 教程

PhantomJS 教程
Three.js 教程

Three.js 教程
Underscore.JS 教程

Underscore.JS 教程
WebGL 教程

WebGL 教程
WebRTC 教程

WebRTC 教程
VueJS 教程

VueJS 教程

数据库教程

SQL 教程

SQL 教程
MySQL 教程

MySQL 教程
MongoDB 教程

MongoDB 教程
PostgreSQL 教程

PostgreSQL 教程
SQLite 教程

SQLite 教程
Redis 教程

Redis 教程
MariaDB 教程

MariaDB 教程

图形图像教程

Vulkan 教程

Vulkan 教程
OpenCV 教程

OpenCV 教程

大数据教程

R语言 教程

R语言 教程

开发工具教程

Git 教程

Git 教程
VSCode 教程

VSCode 教程
Docker 教程

Docker 教程
Gerrit 教程

Gerrit 教程
Excel 教程

Excel 教程

计算机教程

Go语言 教程

Go语言 教程
C++ 教程

C++ 教程

Fastapi 教程

登录

找回密码？

注册