mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Merge branch 'document-store' into document-entity-write
This commit is contained in:
phernandez
@@ -0,0 +1,13 @@
|
||||
"""Models package for basic-memory."""
|
||||
|
||||
from basic_memory.models.base import Base
|
||||
from basic_memory.models.documents import Document
|
||||
from basic_memory.models.knowledge import Entity, Observation, Relation
|
||||
|
||||
__all__ = [
|
||||
'Base',
|
||||
'Document',
|
||||
'Entity',
|
||||
'Observation',
|
||||
'Relation'
|
||||
]
|
||||
@@ -0,0 +1,9 @@
|
||||
"""Base model class for SQLAlchemy models."""
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncAttrs
|
||||
from sqlalchemy.orm import DeclarativeBase
|
||||
|
||||
|
||||
class Base(AsyncAttrs, DeclarativeBase):
|
||||
"""Base class for all models"""
|
||||
pass
|
||||
@@ -0,0 +1,40 @@
|
||||
"""Document model for tracking files in the knowledge base."""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Optional, List
|
||||
|
||||
from sqlalchemy import String, DateTime, text, JSON
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from basic_memory.models.base import Base
|
||||
|
||||
|
||||
class Document(Base):
|
||||
"""
|
||||
Tracks documents in the filesystem.
|
||||
|
||||
Documents are the source of truth for content, while this table
|
||||
provides indexing and metadata storage. Like git, the filesystem
|
||||
is the real source of truth.
|
||||
"""
|
||||
|
||||
__tablename__ = "documents"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
path: Mapped[str] = mapped_column(String, unique=True, nullable=False)
|
||||
checksum: Mapped[str] = mapped_column(String, nullable=False)
|
||||
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, server_default=text("CURRENT_TIMESTAMP"), onupdate=text("CURRENT_TIMESTAMP")
|
||||
)
|
||||
|
||||
# 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}')"
|
||||
@@ -0,0 +1,165 @@
|
||||
"""Knowledge graph models for basic-memory."""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlalchemy import (
|
||||
String, DateTime, ForeignKey, Text, Integer,
|
||||
text, UniqueConstraint
|
||||
)
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from basic_memory.models.base import Base
|
||||
from basic_memory.utils import sanitize_name
|
||||
|
||||
|
||||
class Entity(Base):
|
||||
"""
|
||||
Core entity in the knowledge graph.
|
||||
|
||||
Entities are the primary nodes in the knowledge graph. Each entity has:
|
||||
- A unique identifier (text, based on type/name path)
|
||||
- A name
|
||||
- An entity type (e.g., "person", "organization", "event")
|
||||
- A description (optional)
|
||||
- A list of observations
|
||||
"""
|
||||
__tablename__ = "entity"
|
||||
__table_args__ = (
|
||||
UniqueConstraint("entity_type", "name", name="uix_entity_type_name"),
|
||||
)
|
||||
|
||||
id: Mapped[str] = mapped_column(String, primary_key=True)
|
||||
name: Mapped[str] = mapped_column(String)
|
||||
entity_type: Mapped[str] = mapped_column(String)
|
||||
description: Mapped[Optional[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")
|
||||
)
|
||||
|
||||
# Link to source document
|
||||
doc_id: Mapped[Optional[int]] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("documents.id", ondelete="SET NULL"),
|
||||
nullable=True
|
||||
)
|
||||
|
||||
# Relationships
|
||||
document: Mapped["Document"] = relationship(
|
||||
"Document",
|
||||
back_populates="entities"
|
||||
)
|
||||
observations: Mapped[List["Observation"]] = relationship(
|
||||
"Observation",
|
||||
back_populates="entity",
|
||||
cascade="all, delete-orphan"
|
||||
)
|
||||
outgoing_relations: Mapped[List["Relation"]] = relationship(
|
||||
"Relation",
|
||||
foreign_keys="[Relation.from_id]",
|
||||
back_populates="from_entity",
|
||||
cascade="all, delete-orphan"
|
||||
)
|
||||
incoming_relations: Mapped[List["Relation"]] = relationship(
|
||||
"Relation",
|
||||
foreign_keys="[Relation.to_id]",
|
||||
back_populates="to_entity",
|
||||
cascade="all, delete-orphan"
|
||||
)
|
||||
|
||||
@property
|
||||
def relations(self):
|
||||
return self.outgoing_relations + self.incoming_relations
|
||||
|
||||
@classmethod
|
||||
def generate_id(cls, entity_type: str, name: str) -> str:
|
||||
"""Generate a filesystem path-based ID for this entity."""
|
||||
# Use common normalization for filesystem safety
|
||||
safe_name = sanitize_name(name)
|
||||
return f"{entity_type}/{safe_name}"
|
||||
|
||||
def get_file_path(self) -> str:
|
||||
"""Get the filesystem path for this entity."""
|
||||
return f"{self.id}.md" # id is already in path format
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"Entity(id='{self.id}', name='{self.name}', type='{self.entity_type}')"
|
||||
|
||||
|
||||
class Observation(Base):
|
||||
"""
|
||||
Observations are discrete pieces of information about an entity. They are:
|
||||
- Stored as strings
|
||||
- Attached to specific entities
|
||||
- Can be added or removed independently
|
||||
- Should be atomic (one fact per observation)
|
||||
"""
|
||||
__tablename__ = "observation"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
entity_id: Mapped[str] = mapped_column(
|
||||
String,
|
||||
ForeignKey("entity.id", ondelete="CASCADE"),
|
||||
index=True
|
||||
)
|
||||
content: Mapped[str] = mapped_column(String)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime,
|
||||
server_default=text("CURRENT_TIMESTAMP")
|
||||
)
|
||||
context: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
|
||||
# Relationships
|
||||
entity: Mapped[Entity] = relationship("Entity", back_populates="observations")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
content = self.content[:50] + "..." if len(self.content) > 50 else self.content
|
||||
return f"Observation(id={self.id}, entity='{self.entity_id}', content='{content}')"
|
||||
|
||||
|
||||
class Relation(Base):
|
||||
"""
|
||||
Relations define directed connections between entities.
|
||||
They are always stored in active voice and describe how entities
|
||||
interact or relate to each other.
|
||||
"""
|
||||
__tablename__ = "relation"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
from_id: Mapped[str] = mapped_column(
|
||||
String,
|
||||
ForeignKey("entity.id", ondelete="CASCADE"),
|
||||
index=True
|
||||
)
|
||||
to_id: Mapped[str] = mapped_column(
|
||||
String,
|
||||
ForeignKey("entity.id", ondelete="CASCADE"),
|
||||
index=True
|
||||
)
|
||||
relation_type: Mapped[str] = mapped_column(String)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime,
|
||||
server_default=text("CURRENT_TIMESTAMP")
|
||||
)
|
||||
context: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
|
||||
# Relationships
|
||||
from_entity: Mapped[Entity] = relationship(
|
||||
"Entity",
|
||||
foreign_keys=[from_id],
|
||||
back_populates="outgoing_relations"
|
||||
)
|
||||
to_entity: Mapped[Entity] = relationship(
|
||||
"Entity",
|
||||
foreign_keys=[to_id],
|
||||
back_populates="incoming_relations"
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"Relation(id={self.id}, from='{self.from_id}', type='{self.relation_type}', to='{self.to_id}')"
|
||||
@@ -0,0 +1,45 @@
|
||||
"""Repository for document operations."""
|
||||
|
||||
from typing import Optional, Sequence, List
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
|
||||
|
||||
from basic_memory.models import Document
|
||||
from basic_memory.repository.repository import Repository
|
||||
|
||||
|
||||
class DocumentRepository(Repository[Document]):
|
||||
"""Repository for managing documents in the database."""
|
||||
|
||||
def __init__(self, session_maker: async_sessionmaker[AsyncSession]):
|
||||
super().__init__(session_maker, Document)
|
||||
|
||||
async def find_by_path(self, path: str) -> Optional[Document]:
|
||||
"""Find a document by its path."""
|
||||
query = select(Document).where(Document.path == path)
|
||||
return await self.find_one(query)
|
||||
|
||||
async def find_by_checksum(self, checksum: str) -> Sequence[Document]:
|
||||
"""Find all documents with a given checksum."""
|
||||
async with self.session_maker() as session:
|
||||
result = await session.execute(
|
||||
select(Document).where(Document.checksum == checksum)
|
||||
)
|
||||
return result.scalars().all()
|
||||
|
||||
async def find_changed(self, checksums: dict[str, str]) -> List[Document]:
|
||||
"""
|
||||
Find documents that have changed based on their checksums.
|
||||
|
||||
Args:
|
||||
checksums: Dict mapping paths to their current checksums
|
||||
|
||||
Returns:
|
||||
List of documents whose checksums don't match (excluding untracked files)
|
||||
"""
|
||||
changed = []
|
||||
for path, checksum in checksums.items():
|
||||
doc = await self.find_by_path(path)
|
||||
if doc and doc.checksum != checksum: # Only include tracked files that changed
|
||||
changed.append(doc)
|
||||
return changed
|
||||
@@ -1,33 +1,15 @@
|
||||
"""Service layer exceptions and imports."""
|
||||
"""Services package."""
|
||||
|
||||
|
||||
class ServiceError(Exception):
|
||||
"""Base exception for service errors"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class DatabaseSyncError(ServiceError):
|
||||
"""Raised when database sync fails"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class RelationError(ServiceError):
|
||||
"""Base exception for relation-specific errors"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
from .entity_service import EntityService
|
||||
from .observation_service import ObservationService
|
||||
from .relation_service import RelationService
|
||||
from basic_memory.services.document_service import DocumentService
|
||||
from basic_memory.services.entity_service import EntityService
|
||||
from basic_memory.services.observation_service import ObservationService
|
||||
from basic_memory.services.relation_service import RelationService
|
||||
from basic_memory.services.service import BaseService
|
||||
|
||||
__all__ = [
|
||||
"ServiceError",
|
||||
"DatabaseSyncError",
|
||||
"RelationError",
|
||||
"EntityService",
|
||||
"ObservationService",
|
||||
"RelationService",
|
||||
]
|
||||
'BaseService',
|
||||
'DocumentService',
|
||||
'EntityService',
|
||||
'ObservationService',
|
||||
'RelationService',
|
||||
]
|
||||
@@ -0,0 +1,226 @@
|
||||
"""Service for managing documents in the system."""
|
||||
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any, List
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from basic_memory.models import Document
|
||||
from basic_memory.repository.document_repository import DocumentRepository
|
||||
from basic_memory.services.service import BaseService
|
||||
|
||||
|
||||
class DocumentError(Exception):
|
||||
"""Base exception for document operations."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class DocumentNotFoundError(DocumentError):
|
||||
"""Raised when a document doesn't exist."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class DocumentWriteError(DocumentError):
|
||||
"""Raised when document file operations fail."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class DocumentService(BaseService[DocumentRepository]):
|
||||
"""
|
||||
Service for managing documents and their metadata.
|
||||
|
||||
Handles both file operations and database tracking, keeping
|
||||
them in sync. The filesystem is the source of truth.
|
||||
"""
|
||||
|
||||
def __init__(self, document_repository: DocumentRepository):
|
||||
super().__init__(document_repository)
|
||||
|
||||
async def compute_checksum(self, content: str) -> str:
|
||||
"""Compute SHA-256 checksum of content."""
|
||||
return hashlib.sha256(content.encode()).hexdigest()
|
||||
|
||||
async def ensure_parent_directory(self, path: Path) -> None:
|
||||
"""
|
||||
Ensure parent directory exists and is writable.
|
||||
|
||||
Args:
|
||||
path: Path to check
|
||||
|
||||
Raises:
|
||||
DocumentWriteError: If directory cannot be created or is not writable
|
||||
"""
|
||||
parent = path.parent
|
||||
try:
|
||||
if not parent.exists():
|
||||
parent.mkdir(parents=True)
|
||||
|
||||
# Verify we can write to it
|
||||
test_file = parent / ".write_test"
|
||||
test_file.touch()
|
||||
test_file.unlink()
|
||||
except Exception as e:
|
||||
raise DocumentWriteError(f"Directory not writable: {parent}: {e}")
|
||||
|
||||
async def list_documents(self) -> List[Document]:
|
||||
"""List all documents in the database."""
|
||||
return await self.repository.find_all()
|
||||
|
||||
async def create_document(
|
||||
self, path: str, content: str, metadata: Optional[Dict[str, Any]] = None
|
||||
) -> Document:
|
||||
"""
|
||||
Create a new document.
|
||||
|
||||
Args:
|
||||
path: Path where to create the document
|
||||
content: Document content
|
||||
metadata: Optional metadata to store
|
||||
|
||||
Returns:
|
||||
Created document record
|
||||
|
||||
Raises:
|
||||
DocumentWriteError: If file cannot be written
|
||||
"""
|
||||
logger.debug(f"Creating document at {path}")
|
||||
|
||||
# Ensure parent directories exist and are writable
|
||||
file_path = Path(path)
|
||||
await self.ensure_parent_directory(file_path)
|
||||
|
||||
# Write file first
|
||||
try:
|
||||
file_path.write_text(content)
|
||||
except Exception as e:
|
||||
raise DocumentWriteError(f"Failed to write document file: {e}")
|
||||
|
||||
# After file is written, create database record
|
||||
try:
|
||||
checksum = await self.compute_checksum(content)
|
||||
doc = await self.repository.create(
|
||||
{"path": str(path), "checksum": checksum, "doc_metadata": metadata}
|
||||
)
|
||||
return doc
|
||||
except Exception:
|
||||
# If database operation fails, clean up the file
|
||||
file_path.unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
async def read_document(self, path: str) -> tuple[Document, str]:
|
||||
"""
|
||||
Read a document and its content.
|
||||
|
||||
Args:
|
||||
path: Path to the document
|
||||
|
||||
Returns:
|
||||
Tuple of (document record, content)
|
||||
|
||||
Raises:
|
||||
DocumentNotFoundError: If document doesn't exist
|
||||
"""
|
||||
logger.debug(f"Reading document at {path}")
|
||||
|
||||
# Check if file exists
|
||||
file_path = Path(path)
|
||||
if not file_path.exists():
|
||||
raise DocumentNotFoundError(f"Document not found: {path}")
|
||||
|
||||
# Read content first since file is source of truth
|
||||
try:
|
||||
content = file_path.read_text()
|
||||
except Exception as e:
|
||||
raise DocumentError(f"Failed to read document: {e}")
|
||||
|
||||
# Get document record
|
||||
doc = await self.repository.find_by_path(str(path))
|
||||
if not doc:
|
||||
# File exists but no DB record - create one
|
||||
checksum = await self.compute_checksum(content)
|
||||
doc = await self.repository.create({"path": str(path), "checksum": checksum})
|
||||
|
||||
return doc, content
|
||||
|
||||
async def update_document(
|
||||
self, path: str, content: str, metadata: Optional[Dict[str, Any]] = None
|
||||
) -> Document:
|
||||
"""
|
||||
Update an existing document.
|
||||
|
||||
Args:
|
||||
path: Path to the document
|
||||
content: New content
|
||||
metadata: Optional new metadata
|
||||
|
||||
Returns:
|
||||
Updated document record
|
||||
|
||||
Raises:
|
||||
DocumentNotFoundError: If document doesn't exist
|
||||
DocumentWriteError: If update fails
|
||||
"""
|
||||
logger.debug(f"Updating document at {path}")
|
||||
|
||||
# Verify file exists
|
||||
file_path = Path(path)
|
||||
if not file_path.exists():
|
||||
raise DocumentNotFoundError(f"Document not found: {path}")
|
||||
|
||||
# Write new content first
|
||||
try:
|
||||
file_path.write_text(content)
|
||||
except Exception as e:
|
||||
raise DocumentWriteError(f"Failed to write document: {e}")
|
||||
|
||||
# Update database record
|
||||
doc = await self.repository.find_by_path(str(path))
|
||||
if not doc:
|
||||
# File exists but no DB record - create one
|
||||
checksum = await self.compute_checksum(content)
|
||||
return await self.repository.create(
|
||||
{"path": str(path), "checksum": checksum, "doc_metadata": metadata}
|
||||
)
|
||||
|
||||
# Update existing record
|
||||
checksum = await self.compute_checksum(content)
|
||||
update_data = {"checksum": checksum}
|
||||
if metadata is not None:
|
||||
update_data["doc_metadata"] = metadata
|
||||
|
||||
updated_document = await self.repository.update(doc.id, update_data)
|
||||
assert updated_document is not None, f"Could not update document {doc.id}"
|
||||
return updated_document
|
||||
|
||||
async def delete_document(self, path: str) -> None:
|
||||
"""
|
||||
Delete a document.
|
||||
|
||||
Args:
|
||||
path: Path to the document
|
||||
|
||||
Raises:
|
||||
DocumentNotFoundError: If document doesn't exist
|
||||
DocumentWriteError: If deletion fails
|
||||
"""
|
||||
logger.debug(f"Deleting document at {path}")
|
||||
|
||||
# Verify file exists
|
||||
file_path = Path(path)
|
||||
if not file_path.exists():
|
||||
raise DocumentNotFoundError(f"Document not found: {path}")
|
||||
|
||||
# Delete file first
|
||||
try:
|
||||
file_path.unlink()
|
||||
except Exception as e:
|
||||
raise DocumentWriteError(f"Failed to delete document: {e}")
|
||||
|
||||
# Delete database record if it exists
|
||||
doc = await self.repository.find_by_path(str(path))
|
||||
if doc:
|
||||
await self.repository.delete(doc.id)
|
||||
@@ -0,0 +1,205 @@
|
||||
"""Service for syncing files with the database."""
|
||||
|
||||
import hashlib
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Set
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from basic_memory.repository.document_repository import DocumentRepository
|
||||
|
||||
|
||||
@dataclass
|
||||
class SyncReport:
|
||||
"""Report of sync results."""
|
||||
new: Set[str]
|
||||
modified: Set[str]
|
||||
deleted: Set[str]
|
||||
|
||||
@property
|
||||
def total_changes(self) -> int:
|
||||
return len(self.new) + len(self.modified) + len(self.deleted)
|
||||
|
||||
def __str__(self) -> str:
|
||||
return (
|
||||
f"Changes detected:\n"
|
||||
f" New files: {len(self.new)}\n"
|
||||
f" Modified: {len(self.modified)}\n"
|
||||
f" Deleted: {len(self.deleted)}"
|
||||
)
|
||||
|
||||
|
||||
class SyncError(Exception):
|
||||
"""Raised when sync operations fail."""
|
||||
pass
|
||||
|
||||
|
||||
class FileSyncService:
|
||||
"""Service for keeping files and database in sync."""
|
||||
|
||||
def __init__(self, document_repository: DocumentRepository):
|
||||
self.repository = document_repository
|
||||
|
||||
async def compute_checksum(self, content: str) -> str:
|
||||
"""Compute SHA-256 checksum of content."""
|
||||
return hashlib.sha256(content.encode()).hexdigest()
|
||||
|
||||
async def scan_files(self, directory: Path) -> dict[str, str]:
|
||||
"""
|
||||
Scan directory for files and their checksums.
|
||||
Only processes files, ignores directories.
|
||||
|
||||
Args:
|
||||
directory: Root directory to scan
|
||||
|
||||
Returns:
|
||||
Dict mapping paths to checksums
|
||||
|
||||
Raises:
|
||||
SyncError: If any file cannot be read
|
||||
"""
|
||||
logger.debug(f"Scanning directory: {directory}")
|
||||
files = {}
|
||||
errors = []
|
||||
|
||||
for path in directory.rglob('*'):
|
||||
if path.is_file():
|
||||
try:
|
||||
content = path.read_text()
|
||||
checksum = await self.compute_checksum(content)
|
||||
# Store path relative to root directory
|
||||
rel_path = str(path.relative_to(directory))
|
||||
files[rel_path] = checksum
|
||||
except Exception as e:
|
||||
errors.append(f"Failed to read {path}: {e}")
|
||||
|
||||
if errors:
|
||||
raise SyncError("Failed to read files:\n" + "\n".join(errors))
|
||||
|
||||
logger.debug(f"Found {len(files)} files")
|
||||
return files
|
||||
|
||||
async def find_changes(self, current_files: dict[str, str]) -> SyncReport:
|
||||
"""
|
||||
Find changes between filesystem and database.
|
||||
|
||||
Args:
|
||||
current_files: Dict mapping paths to checksums
|
||||
|
||||
Returns:
|
||||
SyncReport detailing changes
|
||||
"""
|
||||
logger.debug("Finding changes")
|
||||
|
||||
# Get all documents from DB
|
||||
db_documents = await self.repository.find_all()
|
||||
db_files = {
|
||||
doc.path: doc.checksum
|
||||
for doc in db_documents
|
||||
}
|
||||
|
||||
# Find changes
|
||||
new = set(current_files.keys()) - set(db_files.keys())
|
||||
deleted = set(db_files.keys()) - set(current_files.keys())
|
||||
modified = {
|
||||
path for path in current_files
|
||||
if path in db_files and current_files[path] != db_files[path]
|
||||
}
|
||||
|
||||
return SyncReport(new=new, modified=modified, deleted=deleted)
|
||||
|
||||
async def sync_new_file(self, path: str, directory: Path) -> None:
|
||||
"""
|
||||
Sync a new file.
|
||||
|
||||
Args:
|
||||
path: Relative path to file
|
||||
directory: Root directory
|
||||
|
||||
Raises:
|
||||
SyncError: If sync fails
|
||||
"""
|
||||
full_path = directory / path
|
||||
try:
|
||||
content = full_path.read_text()
|
||||
checksum = await self.compute_checksum(content)
|
||||
await self.repository.create({
|
||||
"path": path,
|
||||
"checksum": checksum
|
||||
})
|
||||
except Exception as e:
|
||||
raise SyncError(f"Failed to sync new file {path}: {e}")
|
||||
|
||||
async def sync_modified_file(self, path: str, directory: Path) -> None:
|
||||
"""
|
||||
Sync a modified file.
|
||||
|
||||
Args:
|
||||
path: Relative path to file
|
||||
directory: Root directory
|
||||
|
||||
Raises:
|
||||
SyncError: If sync fails
|
||||
"""
|
||||
full_path = directory / path
|
||||
try:
|
||||
content = full_path.read_text()
|
||||
checksum = await self.compute_checksum(content)
|
||||
doc = await self.repository.find_by_path(path)
|
||||
if doc:
|
||||
await self.repository.update(doc.id, {"checksum": checksum})
|
||||
else:
|
||||
await self.repository.create({
|
||||
"path": path,
|
||||
"checksum": checksum
|
||||
})
|
||||
except Exception as e:
|
||||
raise SyncError(f"Failed to sync modified file {path}: {e}")
|
||||
|
||||
async def sync(self, directory: Path) -> SyncReport:
|
||||
"""
|
||||
Sync filesystem with database.
|
||||
Filesystem is source of truth.
|
||||
|
||||
Args:
|
||||
directory: Root directory to sync
|
||||
|
||||
Returns:
|
||||
SyncReport detailing changes
|
||||
|
||||
Raises:
|
||||
SyncError: If sync fails
|
||||
"""
|
||||
logger.info(f"Starting sync of {directory}")
|
||||
|
||||
# Get current state
|
||||
current_files = await self.scan_files(directory)
|
||||
|
||||
# Find changes
|
||||
changes = await self.find_changes(current_files)
|
||||
logger.info(f"Found changes: {changes}")
|
||||
|
||||
if changes.total_changes == 0:
|
||||
logger.info("No changes detected")
|
||||
return changes
|
||||
|
||||
# Process new files
|
||||
for path in changes.new:
|
||||
logger.debug(f"Processing new file: {path}")
|
||||
await self.sync_new_file(path, directory)
|
||||
|
||||
# Process modified files
|
||||
for path in changes.modified:
|
||||
logger.debug(f"Processing modified file: {path}")
|
||||
await self.sync_modified_file(path, directory)
|
||||
|
||||
# Process deleted files
|
||||
for path in changes.deleted:
|
||||
logger.debug(f"Processing deleted file: {path}")
|
||||
doc = await self.repository.find_by_path(path)
|
||||
if doc:
|
||||
await self.repository.delete(doc.id)
|
||||
|
||||
logger.info("Sync completed successfully")
|
||||
return changes
|
||||
Reference in New Issue
Block a user