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

This commit is contained in:
2026-02-04 12:49:08 +00:00
parent 8639f988b8
commit d8ecd258e9

View File

@@ -0,0 +1,153 @@
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):
"""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:
"""Add an entry to the registry.
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]:
"""Get an entry by ID.
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 entries:
score = 0
if query_lower in entry.name.lower():
score += 10
if entry.description and query_lower in entry.description.lower():
score += 5
if any(query_lower in tag for tag in entry.tags):
score += 3
if score > 0:
results.append(RegistrySearchResult(
entry=entry,
relevance_score=score,
))
return sorted(results, key=lambda r: r.relevance_score, reverse=True)
def count(self) -> int:
"""Get total number of entries."""
index = self._load_index()
return len(index)