Flask 项目构建

exts.py

# flask_sqlalchemy
from flask_sqlalchemy import SQLAlchemy
from flask_mail import Mail
from flask_caching import Cache
from flask_wtf import CSRFProtect
from flask_avatars import Avatars
from flask_jwt_extended import JWTManager
from flask_cors import CORSdb = SQLAlchemy()
mail = Mail()
cache = Cache()
csrf = CSRFProtect()
avatars = Avatars()
jwt = JWTManager()
cors = CORS()from gevent import pywsgifrom geventwebsocket.handler import WebSocketHandlerserver = pywsgi.WSGIServer(('0.0.0.0', 8099), app, handler_class=WebSocketHandler)server.serve_forever()

config.py

import os#===============================================================================
# 基础路由配置
#===============================================================================
BASE_DIR = os.path.dirname(os.path.abspath(__file__))#===============================================================================
# 随机密钥
#===============================================================================
SECRET_KEY = "dfasdfsdflasdjfl"#===============================================================================
# msyql 数据库配置
#===============================================================================
from utils.configUtils import get_section_dictcfg_msyql = get_section_dict(file_path=os.path.join(BASE_DIR,"config.ini"),section="mysql"
)DB_URI = 'mysql+pymysql://%s:%s@%s:%s/%s?charset=utf8mb4' % (cfg_msyql.get("user","root"),cfg_msyql.get("password","123456"),cfg_msyql.get("host","localhost"),cfg_msyql.get("port","3306"),cfg_msyql.get("database","employee_face_recognition_and_behavior_detection"),)SQLALCHEMY_DATABASE_URI = DB_URI#===============================================================================
# redis 数据库配置
#===============================================================================
import redis
cfg_redis = get_section_dict(file_path=os.path.join(BASE_DIR,"config.ini"),section="redis"
)redis_con = redis.Redis(host=cfg_redis.get("host","localhost"), port=cfg_redis.get("port",6379), db=cfg_redis.get("db",1), decode_responses=cfg_redis.get("decode_responses",False)
)cfg_flask = get_section_dict(file_path=os.path.join(BASE_DIR,"config.ini"),section="flask"
)

apps包中的__init__.py

from flask import Flask
from flask_migrate import Migrate
import config
import pymysql
from exts import db,login_manager,cors
import redis# ===========================================================================
# 这里需要将自定义模型导入到这里
# from apps.models.employee import Employee,Attendance
# from apps.models.user import Userdef create_app(config_class=config):app = Flask(__name__)app.config.from_object(config_class)# 初始化扩展db.init_app(app)migrate = Migrate(app, db)cors.init_app(app=app,resources={r"/api/*": {"origins": "*"}},supports_credentials=True)login_manager.init_app(app)login_manager.login_view = 'api.login'# @login_manager.user_loader# def load_user(user_id: str):#     return User.query.get(int(user_id))# 蓝图注册# from app.routes.main import main_bp# from app.routes.api import api_bp# app.register_blueprint(main_bp)# app.register_blueprint(api_bp,url_prefix='/api')# 创建数据库表 - 在设置了URI后才创建with app.app_context():db.create_all()pymysql.install_as_MySQLdb()return app 

main.py

from apps import create_appapp = create_app()if __name__ == '__main__':app.run(debug=True)

config.ini

[mysql]
# 连接数据库基本参数
host        = localhost
port        = 3306
user        = root
password    = 123456
database    = databaseName
charset     = utf8mb4# 可选：连接池／超时配置
connect_timeout = 10
pool_size       = 5[redis]
host             = localhost
port             = 6379
db               = 1
decode_responses = False[flask]
APP_NAME = 'flask_app'

utils.configUtils.py

import configparser
from typing import Dict
from typing import Dict, Any
import configparser
import re,random
from app.models.employee import BehaviorTypeEnum,Behavior
from datetime import datetime,timedelta
import pytzdef get_section_dict(file_path: str,section: str,encoding: str = 'utf-8'
) -> Dict[str, Any]:"""从指定的 ini 文件中读取一个区块，将该区块的 key/value 封装为字典返回，并自动将字符串形式的布尔值、整数和浮点数转换为相应的 Python 类型。:param file_path: ini 文件路径:param section: 要读取的区块名称:param encoding: 文件编码，默认为 'utf-8':return: 区块键值对字典，value 类型可能为 str, int, float, 或 bool:raises KeyError: 如果指定区块在文件中不存在"""config = configparser.ConfigParser()config.optionxform = str  # 保持 key 的大小写config.read(file_path, encoding=encoding)if section not in config:raise KeyError(f"Section '{section}' not found in '{file_path}'.")raw = config[section]parsed: Dict[str, Any] = {}for key, value in raw.items():v = value.strip()low = v.lower()# 布尔类型if low in ('true', 'yes', 'on'):parsed_val: Any = Trueelif low in ('false', 'no', 'off'):parsed_val = False# 整数elif re.fullmatch(r'-?\d+', v):parsed_val = int(v)# 浮点数elif re.fullmatch(r'-?\d+\.\d+', v):parsed_val = float(v)# 其余保持字符串else:parsed_val = vparsed[key] = parsed_valreturn parsed

utils.restful.py

# Restful API
from flask import jsonifyclass HttpCode(object):# 响应正常ok = 200# 没有登陆错误unloginerror = 401# 没有权限错误permissionerror = 403# 客户端参数错误paramserror = 400# 服务器错误servererror = 500def _restful_result(code, message, data):return jsonify({"message": message or "", "data": data or {}, "code": code})def ok(message=None, data=None):return _restful_result(code=HttpCode.ok, message=message, data=data)def unlogin_error(message="没有登录！"):return _restful_result(code=HttpCode.unloginerror, message=message, data=None)def permission_error(message="没有权限访问！"):return _restful_result(code=HttpCode.paramserror, message=message, data=None)def params_error(message="参数错误！"):return _restful_result(code=HttpCode.paramserror, message=message, data=None)def server_error(message="服务器开小差啦！"):return _restful_result(code=HttpCode.servererror, message=message or '服务器内部错误', data=None)