使用python實現(xiàn)接口的方法

更新時間：2017年07月07日 08:50:33 作者：修仙小白

接口只是定義了一些方法，而沒有去實現(xiàn)，多用于程序設計時，只是設計需要有什么樣的功能，但是并沒有實現(xiàn)任何功能，這些功能需要被另一個類（B）繼承后，由類B去實現(xiàn)其中的某個功能或全部功能。

接口基礎知識：

簡單說下接口測試，現(xiàn)在常用的2種接口就是http api和rpc協(xié)議的接口，今天主要說：http api接口是走http協(xié)議通過路徑來區(qū)分調用的方法，請求報文格式都是key-value形式，返回報文一般是json串；

接口協(xié)議：http、webservice、rpc等。

請求方式：get、post方式

請求參數(shù)格式：

　　a. get請求都是通過url?param=xxx&param1=xxx

　　b. post請求的請求參數(shù)常用類型有：application/json、application/x-www-form-urlencoded、multipart/form-data、text/html等。

還需要知道接口的url、參數(shù)類型、返回結果的數(shù)據(jù)格式、了解接口是否有header、cookie等信息。

接口的實現(xiàn)：請求方式-get，接口的寫法：

 import flask
 from flask import request
 from flask import jsonify
 import tools
 import OP_db
 import settings
 '''
 flask： web框架，可以通過flask提供的裝飾器@server.route()將普通函數(shù)轉換為服務
 登錄接口，需要傳url、username、passwd
 '''
 #創(chuàng)建一個服務，把當前這個python文件當做一個服務
 server = flask.Flask(__name__)
 #server.config['JSON_AS_ASCII'] = False
 
 # @server.route()可以將普通函數(shù)轉變?yōu)榉?登錄接口的路徑、請求方式
 @server.route('/login', methods=['get'])
 def login():
  # 獲取通過url請求傳參的數(shù)據(jù)
  username = request.values.get('name')
  # 獲取url請求傳的密碼，明文
  pwd = request.values.get('pwd')
  # 判斷用戶名、密碼都不為空，如果不傳用戶名、密碼則username和pwd為None
  if username and pwd:
   # 獲取加密后的密碼
   password = tools.md5_pwd(pwd)
   #執(zhí)行sql，如果查詢的username和password不為空，說明數(shù)據(jù)庫存在admin的賬號
   sql = 'select name,password from test where name= "%s" and password= "%s";' %(username, password)
   # 從數(shù)據(jù)查詢結果后，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': '必填參數(shù)未填寫'}
   return jsonify(res)
 
 if __name__ == '__main__':
  server.run(debug=True, port=8888, host=0.0.0.0) #指定端口、host,0.0.0.0代表不管幾個網(wǎng)卡，任何ip都可以訪問

md5加密、數(shù)據(jù)庫mysql的操作詳見我的其他博客~~~~~

get訪問接口：

項目啟動后，接口的地址是：http://127.0.0.1:5000/，默認端口是5000。

打開瀏覽器，輸入urlhttp://127.0.0.1:5000/xxx?name=xxx&pwd=123456，后面跟上接口的地址login，參數(shù)跟url直接使用？相連，每個請求參數(shù)直接使用&相連。請求成功，則返回{'code': 200, 'message': '登錄成功'}。

請求方式-post，接口的寫法：

 import flask
 from flask import jsonify
 from flask import request
 from conf import opMysql
 from conf import md5_create
 '''
 注冊接口：
 post請求，請求參數(shù)入?yún)㈩愋蚸son
 {
  "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':
   # 獲取請求參數(shù)是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
    # 查詢注冊的用戶是否存在數(shù)據(jù)庫，如果存在，則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的話，其他人可以訪問，代表監(jiān)聽多塊網(wǎng)卡上面，默認是127.0.0.1
  server.run(debug=True, port=8899, host='0.0.0.0')

post訪問接口：

項目啟動后，接口的地址是：http://127.0.0.1:5000/，默認端口是5000。

打開瀏覽器，輸入urlhttp://127.0.0.1:5000/xxx，后面跟上接口的地址register，參數(shù)使用postman或jmeter進行請求，參數(shù)類型是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請求，請求參數(shù)入?yún)㈩愋蚸son
 {
  "username":"aaa",
  "pwd":"123456",
  "c_pwd":"123456"
 }
 '''
 server = flask.Flask(__name__)
 @server.route('/register', methods=['get', 'post'])
 def registerPost():
  #post請求獲取請求的參數(shù)，返回結果類型是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
   # 查詢注冊的用戶是否存在數(shù)據(jù)庫，如果存在，則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的話，其他人可以訪問，代表監(jiān)聽多塊網(wǎng)卡上面，
  server.run(debug=True, port=8899, host='0.0.0.0')

