HTML CSS JAVASCRIPT PYTHON PYTHON2 JAVA C C++ C# SQL Excel Linux AI BOOTSTRAP PHP 教程库 参考手册 技术文章 测验 练习 HOWTO FAQ
❮ 上一节 下一节 ❯

Python Falcon - SQLAlchemy 模型

为了演示 Falcon 的响应器功能（on_post()、on_get()、on_put() 和 on_delete()），我们做了 CRUD（代表 以 Python 字典对象列表的形式在内存数据库上创建、检索、更新和删除）操作。 相反，我们可以使用任何关系数据库（如 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})

为了与数据库进行交互，我们需要获取它的句柄。 Session 会话对象是数据库的句柄。 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()

模型类

Students，Base的子类映射到数据库中的students表。 Books 类中的属性对应于目标表中列的数据类型。 请注意，id 属性对应于 book 表中的主键。

class Students(Base):
   __tablename__ = 'student'
   id = Column(Integer, primary_key=True, nullable=False)
   name = Column(String(63), unique=True)
   marks = Column(Integer)
Base.metadata.create_all(bind=engine)

create_all() 方法在数据库中创建相应的表。 可以使用 SQLiteStudio 等 SQLite Visual 工具进行确认。

Sqlite

我们现在需要声明一个 StudentResource 类，其中定义了 HTTP 响应器方法以对 students 表执行 CRUD 操作。 此类的对象与路由相关联，如以下代码片段所示 −

import falcon
import json
from waitress import serve
class StudentResource:
   def on_get(self, req, resp):
      pass
   def on_post(self, req, resp):
      pass
   def on_put_student(self, req, resp, id):
      pass
   def on_delete_student(self, req, resp, id):
      pass
app = falcon.App()
app.add_route("/students", StudentResource())
app.add_route("/students/{id:int}", StudentResource(), suffix='student')

on_post()

其余代码与内存CRUD操作类似，不同之处在于操作函数通过SQLalchemy接口与数据库交互。

on_post() 响应器方法首先根据请求参数构造一个 Students 类的对象，并将其添加到 Students 模型中。 由于此模型映射到数据库中的 students 表，因此添加了对应的行。 on_post()方法如下所示 −

def on_post(self, req, resp):
   data = json.load(req.bounded_stream)
   student=Students(id=data['id'], name=data['name'], marks=data['marks'])
   session.add(student)
   session.commit()
   resp.text = "Student added successfully."
   resp.status = falcon.HTTP_OK
   resp.content_type = falcon.MEDIA_TEXT

如前所述，on_post() 响应程序在收到 POST 请求时被调用。 我们将使用 Postman 应用程序来传递 POST 请求。

启动 Postman，选择 POST 方法并传递值（id=1，name="Manan" 和 marks=760 作为 body 参数。请求处理成功，并向 students 表中添加一行。

Postman

继续发送多个 POST 请求以添加记录。

on_get()

此响应程序旨在检索 Students 模型中的所有对象。 Session 对象上的 query() 方法检索对象。

rows = session.query(Students).all()

由于 Falcon 响应器的默认响应是 JSON 格式，我们必须将上述查询的结果转换为 dict 对象列表。

data=[]
for row in rows:
   data.append({"id":row.id, "name":row.name, "marks":row.marks})

在 StudentResource 类中，让我们添加执行此操作并发送其 JSON 响应的 on_get() 方法，如下所示 −

def on_get(self, req, resp):
   rows = session.query(Students).all()
   data=[]
   for row in rows:
      data.append({"id":row.id, "name":row.name, "marks":row.marks})
      resp.text = json.dumps(data)
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON

GET 请求操作可以在 Postman 应用程序中进行测试。 /students URL 将导致显示 JSON 响应，其中显示了 students 学生模型中所有对象的数据。

Postman 示例

Postman 应用结果面板中显示的两条记录也可以在 SQLiteStudio 的数据视图中验证。

Python Sqlite1

on_put()

on_put() 响应程序执行更新操作。 它响应 URL /students/id。 为了从 Students 模型中获取具有给定 id 的对象，我们将过滤器应用于查询结果，并使用从客户端接收到的数据更新其属性的值。

student = session.query(Students).filter(Students.id == id).first()

on_put()方法的代码如下 −

def on_put_student(self, req, resp, id):
   student = session.query(Students).filter(Students.id == id).first()
   data = json.load(req.bounded_stream)
   student.name=data['name']
   student.marks=data['marks']
   session.commit()
   resp.text = "Student updated successfully."
   resp.status = falcon.HTTP_OK
   resp.content_type = falcon.MEDIA_TEXT

让我们在 Postman 的帮助下用 id=2 更新 Students 模型中的对象，并更改名称和标记。 请注意，这些值作为正文参数传递。

Onget

SQLiteStudio 中的数据视图显示修改已经生效。

Onput

on_delete()

最后，DELETE 操作也很简单。 我们需要获取给定 id 的对象并调用 delete() 方法。

def on_delete_student(self, req, resp, id):
   try:
      session.query(Students).filter(Students.id == id).delete()
      session.commit()
   except Exception as e:
      raise Exception(e)
      resp.text = "deleted successfully"
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_TEXT

作为 on_delete() 响应器的测试，让我们在 Postman 的帮助下删除 id=2 的对象，如下所示 −

Ondelete
❮ 上一节 下一节 ❯