python使用笔记22--mock接口开发

1.mock接口开发

mock是模拟一个接口的意思

为了不阻止测试，开发一个接口，返回你想要的数据，模拟各种场景

需要安装第三方模块flask，flask是web轻量级开发框架

1.1 flask

pip install flask

#mock 模拟一个接口的意思
#http://test.pay.alibab.com/payment
#1、为了不阻碍测试，开发了一个接口，返回你想要的数据，模拟各种场景

import flask
import json
import pymysql
import hashlib
import redis
import time
#flask web轻量级开发框架
server = flask.Flask(__name__)


mysql_info = {'host':'127.0.0.1',
              'port':3306,
              'user':'root',
              'password':'123456',
              'autocommit':True,
              'db':'db001',
              'charset':'utf8'}


#md5加密之后，是固定的，这样的安全性不够高
#可以加盐，salt='xiaoxiaolin_rew43!@'，设置一串复杂的字符串，加强安全性
def md5(s,salt='xiaoxiaolin_rew43!@'):#md5加密
    s = str(s+salt)
    m = hashlib.md5(s.encode())
    return m.hexdigest()

def op_redis(key,value=None,expire=60*60*2):
    r = redis.Redis(host="127.0.0.1", port=6379, password="123456&*", decode_responses=True)
    if value:#如果传入值不为空，则保存值
        r.set(key,value,expire)
    else:#value值为空，则获取值
        return r.get(key)



def select(sql,oneTag=False):
        connect = pymysql.connect(**mysql_info)#**表示将字典转成host='xxx',port=3306
        cur = connect.cursor(pymysql.cursors.DictCursor)#获取游标,获取表头
        cur.execute(sql)
        if oneTag:
            result = cur.fetchone()
        else:
            result = cur.fetchall()#获取查询结果
        print(result)
        cur.close()
        connect.close()
        return result


@server.route('/api/payment',methods=['get'])#methods:用来指定请求方式，get/post
def payment():
    data = {"code":0,"msg":"支付成功","amount":30000}
    return json.dumps(data,ensure_ascii=False)

@server.route('/api/account_data',methods=['get'])
def account_data():
    account_name = flask.request.values.get('account_name')#获取请求参数，key-value形式的
    #flask.request.json.get()#获取json格式的参数
    print(account_name)
    result = select("select * from account_info where uname = '%s';" %account_name)
    print(result)
    data = {'code':0,'msg':'success','data':result}
    return json.dumps(data,ensure_ascii=False)

@server.route('/api/account_register',methods=['post'])
def register():#注册
    uname = flask.request.values.get('uname')
    password = flask.request.values.get('password')
    cpwd = flask.request.values.get('cpwd')
    select_sql = "select * from account_info where uname = '%s';" % uname
    if uname and password and cpwd:#输入参数均不为空
        if password != cpwd:
            data = {'code': 401, 'msg': '两次输入密码不一致'}
        elif select(select_sql):#如果查询结果不为空，则不能注册
            data = {'code': 402, 'msg': '输入的账号已存在'}
        else:
            md5_pwd = md5(password)
            insert_sql = "insert into account_info(uname,password) values('%s','%s');"%(uname,md5_pwd)
            select(insert_sql)
            data = {'code': 200, 'msg': '注册成功'}

    else:
        data = {'code':400,'msg':'输入参数不能为空'}

    return json.dumps(data,ensure_ascii=False)

#登录成功后要返回token，token=username+timestamp md5加密
@server.route('/api/login',methods=['post'])
def login():
    uname = flask.request.values.get('uname')
    password = flask.request.values.get('password')
    if uname and password:
        select_sql = "select * from account_info where uname = '%s';" % uname
        result = select(select_sql,True)
        print(md5(password))
        if result:
            if md5(password) == result.get('password'):
                token = md5(uname+str(time.time()))#token:uname+timestamp md5加密，保存到redis中，设置有效时间
                info = {'username':uname,'id':result.get('id')}
                op_redis(token,json.dumps(info))
                data = {'code': 200, 'msg': '登录成功','token':token}
            else:
                data = {'code': 401, 'msg': '用户不存在'}
        else:
            data = {'code': 403, 'msg': '输入密码错误'}
    else:
        data = {'code': 402, 'msg': '账号/密码不能为空'}

    return json.dumps(data,ensure_ascii=False)

