Flask接口开发与Redis数据持久化实战
前言
近年来，由于Restful协议定义的接口使用json格式的数据进行前后端交互，使得前后端接口实现了分离，大大提高了开发效率，目前很多公司开发的接口都采用这种协议。关于如何实现一个Restful格式的api接口，不同语言有不同的实现框架，在java语言中主要借助于SpringBoot框架，而在python语言中可以使用Django和Flask两种框架，这两种框架的区别在于：Django框架是一个非常重量级的Web开发框架，它包含了很多开发中包含的各种框架，非常全面但是不够灵活；Flask框架是一个轻量级Web开发框架，它只包含很少的核心库，如果要用到其他框架，例如实现数据持久化保存，那么只需要引入该框架即可，能够快速高效的开发接口。因此，本文中我们选择了Flask框架来开发接口。此外，接口请求的数据一般是进行持久化保存，这样一来方便后续查找，也比较灵活。常用的数据库有Mysql与Redis，这两种数据库都是开源的，Mysql是关系型数据库，Redis是内存数据库或者缓存数据库，通过开辟一块内存空间来保存数据，由于本文随后开发的接口个数很少，数据也不多，用不着使用Mysql数据库去创建一张表来存储，因此我们数据保存采用Redis数据库来实现，好了，接下来就让我们去开始Restful格式接口的开发之旅吧。
Flask
这里我展示一段已经开发完全的接口代码，我们使用这段代码来为大家详细讲解下接口开发的流程与内容
# -*-coding:utf-8 -*-
"""
FileName: flask_api
Author: raphealwu
Date: 2021/2/7 10:23 下午
"""
import time
from flask import Flask, request
from api.token_utils.token_utils import toc
from api.flask_apis.redis_manager import RedisDBManager
import logging
redis_host = 'localhost'
server = Flask(__name__)
user_info = "user_info"
# 注册接口
@server.route("/api/v1/register", methods=["POST"])
def register():
"""
注册接口，使用用户名和密码
:return:
"""
username = request.form.get("username")
password = request.form.get("password")
if not str(username) or not str(password):
return {"code": 500, "message": "用户名或密码不能为空"}
with RedisDBManager(redis_host) as r:
if r.hexists(user_info, str(username)):
if bytes.decode(r.hget(user_info, str(username))) == str(password):
return {"code": 401, "message": "用户已经注册，无需再次注册"}
r.hset(user_info, username, password)
logging.info(r.hget(user_info, username))
return {"code": 200, "message": "注册成功"}
# 登陆接口
@server.route("/api/v1/login", methods=["POST"])
def login():
"""
登陆接口，使用用户名和密码
:return:
"""
username = request.form.get("username")
password = request.form.get("password")
if not str(username) or not str(password):
return {"code": 500, "message": "用户名或密码不能为空"}
with RedisDBManager(redis_host) as r:
logging.info(r.hget(user_info, username))
if r.hexists(user_info, username):
logging.info(bytes.decode(r.hget(user_info, username)))
if bytes.decode(r.hget(user_info, username)) == password:
return {"code": 200, "message": "登陆成功", "token": toc.generate_token()}
else:
return {"code": 500, "message": "用户名或密码不存在"}
else:
return {"code": 401, "message": "用户还没有注册，请先注册"}
# 产品创建接口
@server.route("/api/v1/createProd", methods=["POST"])
def create_products():
"""
产品创建接口
:return:
"""
token = request.headers.get('token')
if not token:
return {"code": 401, "message": "无访问权限，拒绝访问"}
if not toc.verify_token(token):
return {"code": 401, "message": "token失效，请重新获取"}
username = request.form.get("username")
product_name = request.form.get("productName")
price = request.form.get("price")
vendor = request.form.get("vendor")
with RedisDBManager(redis_host) as r:
if r.hexists(user_info, username):
products_info = {f"{username}": f"{product_name}-{price}-{vendor}"}
if r.hexists("products_info", username):
if bytes.decode(r.hget("products_info", username)) == f"{product_name}-{price}-{vendor}":
return {"code": 500, "message": "产品信息已经存在，新增产品失败"}
else:
r.hmset("products_info", products_info)
return {"code": 200, "message": "新增产品成功"}
else:
return {"code": 500, "message": "用户信息不存在"}
# 产品查询接口
@server.route("/api/v1/queryProds", methods=["GET"])
def get_products_info():
"""
产品创建接口
:return:
"""
token = request.headers.get('token')
if not token:
return {"code": 401, "message": "无访问权限，拒绝访问"}
# if not toc.verify_token(token):
# return {"code": 401, "message": "token失效，请重新获取"}
username = request.args.get("username")
if not username:
return {"code": 500, "message": "用户名不能为空"}
with RedisDBManager(redis_host) as r:
if r.hexists("products_info", username):
data = bytes.decode(r.hget("products_info", username))
return {"code": 200, "message": "查询成功", "data": data}
else:
return {"code": 500, "message": "用户信息不存在"}
# 产品修改接口
@server.route("/api/v1/modify", methods=["PUT"])
def modify_products_info():
"""
产品修改接口
:return:
"""
token = request.headers.get('token')
if not token:
return {"code": 401, "message": "无访问权限，拒绝访问"}
# if not toc.verify_token(token):
# return {"code": 401, "message": "token失效，请重新获取"}
username = request.form.get("username")
product_name = request.form.get("productName")
price = request.form.get("price")
vendor = request.form.get("vendor")
with RedisDBManager(redis_host) as r:
if bytes.decode(r.hget(user_info, str(username))):
products_info = {f"{username}": f"{product_name}-{price}-{vendor}"}
r.hmset("products_info", products_info)
return {"code": 200, "message": "产品修改成功"}
# 产品删除接口
@server.route("/api/v1/delete", methods=["DELETE"])
def delete_products_info():
"""
产品删除接口
:return:
"""
token = request.headers.get('token')
if not token:
return {"code": 401, "message": "无访问权限，拒绝访问"}
# if not toc.verify_token(token):
# return {"code": 401, "message": "token失效，请重新获取"}
username = request.form.get("username")
with RedisDBManager(redis_host) as r:
if bytes.decode(r.hget(user_info, str(username))):
r.hdel("products_info", username)
return {"code": 200, "message": "产品删除成功"}
else:
return {"code": 500, "message": "产品不存在或者已经被删除"}
if __name__ == '__main__':
server.run(debug=True)
上述代码中，我们总共开发了6个接口，代码开始处有一行代码server = Flask(__name__)，这段代码的意思是注册了一个Flask框架实例，接下来注册接口处有一个@server.route("/api/v1/register", methods=["POST"])，其中@server.route表示一个路由，它匹配了/api/v1/register这个路径，允许的请求体方法为post方法，也就是methods=["POST"]中的POST方法，接下来再接着看这个代码中这两行代码
username = request.form.get("username")
password = request.form.get("password")
由于该接口请求方法为post，因此我们需要从form表单中去获取数据，"username"和"password"表示请求体中包含的参数，对该接口发送post请求时，form表单必须带有这两个参数。
Redis上下文处理器
对于注册接口，我们需要将注册信息保存到redis数据库中，这里我使用了with语法来动态的去释放连接资源，关于RedisDBManager(redis_host)代码信息如下所示：
# -*-coding:utf-8 -*-
"""
FileName: redis_manager
Author: raphealwu
Date: 2021/2/7 9:44 下午
"""
from redis import StrictRedis
class RedisDBManager(object):
"""
redis数据库操作
"""
def __init__(self, host):
self._host = host
self.db_connect = None
def __enter__(self):
self.db_connect = StrictRedis(host='localhost', port=6379, db=0)
return self.db_connect
def __exit__(self, exc_type, exc_val, exc_tb):
if self.db_connect:
self.db_connect.close()
这里是我自己定义的一个上下文处理器，如果要使用with语法来动态关闭或者释放资源，那么你需要在自己定义的类中实现__enter__和__exit__方法即可，分别对应着资源链接或者打开与资源关闭或者释放。
这里我首先使用hexists判断user_info字典中键为username是否存在，存在然后通过hget获取该键对应的值与请求的值做等值判断，如果值相等说明已经注册过了，不需要再次注册，否则予以注册。
token权限控制
接下来的登陆接口也类似，在产品创建接口中，我们看到多了这样一行代码token = request.headers.get('token')，在实际接口开发中，权限控制是一个非常重要的东西，我们这里的token就代表了权限的一种形式，只有在登陆成功之后，才会返回一个token，下次用户带着这些token就可以访问我们后面四个接口了。关于token的生成，我们采用了python中jwt库，使用HS256算法进行加密，具体的加密解密代码如下图所示：
# -*-coding:utf-8 -*-
"""
FileName: token_utils
Author: raphealwu
Date: 2021/2/7 9:21 下午
"""
import time
import jwt
class TokenUtils(object):
"""
jwt token utils class
"""
def __init__(self):
self._algo = 'HS256'
self._secret = 'api-test-demo'
self._expire = 7200
self._payload = {
"iat": time.time(),
"exp": time.time() + self._expire
}
@property
def algo(self):
return self._algo
@algo.setter
def algo(self, value):
if not isinstance(value, str):
raise ValueError('algo must be string')
if value == '':
raise ValueError("algo must not be null string")
else:
self._algo = value
@property
def secret(self):
return self._secret
@secret.setter
def secret(self, value):
if not isinstance(value, str):
raise ValueError("secret should be string")
if value != '':
self._secret = value
else:
raise ValueError("secret should not be null str")
@property
def payload(self):
return self._payload
@payload.setter
def payload(self, value):
if not isinstance(value, dict):
raise ValueError("payload must be dict")
self._payload = value
@property
def expire(self):
return self._expire
@expire.setter
def expire(self, value):
if not isinstance(value, int):
raise ValueError("expire must be integer")
if 0 > value > 7200:
raise ValueError("expire value must be between 1 and 7200")
else:
self._expire = value
def generate_token(self):
# 生成token
try:
token = jwt.encode(self._payload, self._secret, algorithm=self._algo)
except Exception as e:
raise e
else:
return token
def parse_token(self, token):
# 解析token中的payload
try:
data = jwt.decode(token, self._secret, algorithms=[self._algo])
except Exception as e:
raise e
else:
print(data)
return data
def verify_token(self, token):
# 验证token是否有效，有效返回为True, 反之返回False
return self.parse_token(token) == self._payload
toc = TokenUtils()
代码中包含token生成函数，token解析函数以及是否解析正确的token函数。此外关于类实例的几个变量，这里采用了@property这个注解和@funname.setter，简单解释下这个注解的作用：通过@property装饰器，通过getter和setter方法来获取某个类实例变量和设置某个类实例变量。比如可以通过tc.expire = 72000来修改expire的值，这样做的好处是避免了在类的实例化时将所有变量全部暴露出去，如果需要修改那个变量，只需要修改那个变量即可，其他变量不受影响，避免过度暴露类信息。
总结
今天我们介绍了使用Flask框架来开发Restful格式的API接口，在这个过程中，又穿插了Redis数据库操作知识，with上下文管理器，@property装饰器和jwt格式token的生成，每个模块只做到了浅尝辄止把自己实践中的一些知识分享给大家，希望能对大家有帮助。