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

以下是包含的方言 -

  • 火鸟

  • 微软SQL服务器

  • MySQL

  • 甲骨文

  • PostgreSQL

  • SQLite

  • 赛贝斯

数据库引擎

由于我们将使用 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()

模型类

Students是 Base 的子类，映射到数据库中的学生表。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 ()方法在数据库中创建相应的表。可以使用 SQLite Visual 工具（例如SQLiteStudio ）进行确认。

sqlite

现在，我们需要声明一个StudentResource类，其中定义了 HTTP 响应程序方法来对学生表执行 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 模型。由于该模型映射到数据库中的学生表，因此添加了相应的行。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

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

启动Postman，选择POST方法并传递值（id = 1，name =“Manan”和marks = 760作为正文参数。请求已成功处理，并向学生表中添加了一行。

邮差

继续发送多个 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 响应，其中显示学生模型中所有对象的数据。

邮递员示例

Postman 应用程序结果窗格中显示的两条记录也可以在SQLiteStudio的数据视图中进行验证。

Python Sqlite1

on_put()

on_put ()响应程序执行 UPDATE 操作。它响应 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 的帮助下更新 Students 模型中id=2的对象，并更改名称和标记。请注意，这些值作为主体参数传递。

翁盖特

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

输出

on_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 的对象，如下所示 -

删除