fix: resolve CI linting and type errors
Some checks failed
CI / test (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / type-check (push) Has been cancelled

This commit is contained in:
2026-02-04 12:58:35 +00:00
parent 5a275b812b
commit 2ce95e406a

View File

@@ -1,67 +1,155 @@
import os """Local prompt registry."""
from pathlib import Path
from typing import List, Optional
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 from ..core.exceptions import RegistryError
class LocalRegistry: class LocalRegistry:
"""Local prompt registry stored as JSON files."""
def __init__(self, registry_path: Optional[str] = None): 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.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: def add(self, entry: RegistryEntry) -> None:
try: """Add an entry to the registry.
entry.to_file(self.registry_path)
except Exception as e:
raise RegistryError(f"Failed to add entry to registry: {e}")
def list(self, tag: Optional[str] = None, limit: int = 20) -> List[RegistryEntry]: Args:
entries = [] entry: Registry entry to add.
for filepath in self.registry_path.glob("*.yaml"): """
try: index = self._load_index()
entry = RegistryEntry.from_file(filepath) entry.id = str(entry.id or uuid.uuid4())
if tag is None or tag in entry.tags: entry.added_at = entry.added_at or datetime.utcnow()
entries.append(entry) entry.updated_at = datetime.utcnow()
except Exception: index[entry.id] = entry
continue self._save_index(index)
return entries[:limit]
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]: def get(self, entry_id: str) -> Optional[RegistryEntry]:
filepath = self.registry_path / f"{entry_id}.yaml" """Get an entry by ID.
if filepath.exists():
return RegistryEntry.from_file(filepath)
return None
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 = [] results = []
query_lower = query.lower()
for entry in self.list(): query_lower = query.lower()
score = 0.0 for entry in entries:
score = 0
if query_lower in entry.name.lower(): if query_lower in entry.name.lower():
score += 2.0 score += 10
if entry.description and query_lower in entry.description.lower(): if entry.description and query_lower in entry.description.lower():
score += 1.0 score += 5
if any(query_lower in tag.lower() for tag in entry.tags): if any(query_lower in tag for tag in entry.tags):
score += 0.5 score += 3
if score > 0: 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) return sorted(results, key=lambda r: r.relevance_score, reverse=True)
def delete(self, entry_id: str) -> bool: def count(self) -> int:
filepath = self.registry_path / f"{entry_id}.yaml" """Get total number of entries."""
if filepath.exists(): index = self._load_index()
filepath.unlink() return len(index)
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