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 驱动程序。
以下是包含的方言:
-
Firebird
-
Microsoft SQL Server
-
MySQL
-
Oracle
-
PostgreSQL
-
SQLite
-
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 视觉工具来确认。
我们现在需要声明一个 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 表中添加一行。
继续并发送多个 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 应用的结果窗格中显示的两条记录也可以在 SQLiteStudio 的数据视图中进行验证。
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 参数传递。
SQLiteStudio 中的数据视图显示这些修改已经生效。
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 的对象,如下所示 −