Spaces:
Running
Running
File size: 2,207 Bytes
2305b9f 4719066 2305b9f 4719066 2305b9f 4719066 2305b9f 4719066 2305b9f | 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 | """Abstract base class for all OrgOS app modules."""
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Tuple
from server.schema_drift import SchemaDriftEngine
class BaseApp(ABC):
APP_NAME: str = ""
def __init__(self, drift: SchemaDriftEngine):
self._drift = drift
# ------------------------------------------------------------------
# Core interface — every app must implement these
# ------------------------------------------------------------------
@abstractmethod
def initialize(self, records: List[Dict]) -> None:
"""Load synthetic records into in-memory state."""
@abstractmethod
def execute(self, operation: str, args: Dict) -> Dict:
"""
Execute an operation.
Returns dict with at minimum:
{"success": bool, "message": str}
May also include:
{"data": ..., "schema_error": str, "schema_adapted": bool, "ticket": dict}
"""
@abstractmethod
def get_state_view(self, max_rows: int = 5) -> str:
"""Return agent-visible snapshot as a compact multi-line string."""
@abstractmethod
def count_open_items(self) -> int:
"""Count pending/open work items (used by grader)."""
# ------------------------------------------------------------------
# Shared helpers available to all concrete apps
# ------------------------------------------------------------------
def _check_schema_drift(self, args: Dict) -> Tuple[Optional[str], bool]:
"""
Delegate to the drift engine to check if args use stale canonical names.
Returns (schema_error_field_or_None, schema_adapted_bool).
"""
return self._drift.check_args_for_drift(args, self.APP_NAME)
def _to_agent_view(self, record: Dict) -> Dict:
"""Translate a canonical record to the agent-visible drifted representation."""
return self._drift.translate_record(record, self.APP_NAME)
def _compact(self, record: Dict, fields: List[str]) -> Dict:
"""Return only the specified fields from a (possibly drifted) record."""
return {k: v for k, v in record.items() if k in fields and v is not None}
|