MRG: Merge branch 'main' into publish-answers-to-question-topic-2

.github/workflows/add-issues-to-octue-board.yaml → .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 }}

.github/workflows/python-ci.yml

@@ -81,11 +81,15 @@ jobs:
       - name: Checkout Repository
         uses: actions/checkout@v3
 
+      - name: Install Poetry
+        uses: snok/[email protected]
+
+      - name: Build a binary wheel and a source tarball
+        run: poetry build
+
       - name: Test package is publishable with PyPI test server
-        uses: JRubics/poetry-[email protected]
+        uses: pypa/gh-action-pypi-publish@v1.8.10
         with:
-          python_version: "3.9"
-          pypi_token: ${{ secrets.TEST_PYPI_TOKEN }}
-          repository_name: "testpypi"
-          repository_url: "https://test.pypi.org/legacy/"
-          ignore_dev_requirements: "yes"
+          repository-url: https://test.pypi.org/legacy/
+          skip-existing: true
+          verbose: true

.github/workflows/release.yml

@@ -83,18 +83,24 @@ jobs:
           prerelease: false
 
   publish:
-    runs-on: ubuntu-latest
     needs: release
+    runs-on: ubuntu-latest
+    permissions:
+      id-token: write
+      contents: read
+
     steps:
       - name: Checkout Repository
         uses: actions/checkout@v3
 
-      - name: Build and publish latest package to PyPI
-        uses: JRubics/[email protected]
-        with:
-          python_version: "3.9"
-          pypi_token: ${{ secrets.PYPI_TOKEN }}
-          ignore_dev_requirements: "yes"
+      - name: Install Poetry
+        uses: snok/[email protected]
+
+      - name: Build a binary wheel and a source tarball
+        run: poetry build
+
+      - name: Publish package distributions to PyPI
+        uses: pypa/[email protected]
 
   docker:
     runs-on: ubuntu-latest

docs/source/asking_questions.rst

@@ -63,7 +63,8 @@ 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.
 
 .. code-block:: python
 
@@ -78,7 +79,16 @@ You can also ask multiple questions to a service in parallel.
             {"output_values": {"different": "result"}, "output_manifest": None},
         ]
 
-This method uses threads, allowing all the questions to be asked at once instead of one after another.
+This method uses multithreading, allowing all the questions to be asked at once instead of one after another.
+
+Options:
+
+- If ``raise_errors=False`` is provided, answers are returned for all successful questions while unraised errors are
+  returned for unsuccessful ones
+- If ``raise_errors=False`` is provided with ``max_retries > 0``, failed questions are retried up to this number of
+  times
+- If ``raise_errors=False`` is provided with ``max_retries > 0`` and ``prevent_retries_when`` is set to a list of
+  exception types, failed questions are retried except for those whose exception types are in the list
 
 
 Asking a question within a service

octue/cloud/deployment/google/cloud_run/Dockerfile-python311

@@ -0,0 +1,49 @@
+FROM windpioneers/gdal-python:little-gecko-gdal-2.4.1-python-3.11-slim
+
+# Ensure print statements and log messages appear promptly in Cloud Logging.
+ENV PYTHONUNBUFFERED True
+
+ENV PROJECT_ROOT=/workspace
+WORKDIR $PROJECT_ROOT
+
+RUN apt-get update -y && apt-get install -y --fix-missing build-essential && rm -rf /var/lib/apt/lists/*
+
+# Install poetry.
+ENV POETRY_HOME=/root/.poetry
+ENV PATH "$POETRY_HOME/bin:$PATH"
+RUN curl -sSL https://install.python-poetry.org | python3 - && poetry config virtualenvs.create false;
+
+# Copy in the dependencies file(s) for caching. One or more of `requirements.txt`, `setup.py`, and `pyproject.toml and
+# `poetry.lock` must be present.
+COPY pyproject.tom[l] poetry.loc[k] setup.p[y] requirements.tx[t] ./
+
+# If `pyproject.toml` is present, install the dependencies only to utilise layer caching for quick rebuilds.
+RUN if [ -f "pyproject.toml" ]; then poetry install  \
+    --no-ansi  \
+    --no-interaction  \
+    --no-cache  \
+    --no-root  \
+    --only main;  \
+    fi
+
+# Copy local code to the application root directory.
+COPY . .
+
+# Install local packages if using poetry. Otherwise, install everything if using `setup.py` or `requirements.txt`.
+RUN if [ -f "pyproject.toml" ]; then poetry install --only main;  \
+    elif [ -f "setup.py" ]; then pip install --upgrade pip && pip install -e .;  \
+    elif [ -f "requirements.txt" ]; then pip install --upgrade pip && pip install -r requirements.txt; fi
+
+EXPOSE $PORT
+
+ENV USE_OCTUE_LOG_HANDLER=1
+ENV COMPUTE_PROVIDER=GOOGLE_CLOUD_RUN
+
+ARG GUNICORN_WORKERS=1
+ENV GUNICORN_WORKERS=$GUNICORN_WORKERS
+
+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.cloud.deployment.google.cloud_run.flask_app:app

octue/cloud/pub_sub/service.py

@@ -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
@@ -294,11 +294,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/service_id.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/metadata/recorded_questions.jsonl

@@ -78,3 +78,6 @@
 {"parent_sdk_version": "0.47.2", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"ffb8a979-5435-4827-a3db-28beb75a0d10\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmp_8dj57yl\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "6f7528e0-435d-4e40-ac15-2f7c8398591e", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.47.2"}}}
 {"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"}}}
+{"parent_sdk_version": "0.50.0", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"2d618d16-2a2b-437b-8db7-d5139e1645f2\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmpr7jpj4sp\"}}, \"children\": null, \"message_number\": 0}", "attributes": {"question_uuid": "7bc572dc-ab8b-4aed-a132-b0b509caceb3", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.50.0"}}}