feat: add multiple STAC licensors TDE-643 (#355)

scripts/cli/cli_helper.py

@@ -2,7 +2,7 @@
 import json
 from datetime import datetime
 from os import environ
-from typing import List
+from typing import List, Optional
 
 from dateutil import parser, tz
 from linz_logger import get_log
@@ -70,3 +70,19 @@ def valid_date(s: str) -> datetime:
     except ValueError as e:
         msg = f"not a valid date: {s}"
         raise argparse.ArgumentTypeError(msg) from e
+
+
+def parse_list(list_s: str, separator: Optional[str] = ";") -> List[str]:
+    """Transform a string representing a list to a list of strings
+    example: "foo; bar; foo bar" -> ["foo", "bar", "foo bar"]
+
+    Args:
+        list_s: string representing a list to transform
+        separator: separator of the list
+
+    Returns:
+        a list of strings
+    """
+    if list_s:
+        return [s.strip() for s in list_s.split(separator) if s != ""]
+    return []

scripts/tests/cli_helper_test.py → scripts/cli/tests/cli_helper_test.py

@@ -1,4 +1,4 @@
-from scripts.cli.cli_helper import format_source
+from scripts.cli.cli_helper import format_source, parse_list
 
 
 def test_format_source_from_basemaps_cli_file() -> None:
@@ -39,3 +39,14 @@ def test_format_source_json_loading_error() -> None:
     assert isinstance(file_list, list)
     assert len(file_list) == 1
     assert file_list == ["[s3://test/image_one.tiff"]
+
+
+def test_parse_list() -> None:
+    str_list = "Auckland Council; Toitū Te Whenua Land Information New Zealand;Nelson Council;"
+    list_parsed = parse_list(str_list)
+    assert list_parsed == ["Auckland Council", "Toitū Te Whenua Land Information New Zealand", "Nelson Council"]
+
+
+def test_parse_list_empty() -> None:
+    list_parsed = parse_list("")
+    assert list_parsed == []

scripts/collection_from_items.py

@@ -6,6 +6,7 @@
 from boto3 import client
 from linz_logger import get_log
 
+from scripts.cli.cli_helper import parse_list
 from scripts.files.fs_s3 import bucket_name_from_path, get_object_parallel_multithreading, list_json_in_uri
 from scripts.logging.time_helper import time_in_ms
 from scripts.stac.imagery.collection import ImageryCollection
@@ -19,17 +20,22 @@ def main() -> None:
     parser.add_argument("--title", dest="title", help="collection title", required=True)
     parser.add_argument("--description", dest="description", help="collection description", required=True)
     parser.add_argument("--producer", dest="producer", help="imagery producer", required=True)
-    parser.add_argument("--licensor", dest="licensor", help="imagery licensor", required=True)
+    parser.add_argument("--licensor", dest="licensor", help="imagery licensor")
+    parser.add_argument("--licensor-list", dest="licensor_list", help="imagery licensor list")
     parser.add_argument(
         "--concurrency", dest="concurrency", help="The number of files to limit concurrent reads", required=True, type=int
     )
 
     arguments = parser.parse_args()
     uri = arguments.uri
-    providers: List[Provider] = [
-        {"name": arguments.producer, "roles": [ProviderRole.PRODUCER]},
-        {"name": arguments.licensor, "roles": [ProviderRole.LICENSOR]},
-    ]
+    licensors: List[str] = parse_list(arguments.licensor_list)
+    if len(licensors) <= 1:
+        licensors = [arguments.licensor]
+
+    providers: List[Provider] = []
+    for licensor_name in licensors:
+        providers.append({"name": licensor_name, "roles": [ProviderRole.LICENSOR]})
+    providers.append({"name": arguments.producer, "roles": [ProviderRole.PRODUCER]})
 
     collection = ImageryCollection(
         title=arguments.title, description=arguments.description, collection_id=arguments.collection_id, providers=providers