Flask接口开发与Redis数据持久化实战
Posted RaphealTestTech
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了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)
以上是关于Flask接口开发与Redis数据持久化实战的主要内容，如果未能解决你的问题，请参考以下文章