python学习笔记（十五）-- flask接口开发

 flask接口开发

import flask,json

#启动服务，__name__代表当前这个python文件
server = flask.Flask(__name__)

#加上装饰器，变成接口
#访问路径 ip:port/login
@server.route('/login')
def welcome():
    data = {'code':0,'msg':'登陆成功','session_id':'ssfwfwrsd2'}
    return json.dumps(data,ensure_ascii=False)#把字典转化成json串

#get请求，url传参
# http://127.0.0.1:9999/urldata?name=nana&passwd=123456
@server.route('/urldata')
def urlData():
    name = flask.request.args.get('name')#传过来的参数
    passwd = flask.request.args.get('passwd')
    data = {'name':name,'passwd':passwd}
    return json.dumps(data,ensure_ascii=False)

#post请求，不加methods默认get请求
@server.route('/post',methods=['post'])
def post():
    #url传参都用这种方式
    name = flask.request.args.get('name')#传过来的参数
    passwd = flask.request.args.get('passwd')

    #k--v,form-data里面的参数
    stu_name = flask.request.values.get('stu_name')
    addr = flask.request.values.get('addr')

    data = {'name':name,'passwd':passwd,'stu_name':stu_name,'addr':addr}
    return json.dumps(data,ensure_ascii=False)

#json参数
@server.route('/add_stu2',methods=['post'])
def add_stu2():
    if flask.request.is_json:
        name = flask.request.json.get('name')
        age = flask.request.json.get('age')
        sex = flask.request.json.get('sex')
        addr = flask.request.json.get('addr')
        phone = flask.request.json.get('phone')
        stu_class = flask.request.json.get('class')
        print(flask.request.json)
        return json.dumps([{'msg':'添加成功'}],ensure_ascii=False)
    else:
        return json.dumps({'msg':'入参请传入json'},ensure_ascii=False)

#上传文件
@server.route('/file',methods=['post'])
def uploadFile():
    file = flask.request.files.get('f')
    print(file.filename)#传入的文件名字
    file.save(file.filename)#保存文件
    # print(dir(file))
    return json.dumps({'msg':'上传成功'},ensure_ascii=False)

#ip 0.0.0.0,同局域网内可以访问
# server.run(debug=True) #debug=True 调试时更改程序后不用重启服务，更改内容就可以生效
server.run(host='0.0.0.0',port=8888,debug=True)#启动服务，默认端口5000

实例

import flask
import pymysql,hashlib,json
"""
1、查询表接口，入参为table_data、limit(可选)，接口类型为get请求
2、注册接口，入参为username,password，接口类型为post请求
"""

def md5(s,salt='$!@#$12232'):
    s = (str(s)+salt).encode()
    m = hashlib.md5(s)#加密
    return m.hexdigest()

#操作数据库函数
#传入一个sql语句，返回结果
def op_mysql(sql:str):
    mysql_info = {
        'host': '127.0.0.1',
        'port': 3306,
        'password': '123456',
        'user': 'xxx',
        'db': 'xxx',
        'charset': 'utf8',
        'autocommit': True
    }
    result = '执行完成'
    conn = pymysql.connect(**mysql_info)
    cur = conn.cursor(pymysql.cursors.DictCursor) #建立游标
    cur.execute(sql)
    if sql.strip().lower().startswith('select'):
        # result  = cur.fetchone()
        result  = cur.fetchall()
    cur.close()
    conn.close()
    return result

server = flask.Flask(__name__)

#查询表内容，入参table_name,limit
@server.route('/table_data')
def table_data():
    tables = ['app_myuser', 'dsk_test', 'app_student', 'app_product']
    table_name = flask.request.args.get('table_name')
    limit = flask.request.args.get('limit','10')
    if not table_name:
        return json.dumps({'msg': 'table_name为必填字段！'}, ensure_ascii=False)
    if table_name not in tables:
        return json.dumps({'msg':'没有权限！'}, ensure_ascii=False)
    if not limit.isdigit():
        return json.dumps({'msg': 'limit必须为整数！'}, ensure_ascii=False)
    else:
        sql = 'select * from %s limit %s; '%(table_name,limit)
        data = op_mysql(sql)
        return json.dumps(data,ensure_ascii=False)

@server.route('/add_mem',methods=['post'])
def add_mem():
    username = flask.request.values.get('username')
    password = flask.request.values.get('password')
    if username and password:
        sql = 'select * from app_myuser where username="%s";' % username
        if op_mysql(sql):
            return json.dumps({'msg': '用户名重复！'}, ensure_ascii=False)
        else:
            new_passwd = md5(password)
            sql2 = 'insert into app_myuser (username,passwd) value ("%s","%s"); '%(username,new_passwd)
            op_mysql(sql2)
            return json.dumps({'msg': '注册成功！'}, ensure_ascii=False)
    else:
        return json.dumps({'msg':'请传入用户名、密码！'},ensure_ascii=False)

server.run(debug=True)