diff --git a/app/api/endpoints/plugin.py b/app/api/endpoints/plugin.py index 4b75c445a..bb41dc1f9 100644 --- a/app/api/endpoints/plugin.py +++ b/app/api/endpoints/plugin.py @@ -979,14 +979,15 @@ def uninstall_plugin(plugin_id: str, _: ApiPrincipal = Depends(get_current_activ remove_plugin_job(plugin_id) # 判断是否为分身 plugin_class = plugin_manager.plugins.get(plugin_id) + # 删除必须晚于停止：停机钩子会重建刚删的自有库；停止同时注销插件类，故删除一律按 force + plugin_manager.stop(plugin_id) if virtual_instance: plugin_manager.delete_plugin_config(plugin_id, force=True) plugin_manager.delete_plugin_data(plugin_id, force=True) plugin_manager.delete_plugin_instance(plugin_id) elif getattr(plugin_class, "is_clone", False): - # 如果是分身插件，则删除分身数据和配置 - plugin_manager.delete_plugin_config(plugin_id) - plugin_manager.delete_plugin_data(plugin_id) + plugin_manager.delete_plugin_config(plugin_id, force=True) + plugin_manager.delete_plugin_data(plugin_id, force=True) # 分身物理目录只能由包文件 owner 删除。 if plugin_manager.remove_plugin_package(plugin_id): plugin_manager.plugins.pop(plugin_id, None) diff --git a/app/application/plugin/management.py b/app/application/plugin/management.py index 38b7a254e..d8386cb2e 100644 --- a/app/application/plugin/management.py +++ b/app/application/plugin/management.py @@ -331,13 +331,17 @@ async def uninstall_plugin_runtime(plugin_id: str) -> dict[str, Any]: was_clone = bool(getattr(plugin_class, "is_clone", False)) clone_files_removed = False + # 删除数据必须发生在插件停止之后：插件的停机钩子只要取一次自有库句柄，就会把刚 + # 删除的数据重新建出来。停止同时注销插件类，故其后的删除一律按 force 执行 + plugin_manager.stop(plugin_id) + if virtual_instance: plugin_manager.delete_plugin_config(plugin_id, force=True) plugin_manager.delete_plugin_data(plugin_id, force=True) plugin_manager.delete_plugin_instance(plugin_id) elif was_clone: - plugin_manager.delete_plugin_config(plugin_id) - plugin_manager.delete_plugin_data(plugin_id) + plugin_manager.delete_plugin_config(plugin_id, force=True) + plugin_manager.delete_plugin_data(plugin_id, force=True) try: clone_files_removed = await to_thread.run_sync( plugin_manager.remove_plugin_package, diff --git a/app/db/engine.py b/app/db/engine.py index eeb137ca9..2d74fac6c 100644 --- a/app/db/engine.py +++ b/app/db/engine.py @@ -86,6 +86,69 @@ def _async_pool_kwargs(pooled: bool) -> dict: } +def apply_sqlite_journal_mode(engine: SyncEngine) -> Any: + """ + 把配置声明的 journal_mode 写入引擎所指的库文件，并返回实际生效值。 + + journal_mode 是库文件级的持久属性，每个库文件都要有人设置一次。宿主库与插件自有库 + 共用这一段，不会出现「宿主是 WAL、插件库是 DELETE」的分裂行为。 + :param engine: 同步引擎 + :return: 生效的 journal_mode + """ + journal_mode = "WAL" if get_runtime_setting("DB_WAL_ENABLE") else "DELETE" + with engine.connect() as connection: + return connection.execute(text(f"PRAGMA journal_mode={journal_mode};")).scalar() + + +def build_sqlite_engine(url: str) -> SyncEngine: + """ + 按宿主 SQLite 连接策略构建指向给定库文件的同步引擎。 + + 连接超时、驱动级参数、连接池、外键约束、错误日志与池指标在这里收口，宿主库与插件 + 自有库因此共享同一套连接语义，不会因为落在不同文件而出现两种行为。 + :param url: SQLite 连接串 + :return: 同步引擎 + """ + # 连接参数 + _connect_args = { + "timeout": get_runtime_setting("DB_TIMEOUT"), + } + # 允许部署侧注入驱动级参数（如 PgBouncer 事务模式下的 statement_cache_size） + _connect_args.update(get_runtime_setting("DB_CONNECT_ARGS") or {}) + # 启用 WAL 模式时的额外配置 + if get_runtime_setting("DB_WAL_ENABLE"): + _connect_args["check_same_thread"] = False + + # 根据池类型设置 poolclass 和相关参数 + _pool_class = NullPool if get_runtime_setting("DB_POOL_TYPE") == "NullPool" else QueuePool + + # 数据库参数 + _db_kwargs = { + "url": url, + "pool_pre_ping": get_runtime_setting("DB_POOL_PRE_PING"), + "echo": get_runtime_setting("DB_ECHO"), + "poolclass": _pool_class, + "pool_recycle": get_runtime_setting("DB_POOL_RECYCLE"), + "connect_args": _connect_args, + } + + # 当使用 QueuePool 时，添加 QueuePool 特有的参数 + if _pool_class == QueuePool: + _db_kwargs.update( + { + "pool_size": get_runtime_setting("DB_SQLITE_POOL_SIZE"), + "pool_timeout": get_runtime_setting("DB_POOL_TIMEOUT"), + "max_overflow": get_runtime_setting("DB_SQLITE_MAX_OVERFLOW"), + } + ) + + engine = create_engine(**_db_kwargs) + _register_sqlite_foreign_keys(engine) + _register_database_error_logging(engine) + _register_database_pool_metrics(engine) + return engine + + def _get_database_engine(is_async: bool = False, pooled: bool = False): """ 获取数据库连接参数并设置WAL模式 @@ -104,46 +167,9 @@ def _get_sqlite_engine(is_async: bool = False, pooled: bool = False): """ 获取SQLite数据库引擎 """ - # 连接参数 - _connect_args = { - "timeout": get_runtime_setting("DB_TIMEOUT"), - } - # 允许部署侧注入驱动级参数（如 PgBouncer 事务模式下的 statement_cache_size） - _connect_args.update(get_runtime_setting("DB_CONNECT_ARGS") or {}) - # 启用 WAL 模式时的额外配置 - if get_runtime_setting("DB_WAL_ENABLE"): - _connect_args["check_same_thread"] = False - # 创建同步引擎 if not is_async: - # 根据池类型设置 poolclass 和相关参数 - _pool_class = NullPool if get_runtime_setting("DB_POOL_TYPE") == "NullPool" else QueuePool - - # 数据库参数 - _db_kwargs = { - "url": get_runtime_setting("DB_SQLITE_URL")(), - "pool_pre_ping": get_runtime_setting("DB_POOL_PRE_PING"), - "echo": get_runtime_setting("DB_ECHO"), - "poolclass": _pool_class, - "pool_recycle": get_runtime_setting("DB_POOL_RECYCLE"), - "connect_args": _connect_args, - } - - # 当使用 QueuePool 时，添加 QueuePool 特有的参数 - if _pool_class == QueuePool: - _db_kwargs.update( - { - "pool_size": get_runtime_setting("DB_SQLITE_POOL_SIZE"), - "pool_timeout": get_runtime_setting("DB_POOL_TIMEOUT"), - "max_overflow": get_runtime_setting("DB_SQLITE_MAX_OVERFLOW"), - } - ) - - # 创建数据库引擎 - engine = create_engine(**_db_kwargs) - _register_sqlite_foreign_keys(engine) - _register_database_error_logging(engine) - _register_database_pool_metrics(engine) + engine = build_sqlite_engine(get_runtime_setting("DB_SQLITE_URL")()) # 设置WAL模式。 # 这是引擎构建里唯一的阻塞 I/O，且发生在 get_engine() 的创建锁内——异步侧因此 @@ -151,13 +177,20 @@ def _get_sqlite_engine(is_async: bool = False, pooled: bool = False): # 设置一次，而同步引擎的首次创建由 lifespan 数据库准备组件中的 init_db() 完成， # 不存在一群线程 # 等在锁上的场面；即便退化到运行期首次访问，阻塞的也只是本地 SQLite 的一次 PRAGMA。 - _journal_mode = "WAL" if get_runtime_setting("DB_WAL_ENABLE") else "DELETE" - with engine.connect() as connection: - current_mode = connection.execute(text(f"PRAGMA journal_mode={_journal_mode};")).scalar() - print(f"SQLite database journal mode set to: {current_mode}") + print(f"SQLite database journal mode set to: {apply_sqlite_journal_mode(engine)}") return engine else: + # 连接参数 + _connect_args = { + "timeout": get_runtime_setting("DB_TIMEOUT"), + } + # 允许部署侧注入驱动级参数（如 PgBouncer 事务模式下的 statement_cache_size） + _connect_args.update(get_runtime_setting("DB_CONNECT_ARGS") or {}) + # 启用 WAL 模式时的额外配置 + if get_runtime_setting("DB_WAL_ENABLE"): + _connect_args["check_same_thread"] = False + # 数据库参数，只能使用 NullPool _db_kwargs = { "url": get_runtime_setting("DB_SQLITE_URL")("aiosqlite"), diff --git a/app/db/plugin/__init__.py b/app/db/plugin/__init__.py new file mode 100644 index 000000000..93ee726c4 --- /dev/null +++ b/app/db/plugin/__init__.py @@ -0,0 +1 @@ +"""按插件隔离的自管理数据库：独立引擎、建表、迁移与销毁。""" diff --git a/app/db/plugin/base.py b/app/db/plugin/base.py new file mode 100644 index 000000000..ee7554647 --- /dev/null +++ b/app/db/plugin/base.py @@ -0,0 +1,24 @@ +"""插件专属声明式基类工厂。""" + +from __future__ import annotations + +from sqlalchemy.orm import DeclarativeBase + +__all__ = ["plugin_declarative_base"] + + +def plugin_declarative_base() -> type[DeclarativeBase]: + """ + 产出携带全新 ``MetaData`` 的声明式基类。 + + 插件模型继承本函数的返回值定义，其表便注册在独立的 ``MetaData`` 上：既不与宿主 + ``app.db.base.Base`` 抢同一份注册表，也允许两个插件各自定义同名表。插件热重载会 + 重新执行插件模块、重新调用本函数，因此每次都拿到干净的注册表，不会在同一 + ``MetaData`` 上重复定义同名表而报错。 + :return: 声明式基类 + """ + + class PluginBase(DeclarativeBase): # type: ignore[misc] # SQLAlchemy 无 py.typed 基类 + """插件自有表的声明式基类。""" + + return PluginBase diff --git a/app/db/plugin/container.py b/app/db/plugin/container.py new file mode 100644 index 000000000..9679d1f3d --- /dev/null +++ b/app/db/plugin/container.py @@ -0,0 +1,45 @@ +"""插件数据库句柄：持有引擎、会话工厂与释放策略。""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +from sqlalchemy.engine import Engine +from sqlalchemy.orm import Session, scoped_session, sessionmaker + +__all__ = ["PluginDatabaseHandle"] + + +@dataclass(frozen=True, slots=True) +class PluginDatabaseHandle: + """ + 一个插件专属数据库的连接要素。 + + SQLite 下引擎由本句柄独占，``owns_engine`` 为真；PostgreSQL 下引擎是宿主引擎按 + ``schema_translate_map`` 派生的外观，``owns_engine`` 为假——只有前者可以 dispose， + 后者一旦 dispose 会连累宿主与其它插件仍在使用的同一个连接池。PostgreSQL 下本句柄的 + 会话与连接在每个事务开始时把 ``search_path`` 限定到插件 schema，未限定的原生 SQL 因此 + 同样解析到插件自己的表。 + """ + + plugin_id: str + engine: Engine + session_factory: sessionmaker + scoped_session_factory: scoped_session + db_path: Path | None + schema: str | None + owns_engine: bool + + def session(self) -> Session: + """新建一个绑定本库的会话，提交、回滚与关闭由调用方负责。""" + return self.session_factory() + + def scoped_session(self) -> Session: + """取当前线程绑定的会话，同一线程内重复调用复用同一个实例。""" + return self.scoped_session_factory() + + def dispose(self) -> None: + """释放本句柄独占的连接池；不拥有引擎时不做任何事。""" + if self.owns_engine: + self.engine.dispose() diff --git a/app/db/plugin/locator.py b/app/db/plugin/locator.py new file mode 100644 index 000000000..c25cbdd8e --- /dev/null +++ b/app/db/plugin/locator.py @@ -0,0 +1,75 @@ +"""插件数据库文件路径与 PostgreSQL schema 名的解析，不持有任何状态。""" + +from __future__ import annotations + +import hashlib +from pathlib import Path + +from app.runtime.settings import get_runtime_setting + +__all__ = [ + "plugin_schema_name", + "sqlite_database_path", + "sqlite_sidecar_paths", +] + +# 库文件名固定：所在目录已按插件标识分段，文件名无需再编码插件身份 +DATABASE_FILENAME = "plugin.db" +# SQLite WAL 模式下与库文件同生共死的边车文件后缀 +SIDECAR_SUFFIXES = ("-wal", "-shm") +# PostgreSQL 标识符上限 63 字节，超出部分会被静默截断成另一个 schema +SCHEMA_NAME_MAX_LENGTH = 63 +# 归一改写了插件标识时用于区分两个插件的哈希长度 +SCHEMA_NAME_HASH_LENGTH = 8 + + +def sqlite_database_path(plugin_id: str) -> Path: + """ + 返回插件 SQLite 库文件路径。 + + 落在与 ``_PluginBase.get_data_path()`` 完全相同的插件数据目录下，插件库因此随插件 + 数据一起被备份、迁移和删除，不会形成第二份需要单独维护的持久化根。本函数不创建 + 目录：``ensure`` 在插件什么都没声明时不得凭空产生目录。 + :param plugin_id: 插件标识 + :return: 库文件绝对路径 + """ + return Path(get_runtime_setting("PLUGIN_DATA_PATH")) / plugin_id / DATABASE_FILENAME + + +def sqlite_sidecar_paths(db_path: Path) -> tuple[Path, ...]: + """ + 返回库文件对应的 WAL/SHM 边车文件路径。 + :param db_path: 库文件路径 + :return: 边车路径元组，不保证文件存在 + """ + return tuple(db_path.with_name(db_path.name + suffix) for suffix in SIDECAR_SUFFIXES) + + +def plugin_schema_name(plugin_id: str) -> str: + """ + 按插件标识拼出 PostgreSQL schema 名。 + + 只保留 ASCII 小写字母、数字与下划线，避免插件标识中的字符逃逸成标识符片段，也让 + 名字的字符数与字节数一致。归一是多对一的：``My-Plugin``、``My_Plugin`` 与 + ``my_plugin`` 会折叠到同一个名字，而卸载一个插件执行的是 ``DROP SCHEMA ... + CASCADE``，折叠意味着删掉另一个插件的全部数据。 + 因此只要归一改写过标识，就追加插件标识的哈希把它们重新分开；超长标识同样追加哈希 + 再截断，截断本身也是一次折叠。 + :param plugin_id: 插件标识 + :return: 合法且与插件标识一一对应的 schema 名 + """ + raw = f"plugin_{plugin_id}" + sanitized = "".join( + character + if (character.isascii() and character.isalnum()) or character == "_" + else "_" + for character in raw.lower() + ) + if sanitized == raw and len(sanitized) <= SCHEMA_NAME_MAX_LENGTH: + return sanitized + digest = hashlib.sha1( + plugin_id.encode("utf-8"), + usedforsecurity=False, + ).hexdigest()[:SCHEMA_NAME_HASH_LENGTH] + suffix = f"_{digest}" + return f"{sanitized[:SCHEMA_NAME_MAX_LENGTH - len(suffix)]}{suffix}" diff --git a/app/db/plugin/migration.py b/app/db/plugin/migration.py new file mode 100644 index 000000000..810232d59 --- /dev/null +++ b/app/db/plugin/migration.py @@ -0,0 +1,40 @@ +"""插件自有库的 Alembic 迁移执行。""" + +from __future__ import annotations + +from configparser import ConfigParser +from pathlib import Path + +from alembic.command import upgrade +from alembic.config import Config + +from app.db.plugin.container import PluginDatabaseHandle + +__all__ = ["run_migrations"] + + +def run_migrations(handle: PluginDatabaseHandle, directory: Path) -> None: + """ + 在句柄对应的库上把 Alembic 迁移跑到 head。 + + SQLite 按库文件 URL 起独立连接；PostgreSQL 复用句柄已被 ``schema_translate_map`` + 限定过的连接，迁移脚本的 env.py 必须从 ``context.config.attributes["connection"]`` + 取用它，否则迁移会落在 public schema 而不是该插件的 schema。 + :param handle: 插件数据库句柄 + :param directory: 迁移脚本目录，须符合 Alembic script_location 布局 + """ + config = Config() + # 关闭 ini 插值：迁移目录里出现的 % 不应被当成插值语法（与宿主 _build_alembic_config 一致） + config.file_config = ConfigParser(interpolation=None) + config.set_main_option("script_location", str(directory)) + if handle.owns_engine: + config.set_main_option( + "sqlalchemy.url", + handle.engine.url.render_as_string(hide_password=False), + ) + upgrade(config, "head") + return + with handle.engine.connect() as connection: + config.attributes["connection"] = connection + upgrade(config, "head") + connection.commit() diff --git a/app/db/plugin/registry.py b/app/db/plugin/registry.py new file mode 100644 index 000000000..40d235542 --- /dev/null +++ b/app/db/plugin/registry.py @@ -0,0 +1,264 @@ +"""插件数据库注册表：按插件标识管理句柄的建立、建表与释放。""" + +from __future__ import annotations + +import threading +from collections.abc import Callable, Sequence +from pathlib import Path + +from sqlalchemy import event, text +from sqlalchemy.engine import Connection +from sqlalchemy.orm import scoped_session, sessionmaker + +from app.db.engine import apply_sqlite_journal_mode, build_sqlite_engine, get_engine +from app.db.plugin.container import PluginDatabaseHandle +from app.db.plugin.locator import ( + plugin_schema_name, + sqlite_database_path, + sqlite_sidecar_paths, +) +from app.runtime.log import logger +from app.runtime.settings import get_runtime_setting + +__all__ = [ + "destroy_database", + "ensure_database", + "get_database", + "release_all_databases", + "release_database", +] + +# 插件的启动、停止与热重载分别来自调度线程、文件监控线程与 HTTP 线程，句柄的建立与 +# 释放必须串行，否则同一插件会同时存在两个各持一份连接池的引擎。销毁的关连接与删载体 +# 同样在锁内完成：只要在删载体前放锁，并发的 get_database 就能注册一个指向同一载体的 +# 新句柄，紧接着的删除会把这个新句柄的库连同数据一起抹掉 +_lock = threading.RLock() +_handles: dict[str, PluginDatabaseHandle] = {} + + +def _is_postgresql() -> bool: + """判断宿主当前是否使用 PostgreSQL。""" + db_type: str = get_runtime_setting("DB_TYPE") + return db_type.lower() == "postgresql" + + +def _search_path_setter(schema: str) -> Callable[[Connection], None]: + """ + 构造把事务内未限定标识符解析到插件 schema 的 ``begin`` 监听器。 + + ``schema_translate_map`` 只改写 SQLAlchemy 生成的 schema 感知语句，``text()`` 一类 + 的原生 SQL 不在其列；不动 ``search_path``，插件按合同写的未限定原生 SQL 会落到 + ``public``。``SET LOCAL`` 的作用域随事务结束，不会经连接池把插件的解析根泄漏给宿主 + 或其它插件。 + :param schema: 插件 schema 名 + :return: 绑定该 schema 的 ``begin`` 事件监听器 + """ + + def _apply_search_path(connection: Connection) -> None: + """在事务开始时把该连接的未限定标识符解析根切到插件 schema。""" + connection.exec_driver_sql(f'SET LOCAL search_path TO "{schema}", public') + + return _apply_search_path + + +def _build_handle(plugin_id: str) -> PluginDatabaseHandle: + """ + 构造插件的数据库句柄。 + + SQLite 每插件一个独立库文件，句柄独占引擎；PostgreSQL 复用宿主引擎并按 + ``schema_translate_map`` 派生出限定单一 schema 的外观，句柄不拥有该引擎，并在派生 + 引擎上按事务限定 ``search_path``，让未限定的原生 SQL 同样落在插件自己的 schema。 + :param plugin_id: 插件标识 + :return: 新建的数据库句柄 + """ + if _is_postgresql(): + schema = plugin_schema_name(plugin_id) + host_engine = get_engine() + with host_engine.begin() as connection: + connection.execute(text(f'CREATE SCHEMA IF NOT EXISTS "{schema}"')) + engine = host_engine.execution_options(schema_translate_map={None: schema}) + # 监听器只挂在派生引擎上：OptionEngine 自带 dispatch，仅单向 _join 宿主引擎已有 + # 的监听器，宿主与其它插件的连接不会因此改变解析根 + event.listen(engine, "begin", _search_path_setter(schema)) + db_path = None + owns_engine = False + else: + db_path = sqlite_database_path(plugin_id) + db_path.parent.mkdir(parents=True, exist_ok=True) + engine = build_sqlite_engine(f"sqlite:///{db_path}") + apply_sqlite_journal_mode(engine) + schema = None + owns_engine = True + + session_factory = sessionmaker(bind=engine) + return PluginDatabaseHandle( + plugin_id=plugin_id, + engine=engine, + session_factory=session_factory, + scoped_session_factory=scoped_session(session_factory), + db_path=db_path, + schema=schema, + owns_engine=owns_engine, + ) + + +def _close_handle(handle: PluginDatabaseHandle) -> None: + """ + 先关闭线程局部会话再释放句柄独占的连接池。 + + 线程局部会话各自扣着一条连接，不先归还就 dispose，会让已经被移出注册表的句柄仍然 + 握着文件描述符。两步都是收尾操作，任一步失败只记日志：一个插件的连接池故障不得让 + 宿主引擎与其余插件的释放整体失败。 + :param handle: 数据库句柄 + """ + try: + handle.scoped_session_factory.remove() + except Exception as error: # noqa: BLE001 会话清理故障不得阻断连接池释放 + logger.warning(f"清理插件 {handle.plugin_id} 的线程局部会话失败：{error}") + try: + handle.dispose() + except Exception as error: # noqa: BLE001 单个插件的释放故障不得阻断其余释放 + logger.warning(f"释放插件 {handle.plugin_id} 的数据库连接失败：{error}") + + +def get_database(plugin_id: str) -> PluginDatabaseHandle: + """ + 取插件的数据库句柄，句柄不存在时按需建立。 + + 不要求插件先声明模型：只打算执行原生 SQL 的插件同样可以直接取会话。PostgreSQL 下 + 句柄的会话与连接在每个事务开始时把 ``search_path`` 限定到插件 schema，未限定的原生 + SQL 因此同样解析到插件自己的表。 + :param plugin_id: 插件标识 + :return: 数据库句柄 + """ + with _lock: + handle = _handles.get(plugin_id) + if handle is None: + handle = _build_handle(plugin_id) + _handles[plugin_id] = handle + return handle + + +def _create_declared_tables(handle: PluginDatabaseHandle, models: Sequence[type]) -> None: + """ + 按声明的模型建表，只创建插件显式列出的表。 + :param handle: 数据库句柄 + :param models: 插件声明的模型类 + """ + # 端口签名只承诺 type：__table__ 由 SQLAlchemy 声明式元类在类构造期动态挂载， + # 静态类型系统看不到这层映射。单表继承的父类与子类共享同一个 __table__，按对象去重 + # 后才不会把同一张表提交给 create_all 两次 + tables = list(dict.fromkeys(model.__table__ for model in models)) # type: ignore[attr-defined] + for metadata in dict.fromkeys(table.metadata for table in tables): + metadata.create_all( + bind=handle.engine, + tables=[table for table in tables if table.metadata is metadata], + ) + + +def ensure_database( + plugin_id: str, + models: Sequence[type] = (), + migrations: Path | None = None, +) -> None: + """ + 按插件的声明建立数据库：声明了迁移目录走 alembic，否则按模型建表。 + + 两者都未声明时不建立句柄、不产生任何库文件——绝大多数插件不使用自有库，不该因为 + 宿主统一调用了本函数就凭空多出一个空库和一个空目录。 + :param plugin_id: 插件标识 + :param models: 插件声明的模型类 + :param migrations: 插件声明的 Alembic 迁移目录 + :raise FileNotFoundError: 声明的迁移目录不存在 + """ + if migrations is not None: + # 目录校验必须早于建句柄：alembic 找不到 script_location 时抛错，而句柄已经把 + # 库文件建了出来，插件下次启动面对的是一个既没有表、也没有版本号的空库 + if not migrations.is_dir(): + raise FileNotFoundError(f"插件 {plugin_id} 声明的迁移目录不存在：{migrations}") + # 迁移目录同时描述建表与后续版本演进，与按模型建表会争夺同一批表，故优先且互斥。 + # alembic 只在插件真的声明了迁移目录时才需要，在函数内导入可让宿主与只声明模型的 + # 插件不为它付出导入代价 + from app.db.plugin.migration import run_migrations + + run_migrations(get_database(plugin_id), migrations) + return + if models: + _create_declared_tables(get_database(plugin_id), models) + + +def release_database(plugin_id: str) -> None: + """ + 释放插件的数据库连接，保留全部数据。 + + 只 dispose 句柄自己拥有的引擎；PostgreSQL 下句柄只是宿主引擎的外观，这里不会、 + 也不能触碰宿主连接池。关连接与摘句柄在同一把锁内完成，句柄一旦离开注册表就不会再 + 被任何线程取到一个正在关闭的连接池。 + :param plugin_id: 插件标识 + """ + with _lock: + handle = _handles.pop(plugin_id, None) + if handle is not None: + _close_handle(handle) + + +def release_all_databases() -> None: + """释放全部插件的数据库连接，供进程关停使用。""" + with _lock: + plugin_ids = list(_handles) + for plugin_id in plugin_ids: + try: + release_database(plugin_id) + except Exception as error: # noqa: BLE001 单个插件的故障不得阻断宿主引擎释放 + logger.warning(f"释放插件 {plugin_id} 的数据库失败：{error}") + + +def _drop_schema(plugin_id: str, schema: str) -> None: + """ + 丢弃插件的 PostgreSQL schema。 + :param plugin_id: 插件标识 + :param schema: schema 名 + """ + try: + with get_engine().begin() as connection: + connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + except Exception as error: # noqa: BLE001 删除数据的收尾故障不得升级为异常 + logger.error(f"销毁插件 {plugin_id} 的数据库 schema {schema} 失败：{error}") + + +def _remove_storage(plugin_id: str, handle: PluginDatabaseHandle | None) -> None: + """ + 删除插件数据库的持久载体：SQLite 删库文件与边车，PostgreSQL 丢弃 schema。 + :param plugin_id: 插件标识 + :param handle: 已释放的句柄，为空时按插件标识重新推导载体位置 + """ + if _is_postgresql(): + # 句柄存在时 schema 必非空（PostgreSQL 分支总会填充）；句柄已释放则重新按插件标识 + # 推导，两种取值路径的类型都靠 or 收窄为确定的 str，不引入运行期新分支 + schema = (handle.schema if handle else None) or plugin_schema_name(plugin_id) + _drop_schema(plugin_id, schema) + return + + db_path = (handle.db_path if handle else None) or sqlite_database_path(plugin_id) + for candidate in (db_path, *sqlite_sidecar_paths(db_path)): + try: + candidate.unlink(missing_ok=True) + except OSError as error: + logger.warning(f"删除插件 {plugin_id} 的数据库文件 {candidate} 失败：{error}") + + +def destroy_database(plugin_id: str) -> None: + """ + 销毁插件的数据库：SQLite 删除库文件与 -wal/-shm 边车，PostgreSQL 丢弃对应 schema。 + + 不可逆，调用方必须确认处在「删除插件数据」而非「停止插件」的路径上。摘句柄、关连接 + 与删载体全程持同一把锁，销毁与同一插件的句柄重建因此互斥，删除不会落到并发建出来的 + 新句柄头上。失败只记日志：删除数据是一次收尾操作，把文件系统或数据库的清理故障升级 + 成异常，只会让已经删掉的宿主数据与仍然存在的插件库停在不一致的中间态。 + :param plugin_id: 插件标识 + """ + with _lock: + handle = _handles.pop(plugin_id, None) + if handle is not None: + _close_handle(handle) + _remove_storage(plugin_id, handle) diff --git a/app/db/session.py b/app/db/session.py index 2e088d09f..ac4e572d9 100644 --- a/app/db/session.py +++ b/app/db/session.py @@ -23,6 +23,7 @@ import app.db.engine as engine_module from app.db.engine import (_async_pool_enabled, _get_database_engine, _database_backend_label, get_engine, get_global_async_engine) +from app.db.plugin.registry import release_all_databases from app.runtime.loop import main_loop_registry from app.runtime.settings import get_runtime_setting from app.runtime.log import logger @@ -294,6 +295,9 @@ async def close_database(): 异步引擎与全部池化引擎就都跳过了释放——一条坏连接拖着其余连接一起泄漏， 正是关停路径最不该出现的失败方式。 """ + # 插件自有库先于宿主引擎释放：PostgreSQL 下插件句柄是宿主引擎派生的外观，宿主引擎 + # 一旦 dispose，插件侧再释放就落在一个已经关掉的连接池上 + release_all_databases() # 只释放确实创建过的引擎：惰性之后，为了 dispose 而先把引擎创建出来毫无意义， # 还会在从未用过数据库的进程里凭空连一次库 sync_engine = engine_module.peek_sync_engine() diff --git a/app/plugins/__init__.py b/app/plugins/__init__.py index 2d00c6fc1..ee9dfdd2a 100644 --- a/app/plugins/__init__.py +++ b/app/plugins/__init__.py @@ -1,12 +1,14 @@ from abc import ABCMeta, abstractmethod from pathlib import Path -from typing import Any, List, Dict, Tuple, Optional, Type +from typing import Any, List, Dict, Tuple, Optional, Type, Union from app.chain import ChainBase from app.core.config import settings from app.core.event import EventManager from app.db.oper.plugindata import PluginDataOper from app.db.oper.systemconfig import SystemConfigOper +from app.db.plugin.container import PluginDatabaseHandle +from app.db.plugin.registry import get_database as get_plugin_database from app.helper.message import MessageHelper from app.schemas import Notification, NotificationType, MessageChannel @@ -238,6 +240,31 @@ class _PluginBase(metaclass=ABCMeta): """ pass + def get_database_models(self) -> Optional[List[Type]]: + """ + 声明插件自有数据库中的模型 + [ModelClass1, ModelClass2, ...] + + 对模型类的要求： + 1、模型类必须继承 app.sdk.database.plugin_declarative_base() 产出的基类 + 2、同一插件的模型应继承同一个基类，它们的表在插件启动时一并建立 + 3、同时声明了迁移目录时本声明被忽略，改由 alembic 建表 + """ + pass + + def get_database_migrations(self) -> Optional[Union[str, Path]]: + """ + 声明插件自有数据库的 Alembic 迁移脚本目录 + 目录须符合 Alembic script_location 布局，且必须是绝对路径 + （相对路径按宿主进程的当前工作目录解析，插件无法预期它指向哪里， + 建议用 Path(__file__).parent / "migrations" 之类的写法取得） + + 声明后插件启动时执行 alembic upgrade head，不再按 get_database_models() 建表。 + PostgreSQL 下迁移脚本的 env.py 必须从 context.config.attributes["connection"] + 取用宿主传入的连接，否则迁移会落在 public schema 而不是本插件的 schema。 + """ + pass + @abstractmethod def stop_service(self): """ @@ -275,6 +302,18 @@ class _PluginBase(metaclass=ABCMeta): data_path.mkdir(parents=True) return data_path + def get_database(self, plugin_id: Optional[str] = None) -> PluginDatabaseHandle: + """ + 获取插件自有数据库句柄，用于取会话读写插件自有表 + 句柄不存在时按需建立，不要求先声明模型 + PostgreSQL 下句柄的会话与连接在每个事务开始时把 search_path 限定到本插件 schema， + 未限定的原生 SQL 因此同样解析到插件自己的表，不会落到 public + :param plugin_id: 插件ID + """ + if not plugin_id: + plugin_id = self.__class__.__name__ + return get_plugin_database(plugin_id) + def save_data(self, key: str, value: Any, plugin_id: Optional[str] = None): """ 保存插件数据 diff --git a/app/runtime/extensions/plugin/database.py b/app/runtime/extensions/plugin/database.py new file mode 100644 index 000000000..d9f5e9de7 --- /dev/null +++ b/app/runtime/extensions/plugin/database.py @@ -0,0 +1,73 @@ +"""插件运行时自有数据库端口。""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from pathlib import Path + +PluginDatabaseEnsure = Callable[[str, Sequence[type], Path | None], None] +PluginDatabaseRelease = Callable[[str], None] +PluginDatabaseDestroy = Callable[[str], None] + + +def _ignore_ensure( + _plugin_id: str, + _models: Sequence[type], + _migrations: Path | None, +) -> None: + """组合根尚未装配时忽略插件建库。""" + + +def _ignore_release(_plugin_id: str) -> None: + """组合根尚未装配时忽略插件数据库释放。""" + + +def _ignore_destroy(_plugin_id: str) -> None: + """组合根尚未装配时忽略插件数据库销毁。""" + + +class PluginDatabase: + """封装插件自有数据库的建立、释放与销毁能力。""" + + def __init__( + self, + *, + ensure: PluginDatabaseEnsure = _ignore_ensure, + release: PluginDatabaseRelease = _ignore_release, + destroy: PluginDatabaseDestroy = _ignore_destroy, + ) -> None: + """保存由启动组合根提供的建库、释放与销毁函数。""" + self._ensure = ensure + self._release = release + self._destroy = destroy + + def ensure( + self, + plugin_id: str, + models: Sequence[type], + migrations: Path | None, + ) -> None: + """按插件声明建立自有数据库，两项声明都为空时不建库。""" + self._ensure(plugin_id, models, migrations) + + def release(self, plugin_id: str) -> None: + """释放插件自有数据库的连接，保留数据。""" + self._release(plugin_id) + + def destroy(self, plugin_id: str) -> None: + """销毁插件自有数据库，仅限删除插件数据的路径调用。""" + self._destroy(plugin_id) + + +_plugin_database = PluginDatabase() + + +def configure_plugin_database(database: PluginDatabase) -> None: + """由启动组合根替换插件自有数据库实现。""" + global _plugin_database + _plugin_database = database + + +def get_plugin_database() -> PluginDatabase: + """返回当前插件自有数据库端口。""" + return _plugin_database diff --git a/app/runtime/extensions/plugin/lifecycle.py b/app/runtime/extensions/plugin/lifecycle.py index 2f9192034..ebd4e8748 100644 --- a/app/runtime/extensions/plugin/lifecycle.py +++ b/app/runtime/extensions/plugin/lifecycle.py @@ -5,10 +5,12 @@ from __future__ import annotations import traceback from collections.abc import Callable from functools import wraps +from pathlib import Path import threading import time from typing import Any, Optional, ParamSpec, TypeVar, cast +from app.runtime.extensions.plugin.database import PluginDatabase from app.runtime.observability import record_metric from app.schemas.plugin import PluginRuntimeStatus @@ -69,10 +71,11 @@ class PluginLifecycle: enable_events: Callable[[Any], None], disable_events: Callable[[Any], None], runtime_status_writer: Callable[[str, PluginRuntimeStatus], None], + database: Callable[[], PluginDatabase], log: Any, event_sender: Callable[..., Any], ) -> None: - """保存注册表、加载器和事件端口。""" + """保存注册表、加载器、数据库和事件端口。""" self._classes = classes self._running = running self._load_plugins = load_plugins @@ -84,6 +87,7 @@ class PluginLifecycle: self._enable_events = enable_events self._disable_events = disable_events self._runtime_status_writer = runtime_status_writer + self._database = database self._logger = log self._event_sender = event_sender self._lifecycle_lock = threading.RLock() @@ -121,6 +125,7 @@ class PluginLifecycle: self._classes[current_id] = plugin instance = plugin() instance.init_plugin(self._plugin_config(current_id)) + self._ensure_database(current_id, instance) self._quiesced_hooks.pop(current_id, None) self._running[current_id] = instance self._logger.info( @@ -137,6 +142,9 @@ class PluginLifecycle: status = PluginRuntimeStatus.LOAD_FAILED self._runtime_status_writer(plugin_id or current_id, status) results[plugin_id or current_id] = status + # 建库发生在进入运行态之前：失败的插件不会出现在 _running 里，卸载路径 + # 因此够不到它，句柄只能在这里释放 + self._release_databases((current_id,)) self._logger.error( f"加载插件 {current_id} 出错：{error} - {traceback.format_exc()}" ) @@ -150,6 +158,32 @@ class PluginLifecycle: self._clear_tools() return results + @staticmethod + def _declaration(instance: Any, hook_name: str) -> Any: + """读取插件的数据库声明钩子，未实现该钩子时视为未声明。""" + hook = getattr(instance, hook_name, None) + return hook() if callable(hook) else None + + def _ensure_database(self, plugin_id: str, instance: Any) -> None: + """按插件声明建立其自有数据库，两项声明都缺失时不建库。""" + migrations = self._declaration(instance, "get_database_migrations") + self._database().ensure( + plugin_id, + tuple(self._declaration(instance, "get_database_models") or ()), + Path(migrations) if migrations else None, + ) + + def _release_databases(self, plugin_ids: tuple[str, ...]) -> None: + """释放已卸载插件的自有数据库连接，单个失败不得阻断其余释放。""" + database = self._database() + for plugin_id in plugin_ids: + try: + database.release(plugin_id) + except Exception as error: # noqa: BLE001 释放故障不得阻断卸载 + self._logger.warning( + f"释放插件 {plugin_id} 的数据库连接时发生错误: {error}" + ) + @observe_plugin_lifecycle("initialize") def initialize(self, plugin_id: str, config: dict) -> None: """重新应用指定插件配置并刷新事件注册状态。""" @@ -293,10 +327,14 @@ class PluginLifecycle: self._classes.pop(runtime_id, None) self._running.pop(runtime_id, None) self._quiesced_hooks.pop(runtime_id, None) + self._release_databases((runtime_id,)) else: + # 启动中途失败的插件只登记在 _classes 里，漏掉它就漏掉它的连接池 + released_ids = tuple(dict.fromkeys((*self._running, *self._classes))) self._classes.clear() self._running.clear() self._quiesced_hooks.clear() + self._release_databases(released_ids) self._logger.info("插件停止完成") return True diff --git a/app/runtime/extensions/plugin/runtime.py b/app/runtime/extensions/plugin/runtime.py index 6c5967945..c30d53c53 100644 --- a/app/runtime/extensions/plugin/runtime.py +++ b/app/runtime/extensions/plugin/runtime.py @@ -14,6 +14,7 @@ from app.runtime.extensions.plugin.access import PluginAccessPolicy from app.runtime.extensions.plugin.admission import PluginMutationAdmission from app.runtime.extensions.plugin.catalog import PluginCatalogFacade from app.runtime.extensions.plugin.clone import PluginCloneService +from app.runtime.extensions.plugin.database import PluginDatabase from app.runtime.extensions.plugin.dependency import PluginDependencyService from app.runtime.extensions.plugin.lifecycle import PluginLifecycle from app.runtime.extensions.plugin.loader import PluginLoader @@ -87,6 +88,7 @@ class PluginRuntimeEnvironment: plugins_root: Path storage: Callable[[], PluginStorage] system: Callable[[], PluginSystemServices] + database: Callable[[], PluginDatabase] catalog_factory: PluginCatalogFactory import_preparer: PluginImportService import_scanner: PluginImportService @@ -132,6 +134,7 @@ def build_plugin_runtime( instances = PluginInstanceStore(storage=environment.storage) configs = PluginConfigStore( storage=environment.storage, + database=environment.database, plugin_exists=lambda plugin_id: bool(registry.classes.get(plugin_id)), ) access = PluginAccessPolicy( @@ -177,6 +180,7 @@ def build_plugin_runtime( enable_events=eventmanager.enable_event_handler, disable_events=eventmanager.disable_event_handler, runtime_status_writer=registry.set_runtime_status, + database=environment.database, log=environment.logger, event_sender=eventmanager.send_event, ) diff --git a/app/runtime/extensions/plugin/storage.py b/app/runtime/extensions/plugin/storage.py index c93579520..94f66ec74 100644 --- a/app/runtime/extensions/plugin/storage.py +++ b/app/runtime/extensions/plugin/storage.py @@ -7,6 +7,7 @@ from typing import Any from pydantic import ValidationError +from app.runtime.extensions.plugin.database import PluginDatabase from app.schemas.plugin import PluginInstance from app.schemas.types import SystemConfigKey @@ -88,11 +89,13 @@ class PluginConfigStore: self, *, storage: Callable[[], "PluginStorage"], + database: Callable[[], PluginDatabase], plugin_exists: PluginExists, key_prefix: str = "plugin.%s", ) -> None: - """保存持久化端口和运行态插件查询端口。""" + """保存持久化端口、自有数据库端口和运行态插件查询端口。""" self._storage = storage + self._database = database self._plugin_exists = plugin_exists self._key_prefix = key_prefix @@ -137,10 +140,11 @@ class PluginConfigStore: return self._storage().delete(self._key(plugin_id)) def delete_data(self, plugin_id: str, force: bool = False) -> bool: - """删除插件业务数据并保持旧的布尔结果合同。""" + """删除插件业务数据与自有数据库，并保持旧的布尔结果合同。""" if not force and not self._plugin_exists(plugin_id): return False self._storage().delete_data(plugin_id) + self._database().destroy(plugin_id) return True diff --git a/app/sdk/database.py b/app/sdk/database.py index a74da2f78..0cee48f42 100644 --- a/app/sdk/database.py +++ b/app/sdk/database.py @@ -1,13 +1,17 @@ -"""插件可使用的数据库备份只读门面。""" +"""插件可使用的数据库备份只读门面和自有数据库入口。""" from app.application.backup import BackupArtifact, BackupVerification from app.application.database import get_database_governance as _get_database_governance +from app.db.plugin.base import plugin_declarative_base +from app.db.plugin.container import PluginDatabaseHandle __all__ = [ "BackupArtifact", "BackupVerification", + "PluginDatabaseHandle", "create_backup", "list_backups", + "plugin_declarative_base", "verify_backup", ] diff --git a/app/startup/initializers/plugins.py b/app/startup/initializers/plugins.py index 684161a2b..dab542d60 100644 --- a/app/startup/initializers/plugins.py +++ b/app/startup/initializers/plugins.py @@ -78,6 +78,11 @@ from app.application.plugin.transaction import ( from app.application.scheduling import update_plugin_job from app.application.site.sites import SitesHelper # pylint: disable=import-error,no-name-in-module from app.db.oper.plugindata import PluginDataOper +from app.db.plugin.registry import ( + destroy_database, + ensure_database, + release_database, +) from app.db.session import SessionFactory from app.db.uow import SqlAlchemyUnitOfWork from app.foundation.version import compare_version @@ -88,6 +93,11 @@ from app.runtime.compat.diagnostics import ( ) from app.runtime.compat.resources import scan_plugin_resource_imports from app.runtime.execution import run_in_threadpool_to_completion +from app.runtime.extensions.plugin.database import ( + PluginDatabase, + configure_plugin_database, + get_plugin_database, +) from app.runtime.extensions.plugin.dependency import PluginDependencyInstallResult from app.runtime.extensions.plugin.manager import ( PluginManager, @@ -145,6 +155,15 @@ def _delete_plugin_data(plugin_id: str) -> None: session.close() +def _build_plugin_database() -> PluginDatabase: + """把插件自有数据库端口装配到 db 层的建库、释放与销毁实现。""" + return PluginDatabase( + ensure=ensure_database, + release=release_database, + destroy=destroy_database, + ) + + def _prepare_legacy_plugin_import(*, plugin_id: str, plugin_dir: Path) -> None: """在执行旧插件顶层代码前准备其静态导入所需的宿主资源。""" for capability_id in scan_plugin_resource_imports(plugin_id, plugin_dir): @@ -162,6 +181,7 @@ def build_plugin_runtime_graph(host: PluginRuntimeHost) -> PluginRuntime: plugins_root=Path(get_runtime_setting('ROOT_PATH')) / "app" / "plugins", storage=lambda: get_plugin_storage(), system=lambda: get_plugin_system(), + database=lambda: get_plugin_database(), catalog_factory=lambda mapper: _build_plugin_catalog(mapper), import_preparer=_prepare_legacy_plugin_import, import_scanner=scan_plugin_legacy_imports, @@ -397,6 +417,7 @@ def configure_plugin_services() -> None: delete=lambda key: get_configured_system_config().delete(key), delete_data=_delete_plugin_data, )) + configure_plugin_database(_build_plugin_database()) def _register_plugin_runtime(plugin_id: str) -> None: diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index c980b6e70..9cc603ec2 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -754,8 +754,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 937 | -| 内部导入边 | 7,842 | +| Python 模块 | 944 | +| 内部导入边 | 7,873 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 56efc28bb..52b904a04 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 937 / 7,842 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 944 / 7,873 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/docs/rules/10-data-and-persistent.md b/docs/rules/10-data-and-persistent.md index ca98266be..34016f8ae 100644 --- a/docs/rules/10-data-and-persistent.md +++ b/docs/rules/10-data-and-persistent.md @@ -242,6 +242,70 @@ oper.delete(sid=1) # Delete by key --- +## Plugin-Owned Databases + +Plugins that need SQL storage beyond `save_data`/`get_data` own an isolated +database rather than a table inside the host database. The framework lives in +`app/db/plugin/`: SQLite deploys one file per plugin at +`PLUGIN_DATA_PATH//plugin.db`; PostgreSQL deploys one schema that +reuses the host engine via `schema_translate_map`. The schema name is +`plugin_` when the plugin id already consists of lowercase ASCII +alphanumerics and underscores and the result fits in 63 bytes; otherwise the +sanitized name carries an 8-hex-character suffix derived from the plugin id, +so `My-Plugin`, `My_Plugin` and `my_plugin` never share one schema — uninstall +issues `DROP SCHEMA ... CASCADE` and a shared name would delete another +plugin's data. + +A plugin declares its schema with `get_database_models()` and/or +`get_database_migrations()`. The host pulls both hooks once, right after +`init_plugin()` returns, and creates nothing when both are empty — most +plugins never pay for this framework. A declared migrations directory takes +precedence over declared models and is applied with `alembic upgrade head`; +it must be an absolute, existing directory — `ensure` raises +`FileNotFoundError` before creating anything rather than leaving behind a +database with neither tables nor a version stamp. + +Models must inherit `app.sdk.database.plugin_declarative_base()`, which mints +a fresh `MetaData` per call so a plugin's tables never collide with +`app.db.base.Base.metadata` or with another plugin's same-named tables. At +runtime, `_PluginBase.get_database()` returns a `PluginDatabaseHandle` for +opening sessions against the plugin's own engine. Declaring nothing is not the +same as never having a database: `get_database()` creates the SQLite file (or +the PostgreSQL schema) on first call, so a plugin that only runs raw SQL still +gets an isolated database. Under PostgreSQL the handle's sessions and +connections issue `SET LOCAL search_path` at the start of every transaction, +so unqualified raw SQL resolves to the plugin's own schema rather than +`public`; `SET LOCAL` ends with the transaction and never leaks back to the +host through the shared pool. + +Lifecycle is strictly ensure/release/destroy: plugin start calls `ensure` +after `init_plugin()`; stop, reload and remove call `release` only, which +closes the handle's thread-local sessions and never touches data. `release` +disposes the connection pool only under SQLite, where the handle owns its +engine; under PostgreSQL the handle is a view over the host engine, and +disposing it would take the host and every other plugin down with it. Only +resetting a plugin's data or uninstalling a clone/virtual instance calls +`destroy` (delete the SQLite file and its `-wal`/`-shm` sidecars, or `DROP +SCHEMA ... CASCADE`). Stopping or uninstalling an ordinary plugin never +destroys its database, mirroring the existing `plugindata` retention +semantics. + +Uninstall stops the plugin before deleting anything: a stop hook that calls +`get_database()` would otherwise recreate the database that was just +destroyed. Because stopping also unregisters the plugin class, the deletions +that follow are issued with `force=True`. + +`db_query` / `db_update` keep their existing automatic-Session fallback bound +to the **host** `ScopedSession()`; they are not aware of plugin-owned +databases. A plugin operating on its own tables must pass +`handle.session()` explicitly — the decorators accept any `Session` argument +supplied by the caller. + +Plugin databases never participate in host Alembic (`database/versions/`) +and their tables never register on `app.db.base.Base.metadata`. + +--- + ## SystemConfig — Runtime Configuration **Purpose:** Runtime business configuration that is user-editable, persisted in the database, and survives application restarts. @@ -362,4 +426,4 @@ can be accepted only once without Application knowing the configured backend. - `settings.API_TOKEN` and other secret fields must not be included in log output or API responses. - The `config list --show-secrets` flag exists specifically to gate secret visibility in the CLI. -*Last Updated: 2026-08-28* +*Last Updated: 2026-09-02* diff --git a/tests/conftest.py b/tests/conftest.py index a368f9224..03f2bca31 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -210,6 +210,7 @@ def configure_plugin_system_services(): from app.runtime.extensions.module.dispatcher import ModuleInvocationDispatcher from app.runtime.extensions.module.manager import ModuleManager from app.runtime.extensions.plugin import manager as plugin_manager_module + from app.runtime.extensions.plugin.database import get_plugin_database from app.runtime.extensions.plugin.manager import ( PluginManager, reset_plugin_runtime_factory, @@ -234,6 +235,7 @@ def configure_plugin_system_services(): plugins_root=settings.ROOT_PATH / "app" / "plugins", storage=get_plugin_storage, system=get_plugin_system, + database=get_plugin_database, catalog_factory=lambda mapper: ( plugin_manager_module._plugin_catalog_factory(mapper) ), diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 116d9743d..6e2f0d0bf 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 7842, - "edge_sha256": "e71d433eeedb7ce13873176f99a593806648c4850ce7db250f7874589bbcd92b", + "edge_count": 7873, + "edge_sha256": "0f2cb35f34e4c4f3e2546e4416595b15ebe804f23f453a40bdda98fda35f2c40", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -5547,8 +5547,24 @@ "app.db.oper.workflow -> app.db.models.workflow", "app.db.oper.workflow -> app.db.oper", "app.db.oper.workflow -> app.db.oper.query", + "app.db.plugin.locator -> app.runtime", + "app.db.plugin.locator -> app.runtime.settings", + "app.db.plugin.migration -> app.db", + "app.db.plugin.migration -> app.db.plugin", + "app.db.plugin.migration -> app.db.plugin.container", + "app.db.plugin.registry -> app.db", + "app.db.plugin.registry -> app.db.engine", + "app.db.plugin.registry -> app.db.plugin", + "app.db.plugin.registry -> app.db.plugin.container", + "app.db.plugin.registry -> app.db.plugin.locator", + "app.db.plugin.registry -> app.db.plugin.migration", + "app.db.plugin.registry -> app.runtime", + "app.db.plugin.registry -> app.runtime.log", + "app.db.plugin.registry -> app.runtime.settings", "app.db.session -> app.db", "app.db.session -> app.db.engine", + "app.db.session -> app.db.plugin", + "app.db.session -> app.db.plugin.registry", "app.db.session -> app.runtime", "app.db.session -> app.runtime.log", "app.db.session -> app.runtime.loop", @@ -7596,6 +7612,9 @@ "app.runtime.extensions.plugin.dependency -> app.schemas", "app.runtime.extensions.plugin.dependency -> app.schemas.plugin", "app.runtime.extensions.plugin.lifecycle -> app.runtime", + "app.runtime.extensions.plugin.lifecycle -> app.runtime.extensions", + "app.runtime.extensions.plugin.lifecycle -> app.runtime.extensions.plugin", + "app.runtime.extensions.plugin.lifecycle -> app.runtime.extensions.plugin.database", "app.runtime.extensions.plugin.lifecycle -> app.runtime.observability", "app.runtime.extensions.plugin.lifecycle -> app.schemas", "app.runtime.extensions.plugin.lifecycle -> app.schemas.plugin", @@ -7659,6 +7678,7 @@ "app.runtime.extensions.plugin.runtime -> app.runtime.extensions.plugin.admission", "app.runtime.extensions.plugin.runtime -> app.runtime.extensions.plugin.catalog", "app.runtime.extensions.plugin.runtime -> app.runtime.extensions.plugin.clone", + "app.runtime.extensions.plugin.runtime -> app.runtime.extensions.plugin.database", "app.runtime.extensions.plugin.runtime -> app.runtime.extensions.plugin.dependency", "app.runtime.extensions.plugin.runtime -> app.runtime.extensions.plugin.lifecycle", "app.runtime.extensions.plugin.runtime -> app.runtime.extensions.plugin.loader", @@ -7673,6 +7693,10 @@ "app.runtime.extensions.plugin.runtime -> app.runtime.extensions.plugin.tools", "app.runtime.extensions.plugin.runtime -> app.schemas", "app.runtime.extensions.plugin.runtime -> app.schemas.types", + "app.runtime.extensions.plugin.storage -> app.runtime", + "app.runtime.extensions.plugin.storage -> app.runtime.extensions", + "app.runtime.extensions.plugin.storage -> app.runtime.extensions.plugin", + "app.runtime.extensions.plugin.storage -> app.runtime.extensions.plugin.database", "app.runtime.extensions.plugin.storage -> app.schemas", "app.runtime.extensions.plugin.storage -> app.schemas.plugin", "app.runtime.extensions.plugin.storage -> app.schemas.types", @@ -7992,6 +8016,10 @@ "app.sdk.database -> app.application", "app.sdk.database -> app.application.backup", "app.sdk.database -> app.application.database", + "app.sdk.database -> app.db", + "app.sdk.database -> app.db.plugin", + "app.sdk.database -> app.db.plugin.base", + "app.sdk.database -> app.db.plugin.container", "app.sdk.events -> app.runtime", "app.sdk.events -> app.runtime.event", "app.sdk.events -> app.runtime.event.snapshot", @@ -8629,6 +8657,8 @@ "app.startup.initializers.plugins -> app.db", "app.startup.initializers.plugins -> app.db.oper", "app.startup.initializers.plugins -> app.db.oper.plugindata", + "app.startup.initializers.plugins -> app.db.plugin", + "app.startup.initializers.plugins -> app.db.plugin.registry", "app.startup.initializers.plugins -> app.db.session", "app.startup.initializers.plugins -> app.db.uow", "app.startup.initializers.plugins -> app.foundation", @@ -8641,6 +8671,7 @@ "app.startup.initializers.plugins -> app.runtime.execution", "app.startup.initializers.plugins -> app.runtime.extensions", "app.startup.initializers.plugins -> app.runtime.extensions.plugin", + "app.startup.initializers.plugins -> app.runtime.extensions.plugin.database", "app.startup.initializers.plugins -> app.runtime.extensions.plugin.dependency", "app.startup.initializers.plugins -> app.runtime.extensions.plugin.manager", "app.startup.initializers.plugins -> app.runtime.extensions.plugin.runtime", @@ -8935,7 +8966,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 937, + "module_count": 944, "modules": [ "app", "app.adapters", @@ -9439,6 +9470,12 @@ "app.db.oper.user", "app.db.oper.userconfig", "app.db.oper.workflow", + "app.db.plugin", + "app.db.plugin.base", + "app.db.plugin.container", + "app.db.plugin.locator", + "app.db.plugin.migration", + "app.db.plugin.registry", "app.db.session", "app.db.uow", "app.db.worker", @@ -9700,6 +9737,7 @@ "app.runtime.extensions.plugin.catalog", "app.runtime.extensions.plugin.clone", "app.runtime.extensions.plugin.contracts", + "app.runtime.extensions.plugin.database", "app.runtime.extensions.plugin.dependency", "app.runtime.extensions.plugin.lifecycle", "app.runtime.extensions.plugin.loader", diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index e4c47a74d..716161589 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -9810,6 +9810,11 @@ "name": "BackupVerification", "target": "app.application.backup.BackupVerification" }, + { + "kind": "import", + "name": "PluginDatabaseHandle", + "target": "app.db.plugin.container.PluginDatabaseHandle" + }, { "kind": "FunctionDef", "name": "create_backup", @@ -9820,6 +9825,11 @@ "name": "list_backups", "target": "" }, + { + "kind": "import", + "name": "plugin_declarative_base", + "target": "app.db.plugin.base.plugin_declarative_base" + }, { "kind": "FunctionDef", "name": "verify_backup", diff --git a/tests/fixtures/architecture/startup-performance-baseline.json b/tests/fixtures/architecture/startup-performance-baseline.json index 94153e728..cdfcc788f 100644 --- a/tests/fixtures/architecture/startup-performance-baseline.json +++ b/tests/fixtures/architecture/startup-performance-baseline.json @@ -6,7 +6,7 @@ "repeat": 3, "targets": { "app.startup.lifecycle": { - "loaded_app_module_count": 510, + "loaded_app_module_count": 515, "max_ms": 1115.407, "median_ms": 1106.606, "min_ms": 1098.166, @@ -17,7 +17,7 @@ ] }, "app.factory": { - "loaded_app_module_count": 522, + "loaded_app_module_count": 527, "max_ms": 1140.393, "median_ms": 1134.83, "min_ms": 1119.004, @@ -28,7 +28,7 @@ ] }, "app.main": { - "loaded_app_module_count": 524, + "loaded_app_module_count": 529, "max_ms": 1180.675, "median_ms": 1165.915, "min_ms": 1161.859, diff --git a/tests/test_db_plugin_framework.py b/tests/test_db_plugin_framework.py new file mode 100644 index 000000000..fd2ce9e50 --- /dev/null +++ b/tests/test_db_plugin_framework.py @@ -0,0 +1,775 @@ +"""插件自管理数据库框架：建表隔离、SQLite 生命周期与 PostgreSQL 所有权边界。""" + +from __future__ import annotations + +import asyncio +import threading +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +import pytest +from sqlalchemy import create_engine, event, text +from sqlalchemy import inspect as sa_inspect +from sqlalchemy.orm import Mapped, mapped_column, scoped_session, sessionmaker + +import app.db.engine as engine_module +import app.db.plugin.locator as locator_module +import app.db.plugin.migration as migration_module +import app.db.plugin.registry as registry_module +import app.db.session as session_module +from app.db.base import Base as HostBase +from app.db.plugin.base import plugin_declarative_base +from app.db.plugin.container import PluginDatabaseHandle +from app.db.plugin.locator import ( + SCHEMA_NAME_MAX_LENGTH, + plugin_schema_name, + sqlite_sidecar_paths, +) + + +@pytest.fixture(autouse=True) +def _isolate_plugin_databases(): + """快照插件数据库句柄，用例结束后释放残留句柄并还原快照。""" + handles = dict(registry_module._handles) + registry_module._handles.clear() + yield + for plugin_id in list(registry_module._handles): + registry_module.release_database(plugin_id) + registry_module._handles.clear() + registry_module._handles.update(handles) + + +def _raise_dispose() -> None: + """模拟连接池释放失败。""" + raise RuntimeError("dispose failed") + + +def _borrowed_engine_handle(engine: Any) -> PluginDatabaseHandle: + """ + 构造一个不拥有引擎的句柄，用于验证 PostgreSQL 分支上的连接路由。 + :param engine: 句柄借用的引擎 + :return: owns_engine 为假的数据库句柄 + """ + session_factory = sessionmaker(bind=engine) + return PluginDatabaseHandle( + plugin_id="demo", + engine=engine, + session_factory=session_factory, + scoped_session_factory=scoped_session(session_factory), + db_path=None, + schema="plugin_demo", + owns_engine=False, + ) + + +def _write_migration_directory(root: Path) -> Path: + """ + 写出一个最小可用的 Alembic script_location，只含一条建表迁移。 + :param root: 承载迁移目录的父目录 + :return: 迁移目录路径 + """ + directory = root / "migrations" + (directory / "versions").mkdir(parents=True) + (directory / "env.py").write_text( + """from alembic import context +from sqlalchemy import create_engine + + +def _run(connection): + context.configure(connection=connection, target_metadata=None) + with context.begin_transaction(): + context.run_migrations() + + +injected = context.config.attributes.get("connection") +if injected is not None: + _run(injected) +else: + engine = create_engine(context.config.get_main_option("sqlalchemy.url")) + with engine.connect() as connection: + _run(connection) + engine.dispose() +""", + encoding="utf-8", + ) + (directory / "versions" / "0001_create_notes.py").write_text( + '''"""建立插件自有表。""" + +import sqlalchemy as sa +from alembic import op + +revision = "0001" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade(): + """建出插件声明的表。""" + op.create_table("notes", sa.Column("id", sa.Integer(), primary_key=True)) + + +def downgrade(): + """回退建表。""" + op.drop_table("notes") +''', + encoding="utf-8", + ) + return directory + + +@pytest.fixture +def plugin_data_root(tmp_path, monkeypatch) -> Path: + """把插件数据库文件隔离到进程私有的临时目录。""" + root = tmp_path / "plugins" + monkeypatch.setattr( + locator_module, + "get_runtime_setting", + lambda key, default=None: root if key == "PLUGIN_DATA_PATH" else default, + ) + return root + + +@pytest.fixture +def sqlite_backend(monkeypatch): + """把宿主数据库类型固定为 SQLite。""" + monkeypatch.setattr( + registry_module, + "get_runtime_setting", + lambda key, default=None: "sqlite" if key == "DB_TYPE" else default, + ) + + +@pytest.fixture +def postgresql_backend(monkeypatch): + """把宿主数据库类型固定为 PostgreSQL，并用替身覆盖宿主引擎。""" + host_engine = MagicMock(name="host_engine") + # 派生引擎必须是真实引擎：注册 begin 监听器要求一个真实的 SQLAlchemy 事件目标， + # 替身对象会被 event.listen 拒绝 + borrowed_engine = create_engine("sqlite://") + derived_engine = borrowed_engine.execution_options() + host_engine.execution_options.return_value = derived_engine + monkeypatch.setattr( + registry_module, + "get_runtime_setting", + lambda key, default=None: "postgresql" if key == "DB_TYPE" else default, + ) + monkeypatch.setattr(registry_module, "get_engine", lambda: host_engine) + yield host_engine, derived_engine + borrowed_engine.dispose() + + +def test_plugin_declarative_base_returns_a_fresh_metadata_per_call(): + """两次调用互不共享 MetaData，两个基类上都能定义同名表而不抛错。""" + base_a = plugin_declarative_base() + base_b = plugin_declarative_base() + assert base_a.metadata is not base_b.metadata + + class ItemA(base_a): + __tablename__ = "items" + id: Mapped[int] = mapped_column(primary_key=True) + + class ItemB(base_b): + __tablename__ = "items" + id: Mapped[int] = mapped_column(primary_key=True) + + assert "items" in base_a.metadata.tables + assert "items" in base_b.metadata.tables + + +def test_declared_models_create_tables_in_the_plugin_own_database(plugin_data_root, sqlite_backend): + """按声明的模型建表，数据可插入读回，且不污染宿主 Base.metadata。""" + base = plugin_declarative_base() + + class Widget(base): + __tablename__ = "widgets" + id: Mapped[int] = mapped_column(primary_key=True) + name: Mapped[str] = mapped_column() + + registry_module.ensure_database("demo", (Widget,)) + handle = registry_module.get_database("demo") + session = handle.session() + try: + session.add(Widget(id=1, name="a")) + session.commit() + assert session.query(Widget).count() == 1 + finally: + session.close() + + assert "widgets" not in HostBase.metadata.tables + + +def test_two_plugins_may_declare_the_same_table_name(plugin_data_root, sqlite_backend): + """两个插件各自的同名表互不冲突，各自库文件不同，数据互不可见。""" + base_a = plugin_declarative_base() + base_b = plugin_declarative_base() + + class ItemA(base_a): + __tablename__ = "items" + id: Mapped[int] = mapped_column(primary_key=True) + label: Mapped[str] = mapped_column() + + class ItemB(base_b): + __tablename__ = "items" + id: Mapped[int] = mapped_column(primary_key=True) + note: Mapped[str] = mapped_column() + + registry_module.ensure_database("plugin_a", (ItemA,)) + registry_module.ensure_database("plugin_b", (ItemB,)) + handle_a = registry_module.get_database("plugin_a") + handle_b = registry_module.get_database("plugin_b") + assert handle_a.db_path != handle_b.db_path + + session_a = handle_a.session() + session_b = handle_b.session() + try: + session_a.add(ItemA(id=1, label="x")) + session_a.commit() + session_b.add(ItemB(id=1, note="y")) + session_b.commit() + assert session_a.query(ItemA).count() == 1 + assert session_b.query(ItemB).count() == 1 + finally: + session_a.close() + session_b.close() + + +def test_only_declared_tables_are_created(plugin_data_root, sqlite_backend): + """只建声明的表，未声明的同基类模型不会被创建。""" + base = plugin_declarative_base() + + class Declared(base): + __tablename__ = "declared" + id: Mapped[int] = mapped_column(primary_key=True) + + class Undeclared(base): + __tablename__ = "undeclared" + id: Mapped[int] = mapped_column(primary_key=True) + + registry_module.ensure_database("demo", (Declared,)) + handle = registry_module.get_database("demo") + tables = set(sa_inspect(handle.engine).get_table_names()) + assert tables == {"declared"} + + +def test_sqlite_database_lands_in_the_plugin_data_directory(plugin_data_root, sqlite_backend): + """SQLite 库文件落在插件数据目录下，句柄独占引擎，不带 schema。""" + handle = registry_module.get_database("demo") + assert handle.db_path == plugin_data_root / "demo" / "plugin.db" + assert handle.owns_engine is True + assert handle.schema is None + assert handle.db_path.exists() + + +def test_undeclared_plugin_creates_no_database_file(plugin_data_root, sqlite_backend): + """两项声明都为空时不建句柄、不落盘。""" + registry_module.ensure_database("demo") + assert "demo" not in registry_module._handles + assert not (plugin_data_root / "demo").exists() + + +def test_release_disposes_the_owned_sqlite_engine(plugin_data_root, sqlite_backend, monkeypatch): + """release 只 dispose 句柄独占的引擎，并把句柄移出注册表。""" + handle = registry_module.get_database("demo") + calls: list[str] = [] + monkeypatch.setattr(handle.engine, "dispose", lambda: calls.append("disposed")) + registry_module.release_database("demo") + assert calls == ["disposed"] + assert "demo" not in registry_module._handles + + +def test_release_all_disposes_every_plugin_database(plugin_data_root, sqlite_backend, monkeypatch): + """release_all 释放全部插件的数据库连接，注册表清空。""" + handle_a = registry_module.get_database("plugin_a") + handle_b = registry_module.get_database("plugin_b") + calls: list[str] = [] + monkeypatch.setattr(handle_a.engine, "dispose", lambda: calls.append("a")) + monkeypatch.setattr(handle_b.engine, "dispose", lambda: calls.append("b")) + registry_module.release_all_databases() + assert set(calls) == {"a", "b"} + assert registry_module._handles == {} + + +def test_destroy_removes_the_database_file_and_sidecars(plugin_data_root, sqlite_backend): + """destroy 删除库文件及其 -wal/-shm 边车文件，并移出注册表。""" + handle = registry_module.get_database("demo") + for sidecar in sqlite_sidecar_paths(handle.db_path): + sidecar.write_text("") + registry_module.destroy_database("demo") + assert not handle.db_path.exists() + for sidecar in sqlite_sidecar_paths(handle.db_path): + assert not sidecar.exists() + assert "demo" not in registry_module._handles + + +def test_destroy_after_release_still_removes_the_database_file(plugin_data_root, sqlite_backend): + """先 release 再 destroy 时库文件仍会被删除，覆盖 reset 流程的真实调用顺序。""" + handle = registry_module.get_database("demo") + db_path = handle.db_path + registry_module.release_database("demo") + assert db_path.exists() + registry_module.destroy_database("demo") + assert not db_path.exists() + + +def test_destroy_never_raises_when_the_file_cannot_be_removed(plugin_data_root, sqlite_backend, monkeypatch): + """删除失败只记日志，不得向上抛出异常。""" + registry_module.get_database("demo") + + def _raise_unlink(self, missing_ok=False): + """模拟文件系统拒绝删除。""" + raise OSError("permission denied") + + monkeypatch.setattr(Path, "unlink", _raise_unlink) + registry_module.destroy_database("demo") + + +def test_postgresql_handle_does_not_own_the_host_engine(postgresql_backend): + """PostgreSQL 下句柄只是宿主引擎按 schema 派生的外观，不拥有它。""" + host_engine, derived_engine = postgresql_backend + handle = registry_module.get_database("demo") + assert handle.owns_engine is False + assert handle.engine is derived_engine + assert handle.db_path is None + host_engine.execution_options.assert_called_once_with( + schema_translate_map={None: handle.schema} + ) + + +def test_postgresql_release_never_disposes_the_host_engine(postgresql_backend, monkeypatch): + """release 在 PostgreSQL 下不得 dispose 派生引擎，也不得触碰宿主引擎。""" + host_engine, derived_engine = postgresql_backend + disposed: list[str] = [] + monkeypatch.setattr(derived_engine, "dispose", lambda: disposed.append("derived")) + registry_module.get_database("demo") + registry_module.release_database("demo") + assert disposed == [] + host_engine.dispose.assert_not_called() + + +def test_postgresql_destroy_drops_the_schema_and_keeps_the_host_engine( + postgresql_backend, + monkeypatch, +): + """destroy 在 PostgreSQL 下丢弃对应 schema，且不 dispose 任何引擎。""" + host_engine, derived_engine = postgresql_backend + disposed: list[str] = [] + monkeypatch.setattr(derived_engine, "dispose", lambda: disposed.append("derived")) + handle = registry_module.get_database("demo") + schema = handle.schema + registry_module.destroy_database("demo") + connection = host_engine.begin.return_value.__enter__.return_value + executed = [str(call.args[0]) for call in connection.execute.call_args_list] + assert any("DROP SCHEMA" in sql and schema in sql for sql in executed) + assert disposed == [] + host_engine.dispose.assert_not_called() + + +def test_declared_migrations_take_precedence_over_models( + plugin_data_root, + sqlite_backend, + monkeypatch, + tmp_path, +): + """同时声明模型与迁移目录时优先走 alembic，且模型表不会被建出。""" + calls: list[tuple[str, Path]] = [] + monkeypatch.setattr( + migration_module, + "run_migrations", + lambda handle, directory: calls.append((handle.plugin_id, directory)), + ) + base = plugin_declarative_base() + + class Widget(base): + __tablename__ = "widgets" + id: Mapped[int] = mapped_column(primary_key=True) + + directory = _write_migration_directory(tmp_path) + registry_module.ensure_database("demo", (Widget,), directory) + assert calls == [("demo", directory)] + handle = registry_module.get_database("demo") + assert "widgets" not in sa_inspect(handle.engine).get_table_names() + + +def test_plugin_schema_name_sanitizes_the_plugin_id(): + """schema 名只保留小写字母、数字与下划线，被改写过的标识再带上区分哈希。""" + schema = plugin_schema_name("Demo-Plugin.v2") + assert schema.startswith("plugin_demo_plugin_v2_") + assert schema.removeprefix("plugin_demo_plugin_v2_").isalnum() + + +def test_plugin_schema_name_keeps_an_already_legal_plugin_id_verbatim(): + """标识本身已是合法 schema 片段时不追加哈希。""" + assert plugin_schema_name("my_plugin") == "plugin_my_plugin" + + +def test_plugin_schema_name_separates_ids_that_normalize_to_the_same_text(): + """归一后同名的三个插件标识各自拿到不同 schema，卸载互不波及。""" + schemas = { + plugin_schema_name(plugin_id) + for plugin_id in ("My-Plugin", "My_Plugin", "my_plugin") + } + assert len(schemas) == 3 + + +def test_plugin_schema_name_fits_the_postgresql_identifier_limit(): + """超长插件标识被截断到 PostgreSQL 标识符上限以内，且仍带区分哈希。""" + schema = plugin_schema_name("p" * 200) + assert len(schema.encode("utf-8")) <= SCHEMA_NAME_MAX_LENGTH + assert plugin_schema_name("p" * 200) != plugin_schema_name("p" * 201) + + +def test_sqlite_sidecar_paths_cover_wal_and_shm(): + """边车路径覆盖 -wal 与 -shm 两个后缀。""" + db_path = Path("/tmp/plugin.db") + assert sqlite_sidecar_paths(db_path) == ( + Path("/tmp/plugin.db-wal"), + Path("/tmp/plugin.db-shm"), + ) + + +def test_release_all_isolates_a_failing_plugin_dispose(plugin_data_root, sqlite_backend, monkeypatch): + """一个插件的连接池释放抛错，其余插件仍被释放，异常不向上传播。""" + handle_a = registry_module.get_database("plugin_a") + handle_b = registry_module.get_database("plugin_b") + calls: list[str] = [] + monkeypatch.setattr(handle_a.engine, "dispose", _raise_dispose) + monkeypatch.setattr(handle_b.engine, "dispose", lambda: calls.append("b")) + + registry_module.release_all_databases() + + assert calls == ["b"] + assert registry_module._handles == {} + + +def test_close_database_disposes_the_host_engine_when_a_plugin_dispose_fails( + plugin_data_root, + sqlite_backend, + monkeypatch, +): + """插件连接池释放抛错时，宿主同步引擎仍然被释放。""" + handle = registry_module.get_database("demo") + monkeypatch.setattr(handle.engine, "dispose", _raise_dispose) + disposed: list[str] = [] + host_engine = MagicMock(name="host_sync_engine") + host_engine.dispose.side_effect = lambda: disposed.append("host") + monkeypatch.setattr(engine_module, "peek_sync_engine", lambda: host_engine) + monkeypatch.setattr(engine_module, "peek_async_engine", lambda: None) + monkeypatch.setattr(session_module, "_pooled_async_engines", {}) + + asyncio.run(session_module.close_database()) + + assert disposed == ["host"] + + +def test_destroy_removes_the_database_file_even_when_dispose_fails( + plugin_data_root, + sqlite_backend, + monkeypatch, +): + """连接池释放抛错不得中断销毁，库文件照样被删除。""" + handle = registry_module.get_database("demo") + monkeypatch.setattr(handle.engine, "dispose", _raise_dispose) + + registry_module.destroy_database("demo") + + assert not handle.db_path.exists() + assert "demo" not in registry_module._handles + + +def test_single_table_inheritance_creates_the_shared_table_once(plugin_data_root, sqlite_backend): + """单表继承的父子类共享同一张表，一并声明只建一次，重复建库幂等。""" + base = plugin_declarative_base() + + class Node(base): + __tablename__ = "nodes" + __mapper_args__ = {"polymorphic_on": "kind", "polymorphic_identity": "node"} + id: Mapped[int] = mapped_column(primary_key=True) + kind: Mapped[str] = mapped_column() + + class Leaf(Node): + __mapper_args__ = {"polymorphic_identity": "leaf"} + + assert Leaf.__table__ is Node.__table__ + + registry_module.ensure_database("demo", (Node, Leaf)) + registry_module.ensure_database("demo", (Node, Leaf)) + + handle = registry_module.get_database("demo") + assert set(sa_inspect(handle.engine).get_table_names()) == {"nodes"} + + +def test_release_closes_the_thread_local_session(plugin_data_root, sqlite_backend): + """release 先清掉线程局部会话，句柄不再扣着已 dispose 引擎上的连接。""" + handle = registry_module.get_database("demo") + session = handle.scoped_session() + session.execute(text("SELECT 1")) + assert handle.scoped_session_factory.registry.has() is True + + registry_module.release_database("demo") + + assert handle.scoped_session_factory.registry.has() is False + + +def test_missing_migrations_directory_is_rejected_before_any_file_is_created( + plugin_data_root, + sqlite_backend, + tmp_path, +): + """迁移目录不存在时直接抛错，不留下空库文件，也不建插件数据目录。""" + with pytest.raises(FileNotFoundError): + registry_module.ensure_database("demo", (), tmp_path / "missing") + + assert "demo" not in registry_module._handles + assert not (plugin_data_root / "demo").exists() + + +def test_postgresql_handle_creates_the_plugin_schema(postgresql_backend): + """PostgreSQL 下建句柄先按插件 schema 执行 CREATE SCHEMA IF NOT EXISTS。""" + host_engine, _ = postgresql_backend + handle = registry_module.get_database("demo") + connection = host_engine.begin.return_value.__enter__.return_value + executed = [str(call.args[0]) for call in connection.execute.call_args_list] + assert f'CREATE SCHEMA IF NOT EXISTS "{handle.schema}"' in executed + + +def test_concurrent_get_database_builds_a_single_handle(plugin_data_root, sqlite_backend): + """八个线程同时取同一插件的句柄时只建出一个句柄，不会并存两份连接池。""" + thread_count = 8 + barrier = threading.Barrier(thread_count) + guard = threading.Lock() + handles: list[PluginDatabaseHandle] = [] + + def _acquire() -> None: + """在同一时刻取句柄并记录结果。""" + barrier.wait() + handle = registry_module.get_database("demo") + with guard: + handles.append(handle) + + threads = [threading.Thread(target=_acquire) for _ in range(thread_count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert len(handles) == thread_count + assert all(handle is handles[0] for handle in handles) + assert list(registry_module._handles) == ["demo"] + + +def test_get_database_rebuilds_a_released_handle(plugin_data_root, sqlite_backend): + """release 之后再取句柄会重建出一个可用的新句柄。""" + first = registry_module.get_database("demo") + registry_module.release_database("demo") + + second = registry_module.get_database("demo") + + assert second is not first + session = second.session() + try: + assert session.execute(text("SELECT 1")).scalar() == 1 + finally: + session.close() + + +def test_get_database_rebuilds_a_destroyed_handle(plugin_data_root, sqlite_backend): + """destroy 之后再取句柄会重建库文件与新句柄。""" + first = registry_module.get_database("demo") + registry_module.destroy_database("demo") + + second = registry_module.get_database("demo") + + assert second is not first + assert second.db_path.exists() + session = second.session() + try: + assert session.execute(text("SELECT 1")).scalar() == 1 + finally: + session.close() + + +def test_run_migrations_upgrades_a_sqlite_plugin_database(plugin_data_root, sqlite_backend, tmp_path): + """声明迁移目录时按 alembic 建表并写入版本号，重复建库幂等。""" + directory = _write_migration_directory(tmp_path) + + registry_module.ensure_database("demo", (), directory) + + handle = registry_module.get_database("demo") + tables = set(sa_inspect(handle.engine).get_table_names()) + assert {"notes", "alembic_version"} <= tables + + registry_module.ensure_database("demo", (), directory) + + assert set(sa_inspect(handle.engine).get_table_names()) == tables + + +def test_run_migrations_routes_the_postgresql_connection_through_the_handle( + monkeypatch, + tmp_path, +): + """PostgreSQL 下迁移复用句柄已限定 schema 的连接，并在结束后提交。""" + captured: dict[str, object] = {} + + def _record_upgrade(config, revision): + """记录 alembic 收到的连接与目标版本。""" + captured["connection"] = config.attributes.get("connection") + captured["revision"] = revision + + monkeypatch.setattr(migration_module, "upgrade", _record_upgrade) + handle = _borrowed_engine_handle(MagicMock(name="derived_engine")) + + migration_module.run_migrations(handle, _write_migration_directory(tmp_path)) + + connection = handle.engine.connect.return_value.__enter__.return_value + assert captured["connection"] is connection + assert captured["revision"] == "head" + connection.commit.assert_called_once() + + +def test_release_after_destroy_keeps_a_database_rebuilt_by_a_stop_hook( + plugin_data_root, + sqlite_backend, +): + """销毁后重新建出的库属于仍在运行的插件，普通停止只释放连接、不得再删一次。""" + registry_module.get_database("demo") + registry_module.destroy_database("demo") + + rebuilt = registry_module.get_database("demo") + session = rebuilt.session() + try: + session.execute(text("CREATE TABLE kept (id INTEGER PRIMARY KEY)")) + session.commit() + finally: + session.close() + + registry_module.release_database("demo") + + assert rebuilt.db_path.exists() + assert "demo" not in registry_module._handles + + +def test_destroy_blocks_a_concurrent_handle_rebuild_until_the_carrier_is_removed( + plugin_data_root, + sqlite_backend, + monkeypatch, +): + """销毁未删完载体前并发的取句柄取不到结果，新句柄因此不会指向被删掉的载体。""" + first = registry_module.get_database("demo") + entered_removal = threading.Event() + resume_removal = threading.Event() + original_remove_storage = registry_module._remove_storage + + def _blocking_remove_storage(plugin_id, handle): + """在删除载体前挂住销毁流程，制造并发窗口。""" + entered_removal.set() + assert resume_removal.wait(10) + original_remove_storage(plugin_id, handle) + + monkeypatch.setattr(registry_module, "_remove_storage", _blocking_remove_storage) + rebuilt: list[PluginDatabaseHandle] = [] + + def _rebuild() -> None: + """在销毁进行中重新取句柄。""" + rebuilt.append(registry_module.get_database("demo")) + + destroyer = threading.Thread(target=registry_module.destroy_database, args=("demo",)) + destroyer.start() + assert entered_removal.wait(10) + rebuilder = threading.Thread(target=_rebuild) + rebuilder.start() + rebuilder.join(0.5) + + assert rebuilder.is_alive() + assert rebuilt == [] + + resume_removal.set() + destroyer.join(10) + rebuilder.join(10) + + assert not rebuilder.is_alive() + assert rebuilt and rebuilt[0] is not first + assert rebuilt[0].db_path.exists() + + +def test_search_path_setter_binds_the_quoted_plugin_schema(): + """监听器在事务开始时执行 SET LOCAL search_path，schema 名带引号且回落 public。""" + connection = MagicMock(name="connection") + + registry_module._search_path_setter("plugin_demo")(connection) + + executed = connection.exec_driver_sql.call_args.args[0] + assert executed == 'SET LOCAL search_path TO "plugin_demo", public' + + +def test_begin_listener_on_a_derived_engine_never_reaches_the_host_engine(tmp_path): + """派生引擎上的 begin 监听器只对派生连接生效，宿主引擎的连接不触发。""" + host_engine = create_engine(f"sqlite:///{tmp_path / 'host.db'}") + derived_engine = host_engine.execution_options() + fired: list[str] = [] + + def _listener(connection) -> None: + """记录触发并在同一连接上执行一条无害语句，验证不会递归或报错。""" + fired.append("begin") + connection.exec_driver_sql("SELECT 1") + + event.listen(derived_engine, "begin", _listener) + + with derived_engine.connect() as connection: + connection.execute(text("SELECT 1")) + connection.commit() + assert fired == ["begin"] + + with host_engine.connect() as connection: + connection.execute(text("SELECT 1")) + connection.commit() + assert fired == ["begin"] + + session = sessionmaker(bind=derived_engine)() + try: + assert session.execute(text("SELECT 1")).scalar() == 1 + session.commit() + finally: + session.close() + assert fired == ["begin", "begin"] + + host_engine.dispose() + + +def test_postgresql_handle_binds_a_search_path_listener_to_the_derived_engine( + postgresql_backend, + monkeypatch, +): + """PostgreSQL 建句柄时把插件 schema 的监听器挂到派生引擎，而不是宿主引擎。""" + _host_engine, derived_engine = postgresql_backend + bound: list[tuple] = [] + build_listener = registry_module._search_path_setter + + def _record(schema: str): + """记录被绑定的 schema 与生成的监听器。""" + listener = build_listener(schema) + bound.append((schema, listener)) + return listener + + monkeypatch.setattr(registry_module, "_search_path_setter", _record) + + handle = registry_module.get_database("demo") + + assert [schema for schema, _ in bound] == [handle.schema] + assert event.contains(derived_engine, "begin", bound[0][1]) + + +def test_alembic_migrations_trigger_the_begin_event_on_a_borrowed_engine(tmp_path): + """借用引擎的迁移在连接上首次执行即触发 begin，search_path 监听器因此覆盖 alembic。""" + engine = create_engine(f"sqlite:///{tmp_path / 'host.db'}").execution_options() + begins: list[str] = [] + event.listen(engine, "begin", lambda _connection: begins.append("begin")) + handle = _borrowed_engine_handle(engine) + + migration_module.run_migrations(handle, _write_migration_directory(tmp_path)) + + assert begins + assert "notes" in set(sa_inspect(engine).get_table_names()) + + engine.dispose() diff --git a/tests/test_plugin_database_hooks.py b/tests/test_plugin_database_hooks.py new file mode 100644 index 000000000..94486c024 --- /dev/null +++ b/tests/test_plugin_database_hooks.py @@ -0,0 +1,172 @@ +"""_PluginBase 自有数据库钩子与插件 SDK 数据库入口。""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import pytest +from sqlalchemy import inspect as sa_inspect +from sqlalchemy.orm import Mapped, mapped_column + +import app.db.plugin.locator as locator_module +import app.db.plugin.registry as registry_module +from app.db.decorators import db_query, db_update +from app.db.plugin.base import plugin_declarative_base +from app.plugins import _PluginBase + + +class _SamplePlugin(_PluginBase): + """只实现最小生命周期合同的插件测试替身。""" + + plugin_name = "示例插件" + + def init_plugin(self, config: dict = None): + """接受宿主初始化配置。""" + + def get_state(self) -> bool: + """保持插件为启用状态。""" + return True + + def get_api(self) -> List[Dict[str, Any]]: + """不注册任何 API。""" + return [] + + def get_form(self) -> Tuple[Optional[List[dict]], Dict[str, Any]]: + """不提供配置页面。""" + return None, {} + + def get_page(self) -> Optional[List[dict]]: + """不提供详情页面。""" + return None + + def stop_service(self): + """无后台资源需要停止。""" + + +@pytest.fixture(autouse=True) +def _isolate_plugin_databases(): + """快照插件数据库句柄，用例结束后释放残留句柄并还原快照。""" + handles = dict(registry_module._handles) + registry_module._handles.clear() + yield + for plugin_id in list(registry_module._handles): + registry_module.release_database(plugin_id) + registry_module._handles.clear() + registry_module._handles.update(handles) + + +@pytest.fixture +def plugin_data_root(tmp_path, monkeypatch) -> Path: + """把插件数据库文件隔离到进程私有的临时目录。""" + root = tmp_path / "plugins" + monkeypatch.setattr( + locator_module, + "get_runtime_setting", + lambda key, default=None: root if key == "PLUGIN_DATA_PATH" else default, + ) + return root + + +def test_database_declaration_hooks_default_to_no_declaration(): + """未重写声明钩子的插件视为既未声明模型也未声明迁移目录。""" + plugin = _SamplePlugin() + assert plugin.get_database_models() is None + assert plugin.get_database_migrations() is None + + +def test_get_database_uses_the_plugin_class_name(plugin_data_root): + """未显式传入插件ID时，句柄按插件类名解析，与 get_data_path() 目录约定一致。""" + plugin = _SamplePlugin() + + handle = plugin.get_database() + + assert handle.plugin_id == "_SamplePlugin" + assert handle.db_path == plugin_data_root / "_SamplePlugin" / "plugin.db" + + +def test_get_database_honours_an_explicit_plugin_id(plugin_data_root): + """显式传入插件ID时按该ID解析句柄，不回退到类名。""" + plugin = _SamplePlugin() + + handle = plugin.get_database(plugin_id="OtherPlugin") + + assert handle.plugin_id == "OtherPlugin" + + +def test_clone_class_name_selects_its_own_database(plugin_data_root): + """分身的类名即插件运行时标识，句柄与源插件互不相同。""" + clone_cls = type("_SamplePluginwork", (_SamplePlugin,), {}) + clone = clone_cls() + source = _SamplePlugin() + + clone_handle = clone.get_database() + source_handle = source.get_database() + + assert clone_handle.plugin_id == "_SamplePluginwork" + assert clone_handle.db_path != source_handle.db_path + + +def test_declared_models_reach_the_framework_through_the_lifecycle_hook(plugin_data_root): + """宿主一次性拉取 get_database_models() 的返回值建表，端到端验证拉取式链路。""" + base = plugin_declarative_base() + + class Widget(base): + __tablename__ = "widgets" + id: Mapped[int] = mapped_column(primary_key=True) + + class _ModelPlugin(_SamplePlugin): + """声明一个模型的插件测试替身。""" + + def get_database_models(self): + """声明 Widget 模型。""" + return [Widget] + + plugin = _ModelPlugin() + registry_module.ensure_database("_ModelPlugin", plugin.get_database_models()) + + handle = plugin.get_database() + tables = set(sa_inspect(handle.engine).get_table_names()) + assert tables == {"widgets"} + + +def test_sdk_database_exports_the_plugin_database_contract(): + """SDK 门面复用 db 层的同一对象，不复制实现或制造第二套定义。""" + import app.db.plugin.base as base_module + import app.db.plugin.container as container_module + import app.sdk.database as sdk_database + + assert sdk_database.plugin_declarative_base is base_module.plugin_declarative_base + assert sdk_database.PluginDatabaseHandle is container_module.PluginDatabaseHandle + assert "plugin_declarative_base" in sdk_database.__all__ + assert "PluginDatabaseHandle" in sdk_database.__all__ + + +def test_host_transaction_decorators_accept_a_plugin_owned_session(plugin_data_root): + """db_update/db_query 的自动兜底不认识插件库，但显式传入的插件会话原样可用。""" + base = plugin_declarative_base() + + class Widget(base): + __tablename__ = "widgets" + id: Mapped[int] = mapped_column(primary_key=True) + name: Mapped[str] = mapped_column() + + registry_module.ensure_database("demo", (Widget,)) + handle = registry_module.get_database("demo") + + @db_update + def _write(db, widget): + """插入一条记录，提交由装饰器负责。""" + db.add(widget) + + @db_query + def _count(db): + """读取记录数。""" + return db.query(Widget).count() + + session = handle.session() + try: + _write(session, Widget(id=1, name="a")) + assert _count(session) == 1 + finally: + session.close() diff --git a/tests/test_plugin_database_lifecycle.py b/tests/test_plugin_database_lifecycle.py new file mode 100644 index 000000000..06749ae20 --- /dev/null +++ b/tests/test_plugin_database_lifecycle.py @@ -0,0 +1,574 @@ +"""插件自有数据库的运行时端口、生命周期时机与组合根接线。""" + +from __future__ import annotations + +import ast +import asyncio +import logging +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import app.api.endpoints.plugin as plugin_endpoint +import app.application.plugin.folders as plugin_folders +import app.application.plugin.management as plugin_management +import app.application.plugin.routes as plugin_routes +import app.application.scheduling as scheduling_module +import app.db.engine as engine_module +import app.db.session as session_module +import app.runtime.extensions.plugin.database as database_module +import app.startup.initializers.plugins as plugins_initializer +from app.runtime.extensions.plugin.database import PluginDatabase +from app.runtime.extensions.plugin.lifecycle import PluginLifecycle +from app.runtime.extensions.plugin.storage import PluginConfigStore, PluginStorage +from app.schemas.plugin import PluginInstance, PluginRuntimeStatus +from app.schemas.types import SystemConfigKey + + +class ModelA: + """充当声明模型占位符的哨兵类，仅用于校验实参透传。""" + + +@pytest.fixture +def restore_plugin_database_port(): + """还原全局插件数据库端口，避免测试间相互污染。""" + original = database_module.get_plugin_database() + yield + database_module.configure_plugin_database(original) + + +def _recording_database(calls: list[tuple]) -> PluginDatabase: + """构造把 ensure/release/destroy 三个操作依次记录到同一列表的插件数据库端口。""" + return PluginDatabase( + ensure=lambda plugin_id, models, migrations: calls.append( + ("ensure", plugin_id, models, migrations) + ), + release=lambda plugin_id: calls.append(("release", plugin_id)), + destroy=lambda plugin_id: calls.append(("destroy", plugin_id)), + ) + + +def _make_plugin_class( + name: str, + *, + models: tuple[type, ...] = (), + migrations: str | Path | None = None, + calls: list[tuple] | None = None, + declare_hooks: bool = True, +) -> type: + """ + 构造一个满足宿主插件最小生命周期合同的类，类名即插件运行时标识。 + + 分身在真实加载路径中被 ``PluginLoader._adapt_instance_class`` 改写 + ``__name__`` 为 ``instance_id``，因此这里直接用类名模拟分身身份，无需额外维度。 + :param name: 插件运行时标识（即 ``__name__``） + :param models: ``get_database_models()`` 的返回值 + :param migrations: ``get_database_migrations()`` 的返回值 + :param calls: 记录 ``init_plugin`` 调用的可选列表 + :param declare_hooks: 是否声明数据库钩子；为假时模拟未实现钩子的旧插件 + :return: 满足最小生命周期合同的插件类 + """ + + def init_plugin(self, _config): + """记录初始化调用，供顺序断言使用。""" + if calls is not None: + calls.append(("init", name)) + + namespace: dict[str, Any] = { + "plugin_name": name, + "plugin_version": "1.0.0", + "init_plugin": init_plugin, + "get_state": staticmethod(lambda: True), + "get_name": lambda self: name, + "close": lambda self: None, + "stop_service": lambda self: None, + } + if declare_hooks: + namespace["get_database_models"] = lambda self: list(models) + namespace["get_database_migrations"] = lambda self: migrations + return type(name, (), namespace) + + +def _build_lifecycle(**overrides: Any) -> PluginLifecycle: + """构造直接可用的生命周期实例，默认端口全为空操作，测试按需覆盖。""" + defaults: dict[str, Any] = dict( + classes={}, + running={}, + load_plugins=lambda _plugin_id, _installed, _check: [], + installed_plugins=lambda: [], + plugin_config=lambda _plugin_id: {}, + auth_checker=lambda _plugin: True, + clear_modules=lambda _plugin_id: None, + clear_tools=lambda: None, + enable_events=lambda _plugin: None, + disable_events=lambda _plugin: None, + runtime_status_writer=lambda _plugin_id, _status: None, + database=lambda: PluginDatabase(), + log=logging.getLogger(__name__), + event_sender=lambda *_args, **_kwargs: None, + ) + defaults.update(overrides) + return PluginLifecycle(**defaults) + + +def test_default_plugin_database_ports_are_safe_noops(restore_plugin_database_port): + """未装配组合根时，三个端口方法均是安全的 no-op。""" + database_module.configure_plugin_database(PluginDatabase()) + database = database_module.get_plugin_database() + assert database.ensure("demo", (), None) is None + assert database.release("demo") is None + assert database.destroy("demo") is None + + +def test_configure_plugin_database_replaces_the_process_port(restore_plugin_database_port): + """组合根替换端口后，三次调用按序进入替身记录的调用列表。""" + calls: list[tuple] = [] + database_module.configure_plugin_database(_recording_database(calls)) + + database = database_module.get_plugin_database() + database.ensure("demo", (), None) + database.release("demo") + database.destroy("demo") + + assert [call[0] for call in calls] == ["ensure", "release", "destroy"] + + +def test_start_ensures_the_database_after_init_plugin(): + """建库发生在 init_plugin 之后，插件桩把两者记到同一 calls 列表验证顺序。""" + calls: list[tuple] = [] + plugin_cls = _make_plugin_class("DemoPlugin", calls=calls) + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: _recording_database(calls), + ) + + lifecycle.start("DemoPlugin") + + kinds = [call[0] for call in calls] + assert kinds.index("init") < kinds.index("ensure") + + +def test_start_passes_the_declared_models_and_migration_directory(): + """声明的模型与字符串迁移目录原样透传给 ensure，字符串被转换为 Path。""" + calls: list[tuple] = [] + plugin_cls = _make_plugin_class("DemoPlugin", models=(ModelA,), migrations="m") + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: _recording_database(calls), + ) + + lifecycle.start("DemoPlugin") + + ensure_call = next(call for call in calls if call[0] == "ensure") + assert ensure_call == ("ensure", "DemoPlugin", (ModelA,), Path("m")) + + +def test_start_reports_empty_declarations_for_plugins_without_a_database(): + """未实现数据库钩子的插件仍触发 ensure，实参为空元组和 None。""" + calls: list[tuple] = [] + plugin_cls = _make_plugin_class("DemoPlugin", declare_hooks=False) + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: _recording_database(calls), + ) + + lifecycle.start("DemoPlugin") + + ensure_call = next(call for call in calls if call[0] == "ensure") + assert ensure_call == ("ensure", "DemoPlugin", (), None) + + +def test_plugin_failing_to_ensure_is_not_registered_as_running(): + """建库失败与 init_plugin 抛错同构：进入 load_failed，不进入运行态。""" + plugin_cls = _make_plugin_class("DemoPlugin") + + def _raise_ensure(_plugin_id, _models, _migrations): + """模拟建库失败。""" + raise RuntimeError("ensure failed") + + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: PluginDatabase(ensure=_raise_ensure), + ) + + result = lifecycle.start("DemoPlugin") + + assert result == {"DemoPlugin": PluginRuntimeStatus.LOAD_FAILED} + assert "DemoPlugin" not in lifecycle._running + + +def test_stop_releases_the_database_and_never_destroys_it(): + """停止单个插件只释放连接，不出现任何 destroy 调用。""" + calls: list[tuple] = [] + plugin_cls = _make_plugin_class("DemoPlugin") + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: _recording_database(calls), + ) + lifecycle.start("DemoPlugin") + + lifecycle.stop("DemoPlugin") + + assert ("release", "DemoPlugin") in calls + assert not any(call[0] == "destroy" for call in calls) + + +def test_stop_without_plugin_id_releases_every_running_plugin(): + """整体停止时，每个运行中的插件都收到一次 release。""" + calls: list[tuple] = [] + plugin_a = _make_plugin_class("PluginA") + plugin_b = _make_plugin_class("PluginB") + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_a, plugin_b], + installed_plugins=lambda: ["PluginA", "PluginB"], + database=lambda: _recording_database(calls), + ) + lifecycle.start() + + lifecycle.stop() + + released = {call[1] for call in calls if call[0] == "release"} + assert released == {"PluginA", "PluginB"} + + +def test_reload_releases_then_ensures_again(): + """热重载先释放旧连接再重新建库，全程不出现 destroy。""" + calls: list[tuple] = [] + plugin_cls = _make_plugin_class("DemoPlugin") + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: _recording_database(calls), + ) + lifecycle.start("DemoPlugin") + calls.clear() + + lifecycle.reload("DemoPlugin", "plugin-reload") + + kinds = [call[0] for call in calls if call[0] in ("release", "ensure", "destroy")] + assert kinds.index("release") < kinds.index("ensure") + assert "destroy" not in kinds + + +def test_release_failure_does_not_block_unloading(): + """release 抛异常时卸载仍收敛完成，插件从运行态移出。""" + plugin_cls = _make_plugin_class("DemoPlugin") + + def _raise_release(_plugin_id): + """模拟释放连接失败。""" + raise RuntimeError("release failed") + + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: PluginDatabase(release=_raise_release), + ) + lifecycle.start("DemoPlugin") + + assert lifecycle.quiesce("DemoPlugin") is True + assert lifecycle.finalize("DemoPlugin") is True + assert "DemoPlugin" not in lifecycle._running + + +def test_delete_plugin_data_destroys_the_plugin_database(): + """重置数据先删宿主业务数据行，再销毁插件自有库。""" + calls: list[tuple] = [] + store = PluginConfigStore( + storage=lambda: PluginStorage( + delete_data=lambda plugin_id: calls.append(("storage.delete_data", plugin_id)) + ), + database=lambda: _recording_database(calls), + plugin_exists=lambda _plugin_id: True, + ) + + result = store.delete_data("DemoPlugin", force=True) + + assert result is True + kinds = [call[0] for call in calls] + assert kinds.index("storage.delete_data") < kinds.index("destroy") + assert ("destroy", "DemoPlugin") in calls + + +def test_delete_plugin_data_without_force_refuses_unknown_plugin(): + """未知插件且非强制删除时拒绝执行，不触发销毁。""" + calls: list[tuple] = [] + store = PluginConfigStore( + storage=lambda: PluginStorage( + delete_data=lambda plugin_id: calls.append(("storage.delete_data", plugin_id)) + ), + database=lambda: _recording_database(calls), + plugin_exists=lambda _plugin_id: False, + ) + + result = store.delete_data("DemoPlugin", force=False) + + assert result is False + assert calls == [] + + +def test_clone_uninstall_destroys_only_the_clone_database(): + """分身卸载只销毁分身自己的库，单键语义下与源插件互不影响。""" + calls: list[tuple] = [] + store = PluginConfigStore( + storage=lambda: PluginStorage( + delete_data=lambda plugin_id: calls.append(("storage.delete_data", plugin_id)) + ), + database=lambda: _recording_database(calls), + plugin_exists=lambda _plugin_id: True, + ) + + store.delete_data("DemoPluginwork", force=True) + + destroyed = [call[1] for call in calls if call[0] == "destroy"] + assert destroyed == ["DemoPluginwork"] + + +def test_remove_plugin_only_releases_the_database(): + """从内存移除插件的内部路径即 stop，只释放不销毁。""" + calls: list[tuple] = [] + plugin_cls = _make_plugin_class("DemoPlugin") + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: _recording_database(calls), + ) + lifecycle.start("DemoPlugin") + + lifecycle.stop("DemoPlugin") + + assert ("release", "DemoPlugin") in calls + assert not any(call[0] == "destroy" for call in calls) + + +def test_close_database_releases_plugin_databases_before_the_host_engine(monkeypatch): + """进程关停时先释放插件库，再释放宿主同步引擎。""" + calls: list[str] = [] + monkeypatch.setattr(session_module, "release_all_databases", lambda: calls.append("plugins")) + fake_sync_engine = MagicMock() + fake_sync_engine.dispose.side_effect = lambda: calls.append("sync engine") + monkeypatch.setattr(engine_module, "peek_sync_engine", lambda: fake_sync_engine) + monkeypatch.setattr(engine_module, "peek_async_engine", lambda: None) + monkeypatch.setattr(session_module, "_pooled_async_engines", {}) + + asyncio.run(session_module.close_database()) + + assert calls == ["plugins", "sync engine"] + + +def test_startup_composition_binds_the_plugin_database_to_the_db_framework(monkeypatch): + """组合根把插件数据库端口装配到 db 层的建库、释放与销毁实现。""" + calls: list[tuple] = [] + monkeypatch.setattr( + plugins_initializer, + "ensure_database", + lambda plugin_id, models, migrations: calls.append(("ensure", plugin_id)), + ) + monkeypatch.setattr( + plugins_initializer, + "release_database", + lambda plugin_id: calls.append(("release", plugin_id)), + ) + monkeypatch.setattr( + plugins_initializer, + "destroy_database", + lambda plugin_id: calls.append(("destroy", plugin_id)), + ) + + database = plugins_initializer._build_plugin_database() + database.ensure("demo", (), None) + database.release("demo") + database.destroy("demo") + + assert [call[0] for call in calls] == ["ensure", "release", "destroy"] + + +def test_plugin_runtime_database_port_does_not_import_the_database_layer(): + """端口模块的 import 语句里不含任何 app.db 前缀，保证运行时不依赖数据库实现。""" + source_path = ( + Path(__file__).resolve().parent.parent + / "app" + / "runtime" + / "extensions" + / "plugin" + / "database.py" + ) + tree = ast.parse(source_path.read_text(encoding="utf-8")) + imported_modules: list[str] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imported_modules.extend(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + imported_modules.append(node.module) + + assert not any(module.startswith("app.db") for module in imported_modules) + + +def test_start_releases_the_database_of_a_plugin_that_failed_to_load(): + """建库失败的插件不会进入运行态，其句柄在失败分支就地释放。""" + calls: list[tuple] = [] + plugin_cls = _make_plugin_class("DemoPlugin") + + def _raise_ensure(plugin_id, _models, _migrations): + """模拟建库失败，但句柄可能已经建立。""" + calls.append(("ensure", plugin_id)) + raise RuntimeError("ensure failed") + + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: PluginDatabase( + ensure=_raise_ensure, + release=lambda plugin_id: calls.append(("release", plugin_id)), + ), + ) + + lifecycle.start("DemoPlugin") + + assert [call for call in calls if call[0] == "release"] == [("release", "DemoPlugin")] + assert "DemoPlugin" not in lifecycle._running + + +def test_stop_all_releases_plugins_that_never_reached_the_running_registry(): + """整体停止会释放启动中途失败、只登记在类注册表里的插件。""" + calls: list[tuple] = [] + plugin_cls = _make_plugin_class("DemoPlugin") + + def _raise_ensure(_plugin_id, _models, _migrations): + """模拟建库失败。""" + raise RuntimeError("ensure failed") + + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: PluginDatabase( + ensure=_raise_ensure, + release=lambda plugin_id: calls.append(("release", plugin_id)), + ), + ) + lifecycle.start() + assert "DemoPlugin" not in lifecycle._running + assert "DemoPlugin" in lifecycle._classes + calls.clear() + + lifecycle.stop() + + assert ("release", "DemoPlugin") in calls + + +def test_stopping_an_already_stopped_plugin_stays_idempotent(): + """卸载编排先停后删，删除之后的 remove_plugin 会再停一次，二次停止必须无害。""" + calls: list[tuple] = [] + plugin_cls = _make_plugin_class("DemoPlugin") + lifecycle = _build_lifecycle( + load_plugins=lambda *_a, **_kw: [plugin_cls], + installed_plugins=lambda: ["DemoPlugin"], + database=lambda: _recording_database(calls), + ) + lifecycle.start("DemoPlugin") + lifecycle.stop("DemoPlugin") + calls.clear() + + lifecycle.stop("DemoPlugin") + + assert calls == [("release", "DemoPlugin")] + assert lifecycle._classes == {} + assert lifecycle._running == {} + + +def _uninstall_manager() -> MagicMock: + """构造卸载分身所需的插件管理器替身，按调用顺序记录全部方法调用。""" + plugin_manager = MagicMock() + plugin_manager.get_plugin_instance.return_value = None + plugin_manager.get_plugin_source_instances.return_value = [] + plugin_manager.plugins = {"DemoPluginwork": MagicMock(is_clone=True)} + plugin_manager.remove_plugin_package.return_value = True + return plugin_manager + + +def _assert_stop_precedes_deletion(plugin_manager: MagicMock) -> None: + """断言插件先停止再删除数据，且删除按 force 执行。""" + method_names = [name for name, _args, _kwargs in plugin_manager.mock_calls] + assert method_names.index("stop") < method_names.index("delete_plugin_config") + assert method_names.index("stop") < method_names.index("delete_plugin_data") + plugin_manager.stop.assert_called_once_with("DemoPluginwork") + plugin_manager.delete_plugin_config.assert_called_once_with( + "DemoPluginwork", + force=True, + ) + plugin_manager.delete_plugin_data.assert_called_once_with( + "DemoPluginwork", + force=True, + ) + + +def test_http_uninstall_stops_the_plugin_before_deleting_its_data(monkeypatch): + """HTTP 卸载先停插件再删数据，停机钩子因此无法重建刚销毁的自有库。""" + plugin_manager = _uninstall_manager() + config = MagicMock() + config.get.return_value = ["DemoPluginwork"] + monkeypatch.setattr(plugin_endpoint, "get_plugin_manager", lambda: plugin_manager) + monkeypatch.setattr(plugin_endpoint, "get_configured_system_config", lambda: config) + monkeypatch.setattr(plugin_endpoint, "remove_plugin_api", MagicMock()) + monkeypatch.setattr(plugin_endpoint, "remove_plugin_job", MagicMock()) + monkeypatch.setattr(plugin_endpoint, "remove_plugin_from_folders", MagicMock()) + + result = plugin_endpoint.uninstall_plugin("DemoPluginwork", None) + + assert result.success is True + _assert_stop_precedes_deletion(plugin_manager) + + +def test_runtime_uninstall_stops_the_plugin_before_deleting_its_data(monkeypatch): + """运行态卸载编排同样先停后删，两条卸载路径的数据删除时机保持一致。""" + plugin_manager = _uninstall_manager() + config = MagicMock() + config.get.return_value = ["DemoPluginwork"] + config.async_set = AsyncMock() + monkeypatch.setattr(plugin_management, "get_plugin_manager", lambda: plugin_manager) + monkeypatch.setattr( + plugin_management, + "get_configured_system_config", + lambda: config, + ) + monkeypatch.setattr(plugin_routes, "remove_plugin_api", MagicMock()) + monkeypatch.setattr(scheduling_module, "remove_plugin_job", MagicMock()) + monkeypatch.setattr(plugin_folders, "remove_plugin_from_folders", MagicMock()) + + result = asyncio.run(plugin_management.uninstall_plugin_runtime("DemoPluginwork")) + + assert result == {"was_clone": True, "clone_files_removed": True} + _assert_stop_precedes_deletion(plugin_manager) + + +def test_uninstall_virtual_instance_also_stops_before_deleting(monkeypatch): + """虚拟实例卸载同样先停后删，force 删除不受插件类注销影响。""" + plugin_manager = MagicMock() + plugin_manager.get_plugin_instance.return_value = PluginInstance( + instance_id="DemoPluginwork", + source_plugin_id="DemoPlugin", + ) + plugin_manager.get_plugin_source_instances.return_value = [] + plugin_manager.plugins = {} + config = MagicMock() + config.get.return_value = ["DemoPlugin"] + monkeypatch.setattr(plugin_endpoint, "get_plugin_manager", lambda: plugin_manager) + monkeypatch.setattr(plugin_endpoint, "get_configured_system_config", lambda: config) + monkeypatch.setattr(plugin_endpoint, "remove_plugin_api", MagicMock()) + monkeypatch.setattr(plugin_endpoint, "remove_plugin_job", MagicMock()) + monkeypatch.setattr(plugin_endpoint, "remove_plugin_from_folders", MagicMock()) + + result = plugin_endpoint.uninstall_plugin("DemoPluginwork", None) + + assert result.success is True + config.set.assert_called_once_with( + SystemConfigKey.UserInstalledPlugins, + ["DemoPlugin"], + ) + _assert_stop_precedes_deletion(plugin_manager) diff --git a/tests/test_plugin_lifecycle_status.py b/tests/test_plugin_lifecycle_status.py index 11a012cef..4e0197872 100644 --- a/tests/test_plugin_lifecycle_status.py +++ b/tests/test_plugin_lifecycle_status.py @@ -3,6 +3,7 @@ from types import SimpleNamespace from unittest.mock import MagicMock +from app.runtime.extensions.plugin.database import PluginDatabase from app.runtime.extensions.plugin.lifecycle import PluginLifecycle from app.schemas.plugin import PluginRuntimeStatus @@ -41,6 +42,7 @@ def _lifecycle(*, plugins, auth=True): enable_events=MagicMock(), disable_events=MagicMock(), runtime_status_writer=statuses.__setitem__, + database=lambda: PluginDatabase(), log=MagicMock(), event_sender=MagicMock(), )