MoviePilot/docs/rules/10-data-and-persistent.md

10 — Data and Persistent Management

Database Models

Location: app/db/models/

Models are SQLAlchemy declarative classes. Each model maps to one database table.

Model	Table Domain
Subscribe	Media subscriptions
SubscribeHistory	Completed subscription records
TransferHistory	File transfer history
DownloadHistory / DownloadFiles	Download task history and file list
MediaServerItem	Media server library item cache
SystemConfig	Runtime key-value configuration store
UserConfig	Per-user configuration store
User	User accounts
Site / SiteIcon / SiteStatistic / SiteUserData	Torrent site records and statistics
Message	Message log
PluginData	Plugin-persisted data
PluginIdentity	Installed physical-plugin source binding and payload provenance
PluginInstance	Per-instance plugin descriptor and configuration, one row per instance
PassKey	Passkey authentication records
Workflow	Workflow definitions
WidgetRegistration	每账号每安装的小组件只读凭据摘要与到期时间，不保存原文

---

Alembic Migrations

Location: database/versions/

Rule: Any change to a SQLAlchemy model schema (adding a column, renaming a column, changing a column type, adding a table, removing a table) requires a new Alembic migration script. Never update models without a corresponding migration.

Generating a migration:

# 根据模型与数据库的差异生成迁移脚本
alembic revision --autogenerate -m "describe the change"

# 创建空迁移脚本，以便填写需要手工控制的 SQL
alembic revision -m "describe the change"

Review the auto-generated migration before committing — auto-generation can miss nullable changes, index modifications, or SQLite-incompatible operations.

---

Data Access Layer (Oper Pattern)

Location: app/db/

Each model has a corresponding file under app/db/oper/ containing the data access class, mirroring app/db/models/ one-for-one. Do not write SQLAlchemy queries directly in chain, module, or endpoint code.

Oper Class	File
AgentChatOper	oper/agentchat.py
AgentTaskOper	oper/agenttask.py
DownloadFailureOper	oper/downloadfailure.py
DownloadHistoryOper	oper/downloadhistory.py
MediaServerOper	oper/mediaserver.py
MessageOper	oper/message.py
PluginDataOper	oper/plugindata.py
PluginIdentityOper	oper/pluginidentity.py
PluginInstanceOper	oper/plugininstance.py
SiteOper	oper/site.py
SubscribeHistoryOper	oper/subscribehistory.py
SubscribeOper	oper/subscribe.py
SystemConfigOper	oper/systemconfig.py
TransferHistoryOper	oper/transferhistory.py
TransferPendingOper	oper/transferpending.py
UserConfigOper	oper/userconfig.py
UserOper	oper/user.py
WorkflowOper	oper/workflow.py

Import by module (from app.db.oper.subscribe import SubscribeOper) — that is the preferred form in this repository. app/db/oper/__init__.py also resolves class names lazily for callers that only want a name, but it deliberately does not eagerly re-export: several tests isolate a single Oper by stubbing it in sys.modules, and an eager re-export would pull in the other fifteen and bypass the stub.

Oper classes accept and return persistence values. Turning a MediaInfo or MetaBase into a row is business logic and lives in app/application/.

Application owns use-case commands and persistence Protocols, but does not import app.db, SQLAlchemy, Session or Oper. Concrete persistence is used in app/db/adapters/: adapters implement those Protocols with explicit Session, UnitOfWork and Oper objects. app/startup/composition/ creates and injects the adapters; it does not retain reusable repository implementations.

音乐识别的临时失败可在 TransferPending.state=accepted 且不存在计划、步骤、执行结果和历史回执时， 以当前有效租约 CAS 写入 execution_state=retry_wait、到期时间及重试计数，并释放租约。 这些重试字段也受原有总重试预算约束；它们不是文件操作已发生的证据。接纳恢复必须重新读取 冻结的音乐范围，已规划任务则继续消费原检查点。人工重试零文件操作的音乐拒绝时，只能在同一事务内 验证唯一成功步骤确为原检查点的 planning/reject，删除该内部拒绝步骤和旧 pending，再创建新准入； 旧失败历史只解除旧任务映射，记录与独立结算回执不删除，避免留下无法维护的悬空任务关联。 任何外部操作步骤或版本/租约冲突都禁止走这条重规划路径。

