Updated publisher logic to support multi mode parameter

Author: TejasMorbagal

CHANGES.md

@@ -33,4 +33,11 @@
 - Introduced build_link_to_jnb method for creating STAC-compatible notebook links with 
   metadata on kernel, environment, and containerization.
 - Added originating application platform metadata to generated OGC API records for 
-  DeepESDL experiments and workflows.
+  DeepESDL experiments and workflows.
+
+## Changes in 0.1.6
+
+- Publisher now supports `mode` parameter, This allows more flexible publishing:
+  - `"dataset"` → publish dataset only
+  - `"workflow"` → publish workflow only
+  - `"all"` → publish both (default)

deep_code/cli/publish.py

@@ -4,11 +4,43 @@
 # Permissions are hereby granted under the terms of the MIT License:
 # https://opensource.org/licenses/MIT.
 
+from pathlib import Path
+
 import click
 
 from deep_code.tools.publish import Publisher
 
 
+def _validate_inputs(dataset_config, workflow_config, mode):
+    mode = mode.lower()
+
+    def ensure_file(path: str, label: str):
+        if path is None:
+            raise click.UsageError(f"{label} is required but was not provided.")
+        if not Path(path).is_file():
+            raise click.UsageError(f"{label} not found: {path}")
+
+    if mode == "dataset":
+        # Need dataset only
+        ensure_file(dataset_config, "DATASET_CONFIG")
+        if workflow_config is not None:
+            click.echo("ℹ️ Ignoring WORKFLOW_CONFIG since mode=dataset.", err=True)
+
+    elif mode == "workflow":
+        # Need workflow config only
+        ensure_file(workflow_config, "WORKFLOW_CONFIG")
+
+    elif mode == "all":
+        # Need both
+        ensure_file(dataset_config, "DATASET_CONFIG")
+        ensure_file(workflow_config, "WORKFLOW_CONFIG")
+
+    else:
+        raise click.UsageError(
+            "Invalid mode. Choose one of: all, dataset, workflow_experiment."
+        )
+
+
 @click.command(name="publish")
 @click.argument("dataset_config", type=click.Path(exists=True))
 @click.argument("workflow_config", type=click.Path(exists=True))
@@ -19,7 +51,14 @@
     default="production",
     help="Target environment for publishing (production, staging, testing)",
 )
-def publish(dataset_config, workflow_config, environment):
+@click.option(
+    "--mode",
+    "-m",
+    type=click.Choice(["all", "dataset", "workflow"], case_sensitive=False),
+    default="all",
+    help="Publishing mode: dataset only, workflow only, or both",
+)
+def publish(dataset_config, workflow_config, environment, mode):
     """Request publishing a dataset along with experiment and workflow metadata to the
     open science catalogue.
     """
@@ -28,4 +67,5 @@ def publish(dataset_config, workflow_config, environment):
         workflow_config_path=workflow_config,
         environment=environment.lower(),
     )
-    publisher.publish_all()
+    result = publisher.publish(mode=mode.lower())
+    click.echo(f"Pull request created: {result}")

deep_code/tools/publish.py

@@ -7,6 +7,7 @@
 import logging
 from datetime import datetime
 from pathlib import Path
+from typing import Any, Literal
 
 import fsspec
 import jsonpickle
