mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
add file_path to models
This commit is contained in:
phernandez
@@ -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
|
||||
|
||||
@@ -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]
|
||||
)
|
||||
Reference in New Issue
Block a user