MRG: Merge pull request #604 from octue/require-children-to-be-registered-if-registry-specified

Require children to be registered in service registry if provided

Author: cortadocodes

.github/workflows/add-issues-to-octue-board.yaml renamed to .github/workflows/add-issues-to-octue-board.yml

@@ -6,6 +6,6 @@ on:
 
 jobs:
   add-issues-to-octue-board:
-    uses: octue/.github/.github/workflows/reusable-add-issues-to-octue-board.yaml@main
+    uses: octue/.github/.github/workflows/reusable-add-issues-to-octue-board.yml@main
     secrets:
       github-token: ${{ secrets.PROJECT_AUTOMATION_GITHUB_TOKEN_2 }}

docs/source/asking_questions.rst

@@ -63,7 +63,9 @@ sent back to the parent. If you're not sure how long a particular analysis might
 
 Asking multiple questions in parallel
 =====================================
-You can also ask multiple questions to a service in parallel.
+You can also ask multiple questions to a service in parallel. By default, if any of the questions fail, an error is
+raised and no answers are returned. If ``raise_errors=False`` is provided, answers are returned for all successful
+questions while errors are returned unraised for unsuccessful ones.
 
 .. code-block:: python
 

docs/source/inter_service_compatibility.rst

@@ -20,6 +20,7 @@
     convert_service_id_to_pub_sub_form,
     create_sruid,
     get_default_sruid,
+    raise_if_revision_not_registered,
     split_service_id,
     validate_sruid,
 )
@@ -77,19 +78,18 @@ def __init__(self, backend, service_id=None, run_function=None, name=None, servi
         self.run_function = run_function
         self.name = name
 
-        self.service_registries = service_registries or [
-            {
-                "name": "Octue Registry",
-                "endpoint": OCTUE_SERVICE_REGISTRY_ENDPOINT,
-            },
-        ]
+        self.service_registries = service_registries
 
         self._pub_sub_id = convert_service_id_to_pub_sub_form(self.id)
         self._local_sdk_version = importlib.metadata.version("octue")
         self._publisher = None
         self._message_handler = None
 
     def __repr__(self):
+        """Represent the service as a string.
+
+        :return str: the service represented as a string
+        """
         return f"<{type(self).__name__}({self.name or self.id!r})>"
 
     @property
@@ -286,11 +286,19 @@ def ask(
         """
         service_namespace, service_name, service_revision_tag = split_service_id(service_id)
 
-        if not service_revision_tag:
-            service_id = get_default_sruid(
-                namespace=service_namespace,
-                name=service_name,
-                service_registries=self.service_registries,
+        if self.service_registries:
+            if service_revision_tag:
+                raise_if_revision_not_registered(sruid=service_id, service_registries=self.service_registries)
+            else:
+                service_id = get_default_sruid(
+                    namespace=service_namespace,
+                    name=service_name,
+                    service_registries=self.service_registries,
+                )
+
+        elif not service_revision_tag:
+            raise octue.exceptions.InvalidServiceID(
+                f"A service revision tag for {service_id!r} must be provided if service registries aren't being used."
             )
 
         if not allow_local_files:

octue/cloud/pub_sub/service.py

@@ -209,11 +209,12 @@ def convert_service_id_to_pub_sub_form(service_id):
     return service_id
 
 
-def split_service_id(service_id):
+def split_service_id(service_id, require_revision_tag=False):
     """Split an SRUID or service ID into its namespace, name, and, if present, its revision tag. The split parts are
     validated before being returned.
 
     :param str service_id: the SRUID or service ID to split
+    :param bool require_revision_tag: if `True`, require the service ID to include a revision tag (i.e. require the service ID to be an SRUID)
     :raise octue.exceptions.InvalidServiceID: if any of the namespace, name, or revision tag are invalid
     :return tuple(str, str, str|None): the namespace, name, and revision tag
     """
@@ -225,7 +226,7 @@ def split_service_id(service_id):
         name = name_and_revision_tag
         revision_tag = None
 
-    if revision_tag is None:
+    if revision_tag is None and not require_revision_tag:
         validate_service_id(namespace=namespace, name=name)
     else:
         validate_sruid(namespace=namespace, name=name, revision_tag=revision_tag)
@@ -257,3 +258,25 @@ def get_default_sruid(namespace, name, service_registries):
         f"No revisions for the service {service_id!r} were found in any of the specified service registries: "
         f"{service_registries!r}"
     )
+
+
+def raise_if_revision_not_registered(sruid, service_registries):
+    """Raise an error if the service revision isn't registered in the given service registries.
+
+    :param str sruid: the SRUID of the service revision
+    :param iter(dict) service_registries: the registries to look for the service revision in
+    :raise octue.exceptions.ServiceNotFound: if the service revision isn't registered in any of the service registries
+    :return None:
+    """
+    namespace, name, revision_tag = split_service_id(sruid, require_revision_tag=True)
+
+    for registry in service_registries:
+        response = requests.get(f"{registry['endpoint']}/{namespace}/{name}?revision_tag={revision_tag}")
+
+        if response.ok:
+            logger.info("Found service revision %r in %r registry.", sruid, registry["name"])
+            return
+
+    raise octue.exceptions.ServiceNotFound(
+        f"Service revision {sruid!r} was not found in any of the specified service registries: {service_registries!r}"
+    )

octue/cloud/service_id.py

@@ -79,3 +79,4 @@
 {"parent_sdk_version": "0.48.0", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"eb846c0b-3bc4-4eb6-896b-81c4404d3d24\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmpy5_c3ggz\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "a1523bd3-f64d-4de3-af4c-823a2cfea204", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.48.0"}}}
 {"parent_sdk_version": "0.49.0", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"34216d6d-e32e-4c40-8b05-ee9bd71966c9\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmppi1e5t2n\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "3b1224da-8e7f-4b1b-8ef3-c5cf1a5e698c", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.49.0"}}}
 {"parent_sdk_version": "0.49.1", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"34216d6d-e32e-4c40-8b05-ee9bd71966c9\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmppi1e5t2n\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "3b1224da-8e7f-4b1b-8ef3-c5cf1a5e698c", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.49.1"}}}
+{"parent_sdk_version": "0.49.2", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"bdadcd12-a51b-4571-9db3-9d240bd15faa\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmpw3232ii0\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "0b57da2c-38e2-4f8b-9368-fe35125da8fb", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.49.2"}}}