可以通過以下2種方式進行post請求，一種如下：

通過url拼接參數(shù)：

第二種訪問方式：通過key-value方式進行訪問：

redis相關操作，添加hash類型的值到redis內，接口實現(xiàn)如下：

 import flask
 from flask import jsonify
 from conf import opRedis
 from flask import request
 '''
 redis添加數(shù)據(jù)，存入數(shù)據(jù)的類型是hash類型，格式如下：
 post請求，請求參數(shù)入?yún)㈩愋蚸son
 {name:{"key":"value"}}
 {"username":"url"}
 '''
 server = flask.Flask(__name__)
 @server.route('/set_sties', methods =['post'])
 def set_sties():
  # 獲取url請求參數(shù)，返回結果是字典{"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的話，其他人可以訪問，代表監(jiān)聽多塊網(wǎng)卡上面，
  server.run(debug=True, port=8899, host='0.0.0.0')

hash類型結構如下：

{name:{key,value}}，接口訪問成功后，redis內數(shù)據(jù)存儲結構如下：

redis添加完數(shù)據(jù)后，讀取redis內的數(shù)據(jù)，接口實現(xiàn)如下：

 import flask
 from flask import jsonify
 from conf import opRedis
 from flask import request
 '''
 讀取redis內的數(shù)據(jù)，redis數(shù)據(jù)存儲類型是hash類型，格式如下
 {name:{"key":"value"}}
 思路： 1.通過redis的hgetall(name)方法讀取redis所有數(shù)據(jù)，返回結果類型是字典
  2. 循環(huán)字典內容，將元素類型轉換為str，并將結果存放到字典內
 '''
 server = flask.Flask(__name__)
 @server.route('/get_sties', methods =['get', 'post'])
 def get_sties():
  #獲取redis內所有的數(shù)據(jù)信息，返回結果類型是字典，里面元素是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的話，其他人可以訪問，代表監(jiān)聽多塊網(wǎng)卡上面，
  server.run(debug=True, port=8899, host='0.0.0.0')

通過postman方法接口，返回數(shù)據(jù)如下：

查詢用戶，需要傳token值，實現(xiàn)方法如下：

登錄接口：

 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
 '''
 登錄接口，需要傳用戶名、密碼，通過查詢數(shù)據(jù)庫判斷用戶是否登錄成功，若登錄成功則將用戶名和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，存入數(shù)據(jù)類型是hash類型
    opRedis.get_hashall('user', name, new_token)
    return jsonify({"code": 200})
   else:
    return jsonify({"code": 204})
  else:
   return jsonify({"code": 304})

查詢用戶，需要傳用戶名和token值，實現(xiàn)方法如下：

 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
 '''
 登錄接口，需要傳用戶名、密碼，通過查詢數(shù)據(jù)庫判斷用戶是否登錄成功，若登錄成功則將用戶名和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:
   #查看數(shù)據(jù)庫，看查詢的用戶是否存在，若存在則返回用戶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的話，其他人可以訪問，代表監(jiān)聽多塊網(wǎng)卡上面，
  server.run(debug=True, port=8899, host='0.0.0.0')

以上就是工作中常用的一些接口場景，測試支付相關接口、或者第三方接口時，可以自己mock接口返回假數(shù)據(jù)操作~~~~

您可能感興趣的文章: