mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
5b4f0eafcc
* configure logging * set mcp output logging also * fix type check errors * fix type check * rename Permalink schema type * fix type errors * add typechecks to ci workflow * pytest coverage setup * add tests for status cli * sync tests coverage * watch_service test coverage * tests for tool_utils.py * clean up imports * file_utils coverage * markdown plugins coverage * 99% test coverage * more test coverage, remove ObservationCategory * more tool coverage * fix type-check * format, upgrade deps --------- Co-authored-by: phernandez <phernandez@basicmachines.co>
49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
"""Types and utilities for file sync."""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Set, Dict, Optional
|
|
|
|
from watchfiles import Change
|
|
|
|
|
|
@dataclass
|
|
class FileChange:
|
|
"""A change to a file detected by the watch service.
|
|
|
|
Attributes:
|
|
change_type: Type of change (added, modified, deleted)
|
|
path: Path to the file
|
|
checksum: File checksum (None for deleted files)
|
|
"""
|
|
|
|
change_type: Change
|
|
path: str
|
|
checksum: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class SyncReport:
|
|
"""Report of file changes found compared to database state.
|
|
|
|
Attributes:
|
|
total: Total number of files in directory being synced
|
|
new: Files that exist on disk but not in database
|
|
modified: Files that exist in both but have different checksums
|
|
deleted: Files that exist in database but not on disk
|
|
moves: Files that have been moved from one location to another
|
|
checksums: Current checksums for files on disk
|
|
"""
|
|
|
|
total: int = 0
|
|
# We keep paths as strings in sets/dicts for easier serialization
|
|
new: Set[str] = field(default_factory=set)
|
|
modified: Set[str] = field(default_factory=set)
|
|
deleted: Set[str] = field(default_factory=set)
|
|
moves: Dict[str, str] = field(default_factory=dict) # old_path -> new_path
|
|
checksums: Dict[str, str] = field(default_factory=dict) # path -> checksum
|
|
|
|
@property
|
|
def total_changes(self) -> int:
|
|
"""Total number of changes."""
|
|
return len(self.new) + len(self.modified) + len(self.deleted) + len(self.moves)
|