283 lines
12 KiB
Python
283 lines
12 KiB
Python
# +======================================================================
|
||
# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ]
|
||
# +----------------------------------------------------------------------
|
||
# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有
|
||
# +----------------------------------------------------------------------
|
||
# | 开源协议: Apache-2.0
|
||
# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。
|
||
# +----------------------------------------------------------------------
|
||
# | 官方网站: https://www.djangoadmin.cn
|
||
# +----------------------------------------------------------------------
|
||
# | 研发团队: 小蚂蚁云团队 荣誉出品
|
||
# +======================================================================
|
||
|
||
"""
|
||
数据库配置模块
|
||
=============================================================
|
||
|
||
负责配置数据库连接及相关参数，包含连接池、SSL 加密等，支持多数据库驱动：
|
||
mysql（默认）/ postgresql（pgsql）/ mssql / sqlite / oracle。
|
||
依赖：config/__init__.py 中的 _load_env_file() 已提前加载 .env
|
||
"""
|
||
|
||
import os
|
||
from urllib.parse import quote_plus
|
||
from config.app import FLASK_SECRET_KEY, FLASK_UPLOAD_DIR
|
||
|
||
# ============================================================
|
||
# 数据库环境变量
|
||
# ============================================================
|
||
|
||
|
||
def _normalize_driver(driver: str) -> str:
|
||
"""驱动别名归一化：别名 → 规范名；已是规范名的驱动原样返回。
|
||
|
||
- 规范名：mysql / postgresql / mssql / sqlite / oracle
|
||
- 别名：postgres、pgsql → postgresql；sqlserver → mssql；sqlite3 → sqlite
|
||
- oracle 本身即规范名，无需映射（经 fallback 直接透传），表中显式列出便于自文档化
|
||
与 scripts/migrate_db.py 共用（make_url 也调用），保证 .env / CLI 两端别名一致。
|
||
"""
|
||
return {
|
||
'postgres': 'postgresql',
|
||
'pgsql': 'postgresql',
|
||
'sqlserver': 'mssql',
|
||
'sqlite3': 'sqlite',
|
||
'oracle': 'oracle', # 规范名之一，显式列出便于自文档化
|
||
}.get(str(driver).lower(), str(driver).lower())
|
||
|
||
|
||
def _sqlite_url(database: str) -> str:
|
||
"""SQLite 连接 URL：DB_DATABASE 即文件路径（内存库用 ':memory:'，Windows 反斜杠转正斜杠）。"""
|
||
if database == ':memory:':
|
||
return 'sqlite://'
|
||
return f"sqlite:///{database.replace(chr(92), '/')}"
|
||
|
||
|
||
# 数据库驱动（mysql / postgresql / mssql / sqlite / oracle，支持别名见 _normalize_driver）
|
||
DB_DRIVER = _normalize_driver(os.getenv('DB_DRIVER', 'mysql'))
|
||
# 数据库地址
|
||
DB_HOST = os.getenv('DB_HOST', '127.0.0.1')
|
||
# 各驱动默认端口（未显式配置 DB_PORT 时按驱动回退）
|
||
_DEFAULT_PORT = {
|
||
'mysql': 3306,
|
||
'pgsql': 5432,
|
||
'postgresql': 5432,
|
||
'mssql': 1433,
|
||
'oracle': 1521,
|
||
'sqlite': 0, # sqlite 无端口
|
||
}
|
||
# 数据库端口（MySQL默认3306，PostgreSQL默认5432，MSSQL默认1433，Oracle默认1521）
|
||
DB_PORT = int(os.getenv('DB_PORT', _DEFAULT_PORT.get(DB_DRIVER, 3306)))
|
||
# 数据库名称
|
||
DB_DATABASE = os.getenv('DB_DATABASE', 'djangoadmin.flask.naivevue')
|
||
# 数据库账号
|
||
DB_USERNAME = os.getenv('DB_USERNAME', 'root')
|
||
# 数据库密码
|
||
DB_PASSWORD = os.getenv('DB_PASSWORD', '')
|
||
# 数据表前缀（用于区分同一数据库中不同应用的数据表）
|
||
DB_PREFIX = os.getenv('DB_PREFIX', 'flask_')
|
||
# 是否开启调试模式：True-打印SQL日志，False-不打印
|
||
DB_DEBUG = os.getenv('DB_DEBUG', 'True') == 'True'
|
||
# 是否启用数据库SSL加密连接（生产环境建议开启，防止数据明文传输）
|
||
DB_SSL = os.getenv('DB_SSL', 'False') == 'True'
|
||
# SSL CA证书路径（用于验证服务端证书，留空则信任系统CA）
|
||
DB_SSL_CA = os.getenv('DB_SSL_CA', '')
|
||
# SSL 客户端证书路径（双向认证时使用）
|
||
DB_SSL_CERT = os.getenv('DB_SSL_CERT', '')
|
||
# SSL 客户端私钥路径（双向认证时使用）
|
||
DB_SSL_KEY = os.getenv('DB_SSL_KEY', '')
|
||
|
||
# ============================================================
|
||
# 连接池配置
|
||
# ============================================================
|
||
|
||
# 连接池常驻连接数（默认10，生产环境建议根据并发量调整）
|
||
DB_POOL_SIZE = int(os.getenv('DB_POOL_SIZE', 10))
|
||
# 超出pool_size后允许的最大额外连接（默认10，总连接数=pool_size+max_overflow）
|
||
DB_MAX_OVERFLOW = int(os.getenv('DB_MAX_OVERFLOW', 10))
|
||
# 连接自动回收时间/秒（默认1800，应小于MySQL的wait_timeout，防止"MySQL server has gone away"）
|
||
DB_POOL_RECYCLE = int(os.getenv('DB_POOL_RECYCLE', 1800))
|
||
# 获取连接的超时时间/秒（默认30，超时抛出TimeoutError）
|
||
DB_POOL_TIMEOUT = int(os.getenv('DB_POOL_TIMEOUT', 30))
|
||
# 使用前是否检测连接存活（默认True，防止使用已被MySQL关闭的连接）
|
||
DB_POOL_PRE_PING = os.getenv('DB_POOL_PRE_PING', 'True') == 'True'
|
||
|
||
# 方言 → SQLAlchemy URL 前缀映射（多库支持：只改 DB_DRIVER + 账号即可切换）
|
||
_DB_URI_PREFIX = {
|
||
'mysql': 'mysql+pymysql',
|
||
'pgsql': 'postgresql+psycopg2',
|
||
'postgresql': 'postgresql+psycopg2',
|
||
'mssql': 'mssql+pymssql',
|
||
'sqlite': 'sqlite',
|
||
'oracle': 'oracle+oracledb',
|
||
}
|
||
|
||
|
||
def _build_database_url() -> str:
|
||
"""
|
||
按 DB_DRIVER 构建数据库连接 URL。
|
||
|
||
支持：mysql（默认）/ postgresql / mssql / sqlite / oracle。
|
||
各库差异（端口、账号、ssl、charset、sqlite 无 host、oracle 用 SERVICE_NAME）在此统一封装。
|
||
"""
|
||
prefix = _DB_URI_PREFIX.get(DB_DRIVER, 'mysql+pymysql')
|
||
|
||
# SQLite：无 host/port/账号，DB_DATABASE 即文件路径（内存库用 ':memory:'）
|
||
if DB_DRIVER == 'sqlite':
|
||
return _sqlite_url(DB_DATABASE)
|
||
|
||
cred = f"{quote_plus(DB_USERNAME)}:{quote_plus(DB_PASSWORD)}@{DB_HOST}:{DB_PORT}"
|
||
|
||
# Oracle：无"数据库名"，DB_DATABASE 即 SERVICE_NAME（如 XEPDB1），走 URL 查询参数
|
||
if DB_DRIVER == 'oracle':
|
||
return f"{prefix}://{cred}/?service_name={quote_plus(DB_DATABASE)}"
|
||
|
||
url = f"{prefix}://{cred}/{DB_DATABASE}"
|
||
|
||
# MySQL（含未知驱动回退到 mysql）：显式 utf8mb4 保证中文正确
|
||
if prefix == 'mysql+pymysql':
|
||
url += "?charset=utf8mb4"
|
||
# PostgreSQL：SSL 走 URL 参数（MySQL/MSSQL 的 SSL 在 _build_engine_options / pymssql 内配置）
|
||
elif DB_DRIVER in ('pgsql', 'postgresql') and DB_SSL:
|
||
url += "?sslmode=require"
|
||
|
||
return url
|
||
|
||
|
||
# 数据库连接URL（方便SQLAlchemy直接使用）
|
||
DATABASE_URL = _build_database_url()
|
||
|
||
|
||
def friendly_db_error(e: Exception) -> str:
|
||
"""
|
||
把数据库连接异常翻译成可读的中文提示。
|
||
|
||
处理 psycopg2 在中文 Windows 下返回 GBK 错误消息、被按 UTF-8 解码抛出
|
||
UnicodeDecodeError 的怪癖：先还原字节并按 GBK 解码，再匹配常见错误。
|
||
"""
|
||
raw = str(e)
|
||
if isinstance(e, UnicodeDecodeError):
|
||
# UnicodeDecodeError.args = (encoding, bytes, start, end, reason)；args[1] 是原始字节
|
||
try:
|
||
raw = e.args[1].decode('gbk', errors='ignore')
|
||
except Exception:
|
||
raw = str(e)
|
||
low = raw.lower()
|
||
if any(k in low for k in ('password authentication failed', '28p01', '认证失败', '密码错误', 'ora-01017')):
|
||
return '数据库认证失败：用户名或密码错误，请检查 .env 的 DB_USERNAME / DB_PASSWORD'
|
||
# 数据库不存在（PG SQLSTATE 3d000 / "database ... does not exist" / "数据库...不存在"）
|
||
if '3d000' in low or ('database' in low and 'does not exist' in low) or ('数据库' in raw and '不存在' in raw):
|
||
return '数据库不存在：请先在目标库创建对应数据库（如 CREATE DATABASE <库名>）'
|
||
# 列/表类型不兼容（PG "type ... does not exist" / "类型...不存在"），区别于"数据库不存在"
|
||
if ('type' in low and 'does not exist' in low) or ('类型' in raw and '不存在' in raw):
|
||
return '目标表结构或列类型与数据库不兼容：请检查模型列类型（如 DATETIME 需用通用 DateTime）'
|
||
# Oracle 服务名/监听解析（ORA-12514 未知服务、ORA-12505 未知 SID、ORA-12154 无法解析连接标识符）
|
||
if any(k in low for k in ('ora-12514', 'ora-12505', 'ora-12154', 'could not resolve the connect identifier')):
|
||
return 'Oracle 服务名/监听解析失败：请检查 .env 的 DB_DATABASE（Oracle 填 SERVICE_NAME，如 XEPDB1）与监听配置'
|
||
if any(k in low for k in ('connection refused', '拒绝连接', 'cannot connect', 'target actively refused',
|
||
'ora-12541', 'ora-12560', 'ora-12518', 'ora-12170', 'ora-01034',
|
||
'oracle not available', 'no listener')):
|
||
return '数据库连接被拒绝：请检查 DB_HOST / DB_PORT 与数据库服务是否启动'
|
||
if any(k in low for k in ('timeout', '超时', 'timed out')):
|
||
return '数据库连接超时：请检查网络与 DB_HOST / DB_PORT'
|
||
if any(k in low for k in ('no module named', 'is not installed', '驱动未安装')):
|
||
return '数据库驱动未安装：请执行 pip install -r requirements.txt 安装对应驱动'
|
||
return f'数据库连接失败：{raw}'
|
||
|
||
|
||
def _build_engine_options():
|
||
"""
|
||
构建SQLAlchemy引擎选项
|
||
|
||
包含连接池配置和SSL加密配置。
|
||
连接池参数通过SQLALCHEMY_ENGINE_OPTIONS传递，
|
||
确保所有连接管理策略集中在一处。
|
||
"""
|
||
options = {
|
||
'pool_size': DB_POOL_SIZE,
|
||
'max_overflow': DB_MAX_OVERFLOW,
|
||
'pool_recycle': DB_POOL_RECYCLE,
|
||
'pool_timeout': DB_POOL_TIMEOUT,
|
||
'pool_pre_ping': DB_POOL_PRE_PING,
|
||
}
|
||
# SSL：MySQL 走 connect_args；PG 走 URL sslmode（_build_database_url）；MSSQL 由 pymssql 处理；
|
||
# Oracle：oracledb thin 模式默认走系统 CA，TLS 用 URL tcp+tls 或在 connect_args 传 ssl_context，本处不做额外处理
|
||
if DB_SSL and DB_DRIVER == 'mysql':
|
||
ssl_config = {}
|
||
if DB_SSL_CA:
|
||
ssl_config['ca'] = DB_SSL_CA
|
||
if DB_SSL_CERT:
|
||
ssl_config['cert'] = DB_SSL_CERT
|
||
if DB_SSL_KEY:
|
||
ssl_config['key'] = DB_SSL_KEY
|
||
# 无证书路径时使用默认SSL（信任系统CA）
|
||
if not ssl_config:
|
||
ssl_config = True
|
||
options['connect_args'] = {'ssl': ssl_config}
|
||
return options
|
||
|
||
|
||
class DatabaseConfig:
|
||
"""
|
||
数据库配置类
|
||
|
||
包含 MySQL 连接、SQLAlchemy 设置、路径等配置
|
||
"""
|
||
|
||
# ============================================================
|
||
# 基础配置
|
||
# ============================================================
|
||
|
||
# 调试模式（从环境变量读取）
|
||
DEBUG = DB_DEBUG
|
||
|
||
# Flask 秘钥（用于 session 加密等）
|
||
# 启动时校验：未配置或过短则拒绝启动
|
||
if not FLASK_SECRET_KEY or len(FLASK_SECRET_KEY) < 16:
|
||
raise RuntimeError(
|
||
"FLASK_SECRET_KEY 未配置或过短！请在 .env 中设置至少16字节的强随机密钥。\n"
|
||
"生成方式：python -c \"import secrets; print(secrets.token_hex(32))\""
|
||
)
|
||
SECRET_KEY = FLASK_SECRET_KEY
|
||
|
||
# ============================================================
|
||
# CSRF 防护配置
|
||
# ============================================================
|
||
|
||
# 本项目使用 JWT（Authorization 头）认证，非 Cookie/Session，
|
||
# 天然防 CSRF，因此 WTForms CSRF 保护无需开启。
|
||
WTF_CSRF_CHECK_DEFAULT = False
|
||
WTF_CSRF_ENABLED = False
|
||
|
||
# ============================================================
|
||
# 数据库连接配置（按 DB_DRIVER 驱动切换，见 _build_database_url）
|
||
# ============================================================
|
||
|
||
SQLALCHEMY_DATABASE_URI = DATABASE_URL
|
||
# 不追踪对象修改（提升性能）
|
||
SQLALCHEMY_TRACK_MODIFICATIONS = False
|
||
# 是否打印 SQL 语句（调试模式下开启）
|
||
SQLALCHEMY_ECHO = DB_DEBUG
|
||
# 数据库连接引擎选项（连接池 + SSL加密）
|
||
SQLALCHEMY_ENGINE_OPTIONS = _build_engine_options()
|
||
|
||
# ============================================================
|
||
# 路径配置
|
||
# ============================================================
|
||
|
||
# 配置目录路径
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
# 静态文件目录
|
||
STATIC_DIR = os.path.join(BASE_DIR, '../static')
|
||
# 模板文件目录
|
||
TEMPLATE_DIR = os.path.join(BASE_DIR, '../templates')
|
||
# 文件上传目录
|
||
FLASK_UPLOAD_DIR = FLASK_UPLOAD_DIR
|
||
|
||
|
||
# 模块测试入口
|
||
if __name__ == '__main__':
|
||
print(f"数据库连接: {DatabaseConfig.SQLALCHEMY_DATABASE_URI}")
|
||
print(f"调试模式: {DatabaseConfig.DEBUG}")
|
||
print(f"文件上传目录: {DatabaseConfig.FLASK_UPLOAD_DIR}")
|