AI 识别修正由执行仓储在单一短事务中校验失败历史归属、结算版本、租约和剩余预算。 仅包含一个成功 planning/reject 内部步骤的电影/电视剧拒绝可以退回 accepted； 原计划和步骤先归档到新输入的 options.ai_replan_evidence，同一任务、历史绑定和独立结算回执保留， 重试次数及世代递增，源、目标和整理方式不变。任何文件操作、复核状态或不确定证据均拒绝替换。 失败结算提交后的自动 AI 唤醒只处理新回执，不依赖心跳或已有自动任务；重放回执不重复唤醒。

Transaction ownership ratchet

• tests/fixtures/architecture/transaction-debt-baseline.json records formal decorators in concrete files under app/db/models/. Their count is zero and must remain zero. Model/Base code may not import app.db.decorators; legacy Model transaction shells have been removed and must not be recreated.
• Every Model method with a db parameter requires an explicit Session or AsyncSession. The parameter may not default to None, accept displaced business arguments, create a Session, or call commit() / rollback().
• Base.create/get/update/delete/list/truncate and their async forms are plain explicit-session primitives. They only query or stage changes in the caller's transaction; they never own transaction lifecycle.
• Host Oper code routes optional-session entry points through _execute_sync_query / _execute_async_query / _execute_*_write. Plugins access host persistence through Oper or a curated SDK contract, never by importing app.db.models.
• The public db_query, db_update, async_db_query, and async_db_update exports remain available only for plugin-owned database functions. They are forbidden on host Model/Base methods.
• Oper receives a caller-owned Session and may query, add, update, delete, or flush. A composable Oper method must not create its own Session and must not commit or roll back.
• API, Scheduler, Agent and Chain consume an injected Application Port; they do not import or create a Session. The concrete app/db/adapters/ implementation creates the Session and adapts it through app/db/uow.py. Application command code decides when the injected UoW commits or rolls back; events, scheduling refresh, reports and other external effects run only after a successful commit.
• A synchronous Session is private to one worker thread. An AsyncSession is private to one asyncio task/operation; neither may be stored in a process singleton or reused by concurrent work.
• Subscription creation is the reference slice: app/application/subscription/write.py owns the command and persistence Port, app/db/adapters/subscription.py creates an exclusive Session and adapts Oper/UoW, and app/startup/composition/subscription.py only wires scopes and post-commit callbacks. SubscribeOper.stage_add() only queries, adds and flushes. Preserve SubscribeOper.add() only for legacy SDK callers; new host code must not use that auto-commit compatibility path.
• The full Subscription boundary is owned by app/application/subscription/contract.py: complete subscription and history records leave the Session only as deeply frozen snapshots, while identity and patch DTOs make query/write inputs explicit. Standalone Chain/Agent work uses TransactionalSubscriptionRepository; request and Application commands bind SessionSubscriptionRepository or SessionSubscriptionHistoryRepository to their existing Session/UoW. The Session-bound adapters stage and project but never commit. Canonical host code must not receive Subscribe, SubscribeHistory, SubscribeOper, or SubscribeHistoryOper; the old module paths and package-root symbols are plugin ABI implemented only by the SDK Legacy/Compat facade.
• A repository method that opens one short transaction per call is not a batch transaction. Subscription workflows that update multiple rows or combine SystemConfig and subscription mutations must use a caller-owned Session/UoW and publish process snapshots only after commit; this remains the S1-L4/S1-L5 acceptance boundary.
• The same rule applies to SiteMutationCommand, history/workflow commands, AgentChatService.delete(), and DeletePluginDataCommand: bind the repository and UoW to one request/operation Session. Legacy plugin-facing Oper methods may remain temporarily, but a new endpoint or startup workflow must call stage_*.
• Site reads and standalone writes use TransactionalSiteRepository; request mutation uses SessionSiteRepository bound to the endpoint AsyncSession and the request UoW. Both adapters must finish deep DTO projection inside their Session. SiteOper remains a DB-internal table DAO; plugin compatibility is exposed only by the SDK Legacy/Compat mapping.
• User create/update/delete is an aggregate command owned by app/application/security/user.py. It uses one request AsyncSession/UoW, locks active superusers before destructive changes, and rejects removal of the last enabled superuser. Query ports return frozen user/auth snapshots rather than ORM rows.
• The database is the final user-identity guard: user.name is unique; UserConfig.username cascades on user rename/delete; PassKey.user_id cascades on user delete; (UserConfig.username, UserConfig.key) is unique and non-null. A schema change to any of these constraints requires a replay-safe migration that repairs legacy duplicates/orphans before creating constraints.
• DownloadHistory and TransferHistory are projected into deeply frozen DTOs within the adapter Session. Their typed query/write ports use short Session/UoW scopes; TransferHistory additionally exposes a staging port for durable settlement inside the caller-owned transaction. Canonical callers must not receive raw history ORM rows or Oper objects.

