Python Falcon SQLAlchemy模型

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

为了与数据库进行交互，我们需要获得其句柄。会话对象是数据库的句柄。会话类是用 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的一个子类被映射到数据库中的一个 学生 表。书籍类中的属性与目标表中的列的数据类型相对应。注意，id属性对应于书本表中的主键。

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 ）来确认 。

现在我们需要声明一个 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() 响应方法首先从请求参数中构造一个学生类的对象，并将其添加到学生模型中。由于这个模型被映射到数据库中的学生表，因此相应的行被添加。 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）作为主体参数。该请求被成功处理，并且有一行被添加到 学生 表中。

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

on_get()

这个响应器的目的是检索 学生 模型中的所有对象。

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

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

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

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

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。 为了从学生模型中获取具有给定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 的对象，并改变名称和标记。注意，这些值是作为主体参数传递的。

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