chore: progress with catalog manager

Author: jeafreezy

pyproject.toml

@@ -4,7 +4,9 @@ version = "0.1.0"
 description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.9"
-dependencies = []
+dependencies = [
+    "pystac-client==0.8.5",
+]
 
 [tool.uv]
 dev-dependencies = [

ruff.toml

@@ -1,5 +1,9 @@
 catalogs:
   Earth Search:
     url: https://earth-search.aws.element84.com/v1
+    summary: 'Element 84 STAC catalog'
+    is_private: False 
   Planetary Computer:
     url: https://planetarycomputer.microsoft.com/api/stac/v1
+    summary: 'Microsoft Planetary Computer STAC catalog'
+    is_private: False 

superstac/.superstac.yml

@@ -1 +1,129 @@
-class CatalogManager: ...
+"""SuperSTAC Catalog Manager"""
+
+from pathlib import Path
+import attr
+from typing import Any, Dict, Optional, Union
+
+from superstac.enums import CatalogOutputFormat
+from superstac.exceptions import (
+    CatalogConfigFileNotFound,
+    InvalidCatalogSchemaError,
+    InvalidCatalogYAMLError,
+)
+from superstac.models import CatalogEntry, AuthInfo
+import yaml
+
+
+@attr.s(auto_attribs=True)
+class CatalogManager:
+    catalogs: Dict[str, CatalogEntry] = attr.Factory(dict)
+
+    def register_catalog(
+        self,
+        name: str,
+        url: str,
+        is_private: Optional[bool] = False,
+        summary: Optional[str] = None,
+        auth: Optional[AuthInfo] = None,
+    ) -> CatalogEntry:
+        """Register a single STAC catalog in state.
+
+        Args:
+            name (str): The name of the catalog.
+            url (str): A valid URL to the catalog.
+            is_private (Optional[bool], optional): Indicates if the catalog requires authentication or not. Defaults to False.
+            summary (Optional[str], optional): A short description of the catalog. Defaults to None.
+            auth (Optional[AuthInfo], optional): Authentication parameters for the catalog. Defaults to None.
+
+        Raises:
+            InvalidCatalogSchemaError: If an invalid parameter is encountered.
+
+        Returns:
+            CatalogEntry: The registered STAC catalog.
+        """
+        if is_private and auth is None:
+            raise InvalidCatalogSchemaError(
+                f"Authentication parameters is required for private catalogs. If this is a mistake, you can set 'is_private' to False or provide the {AuthInfo.__annotations__} parameters."
+            )
+        self.catalogs[name] = CatalogEntry(
+            name=name,
+            url=url,
+            summary=summary,
+            is_private=is_private,
+            auth=AuthInfo(**auth.__dict__) if auth and not is_private else None,
+        )
+        return self.catalogs[name]
+
+    def get_available_catalogs(
+        self, format: Union[str, CatalogOutputFormat] = CatalogOutputFormat.DICT
+    ) -> list[Union[dict[str, Any], str]]:
+        """Get the available STAC catalogs.
+
+        Raises:
+            ValueError: When an invalid format is provided.
+
+        Returns:
+            list[CatalogEntry]: The list of all available STAC catalogs.
+        """
+        if isinstance(format, str):
+            try:
+                format = CatalogOutputFormat(format.lower())
+            except ValueError:
+                raise ValueError(f"Invalid format: {format}")
+
+        return [
+            c.as_dict() if format == CatalogOutputFormat.DICT else c.as_json()
+            for c in self.catalogs.values()
+            if c.is_available
+        ]
+
+    def load_catalogs_from_config(
+        self, file: Union[str, Path, None] = None
+    ) -> Dict[str, CatalogEntry]:
+        if file is None:
+            base_dir = Path(__file__).parent
+            file = base_dir / ".superstac.yml"
+
+        path = Path(file).expanduser().resolve()
+
+        if not path.exists():
+            raise CatalogConfigFileNotFound(f"Config file not found at {path}")
+
+        try:
+            with open(path, "r") as f:
+                data = yaml.safe_load(f) or {}
+        except yaml.YAMLError as e:
+            raise InvalidCatalogYAMLError(f"YAML parsing failed: {e}") from e
+        except Exception as e:
+            raise InvalidCatalogYAMLError(
+                f"Unexpected error reading config: {e}"
+            ) from e
+
+        catalogs = data.get("catalogs")
+        if not isinstance(catalogs, dict):
+            raise InvalidCatalogSchemaError(
+                f"Missing or invalid 'catalogs' section in config file: {path}"
+            )
+
+        # Register each catalog
+        for name, spec in catalogs.items():
+            self.register_catalog(
+                name=name,
+                url=spec.get("url"),
+                is_private=spec.get("is_private", False),
+                summary=spec.get("summary"),
+                auth=AuthInfo(**spec["auth"]) if "auth" in spec else None,
+            )
+
+        return self.catalogs
+
+
+## TEST
+
+
+if __name__ == "__main__":
+    # cm = CatalogManager()
+    # cm.register_catalog(name="My Catalog", url="https://example.com/stac")
+    # print(cm.load_catalogs_from_config())
+    # print(cm.get_available_catalogs())
+    ...

superstac/catalog.py

@@ -5,3 +5,8 @@ class AuthType(str, Enum):
     BEARER = "bearer"
     BASIC = "basic"
     API_KEY = "apikey"
+
+
+class CatalogOutputFormat(Enum):
+    JSON = "json"
+    DICT = "dict"

superstac/enums.py

@@ -0,0 +1,14 @@
+class InvalidCatalogSchemaError(Exception):
+    """Raised when an invalid catalog schema is provided."""
+
+
+class ParametersError(Exception):
+    """Raised when invalid parameters are used in a query."""
+
+
+class CatalogConfigFileNotFound(FileNotFoundError):
+    """Raised when the provided catalog config file is not found."""
+
+
+class InvalidCatalogYAMLError(Exception):
+    """Raised when invalid catalog yaml is passed or if does not conform to the schema."""

superstac/exceptions.py

@@ -1,23 +1,62 @@
-from dataclasses import dataclass, field, asdict
-from typing import Optional, List
+"""SuperSTAC Models"""
 
+from dataclasses import asdict, dataclass
+import json
+from typing import Optional
+from urllib.parse import urlparse
 
+from superstac.enums import AuthType
+
+
+# Todo - make auth type have a default ?
 @dataclass
 class AuthInfo:
-    type: str
+    """AuthInfo"""
+
+    type: AuthType
     token: Optional[str] = None
     username: Optional[str] = None
     password: Optional[str] = None
     header_key: Optional[str] = None
 
+    def as_dict(self):
+        return asdict(self)
+
+    def as_json(self):
+        return json.dumps(self.as_dict(), indent=2)
+
 
 @dataclass
 class CatalogEntry:
+    """CatalogEntry"""
+
     name: str
     url: str
+    """URL to the STAC catalog."""
+    is_private: Optional[bool] = False
+    """Indicates whether the catalog is a private catalog or not."""
+    summary: Optional[str] = None
+    """Short description of the catalog."""
     auth: Optional[AuthInfo] = None
-    is_available: bool = False
-    latency_ms: Optional[float] = None
-    conforms_to: Optional[List[str]] = field(default_factory=list)
-    collections: Optional[List[str]] = field(default_factory=list)
-    extensions: Optional[List[str]] = field(default_factory=list)
+    """Authentication parameters."""
+    is_available: Optional[bool] = True
+    """Defaults to True on instantiation. It will be updated based on the status code of the catalog. If 200 True and False if otherwise."""
+
+    def __post_init__(self):
+        # Validate the URL
+        parsed_url = urlparse(self.url)
+        if not all([parsed_url.scheme, parsed_url.netloc]):
+            raise ValueError(
+                f"Invalid URL: '{self.url}' - Must have a scheme (e.g., http/https) and network location."
+            )
+
+        if parsed_url.scheme not in ("http", "https"):
+            raise ValueError(
+                f"Invalid URL scheme: '{parsed_url.scheme}' for '{self.url}' - Only http and https are allowed."
+            )
+
+    def as_dict(self):
+        return asdict(self)
+
+    def as_json(self):
+        return json.dumps(self.as_dict(), indent=2)