feat: setup logging + unique id

Author: jeafreezy

superstac/_logging.py

@@ -0,0 +1,58 @@
+"""SuperSTAC logging module"""
+
+import logging
+
+
+PACKAGE_NAME = "superstac"
+LOGGING_FORMAT = f"{PACKAGE_NAME}:%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+
+logger = logging.getLogger(PACKAGE_NAME)
+
+
+# Set the default logging level to INFO.
+logger.setLevel(logging.INFO)
+
+
+console_handler = logging.StreamHandler()
+
+console_formatter = logging.Formatter(LOGGING_FORMAT)
+console_handler.setFormatter(console_formatter)
+
+logger.addHandler(console_handler)
+
+
+def add_file_logging(file_path=f"{PACKAGE_NAME}.log"):
+    """
+    Add file logging to the logger.
+
+    Args:
+        file_path (str): Path to the log file. Defaults to `superstac.log`.
+
+    Example:
+        from superstac import add_file_logging
+        # Configure custom file log
+        add_file_logging("my_log.log")
+    """
+    file_handler = logging.FileHandler(file_path)
+    file_formatter = logging.Formatter(LOGGING_FORMAT)
+    file_handler.setFormatter(file_formatter)
+    logger.addHandler(file_handler)
+
+
+def configure_logging(level=logging.WARNING):
+    """
+    Configure the logging level for the package.
+
+    Args:
+        level (int): Logging level (e.g., logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR)
+
+    Example:
+        # Set logging level to INFO
+        from superstac import configure_logging
+        import logging
+        # Configure logging to INFO level
+        configure_logging(logging.INFO)
+    """
+    logger.setLevel(level)
+    for handler in logger.handlers:
+        handler.setLevel(level)

superstac/catalog.py

@@ -4,6 +4,7 @@
 import attr
 from typing import Any, Dict, Optional, Union
 
+from superstac._logging import logger
 from superstac.enums import CatalogOutputFormat
 from superstac.exceptions import (
     CatalogConfigFileNotFound,
@@ -16,6 +17,7 @@
 
 @attr.s(auto_attribs=True)
 class CatalogManager:
+    logger.info("Initialized superstac")
     catalogs: Dict[str, CatalogEntry] = attr.Factory(dict)
 
     def register_catalog(
@@ -41,10 +43,19 @@ def register_catalog(
         Returns:
             CatalogEntry: The registered STAC catalog.
         """
+        logger.info(f"Registering catalog: {name}")
+        logger.debug(
+            f"Params - url: {url}, is_private: {is_private}, summary: {summary}, auth: {auth}"
+        )
         if is_private and auth is None:
+            logger.error(
+                f"Private catalog '{name}' requires authentication but none was provided."
+            )
             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."
             )
+
+        logger.info(f"Catalog '{name}' registered successfully.")
         self.catalogs[name] = CatalogEntry(
             name=name,
             url=url,
@@ -65,56 +76,86 @@ def get_available_catalogs(
         Returns:
             list[CatalogEntry]: The list of all available STAC catalogs.
         """
+        logger.info("Retrieving available catalogs.")
         if isinstance(format, str):
             try:
                 format = CatalogOutputFormat(format.lower())
             except ValueError:
+                logger.error(f"Invalid output format: {format}")
                 raise ValueError(f"Invalid format: {format}")
 
-        return [
+        available = [
             c.as_dict() if format == CatalogOutputFormat.DICT else c.as_json()
             for c in self.catalogs.values()
             if c.is_available
         ]
 
+        logger.info(f"{len(available)} catalogs available in format '{format.value}'.")
+        return available
+
     def load_catalogs_from_config(
         self, file: Union[str, Path, None] = None
     ) -> Dict[str, CatalogEntry]:
+        """Load catalogs from configuration file.
+
+        Args:
+            file (Union[str, Path, None], optional): Path to the configuration file. Defaults to None.
+
+        Raises:
+            CatalogConfigFileNotFound: Raised when the catalog config file is not founds.
+            InvalidCatalogYAMLError: Raised when the yaml file is invalid.
+            InvalidCatalogSchemaError: Raised when there is a schema error in the provided config file.
+
+        Returns:
+            Dict[str, CatalogEntry]: The registered catalogs.
+        """
+        logger.info("Loading catalogs from configuration file.")
         if file is None:
             base_dir = Path(__file__).parent
             file = base_dir / ".superstac.yml"
 
         path = Path(file).expanduser().resolve()
+        logger.debug(f"Resolved config path: {path}")
 
         if not path.exists():
+            logger.error(f"Config file not found at path: {path}")
             raise CatalogConfigFileNotFound(f"Config file not found at {path}")
 
         try:
             with open(path, "r") as f:
                 data = yaml.safe_load(f) or {}
+            logger.info(f"Successfully loaded YAML config from: {path}")
         except yaml.YAMLError as e:
+            logger.exception("YAML parsing failed.")
             raise InvalidCatalogYAMLError(f"YAML parsing failed: {e}") from e
         except Exception as e:
+            logger.exception("Unexpected error while reading config.")
             raise InvalidCatalogYAMLError(
                 f"Unexpected error reading config: {e}"
             ) from e
 
         catalogs = data.get("catalogs")
         if not isinstance(catalogs, dict):
+            logger.error(
+                f"Missing or invalid 'catalogs' section in config file: {path}"
+            )
             raise InvalidCatalogSchemaError(
                 f"Missing or invalid 'catalogs' section in config file: {path}"
             )
 
-        # Register each catalog
+        logger.info(f"Found {len(catalogs)} catalogs to register.")
         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,
-            )
-
+            try:
+                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,
+                )
+            except Exception as e:
+                logger.warning(f"Failed to register catalog '{name}': {e}")
+        logger.info("All catalogs loaded and registered.")
         return self.catalogs
 
 

superstac/logger.py

@@ -6,6 +6,7 @@
 from urllib.parse import urlparse
 
 from superstac.enums import AuthType
+from superstac.utils import compute_catalog_id
 
 
 # Todo - make auth type have a default ?
@@ -32,6 +33,8 @@ class CatalogEntry:
 
     name: str
     url: str
+    id: Optional[str] = None
+    """Internal unique ID for the catalog. Will be autogenerated post init."""
     """URL to the STAC catalog."""
     is_private: Optional[bool] = False
     """Indicates whether the catalog is a private catalog or not."""
@@ -54,6 +57,7 @@ def __post_init__(self):
             raise ValueError(
                 f"Invalid URL scheme: '{parsed_url.scheme}' for '{self.url}' - Only http and https are allowed."
             )
+        self.id = compute_catalog_id(self.name, self.url)
 
     def as_dict(self):
         return asdict(self)

superstac/models.py

@@ -0,0 +1,15 @@
+import uuid
+
+
+def compute_catalog_id(name: str, url: str) -> str:
+    """Compute a unique catalog id.
+
+    Args:
+        name (str): The name of the catalog.
+        url (str): The url of the catalog.
+
+    Returns:
+        str: A unique uuid for the catalog.
+    """
+    uid = uuid.uuid4().hex[:10]
+    return f"{name.lower().replace(' ', '_')}_{uid}"