JiuContinent
d7db84571d
feat: 添加飞书用户模块和订阅功能支持 - 新增feishu_users模块用于处理飞书用户身份验证和权限管理 - 新增subscriptions模块用于处理订阅相关功能 - 新增personalization模块用于个性化服务 - 在alembic迁移配置中注册新的模型模块 - 在API路由器中添加feishu_users和subscriptions路由 - 实现事件调度服务的改进，包括错误处理和状态更新优化 - 添加飞书命令处理的权限检查机制 - 实现飞书应用票据事件处理 - 改进审计日志记录功能 ```
63 lines
1.4 KiB
Python
63 lines
1.4 KiB
Python
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from sqlalchemy.engine import RowMapping
|
|
|
|
from app.modules.legacy_mysql.constants import LEGACY_SQL_TRAILING_TERMINATOR, LegacyQueryName
|
|
|
|
FORBIDDEN_SQL_TOKENS = {
|
|
"benchmark",
|
|
"call",
|
|
"insert",
|
|
"update",
|
|
"delete",
|
|
"drop",
|
|
"alter",
|
|
"truncate",
|
|
"create",
|
|
"do",
|
|
"dumpfile",
|
|
"execute",
|
|
"replace",
|
|
"grant",
|
|
"get_lock",
|
|
"handler",
|
|
"into",
|
|
"load_file",
|
|
"lock",
|
|
"outfile",
|
|
"release_lock",
|
|
"revoke",
|
|
"set",
|
|
"sleep",
|
|
}
|
|
|
|
|
|
def _jsonable(value: Any) -> Any:
|
|
"""Convert database scalar values into JSON-friendly values."""
|
|
|
|
if isinstance(value, (datetime, date)):
|
|
return value.isoformat()
|
|
if isinstance(value, Decimal):
|
|
return float(value)
|
|
return value
|
|
|
|
|
|
def _row_to_dict(row: RowMapping) -> dict[str, Any]:
|
|
"""Convert a SQLAlchemy row mapping to a serializable dictionary."""
|
|
|
|
return {key: _jsonable(value) for key, value in row.items()}
|
|
|
|
|
|
def _normalize_sql(sql: str) -> str:
|
|
return " ".join(sql.strip().rstrip(LEGACY_SQL_TRAILING_TERMINATOR).split()).lower()
|
|
|
|
|
|
def _query_name_text(query_name: str | LegacyQueryName | None) -> str:
|
|
if query_name is None:
|
|
return LegacyQueryName.PROJECTS.value
|
|
if isinstance(query_name, LegacyQueryName):
|
|
return query_name.value
|
|
return str(query_name)
|