148 lines
5.1 KiB
Python
148 lines
5.1 KiB
Python
"""
|
|
Context Database - Storage and retrieval of context blobs.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from typing import List, Optional, Dict, Any
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text, Float
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
class ContextBlob(Base):
|
|
"""Database model for context blobs."""
|
|
|
|
__tablename__ = "context_blobs"
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
path = Column(String(512), nullable=False, index=True)
|
|
content = Column(Text, nullable=False)
|
|
summary = Column(Text)
|
|
embedding_model = Column(String(100))
|
|
embedding_vector = Column(Text) # JSON serialized vector
|
|
author = Column(String(100))
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
version = Column(Integer, default=1)
|
|
|
|
|
|
@dataclass
|
|
class Context:
|
|
"""Context data structure."""
|
|
id: Optional[int]
|
|
path: str
|
|
content: str
|
|
summary: Optional[str] = None
|
|
author: Optional[str] = None
|
|
created_at: Optional[datetime] = None
|
|
updated_at: Optional[datetime] = None
|
|
version: int = 1
|
|
|
|
|
|
class ContextDatabase:
|
|
"""Main interface for context storage and retrieval."""
|
|
|
|
def __init__(self, db_path: str = "hcfs_context.db"):
|
|
self.db_path = db_path
|
|
self.engine = create_engine(f"sqlite:///{db_path}")
|
|
Base.metadata.create_all(self.engine)
|
|
self.SessionLocal = sessionmaker(bind=self.engine)
|
|
|
|
def get_session(self) -> Session:
|
|
"""Get database session."""
|
|
return self.SessionLocal()
|
|
|
|
def store_context(self, context: Context) -> int:
|
|
"""Store a context blob and return its ID."""
|
|
with self.get_session() as session:
|
|
blob = ContextBlob(
|
|
path=context.path,
|
|
content=context.content,
|
|
summary=context.summary,
|
|
author=context.author,
|
|
version=context.version
|
|
)
|
|
session.add(blob)
|
|
session.commit()
|
|
session.refresh(blob)
|
|
return blob.id
|
|
|
|
def get_context_by_path(self, path: str, depth: int = 1) -> List[Context]:
|
|
"""Retrieve contexts for a path and optionally parent paths."""
|
|
contexts = []
|
|
current_path = Path(path)
|
|
|
|
with self.get_session() as session:
|
|
# Get contexts for current path and parents up to depth
|
|
for i in range(depth + 1):
|
|
search_path = str(current_path) if current_path != Path(".") else "/"
|
|
|
|
blobs = session.query(ContextBlob).filter(
|
|
ContextBlob.path == search_path
|
|
).order_by(ContextBlob.created_at.desc()).all()
|
|
|
|
for blob in blobs:
|
|
contexts.append(Context(
|
|
id=blob.id,
|
|
path=blob.path,
|
|
content=blob.content,
|
|
summary=blob.summary,
|
|
author=blob.author,
|
|
created_at=blob.created_at,
|
|
updated_at=blob.updated_at,
|
|
version=blob.version
|
|
))
|
|
|
|
if current_path.parent == current_path: # Root reached
|
|
break
|
|
current_path = current_path.parent
|
|
|
|
return contexts
|
|
|
|
def list_contexts_at_path(self, path: str) -> List[Context]:
|
|
"""List all contexts at a specific path."""
|
|
with self.get_session() as session:
|
|
blobs = session.query(ContextBlob).filter(
|
|
ContextBlob.path == path
|
|
).order_by(ContextBlob.created_at.desc()).all()
|
|
|
|
return [Context(
|
|
id=blob.id,
|
|
path=blob.path,
|
|
content=blob.content,
|
|
summary=blob.summary,
|
|
author=blob.author,
|
|
created_at=blob.created_at,
|
|
updated_at=blob.updated_at,
|
|
version=blob.version
|
|
) for blob in blobs]
|
|
|
|
def update_context(self, context_id: int, content: str, summary: str = None) -> bool:
|
|
"""Update an existing context."""
|
|
with self.get_session() as session:
|
|
blob = session.query(ContextBlob).filter(ContextBlob.id == context_id).first()
|
|
if blob:
|
|
blob.content = content
|
|
if summary:
|
|
blob.summary = summary
|
|
blob.version += 1
|
|
blob.updated_at = datetime.utcnow()
|
|
session.commit()
|
|
return True
|
|
return False
|
|
|
|
def delete_context(self, context_id: int) -> bool:
|
|
"""Delete a context by ID."""
|
|
with self.get_session() as session:
|
|
blob = session.query(ContextBlob).filter(ContextBlob.id == context_id).first()
|
|
if blob:
|
|
session.delete(blob)
|
|
session.commit()
|
|
return True
|
|
return False |