Python Flask接口开发实战：GET/POST请求与Redis数据交互

小鹿同学

发布时间：2026-10-13| 4人浏览过

原创

本文详解使用Python Flask框架实现HTTP API接口的核心方法。涵盖GET与POST请求的路由配置、参数解析及JSON响应处理，结合MySQL数据库查询与MD5加密逻辑。重点演示如何通过Flask接口向Redis存储和读取Hash类型数据，并介绍在支付及第三方接口测试中利用Mock返回假数据的实用技巧，帮助开发者快速掌握基础接口构建流程。

接口基础知识：

简要介绍接口测试，目前主流接口分为HTTP API和RPC协议两类。本文重点讲解HTTP API接口，其基于HTTP协议，通过URL路径区分调用的方法。请求报文通常采用键值对（key-value）格式，而返回报文则多为JSON字符串。

常见的接口协议包括HTTP、WebService以及RPC等。

常用的请求方式包括GET和POST。

请求参数的格式因请求方式而异：

对于GET请求，参数通常通过URL查询字符串传递，格式为url?param=xxx&param1=xxx。

POST请求支持多种参数类型，常用的包括application/json、application/x-www-form-urlencoded、multipart/form-data以及text/html等。

在开发接口时，需明确接口的URL地址、参数类型、返回结果的数据格式，并确认是否涉及Header或Cookie等额外信息。

以下是接口的具体实现方式，首先展示GET请求的写法：

 import flask
 from flask import request
 from flask import jsonify
 import tools
 import OP_db
 import settings
 '''
 flask： web框架，可以通过flask提供的装饰器@server.route()将普通函数转换为服务
 登录接口，需要传url、username、passwd
 '''
 #创建一个服务，把当前这个python文件当做一个服务
 server = flask.Flask(__name__)
 #server.config['JSON_AS_ASCII'] = False
 
 # @server.route()可以将普通函数转变为服务 登录接口的路径、请求方式
 @server.route('/login', methods=['get'])
 def login():
  # 获取通过url请求传参的数据
  username = request.values.get('name')
  # 获取url请求传的密码，明文
  pwd = request.values.get('pwd')
  # 判断用户名、密码都不为空，如果不传用户名、密码则username和pwd为None
  if username and pwd:
   # 获取加密后的密码
   password = tools.md5_pwd(pwd)
   #执行sql，如果查询的username和password不为空，说明数据库存在admin的账号
   sql = 'select name,password from test where name= "%s" and password= "%s";' %(username, password)
   # 从数据查询结果后，res返回是元组
   res = OP_db.getconn(
    host=settings.mysql_info['host'],
    user=settings.mysql_info['user'],
    passwd=settings.mysql_info['pwd'],
    db=settings.mysql_info['db'],
    port=settings.mysql_info['port'],
    sql=sql
   )
   if res:  #res的结果不为空，说明找到了username=admin的用户，且password为加密前的123456
    resu = {'code': 200, 'message': '登录成功'}
    return jsonify(resu) #将字典转换为json串, json是字符串
   else:
    resu = {'code': -1, 'message': '账号/密码错误'}
    return jsonify(resu)
  else:
   res = {'code': 999, 'message': '必填参数未填写'}
   return jsonify(res)
 
 if __name__ == '__main__':
  server.run(debug=True, port=8888, host=0.0.0.0) #指定端口、host,0.0.0.0代表不管几个网卡，任何ip都可以访问

关于MD5加密及MySQL数据库的具体操作，详见我的其他博客文章。

GET方式访问接口：

项目启动后，接口的基础地址为http://127.0.0.1:5000/，默认监听端口为5000。

在浏览器中输入URL http://127.0.0.1:5000/xxx?name=xxx&pwd=123456，其中xxx为接口路径（如login），参数通过?连接，多个参数间用&分隔。请求成功后，返回结果示例为{'code': 200, 'message': '登录成功'}。