Durable post-commit side effects

Business mutations that must survive process interruption stage their durable intent through OutboxStager in the same Session/UoW as the business row. OutboxDispatchStore owns separate short transactions for claim, complete and retry; a business Session must never call those self-committing operations. app/db/adapters/outbox.py implements both roles, and startup composition supplies the stager, store factory, transaction scope and topic handlers.

Immediate delivery and the dispatcher both claim before execution. Claim is atomic and complete/retry is fenced by the claimed attempt, so an expired owner cannot settle a newer lease. PostCommitResult separately reports the committed business value plus completed and pending effects; a post-commit failure cannot be represented as a rollback of already committed business data.

This boundary is at-least-once, not exactly-once. If an external sink succeeds and the process stops before complete is persisted, the intent can be replayed. Event payloads and the host correlation context therefore carry the stable event key, and consumers that support deduplication should use it. Legacy notification plugins retain their existing method signature and remain an at-least-once boundary where duplicate provider delivery is possible. The dispatcher records bounded retries or dead-letter state. The shared data-maintenance policy controls bounded terminal-history cleanup, with user-configurable 30-day completed and 90-day dead-letter defaults; 0 disables either cleanup. It must not delete pending or leased processing rows. The app/runtime/tasks.py TaskRegistry is only the owner for in-process work and bounded shutdown waiting; it is not a durable queue or a replacement for an Outbox/persistent task table.

All append-only or snapshot history owned by the host must participate in the shared DATA_CLEANUP_ENABLE policy when it has a safe time boundary:

• message, downloadhistory and orphaned downloadfiles, siteuserdata, transferhistory, downloadfailure, and subscribehistory use their own user-configurable retention periods.
• agentchat removes only expired sessions not referenced by an agenttask; Agent 原始会话存于独立 config/agent/runtime/history/users/<user-key>/state.db， 不新增主库模型或 Alembic 表。独立库沿用 DATA_CLEANUP_ENABLE 和 Agent 会话保留期， 活跃用户后台维护按 last_active 分批删除；显式删除同步收回原文和 FTS，墓碑阻止在途复活。 恢复快照与原始证据分开，压缩不删除原文。旧快照只读分批导出并标记 legacy_snapshot， 导入完成后查询不依赖主库。SQLite schema 由 Agent 独立版本管理。 agenttaskrun removes only expired terminal runs that are neither running nor the task's current last_run_id.
• agentinvocation stores only write-call identity, argument digests and fixed host status summaries. Successful/failed receipts and pending receipts that confirm an asynchronous submission are deleted in the same transaction as their owning chat, including shared chat retention cleanup. Pending means submission is confirmed while downstream completion has not been observed; it is history rather than a queue to execute. Replaying the same invocation never repeats submission, but a new user intent may submit again. These history receipts without a chat (including background task calls) use the same Agent chat retention period, based on their UTC settlement timestamp; both the shared cleanup switch and a zero-day retention disable this cleanup. Running/unknown receipts are recovery state and have no age-based deletion; they also protect their chat from automatic retention cleanup. Cold startup changes running receipts to unknown and rotates their fencing token; elapsed time never grants permission to repeat the write.
• outboxmessage has separate completed and dead-letter retention periods; pending and processing intents are recovery state and are never age-deleted.

