mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
Implement global default storage pattern for ProvenanceManager
This commit is contained in:
@@ -90,6 +90,15 @@ class Semantica:
|
||||
)
|
||||
self.lifecycle_manager.register_component("config_manager", self.config_manager)
|
||||
|
||||
# Configure global provenance storage path if specified
|
||||
try:
|
||||
prov_storage_path = self.config.get("provenance", {}).get("storage_path")
|
||||
if prov_storage_path:
|
||||
from ..provenance import ProvenanceManager
|
||||
ProvenanceManager.set_default_storage_path(prov_storage_path)
|
||||
except Exception as e:
|
||||
self.logger.warning(f"Failed to configure provenance storage: {e}")
|
||||
|
||||
# Module placeholders (to be initialized)
|
||||
self._modules: Dict[str, Any] = {}
|
||||
self._initialized: bool = False
|
||||
|
||||
@@ -54,10 +54,19 @@ class ProvenanceManager:
|
||||
>>> lineage = prov_mgr.get_lineage("entity_1")
|
||||
"""
|
||||
|
||||
_default_storage_path: Optional[str] = None
|
||||
|
||||
@classmethod
|
||||
def set_default_storage_path(cls, path: str) -> None:
|
||||
"""Set a global default storage path for all new instances."""
|
||||
cls._default_storage_path = path
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
storage: Optional[ProvenanceStorage] = None,
|
||||
storage_path: Optional[str] = None
|
||||
storage_path: Optional[str] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
**kwargs
|
||||
):
|
||||
"""
|
||||
Initialize provenance manager.
|
||||
@@ -65,10 +74,19 @@ class ProvenanceManager:
|
||||
Args:
|
||||
storage: Custom storage backend (optional)
|
||||
storage_path: Path to SQLite database (optional, uses in-memory if None)
|
||||
config: Configuration dictionary (optional)
|
||||
"""
|
||||
if storage:
|
||||
self.storage = storage
|
||||
elif storage_path:
|
||||
return
|
||||
|
||||
if not storage_path and config:
|
||||
storage_path = config.get("provenance", {}).get("storage_path")
|
||||
|
||||
if not storage_path and self._default_storage_path:
|
||||
storage_path = self._default_storage_path
|
||||
|
||||
if storage_path:
|
||||
self.storage = SQLiteStorage(storage_path)
|
||||
else:
|
||||
self.storage = InMemoryStorage()
|
||||
|
||||
Reference in New Issue
Block a user