MRG: Merge pull request #143 from octue/release/0.1.16

Release/0.1.16

Author: cortadocodes

docs/source/deploying_services.rst

@@ -15,7 +15,7 @@ Automated deployment with Octue means:
 
 All you need to enable automated deployments are the following files in your repository root:
 
-* A ``requirements.txt`` file that includes ``octue>=0.1.15`` and the rest of your service's dependencies
+* A ``requirements.txt`` file that includes ``octue>=0.1.16`` and the rest of your service's dependencies
 * A ``twine.json`` file
 * A ``deployment_configuration.json`` file (optional)
 

octue/cli.py

@@ -224,7 +224,7 @@ def start(app_dir, data_dir, config_dir, service_id, twine, timeout, delete_topi
     backend_configuration_values = runner.configuration["configuration_values"]["backend"]
     backend = service_backends.get_backend(backend_configuration_values.pop("name"))(**backend_configuration_values)
 
-    service = Service(id=service_id, backend=backend, run_function=runner.run)
+    service = Service(service_id=service_id, backend=backend, run_function=runner.run)
     service.serve(timeout=timeout, delete_topic_and_subscription_on_exit=delete_topic_and_subscription_on_exit)
 
 

octue/deployment/__init__.py octue/cloud/deployment/__init__.py

@@ -28,4 +28,4 @@ ARG _GUNICORN_THREADS=8
 ENV _GUNICORN_THREADS=$_GUNICORN_THREADS
 
 # Timeout is set to 0 to disable the timeouts of the workers to allow Cloud Run to handle instance scaling.
-CMD exec gunicorn --bind :$PORT --workers $_GUNICORN_WORKERS --threads $_GUNICORN_THREADS --timeout 0 octue.deployment.google.cloud_run:app
+CMD exec gunicorn --bind :$PORT --workers $_GUNICORN_WORKERS --threads $_GUNICORN_THREADS --timeout 0 octue.cloud.deployment.google.cloud_run:app

octue/deployment/google/Dockerfile octue/cloud/deployment/google/Dockerfile

@@ -5,6 +5,7 @@
 from flask import Flask, request
 
 from octue.cloud.pub_sub.service import Service
+from octue.exceptions import MissingServiceID
 from octue.logging_handlers import apply_log_handler
 from octue.resources.service_backends import GCPPubSubBackend
 from octue.runner import Runner
@@ -13,6 +14,10 @@
 logger = logging.getLogger(__name__)
 apply_log_handler(logger, log_level=logging.INFO)
 
+
+DEPLOYMENT_CONFIGURATION_PATH = "deployment_configuration.json"
+
+
 app = Flask(__name__)
 
 
@@ -64,13 +69,18 @@ def answer_question(project_name, data, question_uuid):
     :param str question_uuid:
     :return None:
     """
-    deployment_configuration_path = "deployment_configuration.json"
+    service_id = os.environ.get("SERVICE_ID")
+
+    if not service_id:
+        raise MissingServiceID(
+            "The ID for the deployed service is missing - ensure SERVICE_ID is available as an environment variable."
+        )
 
     try:
-        with open(deployment_configuration_path) as f:
+        with open(DEPLOYMENT_CONFIGURATION_PATH) as f:
             deployment_configuration = json.load(f)
 
-        logger.info("Deployment configuration loaded from %r.", os.path.abspath(deployment_configuration_path))
+        logger.info("Deployment configuration loaded from %r.", os.path.abspath(DEPLOYMENT_CONFIGURATION_PATH))
 
     except FileNotFoundError:
         deployment_configuration = {}
@@ -91,7 +101,7 @@ def answer_question(project_name, data, question_uuid):
     )
 
     service = Service(
-        id=os.environ["SERVICE_ID"],
+        service_id=service_id,
         backend=GCPPubSubBackend(project_name=project_name, credentials_environment_variable=None),
         run_function=runner.run,
     )

octue/deployment/google/__init__.py octue/cloud/deployment/google/__init__.py

@@ -13,6 +13,7 @@
 from octue.exceptions import FileLocationError
 from octue.mixins import CoolNameable
 from octue.resources.manifest import Manifest
+from octue.utils.encoders import OctueJSONEncoder
 
 
 logger = logging.getLogger(__name__)
@@ -58,10 +59,21 @@ class Service(CoolNameable):
 
     Services communicate entirely via Google Pub/Sub and can ask and/or respond to questions from any other Service that
     has a corresponding topic on Google Pub/Sub.
+
+    :param octue.resources.service_backends.ServiceBackend backend:
+    :param str|None service_id:
+    :param callable|None run_function:
+    :return None:
     """
 
-    def __init__(self, backend, id=None, run_function=None):
-        self.id = id or str(uuid.uuid4())
+    def __init__(self, backend, service_id=None, run_function=None):
+        if service_id is None:
+            self.id = str(uuid.uuid4())
+        elif not service_id:
+            raise ValueError(f"service_id should be None or a non-falsey value; received {service_id!r} instead.")
+        else:
+            self.id = service_id
+
         self.backend = backend
         self.run_function = run_function
 
@@ -121,7 +133,8 @@ def answer(self, data, question_uuid, timeout=30):
         self.publisher.publish(
             topic=topic.path,
             data=json.dumps(
-                {"output_values": analysis.output_values, "output_manifest": serialised_output_manifest}
+                {"output_values": analysis.output_values, "output_manifest": serialised_output_manifest},
+                cls=OctueJSONEncoder,
             ).encode(),
             retry=create_custom_retry(timeout),
         )

octue/deployment/google/cloud_run.py octue/cloud/deployment/google/cloud_run.py

@@ -78,3 +78,7 @@ class AttributeConflict(OctueSDKException):
     """Raise if, when trying to set an attribute whose current value has a significantly higher confidence than the new
     value, the new value conflicts with the current value.
     """
+
+
+class MissingServiceID(OctueSDKException):
+    """Raise when a specific ID for a service is expected to be provided, but is missing or None."""

octue/cloud/pub_sub/service.py

@@ -5,6 +5,8 @@
 the "oneOf" field of the "backend" key of the children schema in `Twined`, which is located at
 `twined/schema/children_schema.json`.
 """
+from abc import ABC
+
 from octue import exceptions
 
 
@@ -17,15 +19,34 @@ def get_backend(backend_name):
     return AVAILABLE_BACKENDS[backend_name]
 
 
-class GCPPubSubBackend:
-    """ A dataclass containing the details needed to use Google Cloud Platform Pub/Sub as a Service backend. """
+class ServiceBackend(ABC):
+    """A dataclass specifying the backend for an Octue Service, including any credentials and other information it
+    needs.
+
+    :param str|None credentials_environment_variable:
+    :return None:
+    """
+
+    def __init__(self, credentials_environment_variable):
+        self.credentials_environment_variable = credentials_environment_variable
+
+
+class GCPPubSubBackend(ServiceBackend):
+    """A dataclass containing the details needed to use Google Cloud Platform Pub/Sub as a Service backend.
+
+    :param str project_name:
+    :param str|None credentials_environment_variable:
+    :return None:
+    """
 
     def __init__(self, project_name, credentials_environment_variable="GOOGLE_APPLICATION_CREDENTIALS"):
         self.project_name = project_name
-        self.credentials_environment_variable = credentials_environment_variable
+        super().__init__(credentials_environment_variable)
 
     def __repr__(self):
         return f"<{type(self).__name__}(project_name={self.project_name!r})>"
 
 
-AVAILABLE_BACKENDS = {key: value for key, value in locals().items() if key.endswith("Backend")}
+AVAILABLE_BACKENDS = {
+    key: value for key, value in locals().items() if key.endswith("Backend") and key != "ServiceBackend"
+}

octue/exceptions.py

@@ -23,7 +23,7 @@ class Runner:
     The Runner class provides a set of configuration parameters for use by your application, together with a range of
     methods for managing input and output file parsing as well as controlling logging.
 
-    :param str twine: string path to the twine file, or a string containing valid twine json
+    :param str|twined.Twine twine: path to the twine file, a string containing valid twine json, or a Twine instance
     :param str|dict paths: If a string, contains a single path to an existing data directory where
         (if not already present), subdirectories 'configuration', 'input', 'tmp', 'log' and 'output' will be created. If a
         dict, it should contain all of those keys, with each of their values being a path to a directory (which will be
@@ -57,15 +57,13 @@ def __init__(
         self.output_manifest_path = output_manifest_path
         self.children = children
         self.skip_checks = skip_checks
-
-        # Store the log level (same log level used for all analyses)
         self._log_level = log_level
         self.handler = handler
 
         if show_twined_logs:
             apply_log_handler(logger=package_logger, handler=self.handler)
 
-        # Ensure the twine is present and instantiate it
+        # Ensure the twine is present and instantiate it.
         if isinstance(twine, Twine):
             self.twine = twine
         else:
@@ -88,15 +86,11 @@ def __init__(
             configuration_manifest,
         )
 
-        # Store the log level (same log level used for all analyses)
-        self._log_level = log_level
-        self.handler = handler
-
         if show_twined_logs:
             apply_log_handler(logger=package_logger, handler=self.handler, log_level=self._log_level)
             package_logger.info(
                 "Showing package logs as well as analysis logs (the package logs are recommended for software "
-                "engineers but may still be useful to app development by scientists."
+                "engineers but may still be useful to app development by scientists)."
             )
 
         self._project_name = project_name

octue/resources/service_backends.py

@@ -1 +1 @@
-octue==0.1.15
+octue==0.1.16

octue/runner.py

@@ -1 +1 @@
-octue==0.1.15
+octue==0.1.16

octue/templates/template-child-services/elevation_service/requirements.txt

@@ -1 +1 @@
-octue==0.1.15
+octue==0.1.16

octue/templates/template-child-services/parent_service/requirements.txt

@@ -1,4 +1,4 @@
-octue==0.1.15
+octue==0.1.16
 
 
 # ----------- Some common libraries  -----------------------------------------------------------------------------------

octue/templates/template-child-services/wind_speed_service/requirements.txt

@@ -1,4 +1,4 @@
-octue==0.1.15
+octue==0.1.16
 
 
 # ----------- Some common libraries  -----------------------------------------------------------------------------------

octue/templates/template-python-fractal/requirements.txt

@@ -17,7 +17,7 @@
 
 setup(
     name="octue",
-    version="0.1.15",  # Ensure all requirements files containing octue are updated, too (e.g. docs build).
+    version="0.1.16",  # Ensure all requirements files containing octue are updated, too (e.g. docs build).
     py_modules=["cli"],
     install_requires=[
         "click>=7.1.2",
@@ -28,7 +28,7 @@
         "google-cloud-storage>=1.35.1",
         "google-crc32c>=1.1.2",
         "gunicorn",
-        "twined>=0.0.18",
+        "twined>=0.0.19",
     ],
     url="https://www.github.com/octue/octue-sdk-python",
     license="MIT",

octue/templates/template-using-manifests/requirements.txt

@@ -1,10 +1,12 @@
 import base64
 import json
+import os
 import uuid
 from unittest import TestCase, mock
 
+from octue.cloud.deployment.google import cloud_run
 from octue.cloud.pub_sub.service import Service
-from octue.deployment.google import cloud_run
+from octue.exceptions import MissingServiceID
 from octue.resources.service_backends import GCPPubSubBackend
 from tests import TEST_PROJECT_NAME
 
@@ -31,7 +33,7 @@ def test_post_to_index_with_invalid_payload_results_in_400_error(self):
     def test_post_to_index_with_valid_payload(self):
         """Test that the Flask endpoint returns a 204 (ok, no content) response to a valid payload."""
         with cloud_run.app.test_client() as client:
-            with mock.patch("octue.deployment.google.cloud_run.answer_question"):
+            with mock.patch("octue.cloud.deployment.google.cloud_run.answer_question"):
 
                 response = client.post(
                     "/",
@@ -48,6 +50,20 @@ def test_post_to_index_with_valid_payload(self):
 
                 self.assertEqual(response.status_code, 204)
 
+    def test_error_is_raised_if_service_id_environment_variable_is_missing_or_empty(self):
+        """Test that a MissingServiceID error is raised if the `SERVICE_ID` environment variable is missing or empty."""
+        with mock.patch.dict(os.environ, clear=True):
+            with self.assertRaises(MissingServiceID):
+                cloud_run.answer_question(
+                    project_name="a-project-name", data={}, question_uuid="8c859f87-b594-4297-883f-cd1c7718ef29"
+                )
+
+        with mock.patch.dict(os.environ, {"SERVICE_ID": ""}):
+            with self.assertRaises(MissingServiceID):
+                cloud_run.answer_question(
+                    project_name="a-project-name", data={}, question_uuid="8c859f87-b594-4297-883f-cd1c7718ef29"
+                )
+
     def test_cloud_run_integration(self):
         """Test that the Google Cloud Run integration works, providing a service that can be asked questions and send
         responses.

setup.py

@@ -181,14 +181,14 @@ class MockService(Service):
     """A mock Google Pub/Sub Service that can send and receive messages synchronously to other instances.
 
     :param octue.resources.service_backends.GCPPubSubBackEnd backend:
-    :param str id:
+    :param str service_id:
     :param callable run_function:
     :param dict(str, MockService)|None children:
     :return None:
     """
 
-    def __init__(self, backend, id=None, run_function=None, children=None):
-        super().__init__(backend, id, run_function)
+    def __init__(self, backend, service_id=None, run_function=None, children=None):
+        super().__init__(backend, service_id, run_function)
         self.children = children or {}
         self.publisher = MockPublisher()
         self.subscriber = MockSubscriber()

tests/deployment/__init__.py tests/cloud/deployment/__init__.py

@@ -1,3 +1,4 @@
+import concurrent.futures
 import uuid
 from unittest.mock import patch
 
@@ -7,7 +8,7 @@
 from octue.resources.service_backends import GCPPubSubBackend
 from tests import TEST_PROJECT_NAME
 from tests.base import BaseTestCase
-from tests.cloud.pub_sub.mocks import MockService, MockSubscription, MockTopic
+from tests.cloud.pub_sub.mocks import MockPullResponse, MockService, MockSubscription, MockTopic
 
 
 class MockAnalysis:
@@ -61,13 +62,34 @@ def test_repr(self):
         asking_service = Service(backend=self.BACKEND)
         self.assertEqual(repr(asking_service), f"<Service({asking_service.name!r})>")
 
+    def test_service_id_cannot_be_non_none_empty_vaue(self):
+        """Ensure that a ValueError is raised if a non-None empty value is provided as the service_id."""
+        with self.assertRaises(ValueError):
+            Service(backend=self.BACKEND, service_id="")
+
+        with self.assertRaises(ValueError):
+            Service(backend=self.BACKEND, service_id=[])
+
+        with self.assertRaises(ValueError):
+            Service(backend=self.BACKEND, service_id={})
+
     def test_ask_on_non_existent_service_results_in_error(self):
         """Test that trying to ask a question to a non-existent service (i.e. one without a topic in Google Pub/Sub)
         results in an error."""
         with patch("octue.cloud.pub_sub.service.Topic", new=MockTopic):
             with self.assertRaises(exceptions.ServiceNotFound):
                 MockService(backend=self.BACKEND).ask(service_id="hello", input_values=[1, 2, 3, 4])
 
+    def test_timeout_error_raised_if_no_messages_received_when_waiting(self):
+        """Test that a concurrent.futures.TimeoutError is raised if no messages are received while waiting."""
+        service = Service(backend=self.BACKEND)
+        mock_topic = MockTopic(name="world", namespace="hello", service=service)
+        mock_subscription = MockSubscription(name="world", topic=mock_topic, namespace="hello", service=service)
+
+        with patch("octue.cloud.pub_sub.service.pubsub_v1.SubscriberClient.pull", return_value=MockPullResponse()):
+            with self.assertRaises(concurrent.futures.TimeoutError):
+                service.wait_for_answer(subscription=mock_subscription)
+
     def test_ask(self):
         """ Test that a service can ask a question to another service that is serving and receive an answer. """
         responding_service = self.make_new_server(self.BACKEND, run_function_returnee=MockAnalysis(), use_mock=True)