@server.route('/api/payment_v1',methods=['post'])
def payment_v1():
    token = flask.request.values.get('token')
    amount = flask.request.values.get('amount')
    if token:#如果token不为空，则去redis中查询用户信息
        result = op_redis(token)
        if result:
            amount = float(amount)
            result = json.loads(result)
            userid = result.get('id')
            update_sql = 'update account_info set balance = balance-%s where id =%s ; '%(amount,userid)
            print(update_sql)
            select(update_sql)
            data = {'code': 0, 'msg': '支付成功'}
        else:
            data = {'code': 400, 'msg': '用户未登录'}
    else:
        data = {'code':400,'msg':'用户未登录'}

    return  json.dumps(data,ensure_ascii=False)



server.run(host='0.0.0.0',port=9999,debug=True)#服务启动只启动一次，服务启动后，不会再执行下面的代码，debug：改代码后不需要重启服务
#host='0.0.0.0',这个不写只能在本机访问，如果写上，别人也可以调用你的接口

1.2 fastapi

pip install fastapi

pip install uvicorn

import fastapi
import uvicorn
from pydantic import BaseModel



server = fastapi.FastAPI()#实例化一个接口
class User(BaseModel):
    username:str
    password:str
    phone:str


@server.post('/user')
def user(u:User):#入参是json类型的，要先定义一个类，继承BaseModel
    print(User)
    return {'code':0}

@server.get('/index')
def index(id:int,sex:str):
    if sex=='nan':
        data = {'id':id,'sex':'男'}
    else:
        data = {'id': id, 'sex': '女'}


@server.post('/order')#@:装饰器
def order(type:str):
    if type == 'success':
        data = {'id':1,'order_amount':10000}
    else:
        data = {'id':0,'order_amount':0}

@server.post('/login')
def login(username:str,password:str):
    if username and password:
        data = {'code':0,'msg':'登录成功'}
    else:
        data = {'code': 0, 'msg': '登录失败'}



if __name__ == '__main__':
    uvicorn.run('fastapi接口开发:server',port=9000,debug=True)#第一个参数是python文件名

2.加密

对称加密：加密和解密用的同样的秘钥

不对称加密：加密和解密用的不同的秘钥

md5加密：md5加密是不可逆转的

#md5加密是不可逆转的
import hashlib
s = '123456'
m = hashlib.md5(s.encode())#md5 传入的值是bytes
#不管字符串多长，加密后都是32位的值
#m = hashlib.sha224(s.encode())

result = m.hexdigest()#加密
print(result)

3.token产生和校验

在登录成功后，都会返回一个token

token是一个唯一的，随机的，且有时效性的值

下面生成token的规则一般是：用户名+timestamp 再md5加密

token保存在redis中，以便于后面接口校验token

3.1.登录接口--生成token

#登录成功后要返回token，token=username+timestamp md5加密
@server.route('/api/login',methods=['post'])
def login():
    uname = flask.request.values.get('uname')
    password = flask.request.values.get('password')
    if uname and password:
        select_sql = "select * from account_info where uname = '%s';" % uname
        result = select(select_sql,True)
        print(md5(password))
        if result:
            if md5(password) == result.get('password'):
                token = md5(uname+str(time.time()))#token:uname+timestamp md5加密，保存到redis中，设置有效时间
                info = {'username':uname,'id':result.get('id')}
                op_redis(token,json.dumps(info))
                data = {'code': 200, 'msg': '登录成功','token':token}
            else:
                data = {'code': 401, 'msg': '用户不存在'}
        else:
            data = {'code': 403, 'msg': '输入密码错误'}
    else:
        data = {'code': 402, 'msg': '账号/密码不能为空'}

    return json.dumps(data,ensure_ascii=False)

3.2.支付接口--校验token

@server.route('/api/payment_v1',methods=['post'])
def payment_v1():
    token = flask.request.values.get('token')
    amount = flask.request.values.get('amount')
    if token:#如果token不为空，则去redis中查询用户信息
        result = op_redis(token)
        if result:
            amount = float(amount)
            result = json.loads(result)
            userid = result.get('id')
            update_sql = 'update account_info set balance = balance-%s where id =%s ; '%(amount,userid)
            print(update_sql)
            select(update_sql)
            data = {'code': 0, 'msg': '支付成功'}
        else:
            data = {'code': 400, 'msg': '用户未登录'}
    else:
        data = {'code':400,'msg':'用户未登录'}

    return  json.dumps(data,ensure_ascii=False)