python flask 项目实践_pyhton flask 的 实践

架构：python+ flask +marshmallow+itsdangerous+pymysql

数据库：mysql

实现功能：登录、验证码、鉴权、用户、上传、下载、错误统一处理

api格式: restful + json

项目目录

web->controller ->uploads.py

import datetime
import os
import time
 
from flask import Blueprint, jsonify, request
 
from web import db
from web.setting import UPLOAD_URL
from web.token_auth import login_required
 
from web.model.uploads import Upload
 
# 创建蓝图
uploads = Blueprint('upload', __name__)
 
 
# 单文件上传
@uploads.route('/file', methods=["POST"])
@login_required()
def file():
    if request.files.to_dict() == {}:
        return jsonify(data={}, msg='file为空', code=500)
 
    year_time = datetime.datetime.now().strftime('%Y')
    month_time = datetime.datetime.now().strftime('%m')
    day_time = datetime.datetime.now().strftime('%d')
 
    pwd = UPLOAD_URL+'\\'+year_time+'\\'+month_time+'\\'+day_time
    save_pwd = 'static\\uploads' + '\\' + year_time + '\\' + month_time + '\\' + day_time
    word_name = os.path.exists(pwd)
 
    f = request.files['file']
    az = os.path.splitext(f.filename)[1]
    if not word_name:
        os.makedirs(pwd)
 
    filename = int(round(time.time() * 1000))
    upload_path = os.path.join(pwd, str(filename)+az)
    save_upload_path = os.path.join(save_pwd, str(filename) + az)
    f.save(upload_path)
 
    data = Upload(name=str(filename)+az, path=save_upload_path)
    db.session.add(data)
    db.session.commit()
 
    return jsonify(data={}, msg='上传成功', code=200)
 
 
# 多文件上传
@uploads.route('/files', methods=["POST"])
@login_required()
def files():
    if len(request.files.getlist('files')) == 0:
        return jsonify(data={}, msg='files为空', code=500)
 
    year_time = datetime.datetime.now().strftime('%Y')
    month_time = datetime.datetime.now().strftime('%m')
    day_time = datetime.datetime.now().strftime('%d')
 
    pwd = UPLOAD_URL + '\\' + year_time + '\\' + month_time + '\\' + day_time
    save_pwd = 'static\\uploads' + '\\' + year_time + '\\' + month_time + '\\' + day_time
 
    for f in request.files.getlist('files'):
        az = os.path.splitext(f.filename)[1]
        word_name = os.path.exists(pwd)
        if not word_name:
            os.makedirs(pwd)
        filename = str(round(time.time() * 1000))
        upload_path = os.path.join(pwd, filename+az)
        save_upload_path = os.path.join(save_pwd, str(filename) + az)
        f.save(upload_path)
 
        data = Upload(name=str(filename) + az, path=save_upload_path)
        db.session.add(data)
 
    db.session.commit()
 
    return jsonify(data={}, msg='上传成功', code=200)

web->controller ->user.py

import hashlib
 
from flask import jsonify, request, Blueprint, session
from marshmallow import ValidationError
 
from web.model.user import User
from web import db
from web.token_auth import create_token, login_required, verify_token
from web.validator.user import UserSchema, RegisterSchema, LoginSchema
 
from web.token_auth import check_token as check_token_auth
 
# 创建蓝图
user = Blueprint('user', __name__)
 
 
# 用户列表
@user.route('/index')
@login_required()
def index():
    total_page = 1
    if request.json is None:
        items = User.query.all()
    else:
        page = request.json.get('page')
        page_size = request.json.get('pageSize')
 
        if page_size is None:
            page_size = 10
 
        if page is None:
            items = User.query.all()
        else:
            data = User.query.paginate(page=page, per_page=page_size)
            items = data.items
            total_page = data.pages
 
    datas = []
    for user in items:
        datas.append({'id': user.id, 'username': user.username, 'password': user.password, 'roleid': user.roleid})
    return jsonify(data={'data': datas, 'total_page': total_page}, msg='成功', code=200)
 
 
# 添加用户
@user.route('/add', methods=["POST"])
@login_required()
def add():
    # 校验字段
    try:
        result = UserSchema().load(request.json)
    except ValidationError as e:
        return jsonify(data=e.messages, msg='', code=500)
 
    username = request.json.get('username')
    password = request.json.get('password')
    roleid = request.json.get('roleid')
 
    # 判断用户名是否重复
    filterData = User.query.filter(User.username == username).all()
    if len(filterData) >= 1:
        return jsonify(data={}, msg='用户名已存在', code=500)
 
    data = User(username=username, password=hashlib.md5(password.encode()).hexdigest(), roleid=roleid)
    db.session.add(data)
    db.session.commit()
    return jsonify(data={}, msg='成功', code=200)
 
 
# 修改用户
@user.route('/edit/<id>', methods=["PUT"])
@login_required()
def edit(id):
    # 校验字段
    try:
        result = UserSchema().load(request.json)
    except ValidationError as e:
        return jsonify(data=e.messages, msg='', code=500)
 
    username = request.json.get('username')
    password = request.json.get('password')
    roleid = request.json.get('roleid')
 
    # 判断是否有此用户
    filterData = User.query.filter(User.id == id).all()
    if len(filterData) == 0:
        return jsonify(data={}, msg='没有此用户', code=500)
 
    oneData = User.query.filter(User.id == id).first()
    oneData.username = username
    oneData.password = hashlib.md5(password.encode()).hexdigest()
    oneData.roleid = roleid
    db.session.commit()
 
    return jsonify(data={}, msg='成功', code=200)
 
 
# 删除用户
@user.route('/delete/<id>', methods=["DELETE"])
@login_required()
def delete(id):
    # 判断是否有此用户
    filterData = User.query.filter(User.id =