transferpending and plugininstallation are recovery queues/journals rather than history. Their age is not proof that they are disposable, so generic retention cleanup must not delete them. Current-state tables keyed by a user, site, plugin, workflow, passkey, or media-library item are likewise outside time-based cleanup; their owning mutation lifecycle must replace or delete them.

Run ./.venv/bin/python scripts/architecture/baseline.py --check-host after persistence changes. A deliberate debt reduction may refresh the low-water mark with --write-host; never refresh it to accept newly introduced debt.

Canonical explicit-session Oper conventions:

with SessionFactory() as session:
    oper = SubscribeOper(session)
    # 查询复用调用方的会话，新增操作只暂存，由调用方的工作单元提交
    subscribe = oper.get(sid=1)
    subscribes = oper.list()
    oper.stage_add(Subscribe(...))

The following no-Session form is legacy plugin ABI only and must not be copied into host code:

oper = SubscribeOper()
subscribe = oper.get(sid=1)
subscribes = oper.list()
oper.add(Subscribe(...))
oper.update(sid=1, name="New Name")
oper.delete(sid=1)

---

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/<PluginId>/plugin.db; PostgreSQL deploys one schema that reuses the host engine via schema_translate_map. The schema name is plugin_<pluginid> 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.

Before the upgrade runs, run_migrations compares the revisions already stamped in the plugin database against the script tree shipped with the installed plugin version. A revision that tree cannot locate means the database was stamped by a newer build — the ordinary outcome of reinstalling an older version — and the host raises PluginMigrationCompatibilityError naming both the revision and the directory instead of letting Alembic fail with a bare Can't locate revision. The host never downgrades automatically: downgrade scripts normally drop tables and columns, so running them unattended would destroy whatever the user produced under the newer version, with no way back. Recovery is to install a plugin version whose migration tree contains that revision, or to have an administrator perform an explicit data migration. A database with no alembic_version table yet reports no revisions, so first creation is never blocked by this check.

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, naming the plugin schema alone. Unqualified raw SQL therefore resolves inside the plugin's own schema, and a name missing there fails instead of falling through to a host table of the same name; 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.

Enum: SystemConfigKey in app/schemas/types.py

Host service: SystemConfigService and get_configured_system_config() in app/application/configuration.py. SystemConfigOper is used behind the composition/persistence boundary and remains available for legacy plugin ABI.

from app.application.configuration import get_configured_system_config
from app.schemas.types import SystemConfigKey

configuration = get_configured_system_config()

rss_urls = configuration.get(SystemConfigKey.RssUrls)

configuration.set(SystemConfigKey.RssUrls, ["https://example.com/rss"])

Rule: Never use raw string literals as SystemConfig keys. Always define a new SystemConfigKey enum entry first. Raw string key lookups are not searchable and cannot be refactored safely.

Rule: Plugin configuration is not SystemConfig. A plugin instance's own settings live on its PluginInstance row (plugininstance.config_data) and are reached through the plugin configuration channel — _PluginBase.get_config() / update_config() for plugins, PluginConfigStore and the PluginStorage instance-config ports for the host. SystemConfigOper treats every key alike and holds no knowledge of plugins; it must never branch on a key's value. A plugin that reads or writes its own configuration directly through SystemConfig (for example under a plugin.<PluginId> key) is unsupported: such a key is an ordinary, unrelated SystemConfig row that the plugin configuration channel neither reads nor writes.

