Agent Zero
cd8ef7500c
Check Cross-Plugin Imports / check (push) Has been cancelled
- Format-Registry (app/core/importexport_registry.py): FormatHandler-Protokoll, available_for() Schnittmenge, Singleton + Testing-Reset - importexport_formats-Plugin (csv/json/xlsx), lifecycle-korrekt: on_activate registriert Handler in der Core-Registry, on_deactivate unregistriert - ContactsContract: importexport-Beitrag (ie_*-Methoden) — Contacts besitzt seine Import/Export-Fachlogik jetzt selbst - import_export_service.py: generische Engine, delegiert generisch ueber registry.list_discovered() an den besitzenden Contract (keine hartcodierten Plugin-Namen mehr); Signaturen identisch - Funktionserhalt bewiesen: 45/45 import_export-Suite passed (inkl. Fehler-Multiplizitaet: 2 failed rows -> 3 total_errors, erreicht via ie_required-Weitergabe + ie_row_valid-Nur-contacts-Early-Return) fixes #359 (Phase 1)
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
"""Import/Export format registry — formats are provided by plugins.
|
|
|
|
A **format plugin** (e.g. ``importexport_formats``) registers handlers for
|
|
its supported file formats (csv, json, xlsx, ...). An **entity module**
|
|
(e.g. contacts) contributes via its contract which formats it supports and
|
|
provides the import/export logic for its data.
|
|
|
|
The core orchestrator (routes + background jobs) resolves the intersection:
|
|
|
|
capabilities(entity) = entity.formats ∩ format_registry.registered
|
|
|
|
The security policy layer (sensitive-data filter, tenant scoping, audit)
|
|
runs in the orchestrator, independently of the module contribution.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Protocol, runtime_checkable
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@runtime_checkable
|
|
class FormatHandler(Protocol):
|
|
"""Handler for one file format (e.g. csv, json, xlsx).
|
|
|
|
Provided by a format plugin via the registry.
|
|
"""
|
|
|
|
format_id: str
|
|
|
|
@staticmethod
|
|
def parse(content: bytes) -> list[dict[str, Any]]:
|
|
"""Parse file content into a list of row dicts."""
|
|
...
|
|
|
|
@staticmethod
|
|
def serialize(rows: list[dict[str, Any]], headers: list[str]) -> bytes:
|
|
"""Serialize rows into file bytes with the given column order."""
|
|
...
|
|
|
|
|
|
class ImportExportFormatRegistry:
|
|
"""Registry for file-format handlers contributed by format plugins."""
|
|
|
|
def __init__(self) -> None:
|
|
self._formats: dict[str, FormatHandler] = {}
|
|
|
|
def register(self, handler: FormatHandler) -> None:
|
|
"""Register (or replace) a format handler."""
|
|
self._formats[handler.format_id] = handler
|
|
logger.debug("Registered import/export format '%s'", handler.format_id)
|
|
|
|
def unregister(self, format_id: str) -> None:
|
|
self._formats.pop(format_id, None)
|
|
logger.debug("Unregistered import/export format '%s'", format_id)
|
|
|
|
def get(self, format_id: str) -> FormatHandler | None:
|
|
return self._formats.get(format_id)
|
|
|
|
def list_formats(self) -> list[str]:
|
|
return sorted(self._formats.keys())
|
|
|
|
def available_for(self, entity_formats: list[str]) -> list[str]:
|
|
"""Return the intersection of an entity's declared formats and
|
|
the currently registered (plugin-provided) format handlers."""
|
|
return sorted(set(entity_formats) & set(self._formats.keys()))
|
|
|
|
def clear(self) -> None:
|
|
"""Clear all registrations (testing only)."""
|
|
self._formats.clear()
|
|
|
|
|
|
# ─── module-level singleton ─────────────────────────────────────────────────
|
|
|
|
_registry: ImportExportFormatRegistry | None = None
|
|
|
|
|
|
def get_format_registry() -> ImportExportFormatRegistry:
|
|
global _registry
|
|
if _registry is None:
|
|
_registry = ImportExportFormatRegistry()
|
|
return _registry
|
|
|
|
|
|
def reset_format_registry_for_testing() -> ImportExportFormatRegistry:
|
|
global _registry
|
|
_registry = ImportExportFormatRegistry()
|
|
return _registry
|