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)