feat: Support AmiProduct download CLI

Add a `download` CLI command to get configuration
files from listings in AWS Marketplace.

The downloaded configuration files have a same
foramt as the local YAML config file. This can
be useful for users who do not have a the full
configuration but request a changes to an existing
listing.

The LiteralString class is introduced to dump strings
in literal block style(|) in the YAML file.

Author: JessicaJang

awsmp/cli.py

@@ -11,7 +11,7 @@
 import yaml
 from botocore.exceptions import ClientError
 
-from . import _driver, models
+from . import _driver, models, yaml_utils
 from .errors import (
     AccessDeniedException,
     NoProductIdProvidedException,
@@ -443,6 +443,46 @@ def ami_product_update(product_id: str, config: TextIO, allow_price_change: bool
         print(f'https://aws.amazon.com/marketplace/management/requests/{response["ChangeSetId"]}')
 
 
+@public_offer.command("download")
+@click.option("--product-id", required=True, prompt=True, help="Product id of the listing")
+@click.option(
+    "--config", type=click.File("w+"), required=True, prompt=True, help="File path of local configuration file"
+)
+def ami_product_download(product_id: str, config: TextIO) -> None:
+    """
+    Download YAML local configuration from AWS Marketplace live listing.
+    :prarm str product_id: Id of listing
+    :param TextIO config: file path of local configuration file to download
+    :return: None
+    :rtype: None
+    """
+
+    # Get product specific details
+    listing_resp = _driver.get_entity_details(product_id)
+    # Remove version output except latest version
+    # Since local yaml file only store one version information
+    versions = listing_resp["Versions"]
+    latest_version = sorted(versions, key=lambda x: x["CreationDate"])[-1]
+    listing_resp["Versions"] = latest_version
+
+    # Get offer specific details
+    offer_id = _driver.get_public_offer_id(product_id)
+    listing_offer_resp = _driver.get_entity_details(offer_id)
+
+    # filtering required term details only
+    listing_resp["Terms"] = []
+    term_order = {"SupportTerm": 0, "UsageBasedPricingTerm": 1, "ConfigurableUpfrontPricingTerm": 2}
+    if "Terms" in listing_offer_resp:
+        listing_resp["Terms"] = sorted(
+            [term for term in listing_offer_resp.get("Terms", []) if term["Type"] in term_order],
+            key=lambda x: term_order.get(x["Type"], 3),
+        )
+
+    yaml_config = models.EntityModel(**listing_resp).get_yaml_from_entity()
+    yaml_utils.dump_yaml(yaml_config, config)
+    print(f"{config.name} has been successfully written")
+
+
 def _load_configuration(config_path: TextIO, required_fields: List[List[str]]) -> Dict:
     """
     Check if keys exist in config file before creating changeset and return config dict

awsmp/models.py

@@ -29,6 +29,7 @@
 )
 
 from .constants import CATEGORIES
+from .yaml_utils import LiteralString
 
 
 class InstanceTypePricing(BaseModel):
@@ -372,8 +373,8 @@ def to_dict(self) -> dict[str, Any]:
         """
         return {
             "product_title": self.ProductTitle,
-            "short_description": self.ShortDescription,
-            "long_description": self.LongDescription,
+            "short_description": LiteralString(self.ShortDescription),
+            "long_description": LiteralString(self.LongDescription),
             "sku": self.Sku,
             "highlights": self.Highlights,
             "search_keywords": self.SearchKeywords,
@@ -439,7 +440,7 @@ def to_dict(self) -> dict[str, Any]:
         :return: Dictionary of support information
         :rtype: dict[str, Any]
         """
-        return {"support_description": self.Description, "support_resources": self.Resources}
+        return {"support_description": LiteralString(self.Description), "support_resources": self.Resources}
 
 
 class RegionAvailabilityModel(BaseModel):
@@ -620,7 +621,7 @@ def to_dict(self) -> dict[str, Any]:
         :return: Dictionary of version information
         :rtype: dict[str, Any]
         """
-        return {**{"usage_instructions": self.Instructions["Usage"]}, **self.Recommendations.to_dict()}
+        return {**{"usage_instructions": LiteralString(self.Instructions["Usage"])}, **self.Recommendations.to_dict()}
 
 
 class VersionModel(BaseModel):
@@ -717,17 +718,17 @@ def to_dict(self) -> dict[str, Any]:
         :rtype: dcit[str, Any]
         """
         description_configs = {
-            **self.Description._to_yaml(),
-            **self.PromotionalResources._to_yaml(),
-            **self.SupportInformation._to_yaml(),
+            **self.Description.to_dict(),
+            **self.PromotionalResources.to_dict(),
+            **self.SupportInformation.to_dict(),
         }
         config = {
             "product": {
                 "description": description_configs,
-                "region": self.RegionAvailability._to_yaml(),
-                "version": self.Versions._to_yaml()
+                "region": self.RegionAvailability.to_dict(),
+                "version": self.Versions.to_dict(),
             },
-            "offer": self._convert_terms_to_yaml(),
+            "offer": self._convert_terms_to_dict(),
         }
 
         return config
@@ -742,7 +743,7 @@ def _convert_terms_to_dict(self) -> dict[str, Any]:
 
         for term in self.Terms:
             if term.Type == "SupportTerm":
-                yaml_config["refund_policy"] = term.RefundPolicy
+                yaml_config["refund_policy"] = LiteralString(term.RefundPolicy)
             else:
                 # Pricing term
                 if term.Type == "UsageBasedPricingTerm":

awsmp/yaml_utils.py

@@ -0,0 +1,23 @@
+import yaml
+from yaml.dumper import SafeDumper
+
+
+class LiteralString(str):
+    pass
+
+
+def literal_str_representer(dumper, data):
+    text = data if data.endswith("\n") else data + "\n"
+    return dumper.represent_scalar("tag:yaml.org,2002:str", text, style="|")
+
+
+class IndentListDumper(SafeDumper):
+    def increase_indent(self, flow=False, indentless=False):
+        return super().increase_indent(flow, False)
+
+
+yaml.add_representer(LiteralString, literal_str_representer, Dumper=IndentListDumper)
+
+
+def dump_yaml(data, config):
+    yaml.dump(data, config, Dumper=IndentListDumper, default_flow_style=False, indent=2, sort_keys=False)

docs/public-offer/index.rst

@@ -241,5 +241,17 @@ To update AMI product listing with multiple requests for product details (Descri
       --config listing_configuration.yaml
 
 
+Download AMI product listing details
+------------------------------------
+
+To download AMI product listing information as a YAML file, run the command below, passing the product ID and product configuration file:
+
+.. code-block:: sh
+
+   awsmp public-offer download \
+      --product-id prod-fwu3xsqup23cs
+      --config listing.yaml
+
+
 .. _`AWS marketplace management portal`: https://aws.amazon.com/marketplace/management/
 .. _`AWS Marketplace update legal resources API reference`: https://docs.aws.amazon.com/marketplace/latest/APIReference/work-with-private-offers.html#update-legal-terms

tests/test_cli.py

@@ -1,7 +1,8 @@
 import json
 import tempfile
+from io import StringIO
 from typing import Any, List
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, mock_open, patch
 
 import pytest
 import yaml
@@ -894,3 +895,90 @@ def test_public_offer_product_update_instance_type_pricing_change_exception(
         ["--product-id", "some-prod-id", "--config", "./tests/test_config.yaml", "--no-allow-price-change"],
     )
     assert res.exit_code == 1 and res.exc_info is not None and "Restricted listings" in res.exc_info[1].args[0]
+
+
+@pytest.mark.parametrize(
+    "key1, key2, value",
+    [
+        ("description", "product_title", "test"),
+        ("description", "categories", ["Migration"]),
+        ("description", "long_description", "test_long_description\n"),
+        ("version", "version_title", "Test Ubuntu AMI"),
+        ("version", "usage_instructions", "test_usage_instruction\n"),
+        ("version", "ami_id", "ami-12345678910"),
+        ("description", "support_description", "test_support_description\n"),
+        ("region", "commercial_regions", ["us-east-1", "us-east-2"]),
+        ("region", "future_region_support", True),
+    ],
+)
+@patch("awsmp._driver.get_entity_details")
+@patch("awsmp._driver.get_public_offer_id")
+def test_public_offer_product_download_product(mock_get_public_offer_id, mock_get_entity_details, key1, key2, value):
+    with open("./tests/test_config.json") as f:
+        mock_prod_resp = json.load(f)
+        mock_prod_resp.pop("Terms")
+
+    mock_prod_resp["Versions"]["CreationDate"] = "2025-01-01"
+    mock_prod_resp["Versions"] = [mock_prod_resp["Versions"]]
+
+    with open("./tests/test_config.json") as f:
+        mock_offer_resp = {"Terms": json.load(f)["Terms"]}
+
+    mock_get_entity_details.side_effect = [mock_prod_resp, mock_offer_resp]
+    mock_get_public_offer_id.return_value = "test-offer-id"
+
+    runner = CliRunner()
+    config_file = tempfile.NamedTemporaryFile()
+    res = runner.invoke(
+        cli.ami_product_download,
+        ["--product-id", "some-prod-id", "--config", config_file.name],
+    )
+
+    with open(config_file.name, "r") as f:
+        config = yaml.safe_load(f)
+
+    assert config["product"][key1][key2] == value
+
+
+@pytest.mark.parametrize(
+    "key1, key2, value",
+    [
+        ("offer", "refund_policy", "test_refund_policy_term\n"),
+        (
+            "offer",
+            "instance_types",
+            [
+                {"name": "a1.large", "hourly": "0.004", "yearly": "24.528"},
+                {"name": "a1.xlarge", "hourly": "0.007", "yearly": "49.056"},
+            ],
+        ),
+        ("offer", "eula_document", [{"type": ""}]),
+    ],
+)
+@patch("awsmp._driver.get_entity_details")
+@patch("awsmp._driver.get_public_offer_id")
+def test_public_offer_product_download_offer(mock_get_public_offer_id, mock_get_entity_details, key1, key2, value):
+    with open("./tests/test_config.json") as f:
+        mock_prod_resp = json.load(f)
+        mock_prod_resp.pop("Terms")
+
+    mock_prod_resp["Versions"]["CreationDate"] = "2025-01-01"
+    mock_prod_resp["Versions"] = [mock_prod_resp["Versions"]]
+
+    with open("./tests/test_config.json") as f:
+        mock_offer_resp = {"Terms": json.load(f)["Terms"]}
+
+    mock_get_entity_details.side_effect = [mock_prod_resp, mock_offer_resp]
+    mock_get_public_offer_id.return_value = "test-offer-id"
+
+    runner = CliRunner()
+    config_file = tempfile.NamedTemporaryFile()
+    res = runner.invoke(
+        cli.ami_product_download,
+        ["--product-id", "some-prod-id", "--config", config_file.name],
+    )
+
+    with open(config_file.name, "r") as f:
+        config = yaml.safe_load(f)
+
+    assert config[key1][key2] == value