---

UserConfig — Per-User Configuration

Purpose: Settings that differ per user account. Host callers use the configured UserConfigurationService; its concrete repository adapts UserConfigOper behind the persistence boundary.

from app.application.security.userconfig import get_configured_user_configuration

configuration = get_configured_user_configuration()
value = configuration.get(username="alice", key="notification_enabled")
configuration.set(username="alice", key="notification_enabled", value=True)

The no-Session UserConfigOper() form is legacy plugin ABI only and must not be copied into host code.

TransactionalUserConfigurationRepository stages a set in a short transaction and publishes the process snapshot only after commit. User rename/delete is first completed by the user aggregate transaction through database cascades; post-commit publication then acquires the write lock and reloads the database fact source so concurrent set/rename/delete operations converge on committed state. If publication fails, the repository reloads that fact source instead of rolling back or hiding the already committed user mutation. Reads and published JSON values are copied so callers cannot mutate shared cache state.

---

Settings / Environment Configuration

Purpose: Deployment-level, environment-level, and startup-time configuration such as ports, paths, proxies, switches, API keys, and third-party service addresses.

Location: ConfigModel and Settings in app/runtime/config.py

These values are loaded at startup from process environment variables, <CONFIG_DIR>/app.env, secret files, and field defaults, in that order. Explicit constructor arguments take precedence. They are not stored in the database. Web/CLI updates write app.env only for locally managed fields; environment and secret-managed fields reject persistence. See Secrets deployment for directory selection, formats, rotation, and migration.

Access:

from app.runtime.config import settings

host = settings.QB_HOST
port = settings.QB_PORT

---

Caching

FileCache / AsyncFileCache

Location: app/runtime/cache.py

Used to cache expensive external API responses to disk. Cache entries have a configurable TTL.

from app.runtime.cache import FileCache, fresh

cache = FileCache(cache_name="tmdb", ttl=3600)

@fresh(cache=cache, key_func=lambda tmdb_id: f"movie_{tmdb_id}")
def get_movie_detail(tmdb_id: int) -> dict:
    """
    查询电影详情，并由装饰器复用一小时内的文件缓存。

    :param tmdb_id: 电影在 TMDB 中的原生标识，同时用于区分缓存项
    :return: 缓存命中时返回已缓存的详情，否则返回客户端查询结果
    """
    return self._tmdb_client.get_movie(tmdb_id)

Redis (Optional)

When REDIS_HOST is configured, app/modules/redis/ provides a distributed cache backend. Prefer FileCache for single-node deployments.

Security-sensitive one-shot state uses AtomicCacheBackend, not a concrete Redis helper. Its strict store() surfaces backend write failures and consume() atomically returns-and-removes a value. Both Memory and Redis backends implement this contract; Passkey receives the capability from startup through PasskeyChallengeCache, so an authentication or registration challenge can be accepted only once without Application knowing the configured backend.

---

Data Lifecycle Rules

• TransferHistory: Records are inserted after every successful file transfer. Do not delete records without user confirmation.
• DownloadHistory: Records are inserted when a download task is added. Linked DownloadFiles records track individual files within a torrent. Host query/write callers use frozen DTOs and the typed DownloadHistory port; ORM rows remain inside the adapter Session.
• SystemConfig: Values may be read and written freely at runtime. Changes to watched config keys trigger on_config_changed() on registered classes via ConfigReloadMixin.
• MediaServerItem: This is a cache of the remote media server library. It is refreshed on media server sync events and can be safely cleared and rebuilt.

---

Sensitive Data Handling

• Never log database record contents that include personal data (user credentials, passkeys, API tokens).
• 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-09-13