接下来是POST请求方式的接口实现：

 import flask
 from flask import jsonify
 from flask import request
 from conf import opMysql
 from conf import md5_create
 '''
 注册接口：
 post请求，请求参数入参类型json
 {
  "username":"aaa",
  "pwd":"123456",
  "c_pwd":"123456"
 }
 '''
 server = flask.Flask(__name__)
 @server.route('/register', methods=['get', 'post'])
 def registerPost():
  #判断接口的请求方式是GET还是POST
  if request.method == 'POST':
   # 获取请求参数是json格式，返回结果是字典
   params = request.json
   username = params.get('username')
   pwd = params.get('pwd')
   confirmpwd = params.get('confirmpwd')
   if username and pwd and confirmpwd: # 判断输入的用户名、密码、确认密码都不为空
    select_sql = 'select username from lhldemo where username = "%s" ;'%username
    # 查询注册的用户是否存在数据库，如果存在，则username不为空，否则username为空
    res_mysql = opMysql.op_select(select_sql)
    if res_mysql:
     return jsonify({"code": 999, "mesg": "用户已注册"})
    else:
     if pwd == confirmpwd: # 判断pwd和confirmpwd一致
      new_pwd = md5_create.md5_test(pwd) # 加密后的密码
      insert_sql = 'insert into lhldemo(username,password) values("%s", "%s") ;' % (username, new_pwd)
      opMysql.op_insert(insert_sql)
      return jsonify({"code": 200, "msg": "注册成功"})
     else:
      return jsonify({"code":998, "msg":"密码不一样"})
   else:
    return jsonify({"code": 504, "msg": "必填项不能为空"})
  else:
   return jsonify({"code": 201, "msg": "请求方式不正确"})
 
 if __name__ == '__main__':
  #port可以指定端口，默认端口是5000
  #host写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，默认是127.0.0.1
  server.run(debug=True, port=8899, host='0.0.0.0')

POST方式访问接口的示例：

项目启动后，接口基础地址同样为http://127.0.0.1:5000/，默认端口为5000。

在浏览器或工具中输入URL http://127.0.0.1:5000/xxx，其中xxx为接口路径（如register）。使用Postman或JMeter发起请求，参数类型为JSON。请求成功后，返回结果示例为{'code': 200, 'message': '登录成功'}。

若接口同时支持GET和POST请求，写法如下：

 import flask
 from flask import jsonify
 from flask import request
 from conf import opMysql
 from conf import md5_create
 '''
 注册接口：
 post请求，请求参数入参类型json
 {
  "username":"aaa",
  "pwd":"123456",
  "c_pwd":"123456"
 }
 '''
 server = flask.Flask(__name__)
 @server.route('/register', methods=['get', 'post'])
 def registerPost():
  #post请求获取请求的参数，返回结果类型是str
  username = request.values.get('username')
  pwd = request.values.get('pwd')
  confirmpwd = request.values.get('confirmpwd')
  if username and pwd and confirmpwd: # 判断输入的用户名、密码、确认密码都不为空
   select_sql = 'select username from lhldemo where username = "%s" ;'%username
   # 查询注册的用户是否存在数据库，如果存在，则username不为空，否则username为空
   res_mysql = opMysql.op_select(select_sql)
   if res_mysql:
    return jsonify({"code": 999, "mesg": "用户已注册"})
   else:
    if pwd == confirmpwd: # 判断pwd和confirmpwd一致
     new_pwd = md5_create.md5_test(pwd) # 加密后的密码
     insert_sql = 'insert into lhldemo(username,password) values("%s", "%s") ;' % (username, new_pwd)
     opMysql.op_insert(insert_sql)
     return jsonify({"code": 200, "msg": "注册成功"})
    else:
     return jsonify({"code": 998, "msg": "密码不一样"})
  else:
   return jsonify({"code": 504, "msg": "必填项不能为空"})
 
 
 if __name__ == '__main__':
  #port可以指定端口，默认端口是5000
  #host默认是127.0.0.1,写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，
  server.run(debug=True, port=8899, host='0.0.0.0')

可以通过以下两种方式进行POST请求，其中一种方式如下：

第一种访问方式是通过URL拼接参数：

第二种访问方式是通过key-value形式传递参数：

涉及Redis相关操作，将Hash类型的值存入Redis的接口实现如下：

 import flask
 from flask import jsonify
 from conf import opRedis
 from flask import request
 '''
 redis添加数据，存入数据的类型是hash类型，格式如下：
 post请求，请求参数入参类型json
 {name:{"key":"value"}}
 {"username":"url"}
 '''
 server = flask.Flask(__name__)
 @server.route('/set_sties', methods =['post'])
 def set_sties():
  # 获取url请求参数，返回结果是字典{"username":"byz","url":"http://www.baidu.com"}
  res_dic = request.json
  if res_dic.get('username') and res_dic.get('url'):
   username = res_dic.get('username')
   url = res_dic.get('url')
   #调用redis的hset方法，将username、url存入redis
   opRedis.get_hashall('sites', username, url)
   return jsonify({"code":20})
  else:
   return jsonify({"code": 204, "msg": "必填项不能为空"})
 
 if __name__ == '__main__':
  #port可以指定端口，默认端口是5000
  #host默认是127.0.0.1,写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，
  server.run(debug=True, port=8899, host='0.0.0.0')

