tegwick
4588cbeee8
Implements optional git-based version control for information spaces: - HistoryConfig model for configuring history tracking - Commit, Branch, HistoryEntry, DiffResult models - IHistoryBackend and IHistoryQuery interfaces - GitHistoryBackend using git CLI for version control - GitHistoryEventHandler for event-driven auto-commits - HistoryEventCoordinator for managing space history - HistoryQueryService for high-level history queries - Automatic commits on DOCUMENT_ADDED/REMOVED/CONTENT_CHANGED events - Support for: * Commit log with pagination and filtering * Diff between versions * File content at specific versions * Branch creation and switching * Version restoration * Uncommitted changes detection - 43 comprehensive unit tests with git availability checks Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
259 lines
8.3 KiB
Python
259 lines
8.3 KiB
Python
"""
|
|
History query service implementation.
|
|
|
|
Provides high-level queries on top of the history backend.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from typing import List, Optional, Dict, Callable
|
|
|
|
from .interfaces import IHistoryBackend, IHistoryQuery
|
|
from .models import HistoryEntry, DiffResult, Commit
|
|
|
|
|
|
class HistoryQueryService(IHistoryQuery):
|
|
"""
|
|
Service for querying history information.
|
|
|
|
Provides convenience methods on top of the raw history backend.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
backend: IHistoryBackend,
|
|
directory_resolver: Callable[[str], Optional[Path]],
|
|
):
|
|
"""
|
|
Initialize the query service.
|
|
|
|
Args:
|
|
backend: History backend to query
|
|
directory_resolver: Function that resolves space_id to directory path
|
|
"""
|
|
self._backend = backend
|
|
self._directory_resolver = directory_resolver
|
|
|
|
def get_history(
|
|
self,
|
|
space_id: str,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
path: Optional[str] = None,
|
|
) -> List[HistoryEntry]:
|
|
"""Get history entries for a space."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory or not self._backend.is_initialized(directory):
|
|
return []
|
|
|
|
commits = self._backend.get_log(directory, limit, offset, path)
|
|
return [self._commit_to_entry(commit, directory) for commit in commits]
|
|
|
|
def _commit_to_entry(self, commit: Commit, directory: Path) -> HistoryEntry:
|
|
"""Convert a commit to a history entry with summary."""
|
|
# Count file changes by type
|
|
files_added = 0
|
|
files_modified = 0
|
|
files_deleted = 0
|
|
|
|
# Get diff to determine change types
|
|
if commit.parent_ids:
|
|
parent = commit.parent_ids[0]
|
|
diffs = self._backend.get_diff(directory, parent, commit.id)
|
|
for diff in diffs:
|
|
if diff.additions > 0 and diff.deletions == 0:
|
|
files_added += 1
|
|
elif diff.additions == 0 and diff.deletions > 0:
|
|
files_deleted += 1
|
|
else:
|
|
files_modified += 1
|
|
else:
|
|
# First commit - all files are additions
|
|
files_added = len(commit.files_changed)
|
|
|
|
# Build summary
|
|
parts = []
|
|
if files_added:
|
|
parts.append(f"+{files_added}")
|
|
if files_modified:
|
|
parts.append(f"~{files_modified}")
|
|
if files_deleted:
|
|
parts.append(f"-{files_deleted}")
|
|
|
|
summary = ", ".join(parts) if parts else "No changes"
|
|
|
|
return HistoryEntry(
|
|
commit=commit,
|
|
summary=summary,
|
|
files_added=files_added,
|
|
files_modified=files_modified,
|
|
files_deleted=files_deleted,
|
|
)
|
|
|
|
def get_version_content(
|
|
self,
|
|
space_id: str,
|
|
document_path: str,
|
|
version: str,
|
|
) -> Optional[str]:
|
|
"""Get document content at a specific version."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory or not self._backend.is_initialized(directory):
|
|
return None
|
|
|
|
# Normalize path (remove leading /)
|
|
if document_path.startswith("/"):
|
|
document_path = document_path[1:]
|
|
|
|
return self._backend.get_file_at_version(directory, document_path, version)
|
|
|
|
def compare_versions(
|
|
self,
|
|
space_id: str,
|
|
document_path: str,
|
|
old_version: str,
|
|
new_version: Optional[str] = None,
|
|
) -> Optional[DiffResult]:
|
|
"""Compare two versions of a document."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory or not self._backend.is_initialized(directory):
|
|
return None
|
|
|
|
# Normalize path
|
|
if document_path.startswith("/"):
|
|
document_path = document_path[1:]
|
|
|
|
diffs = self._backend.get_diff(directory, old_version, new_version, document_path)
|
|
return diffs[0] if diffs else None
|
|
|
|
def search_history(
|
|
self,
|
|
space_id: str,
|
|
query: str,
|
|
limit: int = 20,
|
|
) -> List[HistoryEntry]:
|
|
"""Search commit messages."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory or not self._backend.is_initialized(directory):
|
|
return []
|
|
|
|
# Get all recent commits
|
|
commits = self._backend.get_log(directory, limit=limit * 2)
|
|
|
|
# Filter by query in message
|
|
matching_commits = [
|
|
c for c in commits
|
|
if query.lower() in c.message.lower()
|
|
][:limit]
|
|
|
|
return [self._commit_to_entry(commit, directory) for commit in matching_commits]
|
|
|
|
def restore_version(
|
|
self,
|
|
space_id: str,
|
|
document_path: str,
|
|
version: str,
|
|
) -> bool:
|
|
"""Restore a document to a specific version."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory or not self._backend.is_initialized(directory):
|
|
return False
|
|
|
|
# Normalize path
|
|
if document_path.startswith("/"):
|
|
document_path = document_path[1:]
|
|
|
|
# Get content at version
|
|
content = self._backend.get_file_at_version(directory, document_path, version)
|
|
if content is None:
|
|
return False
|
|
|
|
# Write to working directory
|
|
file_path = directory / document_path
|
|
file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
file_path.write_text(content)
|
|
return True
|
|
|
|
def get_latest_commit(self, space_id: str) -> Optional[Commit]:
|
|
"""Get the latest commit for a space."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory or not self._backend.is_initialized(directory):
|
|
return None
|
|
|
|
commits = self._backend.get_log(directory, limit=1)
|
|
return commits[0] if commits else None
|
|
|
|
def get_commit_by_id(self, space_id: str, commit_id: str) -> Optional[Commit]:
|
|
"""Get a specific commit by ID."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory:
|
|
return None
|
|
|
|
return self._backend.get_commit(directory, commit_id)
|
|
|
|
def get_file_history(
|
|
self,
|
|
space_id: str,
|
|
document_path: str,
|
|
limit: int = 20,
|
|
) -> List[HistoryEntry]:
|
|
"""Get history for a specific file."""
|
|
# Normalize path
|
|
if document_path.startswith("/"):
|
|
document_path = document_path[1:]
|
|
|
|
return self.get_history(space_id, limit=limit, path=document_path)
|
|
|
|
def has_history(self, space_id: str) -> bool:
|
|
"""Check if a space has history tracking enabled."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory:
|
|
return False
|
|
return self._backend.is_initialized(directory)
|
|
|
|
def get_branches(self, space_id: str) -> List[str]:
|
|
"""Get list of branch names for a space."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory or not self._backend.is_initialized(directory):
|
|
return []
|
|
|
|
branches = self._backend.list_branches(directory)
|
|
return [b.name for b in branches]
|
|
|
|
def get_current_branch(self, space_id: str) -> Optional[str]:
|
|
"""Get the current branch name."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory:
|
|
return None
|
|
return self._backend.get_current_branch(directory)
|
|
|
|
def create_branch(
|
|
self,
|
|
space_id: str,
|
|
name: str,
|
|
start_point: Optional[str] = None,
|
|
) -> bool:
|
|
"""Create a new branch."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory or not self._backend.is_initialized(directory):
|
|
return False
|
|
|
|
try:
|
|
self._backend.create_branch(directory, name, start_point)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def switch_branch(self, space_id: str, name: str) -> bool:
|
|
"""Switch to a different branch."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory:
|
|
return False
|
|
return self._backend.switch_branch(directory, name)
|
|
|
|
def has_uncommitted_changes(self, space_id: str) -> bool:
|
|
"""Check if space has uncommitted changes."""
|
|
directory = self._directory_resolver(space_id)
|
|
if not directory:
|
|
return False
|
|
return self._backend.has_changes(directory)
|