Add registry module (local, remote, models)
This commit is contained in:
67
src/promptforge/registry/local.py
Normal file
67
src/promptforge/registry/local.py
Normal file
@@ -0,0 +1,67 @@
|
|||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import List, Optional
|
||||||
|
|
||||||
|
from .models import RegistryEntry, SearchResult
|
||||||
|
from ..core.exceptions import RegistryError
|
||||||
|
|
||||||
|
|
||||||
|
class LocalRegistry:
|
||||||
|
def __init__(self, registry_path: Optional[str] = None):
|
||||||
|
self.registry_path = Path(registry_path or os.path.expanduser("~/.promptforge/registry"))
|
||||||
|
self.registry_path.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
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}")
|
||||||
|
|
||||||
|
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]
|
||||||
|
|
||||||
|
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
|
||||||
|
|
||||||
|
def search(self, query: str) -> List[SearchResult]:
|
||||||
|
results = []
|
||||||
|
query_lower = query.lower()
|
||||||
|
|
||||||
|
for entry in self.list():
|
||||||
|
score = 0.0
|
||||||
|
if query_lower in entry.name.lower():
|
||||||
|
score += 2.0
|
||||||
|
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
|
||||||
|
|
||||||
|
if score > 0:
|
||||||
|
results.append(SearchResult(entry, 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
|
||||||
Reference in New Issue
Block a user