MRG: Merge pull request #581 from octue/fix/allow-sets-and-datetimes-as-input-and-output-values

Allow sets and datetimes as service input/output values

Author: cortadocodes

docs/source/inter_service_compatibility.rst

@@ -15,6 +15,7 @@
 from octue.definitions import GOOGLE_COMPUTE_PROVIDERS
 from octue.log_handlers import COLOUR_PALETTE
 from octue.resources.manifest import Manifest
+from octue.utils.decoders import OctueJSONDecoder
 from octue.utils.threads import RepeatingTimer
 
 
@@ -226,7 +227,7 @@ def _pull_and_enqueue_message(self, timeout):
             if not self._child_sdk_version:
                 self._heartbeat_checker.cancel()
 
-        message = json.loads(answer.message.data.decode())
+        message = json.loads(answer.message.data.decode(), cls=OctueJSONDecoder)
 
         message_number = int(message["message_number"])
         self._waiting_messages[message_number] = message
@@ -354,7 +355,7 @@ def _handle_monitor_message(self, message):
         logger.debug("%r received a monitor message.", self.receiving_service)
 
         if self.handle_monitor_message is not None:
-            self.handle_monitor_message(json.loads(message["data"]))
+            self.handle_monitor_message(json.loads(message["data"], cls=OctueJSONDecoder))
 
     def _handle_log_message(self, message):
         """Deserialise the message into a log record and pass it to the local log handlers, adding [<service-name>] to

octue/cloud/pub_sub/message_handler.py

@@ -489,15 +489,15 @@ def _parse_question(self, question):
 
         try:
             # Parse question directly from Pub/Sub or Dataflow.
-            data = json.loads(question.data.decode())
+            data = json.loads(question.data.decode(), cls=OctueJSONDecoder)
 
             # Acknowledge it if it's directly from Pub/Sub
             if hasattr(question, "ack"):
                 question.ack()
 
         except Exception:
             # Parse question from Google Cloud Run.
-            data = json.loads(base64.b64decode(question["data"]).decode("utf-8").strip())
+            data = json.loads(base64.b64decode(question["data"]).decode("utf-8").strip(), cls=OctueJSONDecoder)
 
         # Keep backwards compatibility with questions from Octue services running `octue<0.41.1`.
         if isinstance(data["input_manifest"], str):

octue/cloud/pub_sub/service.py

@@ -70,3 +70,4 @@
 {"parent_sdk_version": "0.44.0", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"8d79f172-5565-4903-8dc2-481b95d37cd8\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmpkyd5lpg9\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "34f946ef-f359-436a-892a-5a91b5991a1c", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.44.0"}}}
 {"parent_sdk_version": "0.45.0", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"ef377a21-8633-4324-ac82-efab087605c8\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmpit8qx3kc\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "5f0a60b6-b770-465b-a7ff-2e7b1854a171", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.45.0"}}}
 {"parent_sdk_version": "0.46.0", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"a289353b-2708-4a69-a72e-54bf67b75e61\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmp56ucgays\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "bb08fdfc-6f67-4942-8fc4-8a4adb7bb93c", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.46.0"}}}
+{"parent_sdk_version": "0.46.1", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"d45acf3c-cc17-4b5c-9d5d-4036341cbcf4\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmp11q86f7l\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "3b880fcc-716b-4bdd-a5f8-044c6163bd20", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.46.1"}}}