admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

finish watch_service.py

Browse Source
This commit is contained in:
phernandez
2025-02-02 19:08:31 -06:00
parent 0af60e7fdc
commit 4afa368a6b
7 changed files with 116 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/sync/file_change_scanner.py
+12 -5
View File
@@ -2,14 +2,14 @@
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, Sequence
from typing import Dict, Sequence, Optional
from loguru import logger
from basic_memory.file_utils import compute_checksum
from basic_memory.models import Entity
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.sync.utils import SyncReport
from basic_memory.sync.utils import SyncReport, FileChange
@dataclass
@@ -87,13 +87,19 @@ class FileChangeScanner:
return result
async def find_changes(
self, directory: Path, db_file_state: Dict[str, FileState]
self,
db_file_state: Dict[str, FileState],
directory: Optional[Path] = None,
) -> SyncReport:
"""Find changes between filesystem and database."""
# Get current files and checksums
scan_result = await self.scan_directory(directory)
current_files = scan_result.files
# scan the directory provided
scan_result = await self.scan_directory(directory)
# the set of all of the current files and their checksums
current_files = scan_result.files
# Build report
report = SyncReport()
@@ -157,5 +163,6 @@ class FileChangeScanner:
async def find_knowledge_changes(self, directory: Path) -> SyncReport:
"""Find changes in knowledge directory."""
db_file_state = await self.get_db_file_state(await self.entity_repository.find_all())
return await self.find_changes(directory=directory, db_file_state=db_file_state)
src/basic_memory/sync/sync_service.py
+4 -22
View File
@@ -61,28 +61,10 @@ class SyncService:
else:
logger.debug(f"No entity found to delete: {file_path}")
async def sync(self, directory: Optional[Path] = None, file_changes: Optional[dict[str, FileChange]] = None) -> SyncReport:
"""Sync knowledge files with database."""
if file_changes is not None:
changes = SyncReport()
for path, file_change in file_changes.items():
logger.debug(f"path {path} file_change {file_change}")
match file_change.change_type:
case Change.added:
changes.new.add(path)
changes.checksums[path] = file_change.checksum
case Change.modified:
changes.modified.add(path)
changes.checksums[path] = file_change.checksum
case Change.deleted:
changes.deleted.add(path)
else:
# Traditional directory scan mode
if directory is None:
raise ValueError("Must provide either directory or file_changes")
changes = await self.scanner.find_knowledge_changes(directory)
logger.info(f"Found {changes.total_changes} knowledge changes")
async def sync(self, directory: Path) -> SyncReport:
"""Sync knowledge files with database."""
changes = await self.scanner.find_knowledge_changes(directory)
logger.info(f"Found {changes.total_changes} knowledge changes")
# Handle moves first
for old_path, new_path in changes.moves.items():
src/basic_memory/sync/utils.py
+6 -1
View File
@@ -55,5 +55,10 @@ class SyncReport:
@property
def total_changes(self) -> int:
"""Total number of files that need attention."""
"""Total number of changes."""
return len(self.new) + len(self.modified) + len(self.deleted) + len(self.moves)
@property
def total_files(self) -> int:
"""Total number of files synced."""
return len(self.new) + len(self.modified) + len(self.moves)
src/basic_memory/sync/watch_service.py
+15 -23
View File
@@ -20,7 +20,7 @@ from basic_memory.sync.utils import FileChange
class WatchEvent(BaseModel):
timestamp: datetime
path: str
action: str # sync, delete, etc
action: str # new, delete, etc
status: str # success, error
error: Optional[str] = None
@@ -32,8 +32,6 @@ class WatchServiceState(BaseModel):
pid: int = dataclasses.field(default_factory=os.getpid)
# Stats
files_synced: int = 0
bytes_processed: int = 0
error_count: int = 0
last_error: Optional[datetime] = None
last_scan: Optional[datetime] = None
@@ -54,6 +52,7 @@ class WatchServiceState(BaseModel):
def record_error(self, error: str):
self.error_count += 1
self.add_event(path="", action="sync", status="error", error=error)
self.last_error = datetime.now()
@@ -79,7 +78,7 @@ class WatchService:
debounce=self.config.sync_delay,
recursive=True,
):
await self.handle_changes(changes)
await self.handle_changes(self.config.home)
except Exception as e:
self.state.record_error(str(e))
@@ -97,35 +96,28 @@ class WatchService:
"""Filter to only watch markdown files"""
return path.endswith(".md") and not Path(path).name.startswith(".")
async def handle_changes(self, changes: set[tuple[Change, str]]):
async def handle_changes(self, directory: Path):
"""Process a batch of file changes"""
# Group changes by file path
changes_by_file = {}
try:
for change_type, path in changes:
file_change = await FileChange.from_path(path, change_type, self.file_service)
# store changes by relative path
changes_by_file[str(file_change.path)] = file_change
# Process changes with timeout
await self.sync_service.sync(file_changes=changes_by_file)
report = await self.sync_service.sync(directory)
self.state.last_scan = datetime.now()
self.state.total_files = report.total_files
# Update stats
self.state.files_synced += len(changes_by_file)
for path, change in changes_by_file.items():
if change.change_type != Change.deleted:
size = self.file_service.path(path,absolute=True).stat().st_size
self.state.bytes_processed += size
self.state.add_event(path=path, action="sync", status="success")
for path in report.new:
self.state.add_event(path=path, action="new", status="success")
for path in report.modified:
self.state.add_event(path=path, action="modified", status="success")
for path in report.moves:
self.state.add_event(path=path, action="moved", status="success")
for path in report.deleted:
self.state.add_event(path=path, action="deleted", status="success")
await self.write_status()
except Exception as e:
self.state.record_error(str(e))
for path in changes_by_file:
self.state.add_event(path=path, action="sync", status="error", error=str(e))
await self.write_status()
raise
tests/conftest.py
+11
View File
@@ -28,6 +28,7 @@ from basic_memory.services.link_resolver import LinkResolver
from basic_memory.services.search_service import SearchService
from basic_memory.sync import FileChangeScanner
from basic_memory.sync.sync_service import SyncService
from basic_memory.sync.watch_service import WatchService
@pytest_asyncio.fixture
@@ -382,3 +383,13 @@ async def test_graph(
"observations": [e.observations for e in entities],
"relations": relations,
}
@pytest_asyncio.fixture
def watch_service(sync_service, file_service, test_config):
return WatchService(
sync_service=sync_service,
file_service=file_service,
config=test_config
)
tests/sync/test_watch_service.py
+68 -80
View File
@@ -1,27 +1,15 @@
"""Tests for the watch service."""
from dataclasses import asdict
import pytest
import pytest_asyncio
from datetime import datetime
from pathlib import Path
from watchfiles import Change
from basic_memory.sync.utils import FileChange
from basic_memory.sync.watch_service import WatchService, WatchEvent, WatchServiceState
from basic_memory.sync.watch_service import WatchServiceState
@pytest_asyncio.fixture
def watch_service(sync_service, file_service, test_config):
return WatchService(
sync_service=sync_service,
file_service=file_service,
config=test_config
)
@pytest_asyncio.fixture
async def sample_markdown(tmp_path):
async def sample_markdown_file(test_config):
content = """---
title: Test Note
type: note
@@ -29,59 +17,82 @@ type: note
# Test Note
This is a test note.
"""
file_path = tmp_path / "test.md"
file_path = test_config.home / "test.md"
file_path.write_text(content)
return file_path
@pytest.mark.asyncio
async def test_handle_file_added(watch_service, sync_service, sample_markdown):
async def test_handle_file_added(test_config, watch_service, sync_service, sample_markdown_file):
"""Test handling a new file event"""
changes = {(Change.added, str(sample_markdown))}
await watch_service.handle_changes(changes)
await watch_service.handle_changes(test_config.home)
# Check stats updated
assert watch_service.state.files_synced == 1
assert watch_service.state.bytes_processed > 0
assert watch_service.state.total_files == 1
assert watch_service.state.last_scan is not None
# Check event recorded
assert len(watch_service.state.recent_events) == 1
event = watch_service.state.recent_events[0]
assert event.path == "test.md"
assert event.action == "sync"
assert event.action == "new"
assert event.status == "success"
@pytest.mark.asyncio
async def test_handle_file_modified(watch_service, sync_service, sample_markdown):
async def test_handle_file_modified(test_config, watch_service, sync_service, sample_markdown_file):
"""Test handling a modified file event"""
# First add the file
await watch_service.handle_changes({(Change.added, str(sample_markdown))})
await watch_service.handle_changes(test_config.home)
# Modify the file
sample_markdown.write_text(sample_markdown.read_text() + "\nModified content")
await watch_service.handle_changes({(Change.modified, str(sample_markdown))})
sample_markdown_file.write_text(sample_markdown_file.read_text() + "\nModified content")
await watch_service.handle_changes(test_config.home)
# Should have two events
assert len(watch_service.state.recent_events) == 2
assert watch_service.state.files_synced == 2
assert watch_service.state.total_files == 1
event = watch_service.state.recent_events[0]
assert event.path == "test.md"
assert event.action == "modified"
assert event.status == "success"
@pytest.mark.asyncio
async def test_handle_file_deleted(watch_service, sync_service, sample_markdown):
async def test_handle_file_moved(test_config, watch_service, sync_service, sample_markdown_file):
"""Test handling a moved file event"""
# First add the file
await watch_service.handle_changes(test_config.home)
# Modify the file
renamed = sample_markdown_file.rename(test_config.home / "moved.md")
await watch_service.handle_changes(test_config.home)
# Should have two events
assert len(watch_service.state.recent_events) == 2
assert watch_service.state.total_files == 1
event = watch_service.state.recent_events[0]
assert event.path == "test.md"
assert event.action == "moved"
assert event.status == "success"
@pytest.mark.asyncio
async def test_handle_file_deleted(test_config, watch_service, sync_service, sample_markdown_file):
"""Test handling a deleted file event"""
# First add the file
await watch_service.handle_changes({(Change.added, str(sample_markdown))})
await watch_service.handle_changes(test_config.home)
# Delete the file
sample_markdown.unlink()
await watch_service.handle_changes({(Change.deleted, str(sample_markdown))})
sample_markdown_file.unlink()
await watch_service.handle_changes(test_config.home)
# Should have two events
assert len(watch_service.state.recent_events) == 2
delete_event = watch_service.state.recent_events[0]
assert delete_event.action == "sync"
assert delete_event.action == "deleted"
assert delete_event.path == "test.md"
@@ -93,30 +104,12 @@ async def test_filter_changes(watch_service):
assert watch_service.filter_changes(Change.added, ".test.md") is False
@pytest.mark.asyncio
async def test_handle_error(watch_service, sync_service, ):
"""Test error handling during sync"""
changes = {(Change.added, "test.md")}
# ValueError raised because "test.md" path does not exist
with pytest.raises(ValueError):
await watch_service.handle_changes(changes)
assert watch_service.state.error_count == 1
assert watch_service.state.last_error is not None
@pytest.mark.asyncio
async def test_recent_events_limit(watch_service):
"""Test that recent events are limited to 100"""
for i in range(150):
watch_service.state.add_event(
path=f"test{i}.md",
action="sync",
status="success"
)
watch_service.state.add_event(path=f"test{i}.md", action="sync", status="success")
assert len(watch_service.state.recent_events) == 100
# Most recent should be at the start
assert watch_service.state.recent_events[0].path == "test149.md"
@@ -127,26 +120,20 @@ async def test_state_serialization(watch_service):
"""Test state serializes to dict correctly"""
# Add some test state
watch_service.state.running = True
watch_service.state.files_synced = 42
watch_service.state.add_event(
path="test.md",
action="sync",
status="success"
)
watch_service.state.add_event(path="test.md", action="sync", status="success")
data = WatchServiceState.model_dump(watch_service.state)
# Check basic fields
assert data['running'] is True
assert data['files_synced'] == 42
assert isinstance(data['start_time'], datetime)
assert data["running"] is True
assert isinstance(data["start_time"], datetime)
# Check events serialized
assert len(data['recent_events']) == 1
event = data['recent_events'][0]
assert event['path'] == "test.md"
assert event['action'] == "sync"
assert isinstance(event['timestamp'], datetime)
assert len(data["recent_events"]) == 1
event = data["recent_events"][0]
assert event["path"] == "test.md"
assert event["action"] == "sync"
assert isinstance(event["timestamp"], datetime)
@pytest.mark.asyncio
@@ -154,13 +141,14 @@ async def test_status_file(watch_service, tmp_path):
"""Test status file writing"""
watch_service.state.running = True
await watch_service.write_status()
status_file = watch_service.status_path
assert status_file.exists()
# Should be valid JSON with our fields
import json
data = json.loads(status_file.read_text())
assert data['running'] is True
assert isinstance(data['start_time'], str)
assert data['pid'] > 0
assert data["running"] is True
assert isinstance(data["start_time"], str)
assert data["pid"] > 0
tests/sync/test_watch_sync.py
-129
View File
@@ -1,129 +0,0 @@
"""Tests for watch service integration with sync service."""
import pytest
import pytest_asyncio
from pathlib import Path
from watchfiles import Change
from basic_memory.sync.utils import FileChange
@pytest.mark.asyncio
async def test_sync_with_file_changes(sync_service, tmp_path):
"""Test sync with file changes from watch service"""
# Create a test file
content = """---
title: Test Note
type: note
---
# Test Note
This is a test."""
file_path = tmp_path / "test.md"
file_path.write_text(content)
# Create FileChange for a new file
changes = {
str(file_path): FileChange(
change_type=Change.added,
path=str(file_path),
checksum="abc123"
)
}
report = await sync_service.sync(file_changes=changes)
assert str(file_path) in report.new
assert report.checksums[str(file_path)] == "abc123"
@pytest.mark.asyncio
async def test_sync_mixed_changes(sync_service, tmp_path):
"""Test sync with multiple types of changes"""
# Setup initial files
new_file = tmp_path / "new.md"
new_file.write_text("New file")
mod_file = tmp_path / "modified.md"
mod_file.write_text("Original content")
del_file = tmp_path / "deleted.md"
changes = {
str(new_file): FileChange(
change_type=Change.added,
path=str(new_file),
checksum="new123"
),
str(mod_file): FileChange(
change_type=Change.modified,
path=str(mod_file),
checksum="mod123"
),
str(del_file): FileChange(
change_type=Change.deleted,
path=str(del_file)
)
}
report = await sync_service.sync(file_changes=changes)
# Verify report contains all changes
assert str(new_file) in report.new
assert str(mod_file) in report.modified
assert str(del_file) in report.deleted
# Check checksums
assert report.checksums[str(new_file)] == "new123"
assert report.checksums[str(mod_file)] == "mod123"
assert str(del_file) not in report.checksums
@pytest.mark.asyncio
async def test_sync_requires_params(sync_service):
"""Test sync requires either directory or file_changes"""
with pytest.raises(ValueError):
await sync_service.sync()
@pytest.mark.asyncio
async def test_sync_with_invalid_change_type(sync_service, tmp_path):
"""Test handling of invalid change types"""
file_path = tmp_path / "test.md"
file_path.touch()
# Type checker won't let us create this directly, so ignore the type
changes = {
str(file_path): FileChange(
change_type="invalid", # type: ignore
path=str(file_path),
checksum="test123"
)
}
with pytest.raises(Exception):
await sync_service.sync(file_changes=changes)
@pytest.mark.asyncio
async def test_end_to_end_watch_sync(watch_service, sync_service, tmp_path):
"""Test complete watch -> sync flow"""
# Create a test file that will trigger a watch event
test_file = tmp_path / "test.md"
test_file.write_text("""---
title: Test Note
type: note
---
# Test
Testing watch -> sync flow
""")
# Simulate watchfiles event
changes = {(Change.added, str(test_file))}
await watch_service.handle_changes(changes)
# Check watch service state
assert watch_service.state.files_synced == 1
assert len(watch_service.state.recent_events) == 1
event = watch_service.state.recent_events[0]
assert event.path == str(test_file)
assert event.status == "success"