@@ -98,13 +99,14 @@ def publish_files(
 
 
 class Publisher:
-    """Publishes products (datasets) to the OSC GitHub repository.
+    """Publishes products (datasets), workflows and experiments to the OSC GitHub
+    repository.
     """
 
     def __init__(
         self,
-        dataset_config_path: str,
-        workflow_config_path: str,
+        dataset_config_path: str | None = None,
+        workflow_config_path: str | None = None,
         environment: str = "production",
     ):
         self.environment = environment
@@ -121,24 +123,37 @@ def __init__(
         self.collection_id = ""
         self.workflow_title = ""
 
-        # Paths to configuration files
+        # Paths to configuration files, may be optional
         self.dataset_config_path = dataset_config_path
         self.workflow_config_path = workflow_config_path
 
+        # Config dicts (loaded lazily)
+        self.dataset_config: dict[str, Any] = {}
+        self.workflow_config: dict[str, Any] = {}
+
+        # Values that may be set from configs (lazy)
+        self.collection_id: str | None = None
+        self.workflow_title: str | None = None
+        self.workflow_id: str | None = None
+
         # Load configuration files
         self._read_config_files()
-        self.collection_id = self.dataset_config.get("collection_id")
-        self.workflow_title = self.workflow_config.get("properties", {}).get("title")
-        self.workflow_id = self.workflow_config.get("workflow_id")
 
-        if not self.collection_id:
-            raise ValueError("collection_id is missing in dataset config.")
+        if self.dataset_config:
+            self.collection_id = self.dataset_config.get("collection_id")
+        if self.workflow_config:
+            self.workflow_title = self.workflow_config.get("properties", {}).get(
+                "title"
+            )
+            self.workflow_id = self.workflow_config.get("workflow_id")
 
     def _read_config_files(self) -> None:
-        with fsspec.open(self.dataset_config_path, "r") as file:
-            self.dataset_config = yaml.safe_load(file) or {}
-        with fsspec.open(self.workflow_config_path, "r") as file:
-            self.workflow_config = yaml.safe_load(file) or {}
+        if self.dataset_config_path:
+            with fsspec.open(self.dataset_config_path, "r") as file:
+                self.dataset_config = yaml.safe_load(file) or {}
+        if self.workflow_config_path:
+            with fsspec.open(self.workflow_config_path, "r") as file:
+                self.workflow_config = yaml.safe_load(file) or {}
 
     @staticmethod
     def _write_to_file(file_path: str, data: dict):
@@ -206,10 +221,18 @@ def _update_variable_catalogs(self, generator, file_dict, variable_ids):
                 )
                 file_dict[var_file_path] = updated_catalog.to_dict()
 
-    def publish_dataset(self, write_to_file: bool = False):
+    def publish_dataset(
+        self,
+        write_to_file: bool = False,
+        mode: Literal["all", "dataset", "workflow"] = "all",
+    ) -> dict[str, Any]:
         """Prepare dataset/product collection for publishing to the specified GitHub
         repository."""
 
+        if not self.dataset_config:
+            raise ValueError(
+                "No dataset config loaded. Provide dataset_config_path to publish dataset."
+            )
         dataset_id = self.dataset_config.get("dataset_id")
         self.collection_id = self.dataset_config.get("collection_id")
         documentation_link = self.dataset_config.get("documentation_link")
@@ -240,7 +263,7 @@ def publish_dataset(self, write_to_file: bool = False):
         )
 
         variable_ids = generator.get_variable_ids()
-        ds_collection = generator.build_dataset_stac_collection()
+        ds_collection = generator.build_dataset_stac_collection(mode=mode)
 
         # Prepare a dictionary of file paths and content
         file_dict = {}
@@ -315,9 +338,19 @@ def _update_base_catalog(
 
         return base_catalog
 
-    def generate_workflow_experiment_records(self, write_to_file: bool = False) -> None:
+    def generate_workflow_experiment_records(
+        self,
+        write_to_file: bool = False,
+        mode: Literal["all", "dataset", "workflow"] = "all",
+    ) -> dict[str, Any]:
         """prepare workflow and experiment as ogc api record to publish it to the
         specified GitHub repository."""
+
+        file_dict = {}
+
+        if mode not in {"workflow", "all"}:
+            return file_dict  # nothing to do for mode="dataset"
+
         workflow_id = self._normalize_name(self.workflow_config.get("workflow_id"))
         if not workflow_id:
             raise ValueError("workflow_id is missing in workflow config.")
@@ -367,85 +400,117 @@ def generate_workflow_experiment_records(self, write_to_file: bool = False) -> N
         exp_record_properties.type = "experiment"
         exp_record_properties.osc_workflow = workflow_id
 
-        dataset_link = link_builder.build_link_to_dataset(self.collection_id)
-
-        experiment_record = ExperimentAsOgcRecord(
-            id=workflow_id,
-            title=self.workflow_title,
-            type="Feature",
-            jupyter_notebook_url=jupyter_notebook_url,
-            collection_id=self.collection_id,
-            properties=exp_record_properties,
-            links=links + theme_links + dataset_link,
-        )
-        # Convert to dictionary and cleanup
-        experiment_dict = experiment_record.to_dict()
-        if "jupyter_notebook_url" in experiment_dict:
-            del experiment_dict["jupyter_notebook_url"]
-        if "collection_id" in experiment_dict:
-            del experiment_dict["collection_id"]
-        if "osc:project" in experiment_dict["properties"]:
-            del experiment_dict["properties"]["osc:project"]
-        # add experiment record to file_dict
-        exp_file_path = f"experiments/{workflow_id}/record.json"
-        file_dict[exp_file_path] = experiment_dict
-
-        # Update base catalogs of experiments and workflows with links
-        file_dict["experiments/catalog.json"] = self._update_base_catalog(
-            catalog_path="experiments/catalog.json",
-            item_id=workflow_id,
-            self_href=EXPERIMENT_BASE_CATALOG_SELF_HREF,
-        )
+        if mode in ["all"]:
+            if not getattr(self, "collection_id", None):
+                raise ValueError(
+                    "collection_id is required to generate the experiment record when mode='all' "
+                    "(the experiment links to the output dataset)."
+                )
+            # generate experiment record only if there is an output dataset
+            dataset_link = link_builder.build_link_to_dataset(self.collection_id)
+
+            experiment_record = ExperimentAsOgcRecord(
+                id=workflow_id,
+                title=self.workflow_title,
+                type="Feature",
+                jupyter_notebook_url=jupyter_notebook_url,
+                collection_id=self.collection_id,
+                properties=exp_record_properties,
+                links=links + theme_links + dataset_link,
+            )
+            # Convert to dictionary and cleanup
+            experiment_dict = experiment_record.to_dict()
+            if "jupyter_notebook_url" in experiment_dict:
+                del experiment_dict["jupyter_notebook_url"]
+            if "collection_id" in experiment_dict:
+                del experiment_dict["collection_id"]
+            if "osc:project" in experiment_dict["properties"]:
+                del experiment_dict["properties"]["osc:project"]
+            # add experiment record to file_dict
+            exp_file_path = f"experiments/{workflow_id}/record.json"
+            file_dict[exp_file_path] = experiment_dict
+
+            # Update base catalogs of experiments and workflows with links
+            file_dict["experiments/catalog.json"] = self._update_base_catalog(
+                catalog_path="experiments/catalog.json",
+                item_id=workflow_id,
+                self_href=EXPERIMENT_BASE_CATALOG_SELF_HREF,
+            )
 
-        file_dict["workflows/catalog.json"] = self._update_base_catalog(
-            catalog_path="workflows/catalog.json",
-            item_id=workflow_id,
-            self_href=WORKFLOW_BASE_CATALOG_SELF_HREF,
-        )
+            file_dict["workflows/catalog.json"] = self._update_base_catalog(
+                catalog_path="workflows/catalog.json",
+                item_id=workflow_id,
+                self_href=WORKFLOW_BASE_CATALOG_SELF_HREF,
+            )
         # Write to files if testing
         if write_to_file:
             for file_path, data in file_dict.items():
                 self._write_to_file(file_path, data)
             return {}
         return file_dict
 
-    def publish_all(self, write_to_file: bool = False):
-        """Publish both dataset and workflow/experiment in a single PR."""
-        # Get file dictionaries from both methods
-        dataset_files = self.publish_dataset(write_to_file=write_to_file)
-        workflow_files = self.generate_workflow_experiment_records(
-            write_to_file=write_to_file
-        )
+    def publish(
+        self,
+        write_to_file: bool = False,
+        mode: Literal["all", "dataset", "workflow"] = "all",
+    ) -> dict[str, Any] | str:
+        """
+        Publish both dataset and workflow/experiment in a single PR.
+        Args:
+            write_to_file: If True, write JSON files locally and return the generated dict(s).
+                           If False, open a PR and return the PR URL.
+            mode: Select which artifacts to publish:
+                  - "dataset": only dataset collection & related catalogs
+                  - "workflow": only workflow records
+                  - "all": both
+        Returns:
+            dict[str, Any] when write_to_file=True (the files written),
+            or str when write_to_file=False (the PR URL).
+        """
 
-        # Combine the file dictionaries
-        combined_files = {**dataset_files, **workflow_files}
+        files: dict[str, Any] = {}
 
-        if not write_to_file:
-            # Create branch name, commit message, PR info
-            branch_name = (
-                f"{OSC_BRANCH_NAME}-{self.collection_id}"
-                f"-{datetime.now().strftime('%Y%m%d%H%M%S')}"
-            )
-            commit_message = (
-                f"Add new dataset collection: {self.collection_id} and "
-                f"workflow/experiment: {self.workflow_config.get('workflow_id')}"
-            )
-            pr_title = (
-                f"Add new dataset collection: {self.collection_id} and "
-                f"workflow/experiment: {self.workflow_config.get('workflow_id')}"
-            )
-            pr_body = (
-                f"This PR adds a new dataset collection: {self.collection_id} and "
-                f"its corresponding workflow/experiment to the repository."
-            )
+        if mode in ("dataset", "all"):
+            ds_files = self.publish_dataset(write_to_file=False, mode=mode)
+            files.update(ds_files)
 
-            # Publish all files in one go
-            pr_url = self.gh_publisher.publish_files(
-                branch_name=branch_name,
-                file_dict=combined_files,
-                commit_message=commit_message,
-                pr_title=pr_title,
-                pr_body=pr_body,
+        if mode in ("workflow", "all"):
+            wf_files = self.generate_workflow_experiment_records(write_to_file=False)
+            files.update(wf_files)
+
+        if not files:
+            raise ValueError(
+                "Nothing to publish. Choose mode='dataset', 'workflow', or 'all'."
             )
 
-            logger.info(f"Pull request created: {pr_url}")
+        if write_to_file:
+            for file_path, data in files.items():
+                # file_path might be a Path (from _update_and_add_to_file_dict) – normalize to str
+                out_path = str(file_path)
+                self._write_to_file(out_path, data)
+            return {}  # consistent with existing write_to_file behavior
+
+        # Prepare PR
+        mode_label = {
+            "dataset": f"dataset: {self.collection_id or 'unknown'}",
+            "workflow": f"workflow: {self.workflow_id or 'unknown'}",
+            "all": f"dataset: {self.collection_id or 'unknown'} + workflow/experiment: {self.workflow_id or 'unknown'}",
+        }[mode]
+
+        branch_name = (
+            f"{OSC_BRANCH_NAME}-{(self.collection_id or self.workflow_id or 'osc')}"
+            f"-{datetime.now().strftime('%Y%m%d%H%M%S')}"
+        )
+        commit_message = f"Publish {mode_label}"
+        pr_title = f"Publish {mode_label}"
+        pr_body = f"This PR publishes {mode_label} to the repository."
+
+        pr_url = self.gh_publisher.publish_files(
+            branch_name=branch_name,
+            file_dict=files,
+            commit_message=commit_message,
+            pr_title=pr_title,
+            pr_body=pr_body,
+        )
+        logger.info(f"Pull request created: {pr_url}")
+        return pr_url