add example that uses .items() and tests for iterating after already accessing an item

Author: Neradoc

examples/json_stream_advanced.py

@@ -0,0 +1,63 @@
+# SPDX-FileCopyrightText: Copyright (c) 2023 Scott Shawcroft for Adafruit Industries
+#
+# SPDX-License-Identifier: Unlicense
+
+import sys
+import time
+
+import adafruit_json_stream as json_stream
+
+# import json_stream
+
+
+class FakeResponse:
+    def __init__(self, file):
+        self.file = file
+
+    def iter_content(self, chunk_size):
+        while True:
+            yield self.file.read(chunk_size)
+
+
+f = open(sys.argv[1], "rb")  # pylint: disable=consider-using-with
+obj = json_stream.load(FakeResponse(f).iter_content(32))
+
+
+def find_keys(obj, keys):
+    """If we don't know the order in which the keys are,
+    go through all of them and pick the ones we want"""
+    out = dict()
+    # iterate on the items of an object
+    for key, value in obj.items():
+        if key in keys:
+            # if it's a sub object, get it all
+            if isinstance(value, json_stream.Transient):
+                value = value.as_object()
+            out[key] = value
+    return out
+
+def time_to_date(stamp):
+    tt = time.localtime(stamp)
+    month = ["January", "February", "March", "April", "May", "June", "July", "August", "September", "October", "November", "December"][tt.tm_mon]
+    return f"{tt.tm_mday:2d}th of {month}"
+
+def ftoc(temp):
+    return (temp - 32) * 5 / 9
+
+currently = obj["currently"]
+print("Currently:")
+print("    ", time_to_date(currently["time"]))
+print("    ", currently["icon"])
+
+# iterate on the content of a list
+for i, day in enumerate(obj["daily"]["data"]):
+    day_items = find_keys(day, ("time", "summary", "temperatureHigh"))
+    date = time_to_date(day_items["time"])
+    print(
+        f'On {date}: {day_items["summary"]},',
+        f'Max: {int(day_items["temperatureHigh"])}F',
+        f'({int(ftoc(day_items["temperatureHigh"]))}C)'
+    )
+
+    if i > 4:
+        break

tests/test_json_stream.py

@@ -688,7 +688,7 @@ def test_as_object_grabbing_multiple_subscriptable_levels_again_after_passed_rai
 
 
 def test_iterating_keys(dict_with_keys):
-    """Iterate through keys of a simple object"""
+    """Iterate through keys of a simple object."""
 
     bytes_io_chunk = BytesChunkIO(dict_with_keys.encode())
     stream = adafruit_json_stream.load(bytes_io_chunk)
@@ -697,10 +697,29 @@ def test_iterating_keys(dict_with_keys):
 
 
 def test_iterating_items(dict_with_keys):
-    """Iterate through items of a simple object"""
+    """Iterate through items of a simple object."""
 
     bytes_io_chunk = BytesChunkIO(dict_with_keys.encode())
     stream = adafruit_json_stream.load(bytes_io_chunk)
     output = list(stream.items())
     assert output == [("field_1", 1), ("field_2", 2), ("field_3", 3)]
 
+
+def test_iterating_keys_after_get(dict_with_keys):
+    """Iterate through keys of a simple object after an item has already been read."""
+
+    bytes_io_chunk = BytesChunkIO(dict_with_keys.encode())
+    stream = adafruit_json_stream.load(bytes_io_chunk)
+    assert stream["field_1"] == 1
+    output = list(stream)
+    assert output == ["field_2", "field_3"]
+
+
+def test_iterating_items_after_get(dict_with_keys):
+    """Iterate through items of a simple object after an item has already been read."""
+
+    bytes_io_chunk = BytesChunkIO(dict_with_keys.encode())
+    stream = adafruit_json_stream.load(bytes_io_chunk)
+    assert stream["field_1"] == 1
+    output = list(stream.items())
+    assert output == [("field_2", 2), ("field_3", 3)]