Use function from datacube

The parse_doc_stream function
has been integrated into datacube
with the 1.9.9 release.

Author: pjonsson

apps/dc_tools/odc/apps/dc_tools/_docs.py

@@ -1,12 +1,11 @@
 """These should probably be in datacube library."""
 
-import json
 import sys
 from typing import Sequence, Union
 from uuid import UUID, uuid5
 
 from datacube.index.hl import Doc2Dataset
-from datacube.utils.documents import parse_yaml
+from datacube.utils.documents import parse_doc_stream
 
 # Some random UUID to be ODC namespace
 ODC_NS = UUID("6f34c6f4-13d6-43c0-8e4e-42b6c13203af")
@@ -73,37 +72,6 @@ def from_metadata_stream(metadata_stream, index, **kwargs):
                 yield (None, f"Error: {uri}, {err}")
 
 
-def parse_doc_stream(doc_stream, on_error=None, transform=None):
-    """
-    Replace doc bytes/strings with parsed dicts.
-
-       Stream[(uri, bytes)] -> Stream[(uri, dict)]
-
-
-    :param doc_stream: sequence of (uri, doc: bytes|string)
-    :param on_error: Callback uri, doc -> None
-    :param transform: dict -> dict if supplied also apply further transform on parsed document
-
-    On output doc is replaced with python dict parsed from yaml, or with None
-    if parsing/transform error occurred.
-    """
-    for uri, doc in doc_stream:
-        try:
-            if uri.endswith(".json"):
-                metadata = json.loads(doc)
-            else:
-                metadata = parse_yaml(doc)
-
-            if transform is not None:
-                metadata = transform(metadata)
-        except Exception:  # pylint: disable=broad-except
-            if on_error is not None:
-                on_error(uri, doc)
-            metadata = None
-
-        yield uri, metadata
-
-
 def from_yaml_doc_stream(doc_stream, index, logger=None, transform=None, **kwargs):
     """
     Stream of yaml documents to a stream of Dataset results.

apps/dc_tools/odc/apps/dc_tools/s3_to_dc.py

@@ -13,8 +13,8 @@
 from datacube import Datacube
 from datacube.index.hl import Doc2Dataset
 from datacube.ui.click import environment_option, pass_config
+from datacube.utils.documents import parse_doc_stream
 from odc.aio import S3Fetcher, s3_find_glob
-from odc.apps.dc_tools._docs import parse_doc_stream
 from odc.apps.dc_tools.utils import (
     IndexingException,
     SkippedException,

apps/dc_tools/pyproject.toml

@@ -7,7 +7,7 @@ authors = [
 ]
 dependencies = [
     "click",
-    "datacube>=1.9.6",
+    "datacube>=1.9.9",
     "datadog",
     "eodatasets3>=1.9",
     "fsspec",

tests/test-env.yml

@@ -10,7 +10,7 @@ dependencies:
   - python=3.12
 
   # Datacube
-  - datacube>=1.9.0
+  - datacube>=1.9.9
   - sqlalchemy>=2.0
 
   # odc.ui