File size: 3,640 Bytes
fec5550 | 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 | """
AETERNA AI — Base Data Source Architecture
Defines the abstract interface that all data connectors must implement.
Every record returned must include provenance metadata so the system
can transparently communicate data origin to users and stakeholders.
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from enum import Enum
from datetime import datetime
class ProvenanceType(str, Enum):
"""
Formal classification of data provenance.
OBSERVED — Directly measured by an authoritative body
DERIVED — Mathematically computed from observed sources
SYNTHETIC — Procedurally generated by simulation
EXTERNAL_REALTIME — Fetched from a live third-party public API
MODEL_OUTPUT — Produced by an ML or simulation model
UNVERIFIED — Origin unclear or not yet validated
"""
OBSERVED = "OBSERVED"
DERIVED = "DERIVED"
SYNTHETIC = "SYNTHETIC"
EXTERNAL_REALTIME = "EXTERNAL_REALTIME"
MODEL_OUTPUT = "MODEL_OUTPUT"
UNVERIFIED = "UNVERIFIED"
@dataclass
class DataRecord:
"""
A single normalized data record with full provenance metadata.
"""
value: Any
field_name: str
provenance: ProvenanceType
source_name: str
source_url: Optional[str] = None
geographic_granularity: Optional[str] = None
temporal_granularity: Optional[str] = None
observation_date: Optional[str] = None
fetched_at: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
limitations: Optional[str] = None
validation_status: str = "UNVALIDATED"
extra: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"value": self.value,
"field_name": self.field_name,
"provenance": self.provenance.value,
"source_name": self.source_name,
"source_url": self.source_url,
"geographic_granularity": self.geographic_granularity,
"temporal_granularity": self.temporal_granularity,
"observation_date": self.observation_date,
"fetched_at": self.fetched_at,
"limitations": self.limitations,
"validation_status": self.validation_status,
**self.extra,
}
class BaseDataSource(ABC):
"""
Abstract base class for all AETERNA AI data connectors.
Every connector must implement:
- is_available(): Check if the source is accessible
- fetch(): Return normalized DataRecord list with provenance
"""
SOURCE_NAME: str = "Unknown"
SOURCE_URL: Optional[str] = None
IS_STUB: bool = True # True if not yet connected to live data
@abstractmethod
def is_available(self) -> bool:
"""
Returns True if the data source is currently accessible.
NEVER fabricate data if the source is unavailable — return False.
"""
...
@abstractmethod
def fetch(self, **kwargs) -> List[DataRecord]:
"""
Fetch data from the source and return normalized DataRecord objects.
NEVER return fabricated records — raise NotImplementedError or return empty list
if the source is unavailable or credentials are missing.
"""
...
def get_status(self) -> Dict[str, Any]:
"""Return connection status metadata for diagnostics."""
return {
"source_name": self.SOURCE_NAME,
"source_url": self.SOURCE_URL,
"is_stub": self.IS_STUB,
"is_available": self.is_available(),
}
|