Hash类型的数据结构示例如下：

数据结构为{name:{key,value}}，接口访问成功后，Redis内的数据存储结构如下：

数据存入Redis后，读取Redis内数据的接口实现如下：

 import flask
 from flask import jsonify
 from conf import opRedis
 from flask import request
 '''
 读取redis内的数据，redis数据存储类型是hash类型，格式如下
 {name:{"key":"value"}}
 思路： 1.通过redis的hgetall(name)方法读取redis所有数据，返回结果类型是字典
  2. 循环字典内容，将元素类型转换为str，并将结果存放到字典内
 '''
 server = flask.Flask(__name__)
 @server.route('/get_sties', methods =['get', 'post'])
 def get_sties():
  #获取redis内所有的数据信息，返回结果类型是字典，里面元素是bytes类型，name=sites
  dic = opRedis.get_hashall('sites')
  redisList = []
  for key, value in dic.items():
   redis_dic = {}
   #将字典内元素的类型由bytes转换为str
   k = key.decode()
   v = value.decode()
   #字典redis_dic内结构{"username:k, "url":v}
   redis_dic['username'] = k
   redis_dic['url'] = v
   redisList.append(redis_dic)
  return jsonify({"code": 200, "msg": redisList})
 
 if __name__ == '__main__':
  #port可以指定端口，默认端口是5000
  #host默认是127.0.0.1,写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，
  server.run(debug=True, port=8899, host='0.0.0.0')

通过Postman调用接口后，返回的数据结果如下：

查询用户信息时，需要传递Token值，具体实现方法如下：

登录接口的具体代码：

 import flask
 from flask import jsonify
 from conf import opRedis
 from conf import opMysql
 from conf import md5_create
 from flask import request
 import time
 '''
 登录接口，需要传用户名、密码，通过查询数据库判断用户是否登录成功，若登录成功则将用户名和token存入redis内
 '''
 server = flask.Flask(__name__)
 @server.route('/login', methods=['get','post'])
 def set_cookies():
  name = request.values.get('username')
  pwd = request.values.get('pwd')
  if name and pwd:
   #加密后的密码
   new_pwd = md5_create.md5_test(pwd)
   sql = 'select username,password from lhldemo where username="%s" and password="%s" ; ' % (name, new_pwd)
   res_sql = opMysql.op_select(sql)
   if res_sql:
    token = name + time.strftime('%Y%m%d%H%M%S')
    new_token = md5_create.md5_test(token)
    #用户登录成功后，将name和token存入redis，存入数据类型是hash类型
    opRedis.get_hashall('user', name, new_token)
    return jsonify({"code": 200})
   else:
    return jsonify({"code": 204})
  else:
   return jsonify({"code": 304})

查询用户时，需传递用户名和Token值，具体实现方法如下：

 import flask
 from flask import jsonify
 from conf import opRedis
 from conf import opMysql
 from conf import md5_create
 from flask import request
 import time
 '''
 登录接口，需要传用户名、密码，通过查询数据库判断用户是否登录成功，若登录成功则将用户名和token存入redis内
 '''
 server = flask.Flask(__name__)
 @server.route('/search_user', methods=['get','post'])
 def set_cookies():
  name = request.values.get('username')
  token = request.values.get('token')
  print('token',token)
  if name and token:
   #查看数据库，看查询的用户是否存在，若存在则返回用户id
   sql = 'select id from lhldemo where username="%s" ; ' % (name)
   res_sql = opMysql.op_select(sql)
   if res_sql:
    #从redis中获取user下的用户名对应的token值
    res_token = opRedis.getRedis('user:'+name)26    if res_token == token:
     return jsonify({"msg": "用户id", "id": res_sql})
    else:
     return jsonify({"msg": "token错误"})
   else:
    return jsonify({"code": "用户不存在"})
  else:
   return jsonify({"code": "必填项不能为空"})
 
 if __name__ == '__main__':
  #port可以指定端口，默认端口是5000
  #host默认是127.0.0.1,写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，
  server.run(debug=True, port=8899, host='0.0.0.0')

以上涵盖了工作中常用的接口场景。在测试支付相关接口或第三方接口时，可以通过Mock接口返回假数据进行模拟操作。