使用python实现接口的方法

m0_70669463

8066人浏览 · 2023-12-15 16:18:10

m0_70669463 · 2023-12-15 16:18:10 发布

接口基础知识：

简单说下接口测试，现在常用的2种接口就是http api和rpc协议的接口，今天主要说：http api接口是走http协议通过路径来区分调用的方法，请求报文格式都是key-value形式，返回报文一般是json串；

接口协议：http、webservice、rpc等。

请求方式：get、post方式

请求参数格式：

　　a. get请求都是通过url?param=xxx&param1=xxx

　　b. 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。

打开浏览器，输入urlhttp://127.0.0.1:5000/xxx?name=xxx&pwd=123456，后面跟上接口的地址login，参数跟url直接使用？相连，每个请求参数直接使用&相连。请求成功，则返回{'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。

打开浏览器，输入urlhttp://127.0.0.1:5000/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')

可以通过以下2种方式进行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接口返回假数据操作~~~~

现在我也找了很多测试的朋友，做了一个分享技术的交流群，共享了很多我们收集的技术文档和视频教程。
如果你不想再体验自学时找不到资源，没人解答问题，坚持几天便放弃的感受
可以加入我们一起交流。而且还有很多在自动化，性能，安全，测试开发等等方面有一定建树的技术大牛
分享他们的经验，还会分享很多直播讲座和技术沙龙
可以免费学习！划重点！开源的！！！
qq群号：485187702【暗号：csdn11】

最后感谢每一个认真阅读我文章的人，看着粉丝一路的上涨和关注，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走！ 希望能帮助到你！【100%无套路免费领取】

GitCode 开源社区

旨在为数千万中国开发者提供一个无缝且高效的云端环境，以支持学习、使用和贡献开源项目。

更多推荐

[转载]在Windows环境下安装GNU Radio

转自：在Windows环境下安装GNURadio_恐弱智_新浪博客GNU Radio是用Python开发的，大部分开源的工程能够在Linux环境下运行良好，而Windows下却运行的很勉强，而且安装配置都很复杂。GNU Radio算是个例外了，不光提供了Windows的二进制安装，还有比较详细的说明。我是Python小白，所以折腾了好久才弄好，特意记录下来，免得以后再装还折腾。GNU Radio的

GitCode 开源社区

centOS 8 使用dnf安装Docker

DNF是什么？CentOS 8使用YUM软件包管理器版本v4.0.4。现在，该版本使用DNF(已删除YUM)。DNF是软件包管理器。它会在Linux发行版上安装，执行更新并删除软件包。使用DNF安装Docker跳过具有损坏依赖性的程序包一个有效的解决方案是使您的CentOS 8系统使用以下--nobest命令安装最符合条件的版本：sudo dnf install docker...

GitCode 开源社区

定时同步数据库表(mysql+linux+crontab)

sync.sh里面的参数需要改变，ip/username/password/database/tablesync.sh#!/bin/sh# Please change the IP and password of the data source db.# Then change the table name.filename=/home/nington/db/$(date +%Y-%m