Python Falcon 简明教程

Python Falcon - SQLAlchemy Models

若要演示 Falcon 的响应器如何发挥作用 ( on_post(), on_get(), on_put()on_delete() ),我们对 Python 词典对象的列表形式的内存数据库执行了 CRUD (表示创建、检索、更新和删除)操作。相反,我们可以使用任何关系数据库(例如 MySQL、Oracle 等)来执行存储、检索、更新和删除操作。

我们不会使用 DB-API 合规的数据库驱动程序,而会使用 SQLAlchemy 作为 Python 代码和数据库之间的界面(我们将使用 SQLite 数据库,因为 Python 已内置对它的支持)。SQLAlchemy 是一个流行的 SQL 工具包和 Object Relational Mapper

对象关系映射是一种编程技术,用于在面向对象编程语言的不同类型系统之间转换数据。通常,像 Python 这种面向对象语言中使用的类型系统包含非标量类型。但是,大多数数据库产品的(例如 Oracle、MySQL 等)数据类型都是基本类型,例如整数和字符串。

在 ORM 系统中,每个类都映射到底层数据库中的一个表。ORM 负责处理这些问题,让你专注于对系统逻辑进行编程,而不再需要自己编写乏味的数据库接口代码。

为了使用 SQLALchemy,我们需要先使用 PIP 安装程序来安装该库。

pip install sqlalchemy

SQLAlchemy 被设计为使用专为某个特定数据库而构建的 DBAPI 实现来运行。它使用方言系统与各种类型的 DBAPI 实现和数据库通信。所有方言都要求已安装相应的 DBAPI 驱动程序。

以下是包含的方言:

  1. Firebird

  2. Microsoft SQL Server

  3. MySQL

  4. Oracle

  5. PostgreSQL

  6. SQLite

  7. Sybase

Database Engine

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

Model class

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 视觉工具来确认。

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 模型中。由于此模型映射到了数据库中的 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" and 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

可以在 Postman 应用中测试 GET 请求操作。 /students URL 将显示 JSON 响应,显示学生模型中所有对象的数据。

postman example

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

sqllite1

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 为 id=2 的对象,并更改 name 和 marks。请注意,这些值作为 body 参数传递。

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