Add registry module (local, remote, models)
This commit is contained in:
53
src/promptforge/registry/remote.py
Normal file
53
src/promptforge/registry/remote.py
Normal file
@@ -0,0 +1,53 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from .local import LocalRegistry
|
||||
from .models import RegistryEntry, SearchResult
|
||||
from ..core.exceptions import RegistryError
|
||||
|
||||
|
||||
class RemoteRegistry:
|
||||
def __init__(self, base_url: str = "https://registry.promptforge.io"):
|
||||
self.base_url = base_url.rstrip('/')
|
||||
|
||||
def publish(self, entry: RegistryEntry) -> RegistryEntry:
|
||||
try:
|
||||
import requests
|
||||
response = requests.post(
|
||||
f"{self.base_url}/api/entries",
|
||||
json=entry.model_dump(),
|
||||
timeout=30
|
||||
)
|
||||
response.raise_for_status()
|
||||
return RegistryEntry(**response.json())
|
||||
except Exception as e:
|
||||
raise RegistryError(f"Failed to publish to remote registry: {e}")
|
||||
|
||||
def pull(self, entry_id: str, local: LocalRegistry) -> bool:
|
||||
try:
|
||||
import requests
|
||||
response = requests.get(
|
||||
f"{self.base_url}/api/entries/{entry_id}",
|
||||
timeout=30
|
||||
)
|
||||
if response.status_code == 404:
|
||||
return False
|
||||
response.raise_for_status()
|
||||
entry = RegistryEntry(**response.json())
|
||||
local.add(entry)
|
||||
return True
|
||||
except Exception as e:
|
||||
raise RegistryError(f"Failed to pull from remote registry: {e}")
|
||||
|
||||
def search(self, query: str) -> List[SearchResult]:
|
||||
try:
|
||||
import requests
|
||||
response = requests.get(
|
||||
f"{self.base_url}/api/search",
|
||||
params={"q": query},
|
||||
timeout=30
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
return [SearchResult(RegistryEntry(**e), 1.0) for e in data.get("results", [])]
|
||||
except Exception:
|
||||
return []
|
||||
Reference in New Issue
Block a user