Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
"""Entity restore registry — declarative configuration for undo/restore.
|
|
|
|
Each registered entity type declares:
|
|
- model_class: SQLAlchemy model to load
|
|
- restore_permission: permission string required to restore
|
|
- excluded_fields: fields never restored from snapshot (id, tenant_id, timestamps, etc.)
|
|
- special_handler: optional async callable for entity-specific restore logic
|
|
|
|
No dynamic ORM loading, no blind snapshot writes — only explicitly registered
|
|
entity types can be restored, and only through their declared configuration.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Default fields excluded from restore for every entity type
|
|
_DEFAULT_EXCLUDED = frozenset({
|
|
"id",
|
|
"tenant_id",
|
|
"created_at",
|
|
"updated_at",
|
|
"deleted_at",
|
|
"search_tsv",
|
|
"embedding",
|
|
})
|
|
|
|
# Type alias for special restore handler
|
|
SpecialRestoreHandler = Callable[
|
|
[AsyncSession, Any, str, Any, dict[str, Any]],
|
|
Awaitable[dict[str, Any]],
|
|
]
|
|
|
|
|
|
@dataclass
|
|
class RestoreConfig:
|
|
"""Configuration for restoring a specific entity type."""
|
|
|
|
entity_type: str
|
|
model_class: type
|
|
restore_permission: str
|
|
excluded_fields: frozenset[str] = field(default_factory=frozenset)
|
|
special_handler: SpecialRestoreHandler | None = None
|
|
|
|
@property
|
|
def all_excluded_fields(self) -> frozenset[str]:
|
|
"""Merge default excluded fields with entity-specific ones."""
|
|
return _DEFAULT_EXCLUDED | self.excluded_fields
|
|
|
|
|
|
class RestoreRegistry:
|
|
"""Singleton registry mapping entity_type → RestoreConfig."""
|
|
|
|
_instance: RestoreRegistry | None = None
|
|
|
|
def __new__(cls) -> RestoreRegistry:
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
cls._instance._configs: dict[str, RestoreConfig] = {}
|
|
return cls._instance
|
|
|
|
def register(self, config: RestoreConfig) -> None:
|
|
"""Register a RestoreConfig for an entity type."""
|
|
if config.entity_type in self._configs:
|
|
logger.warning("Overwriting restore config for entity_type: %s", config.entity_type)
|
|
self._configs[config.entity_type] = config
|
|
logger.debug("Registered restore config for: %s", config.entity_type)
|
|
|
|
def unregister(self, entity_type: str) -> None:
|
|
"""Remove a RestoreConfig (e.g. when plugin is deactivated)."""
|
|
if entity_type in self._configs:
|
|
del self._configs[entity_type]
|
|
logger.debug("Unregistered restore config for: %s", entity_type)
|
|
|
|
def get(self, entity_type: str) -> RestoreConfig | None:
|
|
"""Get RestoreConfig for entity_type, or None if not registered."""
|
|
return self._configs.get(entity_type)
|
|
|
|
def is_registered(self, entity_type: str) -> bool:
|
|
"""Check if entity_type is registered for restore."""
|
|
return entity_type in self._configs
|
|
|
|
def list_registered(self) -> list[str]:
|
|
"""Return all registered entity types."""
|
|
return sorted(self._configs.keys())
|
|
|
|
def _reset_for_testing(self) -> None:
|
|
"""Clear all registrations — for unit tests only."""
|
|
self._configs.clear()
|
|
|
|
|
|
def get_restore_registry() -> RestoreRegistry:
|
|
"""Return the global RestoreRegistry singleton."""
|
|
return RestoreRegistry()
|
|
|
|
|
|
def reset_restore_registry_for_testing() -> RestoreRegistry:
|
|
"""Return a fresh singleton — for unit tests only."""
|
|
reg = get_restore_registry()
|
|
reg._reset_for_testing()
|
|
return reg
|
|
|
|
|
|
# ─── Default entity registrations ───
|
|
|
|
|
|
def register_default_entities() -> None:
|
|
"""Register Core entity types for restore.
|
|
|
|
Called during app startup. Plugin entities register themselves
|
|
in their on_activate() lifecycle hook.
|
|
"""
|
|
# Contact is registered by ContactsPlugin.on_activate() — do not register
|
|
# it here to avoid double registration. This function remains for future
|
|
# Core entities that have no plugin.
|
|
|
|
# Plugin entities (task, calendar_entry, dms_file, mail) are registered
|
|
# by their respective plugins in on_activate(). See P0-7 fix.
|