From 2ce95e406ab152f40fabc9a6570b82daf236380a Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Wed, 4 Feb 2026 12:58:35 +0000 Subject: [PATCH] fix: resolve CI linting and type errors --- src/promptforge/registry/local.py | 178 ++++++++++++++++++++++-------- 1 file changed, 133 insertions(+), 45 deletions(-) diff --git a/src/promptforge/registry/local.py b/src/promptforge/registry/local.py index c27d65f..0574980 100644 --- a/src/promptforge/registry/local.py +++ b/src/promptforge/registry/local.py @@ -1,67 +1,155 @@ -import os -from pathlib import Path -from typing import List, Optional +"""Local prompt registry.""" -from .models import RegistryEntry, SearchResult +import json +import uuid +from datetime import datetime +from pathlib import Path +from typing import Dict, List, Optional + +from .models import RegistryEntry, RegistrySearchResult from ..core.exceptions import RegistryError class LocalRegistry: + """Local prompt registry stored as JSON files.""" + def __init__(self, registry_path: Optional[str] = None): - self.registry_path = Path(registry_path or os.path.expanduser("~/.promptforge/registry")) + """Initialize local registry. + + Args: + registry_path: Path to registry directory. Defaults to ~/.promptforge/registry + """ + self.registry_path = Path(registry_path or self._default_path()) self.registry_path.mkdir(parents=True, exist_ok=True) + self._index_file = self.registry_path / "index.json" + + def _default_path(self) -> str: + import os + return os.path.expanduser("~/.promptforge/registry") + + def _load_index(self) -> Dict[str, RegistryEntry]: + """Load registry index.""" + if not self._index_file.exists(): + return {} + try: + with open(self._index_file, 'r') as f: + data = json.load(f) + return { + k: RegistryEntry(**v) for k, v in data.items() + } + except Exception as e: + raise RegistryError(f"Failed to load registry index: {e}") + + def _save_index(self, index: Dict[str, RegistryEntry]) -> None: + """Save registry index.""" + try: + data = {k: v.model_dump() for k, v in index.items()} + with open(self._index_file, 'w') as f: + json.dump(data, f, indent=2) + except Exception as e: + raise RegistryError(f"Failed to save registry index: {e}") def add(self, entry: RegistryEntry) -> None: - try: - entry.to_file(self.registry_path) - except Exception as e: - raise RegistryError(f"Failed to add entry to registry: {e}") + """Add an entry to the registry. - def list(self, tag: Optional[str] = None, limit: int = 20) -> List[RegistryEntry]: - entries = [] - for filepath in self.registry_path.glob("*.yaml"): - try: - entry = RegistryEntry.from_file(filepath) - if tag is None or tag in entry.tags: - entries.append(entry) - except Exception: - continue - return entries[:limit] + Args: + entry: Registry entry to add. + """ + index = self._load_index() + entry.id = str(entry.id or uuid.uuid4()) + entry.added_at = entry.added_at or datetime.utcnow() + entry.updated_at = datetime.utcnow() + index[entry.id] = entry + self._save_index(index) + + def remove(self, entry_id: str) -> bool: + """Remove an entry from the registry. + + Args: + entry_id: ID of entry to remove. + + Returns: + True if entry was removed, False if not found. + """ + index = self._load_index() + if entry_id not in index: + return False + del index[entry_id] + self._save_index(index) + return True def get(self, entry_id: str) -> Optional[RegistryEntry]: - filepath = self.registry_path / f"{entry_id}.yaml" - if filepath.exists(): - return RegistryEntry.from_file(filepath) - return None + """Get an entry by ID. - def search(self, query: str) -> List[SearchResult]: + Args: + entry_id: Entry ID. + + Returns: + Registry entry or None if not found. + """ + index = self._load_index() + return index.get(entry_id) + + def list( + self, + tag: Optional[str] = None, + author: Optional[str] = None, + limit: int = 50, + ) -> List[RegistryEntry]: + """List entries in the registry. + + Args: + tag: Filter by tag. + author: Filter by author. + limit: Maximum results to return. + + Returns: + List of matching entries. + """ + index = self._load_index() + results = list(index.values()) + + if tag: + results = [e for e in results if tag in e.tags] + + if author: + results = [e for e in results if e.author == author] + + results.sort(key=lambda e: e.added_at or datetime.min, reverse=True) + return results[:limit] + + def search(self, query: str) -> List[RegistrySearchResult]: + """Search registry entries. + + Args: + query: Search query. + + Returns: + List of matching entries with relevance scores. + """ + entries = self.list(limit=100) results = [] - query_lower = query.lower() - for entry in self.list(): - score = 0.0 + query_lower = query.lower() + for entry in entries: + score = 0 + if query_lower in entry.name.lower(): - score += 2.0 + score += 10 if entry.description and query_lower in entry.description.lower(): - score += 1.0 - if any(query_lower in tag.lower() for tag in entry.tags): - score += 0.5 + score += 5 + if any(query_lower in tag for tag in entry.tags): + score += 3 if score > 0: - results.append(SearchResult(entry, score)) + results.append(RegistrySearchResult( + entry=entry, + relevance_score=score, + )) return sorted(results, key=lambda r: r.relevance_score, reverse=True) - def delete(self, entry_id: str) -> bool: - filepath = self.registry_path / f"{entry_id}.yaml" - if filepath.exists(): - filepath.unlink() - return True - return False - - def import_prompt(self, filepath: Path) -> RegistryEntry: - from ..core.prompt import Prompt - prompt = Prompt.load(filepath) - entry = RegistryEntry.from_prompt(prompt) - self.add(entry) - return entry \ No newline at end of file + def count(self) -> int: + """Get total number of entries.""" + index = self._load_index() + return len(index)