一、为什么后端选择flask框架？
1.因为之前工作中flask接触的不多，这次选择flask作为后端框架也是一个学习的机会。
2.flask框架相比Django比较轻量级，相对比较灵活，符合我开发的要求。
二、项目目录设计
以上是我的项目目录，接下来介绍每个目录的作用。
basic：主要存放一些项目基础或通用功能的蓝图及功能实现文件
conf：存放项目的配置文件
models：存放SQLAlchemy的model文件
permission：存放权限管理模块的蓝图及功能实现文件
static:存放静态文件
utils:存放通用的方法以供项目调用
app.py:项目启动文件
db.py:数据库初始化文件
requirements.text:
autopep8==1.5 certifi==2019.11.28 chardet==3.0.4 Click==7.0 docopt==0.6.2 Flask==1.1.1 Flask-Cors==3.0.8 Flask-JWT-Extended==3.24.1 flask-redis==0.4.0 Flask-SQLAlchemy==2.4.1 idna==2.9 itsdangerous==1.1.0 Jinja2==2.11.1 MarkupSafe==1.1.1 mysqlclient==1.4.6 pipreqs==0.4.10 pycodestyle==2.5.0 PyJWT==1.7.1 pywin32==227 PyYAML==5.3 redis==3.4.1 requests==2.23.0 six==1.14.0 SQLAlchemy==1.3.13 urllib3==1.25.8 Werkzeug==1.0.0 WMI==1.4.9 yarg==0.1.9
三、配置文件
在conf文件夹下新建conf.py，该文件是项目的配置文件，配置数据库连接等信息。
# !/usr/bin/python3 # -*- coding: utf-8 -*- """ @Author : Huguodong @Version : ------------------------------------ @File : config.py @Description : @CreateTime : 2020/3/7 14:36 ------------------------------------ @ModifyTime : """ # 日志 LOG_LEVEL = "DEBUG" LOG_DIR_NAME = "logs" # mysql MYSQL = {"HOST": "192.168.68.133", 'PORT': "3306", 'USER': "root", 'PASSWD': "root", 'DB': "devops"} REDIS = { 'HOST': '192.168.68.133', 'PORT': 6379, 'PASSWD': '', 'DB': 0, "EXPIRE": 60000 } # token SECRET_KEY = "jinwandalaohu" EXPIRES_IN = 9999 # 上传文件 UPLOAD_HEAD_FOLDER = "static/uploads/avatar" app_url = "http://localhost:5000"
四、日志封装
日志是每个项目必不可少的，这里对logger做个简单的封装。
1.utils文件夹下新建conf_log.py文件
# !/usr/bin/python3 # -*- coding: utf-8 -*- """ @Author : Huguodong @Version : ------------------------------------ @File : log_conf.py @Description : log配置，实现日志自动按日期生成日志文件 @CreateTime : 2020/3/7 19:17 ------------------------------------ @ModifyTime : """ import os import time import logging from conf import config def make_dir(make_dir_path): """ 文件生成 :param make_dir_path: :return: """ path = make_dir_path.strip() if not os.path.exists(path): os.makedirs(path) return path log_dir_name = config.LOG_DIR_NAME # 日志文件夹 log_file_name = 'logger-' + time.strftime('%Y-%m-%d', time.localtime(time.time())) + '.log' # 文件名 log_file_folder = os.path.abspath( os.path.join(os.path.dirname(__file__), os.pardir, os.pardir)) + os.sep + log_dir_name make_dir(log_file_folder) log_file_str = log_file_folder + os.sep + log_file_name # 输出格式 log_level = config.LOG_LEVEL # 日志等级 handler = logging.FileHandler(log_file_str, encoding='UTF-8') handler.setLevel(log_level) logging_format = logging.Formatter( '%(asctime)s - %(levelname)s - %(filename)s - %(funcName)s - %(lineno)s - %(message)s') handler.setFormatter(logging_format)
2.app.py初始化
# 加载日志 app.logger.addHandler(handler)
3.想要使用日志只要引入current_app 就行了
from flask import current_app as app app.logger.error("xxxxx")
五、flask-sqlalchemy封装
1.新建db.py
# !/usr/bin/python3 # -*- coding: utf-8 -*- from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy()
2.models文件夹下新建BaseModel.py,这样后面所有的model就可以继承这个基类，不用每个model再写单独的新增修改删除方法。
# !/usr/bin/python3 # -*- coding: utf-8 -*- """ @Author : Huguodong @Version : ------------------------------------ @File : BaseModel.py @Description : ORM封装 @CreateTime : 2020/3/8 15:13 ------------------------------------ @ModifyTime : """ from sqlalchemy import func from db import db class BaseModel(db.Model): __abstract__ = True ## 声明当前类为抽象类，被继承，调用不会被创建 id = db.Column(db.Integer, primary_key=True, autoincrement=True) create_by = db.Column(db.String(64), comment="创建者") created_at = db.Column(db.TIMESTAMP(True), comment="创建时间", nullable=False, server_default=func.now()) update_by = db.Column(db.String(64), comment="更新者") updated_at = db.Column(db.TIMESTAMP(True), comment="更新时间", nullable=False, server_default=func.now(), onupdate=func.now()) remark = db.Column(db.String(500), comment="备注") def save(self): ''' 新增数据 :return: ''' db.session.add(self) db.session.commit() def update(self): ''' 更新数据 :return: ''' db.session.merge(self) db.session.commit() def delete(self): ''' 删除数据 :return: ''' db.session.delete(self) db.session.commit() def save_all(self, data): ''' 保存多条数据 :param data: :return: ''' db.session.execute( self.__table__.insert(), data ) db.session.commit()
3.初始化数据库
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://{}:{}@{}:{}/{}'.format(config.MYSQL['USER'], config.MYSQL['PASSWD'], config.MYSQL['HOST'], config.MYSQL['PORT'], config.MYSQL['DB']) app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # 跟踪对象的修改，在本例中用不到调高运行效率，所以设置为False # app.config['SQLALCHEMY_ECHO'] = True db.init_app(app)
五、reids封装
1.在utils文件夹下新建redis_utils.py
# !/usr/bin/python3 # -*- coding: utf-8 -*- """ @Author : Huguodong @Version : ------------------------------------ @File : redis_utils.py @Description : 封装redis类 @CreateTime : 2020/3/23 22:04 ------------------------------------ @ModifyTime : """ import pickle import redis from flask import current_app as app class Redis(object): """ redis数据库操作 """ @staticmethod def _get_r(): host = app.config['REDIS_HOST'] port = app.config['REDIS_PORT'] db = app.config['REDIS_DB'] passwd = app.config['REDIS_PWD'] r = redis.StrictRedis(host=host, port=port, db=db, password=passwd) return r @classmethod def write(self, key, value, expire=None): """ 写入键值对 """ # 判断是否有过期时间，没有就设置默认值 if expire: expire_in_seconds = expire else: expire_in_seconds = app.config['REDIS_EXPIRE'] r = self._get_r() r.set(key, value, ex=expire_in_seconds) @classmethod def write_dict(self, key, value, expire=None): ''' 将内存数据二进制通过序列号转为文本流，再存入redis ''' if expire: expire_in_seconds = expire else: expire_in_seconds = app.config['REDIS_EXPIRE'] r = self._get_r() r.set(pickle.dumps(key), pickle.dumps(value), ex=expire_in_seconds) @classmethod def read_dict(self, key): ''' 将文本流从redis中读取并反序列化，返回 ''' r = self._get_r() data = r.get(pickle.dumps(key)) if data is None: return None return pickle.loads(data) @classmethod def read(self, key): """ 读取键值对内容 """ r = self._get_r() value = r.get(key) return value.decode('utf-8') if value else value @classmethod def hset(self, name, key, value): """ 写入hash表 """ r = self._get_r() r.hset(name, key, value) @classmethod def hmset(self, key, *value): """ 读取指定hash表的所有给定字段的值 """ r = self._get_r() value = r.hmset(key, *value) return value @classmethod def hget(self, name, key): """ 读取指定hash表的键值 """ r = self._get_r() value = r.hget(name, key) return value.decode('utf-8') if value else value @classmethod def hgetall(self, name): """ 获取指定hash表所有的值 """ r = self._get_r() return r.hgetall(name) @classmethod def delete(self, *names): """ 删除一个或者多个 """ r = self._get_r() r.delete(*names) @classmethod def hdel(self, name, key): """ 删除指定hash表的键值 """ r = self._get_r() r.hdel(name, key) @classmethod def expire(self, name, expire=None): """ 设置过期时间 """ if expire: expire_in_seconds = expire else: expire_in_seconds = app.config['REDIS_EXPIRE'] r = self._get_r() r.expire(name, expire_in_seconds)
2.app.py初始化
app.config['REDIS_HOST'] = config.REDIS['HOST'] app.config['REDIS_PORT'] = config.REDIS['PORT'] app.config['REDIS_DB'] = config.REDIS['DB'] app.config['REDIS_PWD'] = config.REDIS['PASSWD'] app.config['REDIS_EXPIRE'] = config.REDIS['EXPIRE']
3.调用
from utils.redis_utils import Redis #写 Redis.write(f"token_{user.user_name}", token) #读 redis_token = Redis.read(key)
五、枚举类
当我们前端请求之后，后端接受前端请求并返回状态码，那么这些状态码可以用一个枚举类保存起来同意管理。
在utils文件夹下新建code_enum.py
# !/usr/bin/python3 # -*- coding: utf-8 -*- """ @Author : Huguodong @Version : ------------------------------------ @File : code_enum.py @Description : 返回码枚举类 @CreateTime : 2020/3/7 19:48 ------------------------------------ @ModifyTime : """ import enum class Code(enum.Enum): # 成功 SUCCESS = 0 # 获取信息失败 REQUEST_ERROR = 400 # 504错误 NOT_FOUND = 404 # 500错误 INTERNAL_ERROR = 500 # 登录超时 LOGIN_TIMEOUT = 50014 # 无效token ERROR_TOKEN = 50008 # 别的客户端登录 OTHER_LOGIN = 50012 # 权限不够 ERR_PERMISSOM = 50013 # 更新数据库失败 UPDATE_DB_ERROR = 1000 # 更新数据库失败 CREATE_DB_ERROR = 1001 # 更新数据库失败 DELETE_DB_ERROR = 1002 # 不能为空 NOT_NULL = 1003 # 缺少参数 NO_PARAMETER = 1004 # 用户密码错误 ERR_PWD = 1005 # 数据不存在 ID_NOT_FOUND = 1006 # 参数错误 PARAMETER_ERROR = 1007 # 文件不存在 FILE_NO_FOUND = 1008 # 无效的格式 ERROR_FILE_TYPE = 1009 # 超出文件限制 OVER_SIZE = 1010 # 上传失败 UPLOAD_FAILD = 1011
五、公共方法
在utils文件夹里面新建common.py，可以供全局调用的方法一般放在这里面，比如返回给前端的方法。
# !/usr/bin/python3 # -*- coding: utf-8 -*- """ @Author : Huguodong @Version : ------------------------------------ @File : common.py @Description : @CreateTime : 2020/3/7 19:01 ------------------------------------ @ModifyTime : """ # 导入依赖包 import functools import hashlib from flask import request, jsonify, current_app as app from itsdangerous import TimedJSONWebSignatureSerializer as Serializer from conf import config from utils.code_enum import Code def create_token(user_id, user_name, role_list): ''' 生成token :param api_user:用户id :return: token ''' # 第一个参数是内部的私钥，这里写在共用的配置信息里了，如果只是测试可以写死 # 第二个参数是有效期(秒) s = Serializer(config.SECRET_KEY, expires_in=config.EXPIRES_IN) # 接收用户id转换与编码 token = None try: token = s.dumps({"id": user_id, "name": user_name, "role": role_list}).decode("ascii") except Exception as e: app.logger.error("获取token失败:{}".format(e)) return token def verify_token(token): ''' 校验token :param token: :return: 用户信息 or None ''' # 参数为私有秘钥，跟上面方法的秘钥保持一致 s = Serializer(config.SECRET_KEY) try: # 转换为字典 data = s.loads(token) return data except Exception as e: app.logger.error(f"token转换失败:{e}") return None def login_required(*role): def decorator(func): @functools.wraps(func) def wrapper(*args, **kw): try: # 在请求头上拿到token token = request.headers["Authorization"] except Exception as e: # 没接收的到token,给前端抛出错误 return jsonify(code=Code.NO_PARAMETER.value, msg='缺少参数token') s = Serializer(config.SECRET_KEY) try: user = s.loads(token) if role: # 获取token中的权限列表如果在参数列表中则表示有权限，否则就表示没有权限 user_role = user['role'] result = [x for x in user_role if x in list(role)] if not result: return jsonify(code=Code.ERR_PERMISSOM.value, msg="权限不够") except Exception as e: return jsonify(code=Code.LOGIN_TIMEOUT.value, msg="登录已过期") return func(*args, **kw) return wrapper return decorator def model_to_dict(result): ''' 查询结果转换为字典 :param result: :return: ''' from collections import Iterable # 转换完成后，删除 '_sa_instance_state' 特殊属性 try: if isinstance(result, Iterable): tmp = [dict(zip(res.__dict__.keys(), res.__dict__.values())) for res in result] for t in tmp: t.pop('_sa_instance_state') else: tmp = dict(zip(result.__dict__.keys(), result.__dict__.values())) tmp.pop('_sa_instance_state') return tmp except BaseException as e: print(e.args) raise TypeError('Type error of parameter') def construct_page_data(data): ''' 分页需要返回的数据 :param data: :return: ''' page = {"page_no": data.page, # 当前页数 "page_size": data.per_page, # 每页显示的属性 "tatal_page": data.pages, # 总共的页数 "tatal_count": data.total, # 查询返回的记录总数 "is_first_page": True if data.page == 1 else False, # 是否第一页 "is_last_page": False if data.has_next else True # 是否最后一页 } # result = menu_to_dict(data.items) result = model_to_dict(data.items) data = {"page": page, "list": result} return data def construct_menu_data(data): ''' 菜单分页需要返回的数据 :param data: :return: ''' page = {"page_no": data.page, # 当前页数 "page_size": data.per_page, # 每页显示的属性 "tatal_page": data.pages, # 总共的页数 "tatal_count": data.total, # 查询返回的记录总数 "is_first_page": True if data.page == 1 else False, # 是否第一页 "is_last_page": False if data.has_next else True # 是否最后一页 } # result = menu_to_dict(data.items) result = menu_to_dict(data.items) data = {"page": page, "list": result} return data def SUCCESS(data=None): return jsonify(code=Code.SUCCESS.value, msg="ok", data=data) def NO_PARAMETER(msg="未接收到参数!"): return jsonify(code=Code.NO_PARAMETER.value, msg=msg) def PARAMETER_ERR(msg="参数错误!"): return jsonify(code=Code.NO_PARAMETER.value, msg=msg) def OTHER_LOGIN(msg="其他客户端登录!"): return jsonify(code=Code.OTHER_LOGIN.value, msg=msg) def AUTH_ERR(msg="身份验证失败!"): return jsonify(code=Code.ERROR_TOKEN.value, msg=msg) def TOKEN_ERROR(msg="Token校验失败!"): return jsonify(code=Code.ERROR_TOKEN.value, msg=msg) def REQUEST_ERROR(msg="请求失败!"): return jsonify(code=Code.REQUEST_ERROR.value, msg=msg) def ID_NOT_FOUND(msg="数据不存在!"): return jsonify(code=Code.ID_NOT_FOUND.value, msg=msg) def CREATE_ERROR(msg="创建失败!"): return jsonify(code=Code.CREATE_DB_ERROR.value, msg=msg) def UPDATE_ERROR(msg="更新失败!"): return jsonify(code=Code.UPDATE_DB_ERROR.value, msg=msg) def DELETE_ERROR(msg="删除失败"): return jsonify(code=Code.DELETE_DB_ERROR.value, msg=msg) def FILE_NO_FOUND(msg="请选择文件!"): return jsonify(code=Code.FILE_NO_FOUND.value, msg=msg) def ERROR_FILE_TYPE(msg="无效的格式!"): return jsonify(code=Code.ERROR_FILE_TYPE.value, msg=msg) def UPLOAD_FAILD(msg="上传失败!"): return jsonify(code=Code.UPLOAD_FAILD.value, msg=msg) def OVER_SIZE(msg="文件大小超出限制!"): return jsonify(code=Code.OVER_SIZE.value, msg=msg) def get_diff(old_list, new_list): # 计算old_list比new_list多的 less_list = list(set(old_list) - set(new_list)) # 计算new_list比old_list多的 add_list = list(set(new_list) - set(old_list)) return add_list, less_list def create_passwd(passwd): # 创建md5对象 m = hashlib.md5() b = passwd.encode(encoding='utf-8') m.update(b) str_md5 = m.hexdigest() return str_md5 def md5_sum(strs): m = hashlib.md5() b = strs.encode(encoding='utf-8') m.update(b) str_md5 = m.hexdigest() return str_md5
五、app.py
最后的app.py
from flask import Flask, jsonify from flask_cors import CORS from db import db from conf import config from permission import dict_data, menu, user, dict, post, dept, role, configs from basic import upload from utils.code_enum import Code from utils.conf_log import handler def create_app(): app = Flask(__name__) # 设置返回jsonify方法返回dict不排序 app.config['JSON_SORT_KEYS'] = False # 设置返回jsonify方法返回中文不转为Unicode格式 app.config['JSON_AS_ASCII'] = False # 配置跨域 CORS(app, resources={r"/api/*": {"origins": "*"}}) # 注册蓝图 register_blueprints(app) # 加载数据库 init_db(app) # 加载redis配置 init_redis(app) # 加载日志 app.logger.addHandler(handler) return app def register_blueprints(app): ''' 创建蓝图 :param app: :return: ''' app.register_blueprint(user.user, url_prefix='/api/user') app.register_blueprint(menu.menu, url_prefix='/api/menu') app.register_blueprint(dict.dict, url_prefix='/api/dict') app.register_blueprint(dict_data.dictData, url_prefix='/api/dictData') app.register_blueprint(post.post, url_prefix='/api/post') app.register_blueprint(dept.dept, url_prefix='/api/dept') app.register_blueprint(role.role, url_prefix='/api/role') app.register_blueprint(configs.configs, url_prefix='/api/configs') app.register_blueprint(upload.upload, url_prefix='/api/upload') def init_db(app): ''' 加载数据库 :param app: :return: ''' app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://{}:{}@{}:{}/{}'.format(config.MYSQL['USER'], config.MYSQL['PASSWD'], config.MYSQL['HOST'], config.MYSQL['PORT'], config.MYSQL['DB']) app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # 跟踪对象的修改，在本例中用不到调高运行效率，所以设置为False # app.config['SQLALCHEMY_ECHO'] = True db.init_app(app) def init_redis(app): ''' 加载redis :param app: :return: ''' app.config['REDIS_HOST'] = config.REDIS['HOST'] app.config['REDIS_PORT'] = config.REDIS['PORT'] app.config['REDIS_DB'] = config.REDIS['DB'] app.config['REDIS_PWD'] = config.REDIS['PASSWD'] app.config['REDIS_EXPIRE'] = config.REDIS['EXPIRE'] app = create_app() @app.errorhandler(Exception) def handle_error(err): """自定义处理错误方法""" # 这个函数的返回值会是前端用户看到的最终结果 try: if err.code == 404: app.logger.error(err) return jsonify(code=Code.NOT_FOUND.value, msg="服务器异常,404") elif err.code == 400: app.logger.error(err) return jsonify(code=Code.REQUEST_ERROR.value, msg="服务器异常,400") elif err.code == 500: app.logger.error(err) return jsonify(code=Code.INTERNAL_ERROR.value, msg="服务器异常,500") else: return jsonify(code=err.code, msg=f"服务器异常,{err.code}") except: return jsonify(code=Code.INTERNAL_ERROR.value, msg=f"服务器异常,{err}") if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)