Spaces:
Sleeping
Sleeping
File size: 4,651 Bytes
92c4ae6 | 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 125 126 127 128 129 130 131 132 | """
Luuna Autoflow Core - Memory Layer
=================================
Simple local storage for execution records.
Uses JSON file storage (dev-safe, no production DB).
Storage location: backend/.autoflow/executions.json
This path is gitignored to prevent committing execution data.
"""
import json
import logging
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime
from .models import ExecutionRecord
logger = logging.getLogger(__name__)
# Default storage directory (gitignored, dev-safe)
DEFAULT_STORAGE_DIR = Path(__file__).parent.parent / ".autoflow"
DEFAULT_STORAGE_FILE = "executions.json"
class MemoryStore:
"""
Simple local JSON storage for execution records.
For development only - production should use proper DB.
Storage path: backend/.autoflow/executions.json
This is intentionally outside repo tracking to keep execution data safe.
"""
def __init__(self, storage_path: Optional[str] = None):
"""
Initialize memory store.
Args:
storage_path: Optional custom path to JSON file for storage.
If not provided, uses backend/.autoflow/executions.json
"""
if storage_path:
self.storage_path = Path(storage_path)
else:
# Default to backend/.autoflow/executions.json (gitignored, dev-safe)
self.storage_path = DEFAULT_STORAGE_DIR / DEFAULT_STORAGE_FILE
# Ensure directory exists
self.storage_path.parent.mkdir(parents=True, exist_ok=True)
# Initialize storage
self._records: Dict[str, dict] = {}
self._load()
logger.info(f"[MemoryStore] Initialized at: {self.storage_path}")
def _load(self) -> None:
"""Load records from disk."""
if self.storage_path.exists():
try:
with open(self.storage_path, "r", encoding="utf-8") as f:
self._records = json.load(f)
logger.debug(f"Loaded {len(self._records)} execution records")
except (json.JSONDecodeError, IOError) as e:
logger.warning(f"Failed to load records: {e}")
self._records = {}
def _save(self) -> None:
"""Save records to disk."""
try:
with open(self.storage_path, "w", encoding="utf-8") as f:
json.dump(self._records, f, indent=2, default=str, ensure_ascii=False)
logger.debug(f"Saved {len(self._records)} execution records")
except IOError as e:
logger.error(f"Failed to save records: {e}")
def store(self, record: ExecutionRecord) -> None:
"""Store an execution record."""
self._records[record.execution_id] = record.model_dump()
self._save()
logger.debug(f"Stored execution record: {record.execution_id}")
def get(self, execution_id: str) -> Optional[ExecutionRecord]:
"""Retrieve an execution record by ID."""
data = self._records.get(execution_id)
if data:
return ExecutionRecord(**data)
return None
def update(self, record: ExecutionRecord) -> None:
"""Update an existing record."""
record.updated_at = datetime.utcnow()
self._records[record.execution_id] = record.model_dump()
self._save()
logger.debug(f"Updated execution record: {record.execution_id}")
def delete(self, execution_id: str) -> bool:
"""Delete a record by ID."""
if execution_id in self._records:
del self._records[execution_id]
self._save()
return True
return False
def list_all(self) -> list:
"""List all execution records."""
return [ExecutionRecord(**data) for data in self._records.values()]
def list_by_status(self, status: str) -> List[ExecutionRecord]:
"""List records by status."""
return [
ExecutionRecord(**data)
for data in self._records.values()
if data.get("status") == status
]
def clear(self) -> None:
"""Clear all records."""
self._records = {}
self._save()
def get_storage_path(self) -> str:
"""Get the current storage path for reporting."""
return str(self.storage_path)
def count(self) -> int:
"""Return total number of stored records."""
return len(self._records)
|