File size: 5,522 Bytes
c14ceee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | """Harness core — the platform contract that turns 'a dashboard' into 'a product'.
Three concepts:
Connector a source adapter (Odoo / QuickBooks / Amazon / …). Declares its capabilities and
answers canonical Queries. Adding a source = subclass Connector.
Tenant one client (Royal Imports, …). Bundles the connectors they've hooked up + their scope
config (business units, excluded accounts, channel map, branding). Multi-tenant lives here.
Skill a portable dashboard/metric written ONCE against the canonical model; it declares the
capabilities it needs so a tenant only ever sees skills its sources can serve.
The UI (app.py) becomes a thin renderer over (tenant → available skills). Nothing above the
connector layer knows what Odoo is.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Callable
from .canonical import Capability, Query, Entity
class Connector(ABC):
"""A data source. Implement `capabilities()` + `aggregate()` (and optionally `records()`), map the
canonical model to the source, and the source is usable by every compatible skill."""
key: str = 'connector'
name: str = 'Connector'
@abstractmethod
def capabilities(self) -> set[Capability]:
"""The canonical capabilities this source can serve for the given account."""
@abstractmethod
def aggregate(self, q: Query) -> list[dict]:
"""Answer an aggregate Query → list of canonical dicts (keys = measures/group_by, + 'count')."""
def records(self, q: Query) -> list[dict]:
"""Optional row-level fetch (canonical dicts). Default: not supported."""
raise NotImplementedError(f'{self.key}: row-level records() not implemented')
def distinct_count(self, q: Query, fieldname: str) -> int:
"""Number of distinct values of a canonical field over the query's scope (e.g. active
customers). Default: derive from an aggregate group_by; connectors may override efficiently."""
from dataclasses import replace
rows = self.aggregate(replace(q, measures=(), aggregate='count', group_by=(fieldname,)))
return sum(1 for r in rows if r.get(fieldname) not in (None, False))
def supports(self, cap: Capability) -> bool:
return cap in self.capabilities()
def entities(self) -> set[Entity]:
"""Canonical entities this connector can materialize (advertised by subclasses)."""
return set()
@dataclass
class Tenant:
"""One client of the platform. `sources` maps a connector key → Connector instance. `config` holds
the client's scope semantics (which is what used to be hard-coded in core/odoo.py):
business_units {label: source_scope_value} e.g. {'Fisch': 5, 'Royal': 6}
confirmed_statuses [...] which order states count as 'real'
excluded_customers [ids] house / inter-co accounts to drop
channels {label: match} e.g. {'Amazon': <partner>} (marketplace split)
branding {...} palette, name, logo
A skill reads tenant.config for scope and asks tenant.source(capability) for the right connector."""
key: str
name: str
sources: dict[str, Connector] = field(default_factory=dict)
config: dict = field(default_factory=dict)
def add(self, conn: Connector) -> 'Tenant':
self.sources[conn.key] = conn
return self
def capabilities(self) -> set[Capability]:
caps: set[Capability] = set()
for c in self.sources.values():
caps |= c.capabilities()
return caps
def source(self, cap: Capability, prefer: str | None = None) -> Connector:
"""The connector that serves `cap` (a tenant may hook several sources; `prefer` picks one,
else first that supports it). Raises if no connected source provides the capability."""
if prefer and prefer in self.sources and self.sources[prefer].supports(cap):
return self.sources[prefer]
for c in self.sources.values():
if c.supports(cap):
return c
raise LookupError(f"tenant {self.key!r} has no source for capability {cap.value!r}")
def has(self, *caps: Capability) -> bool:
mine = self.capabilities()
return all(c in mine for c in caps)
# --------------------------------------------------------------------------- skill registry
@dataclass(frozen=True)
class Skill:
"""A portable dashboard. `requires` are the capabilities it needs; the harness only offers it to a
tenant whose connected sources cover them. `render` is the UI entrypoint (given tenant + context)."""
key: str
label: str
requires: tuple[Capability, ...]
render: Callable | None = None
hq: bool = False # consolidated / company-level (ignores the business-unit filter)
def available_for(self, tenant: Tenant) -> bool:
return tenant.has(*self.requires)
_SKILLS: dict[str, Skill] = {}
def register_skill(skill: Skill) -> Skill:
_SKILLS[skill.key] = skill
return skill
def skills_for(tenant: Tenant) -> list[Skill]:
"""The skills a tenant can actually run, given its connected sources — this is the dynamic nav."""
return [s for s in _SKILLS.values() if s.available_for(tenant)]
def all_skills() -> list[Skill]:
return list(_SKILLS.values())
|