diff --git a/src/promptforge/registry/remote.py b/src/promptforge/registry/remote.py index 0d1daf7..c9b6ee2 100644 --- a/src/promptforge/registry/remote.py +++ b/src/promptforge/registry/remote.py @@ -1,53 +1,149 @@ -from typing import List, Optional +"""Remote registry via HTTP.""" -from .local import LocalRegistry -from .models import RegistryEntry, SearchResult +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +import requests + +from .models import RegistryEntry, RegistrySearchResult from ..core.exceptions import RegistryError +if TYPE_CHECKING: + from .local import LocalRegistry + class RemoteRegistry: - def __init__(self, base_url: str = "https://registry.promptforge.io"): + """Remote prompt registry accessed via HTTP API.""" + + def __init__( + self, + base_url: str = "https://registry.promptforge.io", + api_key: Optional[str] = None, + ): + """Initialize remote registry. + + Args: + base_url: Base URL of the registry API. + api_key: API key for authentication. + """ self.base_url = base_url.rstrip('/') + self.api_key = api_key + self._session = requests.Session() + if api_key: + self._session.headers.update({"Authorization": f"Bearer {api_key}"}) + + def _request( + self, + method: str, + endpoint: str, + data: Optional[Dict] = None, + ) -> Any: + """Make HTTP request to registry. + + Args: + method: HTTP method. + endpoint: API endpoint. + data: Request data. + + Returns: + Response JSON. + + Raises: + RegistryError: If request fails. + """ + url = f"{self.base_url}/api/v1{endpoint}" + try: + response = self._session.request(method, url, json=data) + response.raise_for_status() + return response.json() + except requests.HTTPError as e: + raise RegistryError(f"Registry API error: {e.response.text}") + except requests.RequestException as e: + raise RegistryError(f"Registry connection error: {e}") + + def search(self, query: str, limit: int = 20) -> List[RegistrySearchResult]: + """Search remote registry. + + Args: + query: Search query. + limit: Maximum results. + + Returns: + List of matching entries. + """ + data = self._request("GET", f"/search?q={query}&limit={limit}") + results = [] + for item in data.get("results", []): + entry = RegistryEntry(**item) + results.append(RegistrySearchResult( + entry=entry, + relevance_score=item.get("score", 0), + )) + return results + + def get(self, entry_id: str) -> Optional[RegistryEntry]: + """Get entry by ID. + + Args: + entry_id: Entry ID. + + Returns: + Registry entry or None. + """ + try: + data = self._request("GET", f"/entries/{entry_id}") + return RegistryEntry(**data) + except RegistryError: + return None + + def pull(self, entry_id: str, local_registry: "LocalRegistry") -> bool: + """Pull entry from remote to local registry. + + Args: + entry_id: Entry ID to pull. + local_registry: Local registry to save to. + + Returns: + True if successful. + """ + entry = self.get(entry_id) + if entry is None: + return False + entry.is_local = True + local_registry.add(entry) + return True 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}") + """Publish entry to remote registry. - def pull(self, entry_id: str, local: LocalRegistry) -> bool: + Args: + entry: Entry to publish. + + Returns: + Published entry with server-assigned ID. + """ + data = self._request("POST", "/entries", entry.model_dump()) + return RegistryEntry(**data) + + def list_popular(self, limit: int = 10) -> List[RegistryEntry]: + """List popular entries. + + Args: + limit: Maximum results. + + Returns: + List of popular entries. + """ + data = self._request("GET", f"/popular?limit={limit}") + return [RegistryEntry(**item) for item in data.get("entries", [])] + + def validate_connection(self) -> bool: + """Validate connection to remote registry. + + Returns: + True if connection successful. + """ 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) + self._request("GET", "/health") 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 [] \ No newline at end of file + except RegistryError: + return False