Python Falcon 简明教程
Python Falcon - Suffixed Responders
为了理解概念和后缀响应者的需求,让我们定义一个 StudentResource 类。它包含一个 on_get() 响应者,它将学生列表转换 dict 对象为 JSON 并返回作为其响应。
我们还添加 on_post() 响应器,它从传入请求中读取数据并在列表中添加一个新的 dict 对象。
import falcon
import json
from waitress import serve
students = [
{"id": 1, "name": "Ravi", "percent": 75.50},
{"id": 2, "name": "Mona", "percent": 80.00},
{"id": 3, "name": "Mathews", "percent": 65.25},
]
class StudentResource:
def on_get(self, req, resp):
resp.text = json.dumps(students)
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_JSON
def on_post(self, req, resp):
student = json.load(req.bounded_stream)
students.append(student)
resp.text = "Student added successfully."
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_TEXT
使用 Falcon‘s App 对象的 add_route() 函数,我们添加 /students 路由。
app = falcon.App()
app.add_route("/students", StudentResource())
在启动服务器后,我们可以从 HTTPie 命令行测试 GET 和 POST 请求 -
http GET localhost:8000/students
HTTP/1.1 200 OK
Content-Length: 187
Content-Type: application/json
Date: Mon, 18 Apr 2022 06:21:02 GMT
Server: waitress
[
{
"id": 1,
"name": "Ravi",
"percent": 75.5
},
{
"id": 2,
"name": "Mona",
"percent": 80.0
},
{
"id": 3,
"name": "Mathews",
"percent": 65.25
}
]
http POST localhost:8000/students id=4 name="Prachi"
percent=59.90
HTTP/1.1 200 OK
Content-Length: 27
Content-Type: text/plain; charset=utf-8
Date: Mon, 18 Apr 2022 06:20:51 GMT
Server: waitress
Student added successfully.
再次调用 on_get() 来确认添加了新的学生资源。
http GET localhost:8000/students
HTTP/1.1 200 OK
Content-Length: 187
Content-Type: application/json
Date: Mon, 18 Apr 2022 06:21:02 GMT
Server: waitress
[
{
"id": 1,
"name": "Ravi",
"percent": 75.5
},
{
"id": 2,
"name": "Mona",
"percent": 80.0
},
{
"id": 3,
"name": "Mathews",
"percent": 65.25
},
{
"id": "4",
"name": "Prachi",
"percent": "59.90"
}
]
在这一阶段,我们希望在 StudentResource 类中有一个 GET 响应器方法,该方法从 URL 中读取 id 参数并从列表中检索相应的 dict 对象。
换句话说,格式为 /student/{id} 的 URL 应当与资源类中的 GET 方法相关联。但显然,一个类不能有两个相同名称的方法。因此,我们定义对 add_route() 方法使用 suffix 参数来区分 on_get() 响应器的两个定义。
通过指定 suffix ='student' ,将一个带有 id 参数的路由添加到 Application 对象。
app.add_route("/students/{id:int}", StudentResource(), suffix='student')
我们现在可以添加 on_get() 方法的另一个定义和此后缀,以便此响应器的名称为 on_get_student() 如下 -
def on_get_student(self, req, resp, id):
resp.text = json.dumps(students[id-1])
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_JSON
在添加新路由和 on_get_student() 响应器后启动 Waitress 服务器,并按如下方式测试此 URL -
http GET localhost:8000/students/2
HTTP/1.1 200 OK
Content-Length: 42
Content-Type: application/json
Date: Mon, 18 Apr 2022 06:21:05 GMTy
Server: waitress
{
"id": 2,
"name": "Mona",
"percent": 80.0
}
请注意,当客户端在使用适当的请求头请求 URL 路由 /students/{id:int} 时, on_put() 响应器(用于更新资源)和 on_delete() 响应器(用于删除资源)也将被调用。
我们已经添加了此路由,学生作为后缀。因此, on_put_student() 方法将路径参数解析为整数变量。给定 id 的项目的 JSON 表示形式被获取并使用 PUT 请求中提供的数据进行更新。
def on_put_student(self, req, resp, id):
student=students[id-1]
data = json.load(req.bounded_stream)
student.update(data)
resp.text = json.dumps(student)
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_JSON
on_delete_student() 响应器只是删除了 DELETE 请求中指定 id 的项目。返回剩余资源的列表。
def on_delete_student(self, req, resp, id):
students.pop(id-1)
resp.text = json.dumps(students)
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_JSON
我们可以使用 HTTPie 命令测试 API 的 PUT 和 DELETE 操作 -
http PUT localhost:8000/students/2 id=3 name="Mathews"
percent=55
HTTP/1.1 200 OK
Content-Length: 46
Content-Type: application/json
Date: Sat, 18 Apr 2022 10:13:00 GMT
Server: waitress
{
"id": "3",
"name": "Mathews",
"percent": "55"
}
http DELETE localhost:8000/students/2
HTTP/1.1 200 OK
Content-Length: 92
Content-Type: application/json
Date: Sat, 18 Apr 2022 10:18:00 GMT
Server: waitress
[
{
"id": 1,
"name": "Ravi",
"percent": 75.5
},
{
"id": 3,
"name": "Mathews",
"percent": 65.25
}
]
此 API ( studentapi.py ) 的完整代码如下 -
import falcon
import json
from waitress import serve
students = [
{"id": 1, "name": "Ravi", "percent": 75.50},
{"id": 2, "name": "Mona", "percent": 80.00},
{"id": 3, "name": "Mathews", "percent": 65.25},
]
class StudentResource:
def on_get(self, req, resp):
resp.text = json.dumps(students)
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_JSON
def on_post(self, req, resp):
student = json.load(req.bounded_stream)
students.append(student)
resp.text = "Student added successfully."
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_TEXT
def on_get_student(self, req, resp, id):
resp.text = json.dumps(students[id-1])
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_JSON
def on_put_student(self, req, resp, id):
student=students[id-1]
data = json.load(req.bounded_stream)
student.update(data)
resp.text = json.dumps(student)
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_JSON
def on_delete_student(self, req, resp, id):
students.pop(id-1)
print (students)
resp.text = json.dumps(students)
resp.status = falcon.HTTP_OK
resp.content_type = falcon.MEDIA_JSON
app = falcon.App()
app.add_route("/students", StudentResource())
app.add_route("/students/{id:int}", StudentResource(), suffix='student')
if __name__ == '__main__':
serve(app, host='0.0.0.0', port=8000)