一、為什么后端選擇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)