检查当前用户在当前租户下是否具有管理员权限。 Args: raise_error: 是否在权限不足时抛出异常，默认为 True。 Returns: bool: 如果有管理员权限返回 True，否则根据 raise_e

根据ID获取知识库。 Args: id (str): 知识库ID Returns: KnowledgeBase: 知识库对象 Raises: ValueError: 当知识库

根据ID获取MCP服务器。 通过MCP服务器ID查询并返回对应的服务器实例。 Args: mcp_server_id (int): MCP服务器的唯一标识符。 Returns: McpServer:

获取目录的总大小。 递归计算目录中所有文件的总大小。 Args: dir_path (str): 目录路径。 Returns: int: 目录总大小（字节），如果出错返回 0。

获取已发布工作流。 Args: app_id (str): 应用ID Returns: Workflow: 已发布工作流实例 Raises: Exception: 当获取失败

检查执行是否成功。 Returns: bool: 是否成功

处理权限错误的基础方法。 Args: raise_error: 是否抛出异常，如果为 False 则返回 False。 Returns: bool: 当 raise_error 为 False 时返回 False。

Create SSE format event string

创建表结构。 在指定数据库中创建新表，包括列定义、约束等，并保存表信息到系统表。 Args: database_id (int): 数据库ID table_name (str): 表名称 co

数据类型转换方法。 根据数据库类型自动选择正确的映射规则，将原始数据类型 转换为统一的通用数据类型。 Args: data_type (str): 数据库原始数据类型 db_engine (Engin

获取通知详情。 获取指定通知的详细信息。 Args: notification_id (str): 通知ID，通过查询参数传递。 Returns: dict: 通知详细信息字典。

统计指定app_id下的各类指标，并缓存到redis，重复访问优先读缓存。 统计指标包括： - 累计token消费总数（CostAudit.token_num求和） - 累计用户数（Conversation.from_who在Account.id中的

根据ID获取模型。 根据模型ID获取模型对象。 Args: model_id (int): 模型ID。 Returns: Lazymodel: 模型对象，如果不存在则返回None。

获取数据集的标签列表（兼容历史版本）。 Returns: list: 数据集关联的标签名称列表。

发送邮件。 发送 HTML 格式的邮件到指定收件人。会验证所有必需的参数， 如果未指定发件人地址，将使用默认发件人地址。 Args: to (str): 收件人邮箱地址。 subject (str):

打印测试头部信息

创建存储目录。 生成标准化的存储路径格式：/app/upload/{path_name}/user_id/args[0]/args[1]/... 如果没有提供额外参数，则生成随机目录名以防止文件名冲突。 Args: path

创建临时文件存储目录。 Args: user_id: 用户 ID。 *args: 额外的路径组件。 **kwargs: 关键字参数。 Returns: str: 临时存

生成列注释的SQL语句。 根据不同数据库方言生成添加或修改列注释的SQL语句。 Args: table_name (str): 表名 column (dict): 列定义字典，包含注释信息 Retu

获取数据库的默认模式名。 根据数据库类型返回相应的默认模式名称。 Args: engine: SQLAlchemy数据库引擎对象 Returns: str: 模式名称，PostgreSQL返回"public"，其他数据库返回空字符串

根据ID获取脚本实例。 Args: script_id (int): 脚本ID。 Returns: Script or None: 脚本实例，如果不存在则返回None。