使用python實(shí)現(xiàn)接口的方法
更新時(shí)間：2017年07月07日 08:50:33 作者：修仙小白
接口只是定義了一些方法，而沒(méi)有去實(shí)現(xiàn)，多用于程序設(shè)計(jì)時(shí)，只是設(shè)計(jì)需要有什么樣的功能，但是并沒(méi)有實(shí)現(xiàn)任何功能，這些功能需要被另一個(gè)類(lèi)（B）繼承后，由 類(lèi)B去實(shí)現(xiàn)其中的某個(gè)功能或全部功能。
接口基礎(chǔ)知識(shí)：
簡(jiǎn)單說(shuō)下接口測(cè)試，現(xiàn)在常用的2種接口就是http api和rpc協(xié)議的接口，今天主要說(shuō)：http api接口是走h(yuǎn)ttp協(xié)議通過(guò)路徑來(lái)區(qū)分調(diào)用的方法，請(qǐng)求報(bào)文格式都是key-value形式，返回報(bào)文一般是json串；
接口協(xié)議：http、webservice、rpc等。
請(qǐng)求方式：get、post方式
請(qǐng)求參數(shù)格式：
a. get請(qǐng)求都是通過(guò)url?param=xxx¶m1=xxx
b. post請(qǐng)求的請(qǐng)求參數(shù)常用類(lèi)型有：application/json、application/x-www-form-urlencoded、multipart/form-data、text/html等。
還需要知道接口的url、參數(shù)類(lèi)型、返回結(jié)果的數(shù)據(jù)格式、了解接口是否有header、cookie等信息。
接口的實(shí)現(xiàn)：請(qǐng)求方式-get，接口的寫(xiě)法：
import flask
from flask import request
from flask import jsonify
import tools
import OP_db
import settings
'''
flask： web框架，可以通過(guò)flask提供的裝飾器@server.route()將普通函數(shù)轉(zhuǎn)換為服務(wù)
登錄接口，需要傳url、username、passwd
'''
#創(chuàng)建一個(gè)服務(wù)，把當(dāng)前這個(gè)python文件當(dāng)做一個(gè)服務(wù)
server = flask.Flask(__name__)
#server.config['JSON_AS_ASCII'] = False
# @server.route()可以將普通函數(shù)轉(zhuǎn)變?yōu)榉?wù) 登錄接口的路徑、請(qǐng)求方式
@server.route('/login', methods=['get'])
def login():
# 獲取通過(guò)url請(qǐng)求傳參的數(shù)據(jù)
username = request.values.get('name')
# 獲取url請(qǐng)求傳的密碼，明文
pwd = request.values.get('pwd')
# 判斷用戶(hù)名、密碼都不為空，如果不傳用戶(hù)名、密碼則username和pwd為None
if username and pwd:
# 獲取加密后的密碼
password = tools.md5_pwd(pwd)
#執(zhí)行sql，如果查詢(xún)的username和password不為空，說(shuō)明數(shù)據(jù)庫(kù)存在admin的賬號(hào)
sql = 'select name,password from test where name= "%s" and password= "%s";' %(username, password)
# 從數(shù)據(jù)查詢(xún)結(jié)果后，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的結(jié)果不為空，說(shuō)明找到了username=admin的用戶(hù)，且password為加密前的123456
resu = {'code': 200, 'message': '登錄成功'}
return jsonify(resu) #將字典轉(zhuǎn)換為json串, json是字符串
else:
resu = {'code': -1, 'message': '賬號(hào)/密碼錯(cuò)誤'}
return jsonify(resu)
else:
res = {'code': 999, 'message': '必填參數(shù)未填寫(xiě)'}
return jsonify(res)
if __name__ == '__main__':
server.run(debug=True, port=8888, host=0.0.0.0) #指定端口、host,0.0.0.0代表不管幾個(gè)網(wǎng)卡，任何ip都可以訪問(wèn)
md5加密、數(shù)據(jù)庫(kù)mysql的操作詳見(jiàn)我的其他博客~~~~~
get訪問(wèn)接口：
項(xiàng)目啟動(dòng)后，接口的地址是：http://127.0.0.1:5000/，默認(rèn)端口是5000。
打開(kāi)瀏覽器，輸入urlhttp://127.0.0.1:5000/xxx?name=xxx&pwd=123456，后面跟上接口的地址login，參數(shù)跟url直接使用？相連，每個(gè)請(qǐng)求參數(shù)直接使用&相連。請(qǐng)求成功，則返回{'code': 200, 'message': '登錄成功'}。
請(qǐng)求方式-post，接口的寫(xiě)法：
import flask
from flask import jsonify
from flask import request
from conf import opMysql
from conf import md5_create
'''
注冊(cè)接口：
post請(qǐng)求，請(qǐng)求參數(shù)入?yún)㈩?lèi)型json
{
"username":"aaa",
"pwd":"123456",
"c_pwd":"123456"
}
'''
server = flask.Flask(__name__)
@server.route('/register', methods=['get', 'post'])
def registerPost():
#判斷接口的請(qǐng)求方式是GET還是POST
if request.method == 'POST':
# 獲取請(qǐng)求參數(shù)是json格式，返回結(jié)果是字典
params = request.json
username = params.get('username')
pwd = params.get('pwd')
confirmpwd = params.get('confirmpwd')
if username and pwd and confirmpwd: # 判斷輸入的用戶(hù)名、密碼、確認(rèn)密碼都不為空
select_sql = 'select username from lhldemo where username = "%s" ;'%username
# 查詢(xún)注冊(cè)的用戶(hù)是否存在數(shù)據(jù)庫(kù)，如果存在，則username不為空，否則username為空
res_mysql = opMysql.op_select(select_sql)
if res_mysql:
return jsonify({"code": 999, "mesg": "用戶(hù)已注冊(cè)"})
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": "注冊(cè)成功"})
else:
return jsonify({"code":998, "msg":"密碼不一樣"})
else:
return jsonify({"code": 504, "msg": "必填項(xiàng)不能為空"})
else:
return jsonify({"code": 201, "msg": "請(qǐng)求方式不正確"})
if __name__ == '__main__':
#port可以指定端口，默認(rèn)端口是5000
#host寫(xiě)成0.0.0.0的話(huà)，其他人可以訪問(wèn)，代表監(jiān)聽(tīng)多塊網(wǎng)卡上面，默認(rèn)是127.0.0.1
server.run(debug=True, port=8899, host='0.0.0.0')
post訪問(wèn)接口：
項(xiàng)目啟動(dòng)后，接口的地址是：http://127.0.0.1:5000/，默認(rèn)端口是5000。
打開(kāi)瀏覽器，輸入urlhttp://127.0.0.1:5000/xxx，后面跟上接口的地址register，參數(shù)使用postman或jmeter進(jìn)行請(qǐng)求，參數(shù)類(lèi)型是json。請(qǐng)求成功，則返回{'code': 200, 'message': '登錄成功'}。
請(qǐng)求方式-get、post都可以訪問(wèn)，寫(xiě)法如下：
import flask
from flask import jsonify
from flask import request
from conf import opMysql
from conf import md5_create
'''
注冊(cè)接口：
post請(qǐng)求，請(qǐng)求參數(shù)入?yún)㈩?lèi)型json
{
"username":"aaa",
"pwd":"123456",
"c_pwd":"123456"
}
'''
server = flask.Flask(__name__)
@server.route('/register', methods=['get', 'post'])
def registerPost():
#post請(qǐng)求獲取請(qǐng)求的參數(shù)，返回結(jié)果類(lèi)型是str
username = request.values.get('username')
pwd = request.values.get('pwd')
confirmpwd = request.values.get('confirmpwd')
if username and pwd and confirmpwd: # 判斷輸入的用戶(hù)名、密碼、確認(rèn)密碼都不為空
select_sql = 'select username from lhldemo where username = "%s" ;'%username
# 查詢(xún)注冊(cè)的用戶(hù)是否存在數(shù)據(jù)庫(kù)，如果存在，則username不為空，否則username為空
res_mysql = opMysql.op_select(select_sql)
if res_mysql:
return jsonify({"code": 999, "mesg": "用戶(hù)已注冊(cè)"})
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": "注冊(cè)成功"})
else:
return jsonify({"code": 998, "msg": "密碼不一樣"})
else:
return jsonify({"code": 504, "msg": "必填項(xiàng)不能為空"})
if __name__ == '__main__':
#port可以指定端口，默認(rèn)端口是5000
#host默認(rèn)是127.0.0.1,寫(xiě)成0.0.0.0的話(huà)，其他人可以訪問(wèn)，代表監(jiān)聽(tīng)多塊網(wǎng)卡上面，
server.run(debug=True, port=8899, host='0.0.0.0')
可以通過(guò)以下2種方式進(jìn)行post請(qǐng)求，一種如下：
通過(guò)url拼接參數(shù)：
第二種訪問(wèn)方式：通過(guò)key-value方式進(jìn)行訪問(wèn)：
redis相關(guān)操作，添加hash類(lèi)型的值到redis內(nèi)，接口實(shí)現(xiàn)如下：
import flask
from flask import jsonify
from conf import opRedis
from flask import request
'''
redis添加數(shù)據(jù)，存入數(shù)據(jù)的類(lèi)型是hash類(lèi)型，格式如下：
post請(qǐng)求，請(qǐng)求參數(shù)入?yún)㈩?lèi)型json
{name:{"key":"value"}}
{"username":"url"}
'''
server = flask.Flask(__name__)
@server.route('/set_sties', methods =['post'])
def set_sties():
# 獲取url請(qǐng)求參數(shù)，返回結(jié)果是字典{"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')
#調(diào)用redis的hset方法，將username、url存入redis
opRedis.get_hashall('sites', username, url)
return jsonify({"code":20})
else:
return jsonify({"code": 204, "msg": "必填項(xiàng)不能為空"})
if __name__ == '__main__':
#port可以指定端口，默認(rèn)端口是5000
#host默認(rèn)是127.0.0.1,寫(xiě)成0.0.0.0的話(huà)，其他人可以訪問(wèn)，代表監(jiān)聽(tīng)多塊網(wǎng)卡上面，
server.run(debug=True, port=8899, host='0.0.0.0')
hash類(lèi)型結(jié)構(gòu)如下：
{name:{key,value}}，接口訪問(wèn)成功后，redis內(nèi)數(shù)據(jù)存儲(chǔ)結(jié)構(gòu)如下：
redis添加完數(shù)據(jù)后，讀取redis內(nèi)的數(shù)據(jù)，接口實(shí)現(xiàn)如下：
import flask
from flask import jsonify
from conf import opRedis
from flask import request
'''
讀取redis內(nèi)的數(shù)據(jù)，redis數(shù)據(jù)存儲(chǔ)類(lèi)型是hash類(lèi)型，格式如下
{name:{"key":"value"}}
思路： 1.通過(guò)redis的hgetall(name)方法讀取redis所有數(shù)據(jù)，返回結(jié)果類(lèi)型是字典
2. 循環(huán)字典內(nèi)容，將元素類(lèi)型轉(zhuǎn)換為str，并將結(jié)果存放到字典內(nèi)
'''
server = flask.Flask(__name__)
@server.route('/get_sties', methods =['get', 'post'])
def get_sties():
#獲取redis內(nèi)所有的數(shù)據(jù)信息，返回結(jié)果類(lèi)型是字典，里面元素是bytes類(lèi)型，name=sites
dic = opRedis.get_hashall('sites')
redisList = []
for key, value in dic.items():
redis_dic = {}
#將字典內(nèi)元素的類(lèi)型由bytes轉(zhuǎn)換為str
k = key.decode()
v = value.decode()
#字典redis_dic內(nèi)結(jié)構(gòu){"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可以指定端口，默認(rèn)端口是5000
#host默認(rèn)是127.0.0.1,寫(xiě)成0.0.0.0的話(huà)，其他人可以訪問(wèn)，代表監(jiān)聽(tīng)多塊網(wǎng)卡上面，
server.run(debug=True, port=8899, host='0.0.0.0')
通過(guò)postman方法接口，返回?cái)?shù)據(jù)如下：
查詢(xún)用戶(hù)，需要傳token值，實(shí)現(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
'''
登錄接口，需要傳用戶(hù)名、密碼，通過(guò)查詢(xún)數(shù)據(jù)庫(kù)判斷用戶(hù)是否登錄成功，若登錄成功則將用戶(hù)名和token存入redis內(nèi)
'''
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)
#用戶(hù)登錄成功后，將name和token存入redis，存入數(shù)據(jù)類(lèi)型是hash類(lèi)型
opRedis.get_hashall('user', name, new_token)
return jsonify({"code": 200})
else:
return jsonify({"code": 204})
else:
return jsonify({"code": 304})
查詢(xún)用戶(hù)，需要傳用戶(hù)名和token值，實(shí)現(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
'''
登錄接口，需要傳用戶(hù)名、密碼，通過(guò)查詢(xún)數(shù)據(jù)庫(kù)判斷用戶(hù)是否登錄成功，若登錄成功則將用戶(hù)名和token存入redis內(nèi)
'''
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ù)庫(kù)，看查詢(xún)的用戶(hù)是否存在，若存在則返回用戶(hù)id
sql = 'select id from lhldemo where username="%s" ; ' % (name)
res_sql = opMysql.op_select(sql)
if res_sql:
#從redis中獲取user下的用戶(hù)名對(duì)應(yīng)的token值
res_token = opRedis.getRedis('user:'+name)26 if res_token == token:
return jsonify({"msg": "用戶(hù)id", "id": res_sql})
else:
return jsonify({"msg": "token錯(cuò)誤"})
else:
return jsonify({"code": "用戶(hù)不存在"})
else:
return jsonify({"code": "必填項(xiàng)不能為空"})
if __name__ == '__main__':
#port可以指定端口，默認(rèn)端口是5000
#host默認(rèn)是127.0.0.1,寫(xiě)成0.0.0.0的話(huà)，其他人可以訪問(wèn)，代表監(jiān)聽(tīng)多塊網(wǎng)卡上面，
server.run(debug=True, port=8899, host='0.0.0.0')
以上就是工作中常用的一些接口場(chǎng)景，測(cè)試支付相關(guān)接口、或者第三方接口時(shí)，可以自己mock接口返回假數(shù)據(jù)操作~~~~
相關(guān)文章
這篇文章主要介紹了Python的gevent框架的入門(mén)教程,示例代碼基于Python2.x版本,需要的朋友可以參考下2015-04-04
PYQT5 實(shí)現(xiàn)給listwidget的滾動(dòng)條添加滾動(dòng)信號(hào)
這篇文章主要介紹了PYQT5 實(shí)現(xiàn)給listwidget的滾動(dòng)條添加滾動(dòng)信號(hào)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2021-03-03
Django使用redis緩存服務(wù)器的實(shí)現(xiàn)代碼示例
這篇文章主要介紹了Django使用redis緩存服務(wù)器的實(shí)現(xiàn)代碼示例，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-04-04
Python利用wxPython實(shí)現(xiàn)ocr識(shí)別圖片漢字程序
在這篇博客中，我們將介紹一個(gè)如何使用wxPython構(gòu)建的簡(jiǎn)單OCR識(shí)別圖片漢字應(yīng)用程序，文章的示例代碼講解詳細(xì)，感興趣的小伙伴可以學(xué)習(xí)一下2023-08-08
這篇文章主要介紹了Java分治歸并排序算法,結(jié)合實(shí)例形式詳細(xì)分析了分治歸并排序算法的原理及java實(shí)現(xiàn)技巧,需要的朋友可以參考下2017-12-12
這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)QQ批量登錄功能，以及python3.6批量登陸QQ，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-06-06
用Python的線(xiàn)程來(lái)解決生產(chǎn)者消費(fèi)問(wèn)題的示例
這篇文章主要介紹了用Python的線(xiàn)程來(lái)解決生產(chǎn)者消費(fèi)問(wèn)題的示例,包括對(duì)使用線(xiàn)程中容易出現(xiàn)的一些問(wèn)題給出了相關(guān)解答,需要的朋友可以參考下2015-04-04
這篇文章主要為大家詳細(xì)介紹了Python實(shí)現(xiàn)單向鏈表，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2022-05-05
python如何通過(guò)pyqt5實(shí)現(xiàn)進(jìn)度條
這篇文章主要介紹了python如何通過(guò)pyqt5實(shí)現(xiàn)進(jìn)度條,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-01-01
Django項(xiàng)目如何給數(shù)據(jù)庫(kù)添加約束
這篇文章主要介紹了Django項(xiàng)目如何給數(shù)據(jù)庫(kù)添加約束，幫助大家更好的理解和學(xué)習(xí)使用Django框架，感興趣的朋友可以了解下2021-04-04