diff --git a/src/basic_memory/models/documents.py b/src/basic_memory/models/documents.py index 8e3c3f35..6cb04921 100644 --- a/src/basic_memory/models/documents.py +++ b/src/basic_memory/models/documents.py @@ -1,42 +1,49 @@ -"""Models for storing documents.""" +"""Document model for tracking files in the knowledge base.""" from datetime import datetime -from typing import Dict, Any, Optional +from typing import Optional, List -from sqlalchemy import String, JSON, DateTime -from sqlalchemy.orm import Mapped, mapped_column -from sqlalchemy.sql import func +from sqlalchemy import String, DateTime, text, JSON, Index +from sqlalchemy.orm import Mapped, mapped_column, relationship -from basic_memory.db import Base +from basic_memory.models.base import Base class Document(Base): - """Document model.""" + """ + Tracks documents in the filesystem. + + Documents files are the source of truth for document content, while this table + provides indexing and metadata storage. Like git, the filesystem + is the real source of truth. + """ + __tablename__ = "document" - __table_args__ = {'extend_existing': True} + __table_args__ = ( + Index("ix_document_created_at", "created_at"), + Index("ix_document_updated_at", "updated_at") + ) id: Mapped[int] = mapped_column(primary_key=True) # Normalized path for URIs - path_id: Mapped[str] = mapped_column(String, unique=True, index=True) + # path_id is unique for documents + path_id: Mapped[str] = mapped_column(String, unique=True,index=True) # Actual filesystem relative path - file_path: Mapped[str] = mapped_column(String, unique=True, index=True) - checksum: Mapped[str] = mapped_column(String, nullable=False) - doc_metadata: Mapped[Optional[Dict[str, Any]]] = mapped_column( - JSON, - nullable=True, - default=None - ) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), - server_default=func.now() - ) + file_path: Mapped[str] = mapped_column(String, unique=True, index=True) + + checksum: Mapped[str] = mapped_column(String, nullable=True) + doc_metadata: Mapped[Optional[dict]] = mapped_column( + JSON, nullable=True + ) # renamed from metadata + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP")) updated_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), - server_default=func.now(), - onupdate=func.now() + DateTime, server_default=text("CURRENT_TIMESTAMP"), onupdate=text("CURRENT_TIMESTAMP") ) - @property - def normalized_path(self) -> str: - """Get path for URI routing.""" - return self.path_id + # Relationships + entities: Mapped[List["Entity"]] = relationship( # pyright: ignore [reportUndefinedVariable] # noqa: F821 + "Entity", back_populates="document", cascade="all, delete-orphan" + ) + + def __repr__(self) -> str: + return f"Document(id={self.id}, path='{self.path}', checksum='{self.checksum}', created_at='{self.created_at}', updated_at='{self.updated_at}')" diff --git a/src/basic_memory/models/knowledge.py b/src/basic_memory/models/knowledge.py index a6f6e279..c2a3bfbf 100644 --- a/src/basic_memory/models/knowledge.py +++ b/src/basic_memory/models/knowledge.py @@ -1,110 +1,160 @@ -"""Models for storing knowledge entities and their relationships.""" +"""Knowledge graph models.""" from datetime import datetime -from typing import List, Optional +from typing import Optional -from sqlalchemy import String, JSON, DateTime, ForeignKey +from sqlalchemy import Integer, String, Text, ForeignKey, UniqueConstraint, text, DateTime, Index from sqlalchemy.orm import Mapped, mapped_column, relationship -from sqlalchemy.sql import func -from basic_memory.db import Base +from basic_memory.models.base import Base +from basic_memory.models.documents import Document +from enum import Enum class Entity(Base): - """Entity model.""" - __tablename__ = "entity" - __table_args__ = {'extend_existing': True} + """ + Core entity in the knowledge graph. - id: Mapped[int] = mapped_column(primary_key=True) + Entities represent semantic nodes maintained by the AI layer. Each entity: + - Has a unique numeric ID (database-generated) + - Maps to a document file on disk (optional) + - Maintains a checksum for change detection + - Tracks both source document and semantic properties + """ + + __tablename__ = "entity" + __table_args__ = ( + UniqueConstraint("entity_type", "name", name="uix_entity_type_name"), + Index("ix_entity_type", "entity_type"), + Index("ix_entity_doc_id", "doc_id"), + Index("ix_entity_created_at", "created_at"), # For timeline queries + Index("ix_entity_updated_at", "updated_at") # For timeline queries + ) + + # Core identity + id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String) entity_type: Mapped[str] = mapped_column(String) # Normalized path for URIs - path_id: Mapped[str] = mapped_column(String, unique=True, index=True) + # (entity_type, path_id) are unique + path_id: Mapped[str] = mapped_column(String, index=True) # Actual filesystem relative path - file_path: Mapped[str] = mapped_column(String, unique=True, index=True) - description: Mapped[str] = mapped_column(String) - checksum: Mapped[str] = mapped_column(String, nullable=False) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), - server_default=func.now() - ) + file_path: Mapped[str] = mapped_column(String, unique=True, index=True) + + # Content and validation + description: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + checksum: Mapped[Optional[str]] = mapped_column(String, nullable=True) + + # Metadata and tracking + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP")) updated_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), - server_default=func.now(), - onupdate=func.now() + DateTime, server_default=text("CURRENT_TIMESTAMP"), onupdate=text("CURRENT_TIMESTAMP") ) - doc_id: Mapped[Optional[int]] = mapped_column(ForeignKey("document.id"), nullable=True) # Relations - observations: Mapped[List["Observation"]] = relationship( - "Observation", - back_populates="entity", - cascade="all, delete-orphan" + doc_id: Mapped[Optional[int]] = mapped_column( + Integer, ForeignKey("document.id", ondelete="SET NULL"), nullable=True ) - outbound_relations: Mapped[List["Relation"]] = relationship( + + # Relationships + observations = relationship( + "Observation", back_populates="entity", cascade="all, delete-orphan" + ) + from_relations = relationship( "Relation", back_populates="from_entity", foreign_keys="[Relation.from_id]", - cascade="all, delete-orphan" + cascade="all, delete-orphan", ) - inbound_relations: Mapped[List["Relation"]] = relationship( + to_relations = relationship( "Relation", back_populates="to_entity", foreign_keys="[Relation.to_id]", - cascade="all, delete-orphan" + cascade="all, delete-orphan", ) - document: Mapped[Optional["Document"]] = relationship("Document") + document: Mapped[Optional[Document]] = relationship(Document, back_populates="entities") @property - def normalized_path(self) -> str: - """Get path for URI routing.""" - return self.path_id + def relations(self): + return self.to_relations + self.from_relations + def __repr__(self) -> str: + return f"Entity(id={self.id}, name='{self.name}', type='{self.entity_type}')" + + +class ObservationCategory(str, Enum): + TECH = "tech" + DESIGN = "design" + FEATURE = "feature" + NOTE = "note" + ISSUE = "issue" + TODO = "todo" class Observation(Base): - """Model for entity observations.""" + """ + An observation about an entity. + + Observations are atomic facts or notes about an entity. + """ + __tablename__ = "observation" - __table_args__ = {'extend_existing': True} - - id: Mapped[int] = mapped_column(primary_key=True) - entity_id: Mapped[int] = mapped_column(ForeignKey("entity.id")) - content: Mapped[str] = mapped_column(String) - metadata: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), - server_default=func.now() + __table_args__ = ( + Index("ix_observation_entity_id", "entity_id"), # Add FK index + Index("ix_observation_category", "category"), # Add category index + Index("ix_observation_created_at", "created_at"), # For timeline queries + Index("ix_observation_updated_at", "updated_at"), # For timeline queries ) - # Relations - entity: Mapped[Entity] = relationship( - "Entity", - back_populates="observations" + id: Mapped[int] = mapped_column(Integer, primary_key=True) + entity_id: Mapped[int] = mapped_column(Integer, ForeignKey("entity.id", ondelete="CASCADE")) + content: Mapped[str] = mapped_column(Text) + category: Mapped[str] = mapped_column( + String, + nullable=False, + default=ObservationCategory.NOTE.value, + server_default=ObservationCategory.NOTE.value ) + context: Mapped[str] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP")) + updated_at: Mapped[datetime] = mapped_column( + DateTime, server_default=text("CURRENT_TIMESTAMP"), onupdate=text("CURRENT_TIMESTAMP") + ) + + # Relationships + entity = relationship("Entity", back_populates="observations") + + def __repr__(self) -> str: + return f"Observation(id={self.id}, entity_id={self.entity_id}, content='{self.content}')" class Relation(Base): - """Model for entity relationships.""" + """ + A directed relation between two entities. + """ + __tablename__ = "relation" - __table_args__ = {'extend_existing': True} + __table_args__ = ( + UniqueConstraint("from_id", "to_id", "relation_type", name="uix_relation"), + Index("ix_relation_type", "relation_type"), + Index("ix_relation_from_id", "from_id"), # Add FK indexes + Index("ix_relation_to_id", "to_id"), + Index("ix_relation_created_at", "created_at"), # For timeline queries + Index("ix_relation_updated_at", "updated_at"), # For timeline queries + ) - id: Mapped[int] = mapped_column(primary_key=True) - from_id: Mapped[int] = mapped_column(ForeignKey("entity.id")) - to_id: Mapped[int] = mapped_column(ForeignKey("entity.id")) + id: Mapped[int] = mapped_column(Integer, primary_key=True) + from_id: Mapped[int] = mapped_column(Integer, ForeignKey("entity.id", ondelete="CASCADE")) + to_id: Mapped[int] = mapped_column(Integer, ForeignKey("entity.id", ondelete="CASCADE")) relation_type: Mapped[str] = mapped_column(String) - metadata: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), - server_default=func.now() + context: Mapped[str] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP")) + updated_at: Mapped[datetime] = mapped_column( + DateTime, server_default=text("CURRENT_TIMESTAMP"), onupdate=text("CURRENT_TIMESTAMP") ) - # Relations - from_entity: Mapped[Entity] = relationship( - "Entity", - back_populates="outbound_relations", - foreign_keys=[from_id] - ) - to_entity: Mapped[Entity] = relationship( - "Entity", - back_populates="inbound_relations", - foreign_keys=[to_id] - ) \ No newline at end of file + # Relationships + from_entity = relationship("Entity", foreign_keys=[from_id], back_populates="from_relations") + to_entity = relationship("Entity", foreign_keys=[to_id], back_populates="to_relations") + + def __repr__(self) -> str: + return f"Relation(id={self.id}, from_id={self.from_id}, to_id={self.to_id}, type='{self.relation_type}')"