diff --git a/src/basic_memory/models/documents.py b/src/basic_memory/models/documents.py index 9aca69e2..8e3c3f35 100644 --- a/src/basic_memory/models/documents.py +++ b/src/basic_memory/models/documents.py @@ -1,44 +1,42 @@ -"""Document model for tracking files in the knowledge base.""" +"""Models for storing documents.""" from datetime import datetime -from typing import Optional, List +from typing import Dict, Any, Optional -from sqlalchemy import String, DateTime, text, JSON, Index -from sqlalchemy.orm import Mapped, mapped_column, relationship +from sqlalchemy import String, JSON, DateTime +from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy.sql import func -from basic_memory.models.base import Base +from basic_memory.db import Base class Document(Base): - """ - 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. - """ - + """Document model.""" __tablename__ = "document" - __table_args__ = ( - Index("ix_document_created_at", "created_at"), - Index("ix_document_updated_at", "updated_at") - ) + __table_args__ = {'extend_existing': True} id: Mapped[int] = mapped_column(primary_key=True) - path: Mapped[str] = mapped_column(String, unique=True, nullable=False, index=True) - checksum: Mapped[str] = mapped_column(String, nullable=False, index=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")) + # Normalized path for URIs + 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() + ) updated_at: Mapped[datetime] = mapped_column( - DateTime, server_default=text("CURRENT_TIMESTAMP"), onupdate=text("CURRENT_TIMESTAMP") + DateTime(timezone=True), + server_default=func.now(), + onupdate=func.now() ) - # 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}')" + @property + def normalized_path(self) -> str: + """Get path for URI routing.""" + return self.path_id diff --git a/src/basic_memory/models/knowledge.py b/src/basic_memory/models/knowledge.py index f9956c90..a6f6e279 100644 --- a/src/basic_memory/models/knowledge.py +++ b/src/basic_memory/models/knowledge.py @@ -1,158 +1,110 @@ -"""Knowledge graph models.""" +"""Models for storing knowledge entities and their relationships.""" from datetime import datetime -from typing import Optional +from typing import List, Optional -from sqlalchemy import Integer, String, Text, ForeignKey, UniqueConstraint, text, DateTime, Index +from sqlalchemy import String, JSON, DateTime, ForeignKey from sqlalchemy.orm import Mapped, mapped_column, relationship +from sqlalchemy.sql import func -from basic_memory.models.base import Base -from basic_memory.models.documents import Document -from enum import Enum +from basic_memory.db import Base class Entity(Base): - """ - Core entity in the knowledge graph. - - 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 - """ - + """Entity model.""" __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 - ) + __table_args__ = {'extend_existing': True} - # Core identity - id: Mapped[int] = mapped_column(Integer, primary_key=True) + id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String) entity_type: Mapped[str] = mapped_column(String) - path_id: Mapped[str] = mapped_column(String, index=True) - - # Content and validation - description: Mapped[Optional[str]] = mapped_column(Text, nullable=True) - # nullable is true so we can insert the row before writing file - # after file write the checksum is updated - checksum: Mapped[Optional[str]] = mapped_column(String, nullable=True, index=True) - - # Metadata and tracking - 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") + # Normalized path for URIs + 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) + 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() ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + onupdate=func.now() + ) + doc_id: Mapped[Optional[int]] = mapped_column(ForeignKey("document.id"), nullable=True) # Relations - doc_id: Mapped[Optional[int]] = mapped_column( - Integer, ForeignKey("document.id", ondelete="SET NULL"), nullable=True + observations: Mapped[List["Observation"]] = relationship( + "Observation", + back_populates="entity", + cascade="all, delete-orphan" ) - - # Relationships - observations = relationship( - "Observation", back_populates="entity", cascade="all, delete-orphan" - ) - from_relations = relationship( + outbound_relations: Mapped[List["Relation"]] = relationship( "Relation", back_populates="from_entity", foreign_keys="[Relation.from_id]", - cascade="all, delete-orphan", + cascade="all, delete-orphan" ) - to_relations = relationship( + inbound_relations: Mapped[List["Relation"]] = 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, back_populates="entities") + document: Mapped[Optional["Document"]] = relationship("Document") @property - def relations(self): - return self.to_relations + self.from_relations + def normalized_path(self) -> str: + """Get path for URI routing.""" + return self.path_id - 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): - """ - An observation about an entity. - - Observations are atomic facts or notes about an entity. - """ - + """Model for entity observations.""" __tablename__ = "observation" - __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 + __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() ) - 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 + # Relations + entity: Mapped[Entity] = relationship( + "Entity", + back_populates="observations" ) - 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): - """ - A directed relation between two entities. - """ - + """Model for entity relationships.""" __tablename__ = "relation" - __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 - ) + __table_args__ = {'extend_existing': True} - 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")) + 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")) relation_type: Mapped[str] = mapped_column(String) - 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") + metadata: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now() ) - # 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}')" + # 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