如何使用 python falcon rest API 为多个用户级别创建令牌的用户身份验证

Question

我使用 python falcon api 创建了一个 rest api。它是访问特定银行ATM的预测值并读取，更新，删除值。

import falcon
import MySQLdb
import json

import re
import mysql.connector
from mysql.connector import  Error
class TesResource:
def on_get(self, req, resp):
try:
atmid=req.get_param('atm_key_id')
datestart=req.get_param('prediction_date_start')
dateend=req.get_param('prediction_date_end')
if atmid is None or atmid=="" or datestart is None or dateend is None:
resp.body=json.dumps({'error': 'Parameter is invalid'})
resp.status=falcon.HTTP_500
return resp
conn = mysql.connector.connect(host='localhost', database='bank', user='root', password='', autocommit=True)
if conn.is_connected():
print('connected')
cursor=conn.cursor()
#q="SELECT prediction_amount FROM prediction WHERE atm_key_id=5 AND (prediction_date BETWEEN '2017-10-01' AND '2017-10-1')"
q="SELECT prediction_amount FROM prediction WHERE atm_key_id=%s AND (prediction_date BETWEEN %s AND %s)" 
#q=("SELECT * FROM prediction")
cursor.execute(q,(atmid, datestart, dateend,))
rows=cursor.fetchall()
output={'tes':[]}
for row in rows:
#data={"key":row[0], "amount":float(row[2])}
data={"amount":float(row[0])}
output['tes'].append(data)
resp.status=falcon.HTTP_200
resp.body=json.dumps(output, encoding='utf-8')
cursor.close()
conn.close()
except Exception as e:
resp.body=json.dumps({'error':str(e)})
resp.status=falcon.HTTP_500
return resp
def on_put(self, req, resp):
try:

atmid=req.get_param('atm_key_id')
date=req.get_param('prediction_date')
amount=req.get_param('prediction_amount')
if atmid is None or atmid=="" or date is None or amount is None:
resp.body=json.dumps({'error': 'Parameter is invalid'})
resp.status=falcon.HTTP_500
return resp
conn = mysql.connector.connect(host='localhost', database='bank', user='root', password='', autocommit=True)
if conn.is_connected():
print('connected')
cursor=conn.cursor()
q="""UPDATE `prediction` SET `prediction_amount`=%s WHERE atm_key_id=%s AND prediction_date=%s """
cursor.execute(q,(amount, atmid, date,))
conn.commit()
cursor.close()
output={'status':"Data successfully updated"}
resp.status=falcon.HTTP_200
data_resp=json.dumps(output, encoding='utf-8')
resp.body=data_resp

except Exception as e:
conn.rollback()
resp.body=json.dumps({'error':str(e)})
resp.status=falcon.HTTP_500
return resp

def on_delete(self, req, resp):
try:
atmid=req.get_param('atm_key_id')
date=req.get_param('prediction_date')
if atmid is None or atmid=="" or date is None:
resp.body=json.dumps({'error': 'Parameter is invalid'})
resp.status=falcon.HTTP_500
return resp
conn = mysql.connector.connect(host='localhost', database='bank', user='root', password='', autocommit=True)
if conn.is_connected():
print('connected')
cursor=conn.cursor()
q="""DELETE FROM `prediction` WHERE atm_key_id=%s AND prediction_date=%s"""
cursor.execute(q, (atmid, date,))
conn.commit()
cursor.close()
output={'status':"Data successfully deleted"}
resp.status=falcon.HTTP_200
data_resp=json.dumps(output, encoding='utf-8')
resp.body=data_resp
except Exception as e:
conn.rollback()
resp.body=json.dumps({'error':str(e)})
resp.status=falcon.HTTP_500
return resp

有两个用户级别。每个人都可以获取特定日期或日期范围(访问级别 1(的预测值。但只有经过授权的人员才能更新或删除预测值(访问级别 2(。我有一个用户 mysql 表。它有用户名、用户 ID 和 access_level(1 或 2(作为列。如何使用令牌创建用户身份验证？任何见解都将非常有帮助。

Answer 1

您可以使用中间件，并在处理请求之前检查路径、令牌和其他参数。这是最干净的方式，因为您可以为每个级别添加多个中间件，并让资源清理并与此特权级别逻辑分离。

查看身份验证中间件示例，您可以分析请求并引发未经授权的异常HTTP_401。与 HTTP 标准保持一致。

要添加多个中间件，您可以使用 falcon API 构造函数。

app = falcon.API(middleware=[
AuthMiddleware(),
UserPrivilegeMiddleware()
])

Answer 2

您可以创建一个身份验证装饰器，该装饰器将检查用户是否具有访问级别 2。 在on_update和on_delete上使用此装饰器。