refactor: clean up code formatting and improve readability in notebook and test files

Author: emmanuelmathot

operator-tools/submit_stac_items_notebook.ipynb

@@ -27,9 +27,7 @@
     }
    ],
    "source": [
-    "import getpass\n",
     "import json\n",
-    "import os\n",
     "from pathlib import Path\n",
     "\n",
     "import pandas as pd\n",
@@ -63,7 +61,15 @@
    "execution_count": null,
    "metadata": {},
    "outputs": [],
-   "source": "# STAC API Configuration\nSTAC_API_URL = \"https://stac.core.eopf.eodc.eu/\"\n\n# Webhook Configuration\nWEBHOOK_URL = \"http://localhost:12000/samples\"\n\nprint(\"✅ Configuration loaded\")"
+   "source": [
+    "# STAC API Configuration\n",
+    "STAC_API_URL = \"https://stac.core.eopf.eodc.eu/\"\n",
+    "\n",
+    "# Webhook Configuration\n",
+    "WEBHOOK_URL = \"http://localhost:12000/samples\"\n",
+    "\n",
+    "print(\"✅ Configuration loaded\")"
+   ]
   },
   {
    "cell_type": "markdown",
@@ -426,7 +432,41 @@
    "execution_count": null,
    "metadata": {},
    "outputs": [],
-   "source": "def submit_item_to_pipeline(item_url: str, target_collection: str) -> bool:\n    \"\"\"\n    Submit a single STAC item to the data pipeline via HTTP webhook.\n\n    Args:\n        item_url: The self-link URL of the STAC item\n        target_collection: The target collection for processing\n\n    Returns:\n        True if successful, False otherwise\n    \"\"\"\n    try:\n        # Create payload\n        payload = {\n            \"source_url\": item_url,\n            \"collection\": target_collection,\n            \"action\": \"convert-v1-s2\",  # specify the action to use the V1 S2 trigger\n        }\n\n        # Submit via HTTP webhook endpoint\n        message = json.dumps(payload)\n        response = requests.post(\n            WEBHOOK_URL,\n            data=message,\n            headers={\"Content-Type\": \"application/json\"},\n        )\n        \n        response.raise_for_status()\n        return True\n\n    except Exception as e:\n        print(f\"❌ Error submitting item: {e}\")\n        return False"
+   "source": [
+    "def submit_item_to_pipeline(item_url: str, target_collection: str) -> bool:\n",
+    "    \"\"\"\n",
+    "    Submit a single STAC item to the data pipeline via HTTP webhook.\n",
+    "\n",
+    "    Args:\n",
+    "        item_url: The self-link URL of the STAC item\n",
+    "        target_collection: The target collection for processing\n",
+    "\n",
+    "    Returns:\n",
+    "        True if successful, False otherwise\n",
+    "    \"\"\"\n",
+    "    try:\n",
+    "        # Create payload\n",
+    "        payload = {\n",
+    "            \"source_url\": item_url,\n",
+    "            \"collection\": target_collection,\n",
+    "            \"action\": \"convert-v1-s2\",  # specify the action to use the V1 S2 trigger\n",
+    "        }\n",
+    "\n",
+    "        # Submit via HTTP webhook endpoint\n",
+    "        message = json.dumps(payload)\n",
+    "        response = requests.post(\n",
+    "            WEBHOOK_URL,\n",
+    "            data=message,\n",
+    "            headers={\"Content-Type\": \"application/json\"},\n",
+    "        )\n",
+    "\n",
+    "        response.raise_for_status()\n",
+    "        return True\n",
+    "\n",
+    "    except Exception as e:\n",
+    "        print(f\"❌ Error submitting item: {e}\")\n",
+    "        return False"
+   ]
   },
   {
    "cell_type": "code",
@@ -3177,4 +3217,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 4
-}
+}

stac/S2A_MSIL2A_20250831T103701_N0511_R008_T31TFL_20250831T145420_with_alternate.json

@@ -600,4 +600,4 @@
     "https://stac-extensions.github.io/alternate-assets/v1.2.0/schema.json",
     "https://stac-extensions.github.io/storage/v2.0.0/schema.json"
   ]
-}
+}

tests/test_alternate_extension.py

@@ -4,8 +4,6 @@
 import json
 from pathlib import Path
 
-from pystac import Item
-
 
 def https_to_s3(https_url: str, endpoint: str) -> str | None:
     """Convert https:// URL back to s3:// URL if it matches the S3 endpoint pattern."""
@@ -53,7 +51,7 @@ def add_alternate_s3_to_item(item_dict: dict, s3_endpoint: str) -> dict:
 
     # Add alternate to each asset
     modified_count = 0
-    for asset_key, asset in item_dict.get("assets", {}).items():
+    for _, asset in item_dict.get("assets", {}).items():
         href = asset.get("href", "")
 
         # Skip non-HTTPS URLs or thumbnails
@@ -85,7 +83,9 @@ def add_alternate_s3_to_item(item_dict: dict, s3_endpoint: str) -> dict:
 def main():
     """Test the alternate extension with the sample JSON."""
     # Read the sample JSON
-    sample_file = Path(__file__).parent / "S2A_MSIL2A_20250831T103701_N0511_R008_T31TFL_20250831T145420.json"
+    sample_file = (
+        Path(__file__).parent / "S2A_MSIL2A_20250831T103701_N0511_R008_T31TFL_20250831T145420.json"
+    )
 
     if not sample_file.exists():
         print(f"❌ Sample file not found: {sample_file}")