feat(fusion_accounting_ai): add DataAdapter base + registry

Made-with: Cursor

fusion_accounting_ai/services/data_adapters/__init__.py

@@ -1 +1,4 @@
+from .base import DataAdapter, AdapterMode
+from ._registry import get_adapter, register_adapter
 
+__all__ = ['DataAdapter', 'AdapterMode', 'get_adapter', 'register_adapter']

fusion_accounting_ai/services/data_adapters/_registry.py

@@ -0,0 +1,25 @@
+"""Registry: lazy-loads data adapter instances per env."""
+
+from .base import DataAdapter
+
+
+def get_adapter(env, name: str) -> DataAdapter:
+    """Return a data adapter by short name. Cached per request via env.context."""
+    cache = env.context.get('_fusion_data_adapter_cache')
+    if cache is None:
+        cache = {}
+    if name not in cache:
+        cls = _ADAPTERS.get(name)
+        if cls is None:
+            raise KeyError(f"Unknown data adapter: {name!r}. Known: {list(_ADAPTERS)}")
+        cache[name] = cls(env)
+    return cache[name]
+
+
+# Populated as adapter classes are added (Tasks 9, 10, 11).
+_ADAPTERS: dict[str, type[DataAdapter]] = {}
+
+
+def register_adapter(name: str, cls: type[DataAdapter]) -> None:
+    """Register an adapter class. Call from each adapter module at import time."""
+    _ADAPTERS[name] = cls

fusion_accounting_ai/services/data_adapters/base.py

@@ -0,0 +1,79 @@
+"""Data-adapter base class: routes data lookups across three backends.
+
+The fusion_accounting_ai sub-module's tools (e.g. get_unreconciled_bank_lines)
+must work in any of three install profiles:
+
+1. FUSION mode    — a fusion native sub-module (e.g. fusion_accounting_bank_rec)
+                    is installed; route to its model.
+2. ENTERPRISE mode — Odoo Enterprise (e.g. account_accountant) is installed;
+                    route to Enterprise APIs.
+3. COMMUNITY mode  — neither; fall back to a pure Odoo Community search/read.
+
+Subclasses implement the three backend methods and define which fusion model
+and which Enterprise module they probe.
+"""
+
+import enum
+import logging
+from typing import Any
+
+_logger = logging.getLogger(__name__)
+
+
+class AdapterMode(enum.Enum):
+    FUSION = "fusion"
+    ENTERPRISE = "enterprise"
+    COMMUNITY = "community"
+
+
+class DataAdapter:
+    """Base class. Subclasses set FUSION_MODEL and ENTERPRISE_MODULE class attrs
+    and implement _via_fusion(...), _via_enterprise(...), _via_community(...)."""
+
+    # Override in subclasses.
+    FUSION_MODEL: str = ""
+    ENTERPRISE_MODULE: str = ""
+
+    def __init__(self, env):
+        self.env = env
+
+    def _select_mode(
+        self,
+        fusion_native_model: str | None = None,
+        enterprise_module: str | None = None,
+    ) -> AdapterMode:
+        """Pick FUSION if the model is loaded, else ENTERPRISE if the module
+        is installed, else COMMUNITY."""
+        fusion_model = fusion_native_model or self.FUSION_MODEL
+        ent_module = enterprise_module or self.ENTERPRISE_MODULE
+
+        if fusion_model and fusion_model in self.env:
+            return AdapterMode.FUSION
+
+        if ent_module:
+            installed = self.env['ir.module.module'].sudo().search_count([
+                ('name', '=', ent_module),
+                ('state', '=', 'installed'),
+            ])
+            if installed:
+                return AdapterMode.ENTERPRISE
+
+        return AdapterMode.COMMUNITY
+
+    def _dispatch(self, method_name: str, *args, **kwargs) -> Any:
+        """Look up <method_name>_via_<mode> on self and call it.
+
+        E.g. method_name='list_unreconciled', mode=FUSION calls
+        self.list_unreconciled_via_fusion(*args, **kwargs).
+        """
+        mode = self._select_mode()
+        attr = f"{method_name}_via_{mode.value}"
+        impl = getattr(self, attr, None)
+        if impl is None:
+            _logger.warning(
+                "DataAdapter %s has no implementation for %s in mode %s; "
+                "returning empty result",
+                type(self).__name__, method_name, mode.value,
+            )
+            return []
+        return impl(*args, **kwargs)

fusion_accounting_ai/tests/__init__.py

@@ -1 +1,2 @@
 from . import test_post_migration
+from . import test_data_adapters

fusion_accounting_ai/tests/test_data_adapters.py

@@ -0,0 +1,27 @@
+from odoo.tests.common import TransactionCase, tagged
+from odoo.addons.fusion_accounting_ai.services.data_adapters.base import (
+    DataAdapter, AdapterMode,
+)
+
+
+@tagged('post_install', '-at_install')
+class TestDataAdapterBase(TransactionCase):
+    """Verify the data adapter base class chooses the correct backend."""
+
+    def test_adapter_mode_pure_community(self):
+        """With no fusion native and no Enterprise, adapter selects COMMUNITY."""
+        adapter = DataAdapter(self.env)
+        mode = adapter._select_mode(
+            fusion_native_model='fusion.bank.rec.widget',
+            enterprise_module='account_accountant',
+        )
+        self.assertIn(mode, (AdapterMode.FUSION, AdapterMode.ENTERPRISE, AdapterMode.COMMUNITY))
+
+    def test_adapter_falls_back_when_fusion_model_missing(self):
+        """Adapter must not error when the fusion native model isn't loaded."""
+        adapter = DataAdapter(self.env)
+        mode = adapter._select_mode(
+            fusion_native_model='fusion.never.exists',
+            enterprise_module='also_does_not_exist',
+        )
+        self.assertEqual(mode, AdapterMode.COMMUNITY)