MRG: Merge pull request #632 from octue/better-support-asynchronous-q…

…uestions

Switch to event-driven infrastructure and improve support for asynchronous questions

.github/workflows/python-ci.yml

@@ -14,15 +14,10 @@ on:
 jobs:
   check-semantic-version:
     if: "!contains(github.event.head_commit.message, 'skipci')"
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-        with:
-          fetch-depth: 0
-      - uses: octue/check-semantic-version@1.0.0.beta-9
-        with:
-          path: pyproject.toml
-          breaking_change_indicated_by: minor
+    uses: octue/workflows/.github/workflows/check-semantic-version.yml@main
+    with:
+      path: pyproject.toml
+      breaking_change_indicated_by: minor
 
   run-tests:
     if: "!contains(github.event.head_commit.message, 'skipci')"
@@ -68,10 +63,11 @@ jobs:
         run: tox -vv -e py
 
       - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
+        uses: codecov/codecov-action@v4
         with:
           files: coverage.xml
           fail_ci_if_error: true
+          token: ${{ secrets.CODECOV_TOKEN }}
 
   test-publish:
     if: "!contains(github.event.head_commit.message, 'skipci')"

.github/workflows/release.yml

@@ -59,10 +59,11 @@ jobs:
         run: tox -vv -e py
 
       - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
+        uses: codecov/codecov-action@v4
         with:
           files: coverage.xml
           fail_ci_if_error: false
+          token: ${{ secrets.CODECOV_TOKEN }}
 
     outputs:
       package_version: ${{ steps.get-package-version.outputs.PACKAGE_VERSION }}

.github/workflows/update-pull-request.yml

@@ -6,21 +6,13 @@
 
 name: update-pull-request
 
-on: pull_request
+on: [pull_request]
 
 jobs:
   description:
-    if: "!contains(github.event.pull_request.body, '<!--- SKIP AUTOGENERATED NOTES --->')"
-    runs-on: ubuntu-latest
-    steps:
-      - uses: octue/generate-pull-request-description@1.0.0.beta-2
-        id: pr-description
-        with:
-          pull_request_url: ${{ github.event.pull_request.url }}
-          api_token: ${{ secrets.GITHUB_TOKEN }}
-
-      - name: Update pull request body
-        uses: riskledger/update-pr-description@v2
-        with:
-          body: ${{ steps.pr-description.outputs.pull_request_description }}
-          token: ${{ secrets.GITHUB_TOKEN }}
+    uses: octue/workflows/.github/workflows/generate-pull-request-description.yml@main
+    secrets:
+      token: ${{ secrets.GITHUB_TOKEN }}
+    permissions:
+      contents: read
+      pull-requests: write

LICENSE

@@ -3,7 +3,7 @@ octue-sdk-python Application SDK for python-based apps on the Octue platform
 
 MIT License
 
-Copyright (c) 2017-2022 Octue Ltd
+Copyright (c) 2017-2024 Octue Ltd
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

docs/source/asking_questions.rst

@@ -4,11 +4,27 @@
 Asking services questions
 =========================
 
-How to ask a question
-=====================
+What is a question?
+===================
+A question is a set of data (input values and/or an input manifest) sent to a child for processing/analysis. Questions
+can be:
+
+- **Synchronous ("ask-and-wait"):** A question whose answer is waited for in real time
+
+- **Asynchronous ("fire-and-forget"):** A question whose answer is not waited for and is instead retrieved later. There
+  are two types:
+
+    - **Regular:** Responses to these questions are automatically stored in an event store where they can be :ref:`retrieved using the Octue SDK <retrieving_asynchronous_answers>`
+
+    - **Push endpoint:** Responses to these questions are pushed to an HTTP endpoint for asynchronous handling using Octue's
+      `django-twined <https://django-twined.readthedocs.io/en/latest/>`_ or custom logic in your own webserver.
+
 Questions are always asked to a *revision* of a service. You can ask a service a question if you have its
-:ref:`SRUID <sruid_definition>`, project name, and the necessary permissions. The question is formed of input values
-and/or an input manifest.
+:ref:`SRUID <sruid_definition>`, project name, and the necessary permissions.
+
+
+Asking a question
+=================
 
 .. code-block:: python
 
@@ -47,19 +63,133 @@ You can also set the following options when you call :mod:`Child.ask <octue.reso
 - ``subscribe_to_logs`` - if true, the child will forward its logs to you
 - ``allow_local_files`` - if true, local files/datasets are allowed in any input manifest you supply
 - ``handle_monitor_message`` - if provided a function, it will be called on any monitor messages from the child
-- ``record_messages_to`` – if given a path to a JSON file, messages received from the parent while it processes the question are saved to it
+- ``record_events`` – if ``True``, events received from the parent while it processes the question are saved to the ``Child.received_events`` property
 - ``save_diagnostics`` – must be one of {"SAVE_DIAGNOSTICS_OFF", "SAVE_DIAGNOSTICS_ON_CRASH", "SAVE_DIAGNOSTICS_ON"}; if turned on, allow the input values and manifest (and its datasets) to be saved by the child either all the time or just if the analysis fails
 - ``question_uuid`` - if provided, the question will use this UUID instead of a generated one
+- ``push_endpoint`` - if provided, the result and other events produced during the processing of the question will be pushed to this HTTP endpoint (a URL)
+- ``asynchronous`` - if ``True``, don't wait for an answer to the question (the result and other events can be :ref:`retrieved from the event store later <retrieving_asynchronous_answers>`)
 - ``timeout`` - how long in seconds to wait for an answer (``None`` by default - i.e. don't time out)
 
+Exceptions raised by a child
+----------------------------
 If a child raises an exception while processing your question, the exception will always be forwarded and re-raised in
 your local service or python session. You can handle exceptions in whatever way you like.
 
-If setting a timeout, bear in mind that the question has to reach the child, the child has to run its analysis on
-the inputs sent to it (this most likely corresponds to the dominant part of the wait time), and the answer has to be
-sent back to the parent. If you're not sure how long a particular analysis might take, it's best to set the timeout to
-``None`` initially or ask the owner/maintainer of the child for an estimate.
+Timeouts
+--------
+If setting a timeout, bear in mind that the question has to reach the child, the child has to run its analysis on the
+inputs sent to it (this will most likely make up the dominant part of the wait time), and the answer has to be sent back
+to the parent. If you're not sure how long a particular analysis might take, it's best to set the timeout to ``None``
+initially or ask the owner/maintainer of the child for an estimate.
+
+
+.. _retrieving_asynchronous_answers:
+
+Retrieving answers to asynchronous questions
+============================================
+To retrieve results and other events from the processing of a question later, make sure you have the permissions to
+access the event store and run:
+
+.. code-block:: python
+
+    from octue.cloud.pub_sub.bigquery import get_events
+
+    events = get_events(
+        table_id="your-project.your-dataset.your-table",
+        sender="octue/test-service:1.0.0",
+        question_uuid="53353901-0b47-44e7-9da3-a3ed59990a71",
+    )
+
+
+**Options**
+
+- ``kind`` - Only retrieve this kind of event if present (e.g. "result")
+- ``include_attributes`` - If ``True``, retrieve all the events' attributes as well
+- ``include_backend_metadata`` - If ``True``, retrieve information about the service backend that produced the event
+- ``limit`` - If set to a positive integer, limit the number of events returned to this
+
+
+.. collapse:: See an example output here...
+
+    .. code-block:: python
+
+        >>> events
+        [
+          {
+            "event": {
+              "datetime": "2024-03-06T15:44:18.156044",
+              "kind": "delivery_acknowledgement"
+            },
+          },
+          {
+            "event": {
+              "kind": "log_record",
+              "log_record": {
+                "args": null,
+                "created": 1709739861.5949728,
+                "exc_info": null,
+                "exc_text": null,
+                "filename": "app.py",
+                "funcName": "run",
+                "levelname": "INFO",
+                "levelno": 20,
+                "lineno": 28,
+                "module": "app",
+                "msecs": 594.9728488922119,
+                "msg": "Finished example analysis.",
+                "name": "app",
+                "pathname": "/workspace/example_service_cloud_run/app.py",
+                "process": 2,
+                "processName": "MainProcess",
+                "relativeCreated": 8560.13798713684,
+                "stack_info": null,
+                "thread": 68328473233152,
+                "threadName": "ThreadPoolExecutor-0_2"
+              }
+            },
+          },
+          {
+            "event": {
+              "datetime": "2024-03-06T15:46:18.167424",
+              "kind": "heartbeat"
+            },
+            "attributes": {
+              "datetime": "2024-04-11T10:46:48.236064",
+              "uuid": "a9de11b1-e88f-43fa-b3a4-40a590c3443f",
+              "order": "7",
+              "question_uuid": "d45c7e99-d610-413b-8130-dd6eef46dda6",
+              "originator": "octue/test-service:1.0.0",
+              "sender": "octue/test-service:1.0.0",
+              "sender_type": "CHILD",
+              "sender_sdk_version": "0.51.0",
+              "recipient": "octue/another-service:3.2.1"
+            }
+          }
+          {
+            "event": {
+              "kind": "result",
+              "output_manifest": {
+                "datasets": {
+                  "example_dataset": {
+                    "files": [
+                      "gs://octue-sdk-python-test-bucket/example_output_datasets/example_dataset/output.dat"
+                    ],
+                    "id": "419bff6b-08c3-4c16-9eb1-5d1709168003",
+                    "labels": [],
+                    "name": "divergent-strange-gharial-of-pizza",
+                    "path": "https://storage.googleapis.com/octue-sdk-python-test-bucket/example_output_datasets/example_dataset/.signed_metadata_files/divergent-strange-gharial-of-pizza",
+                    "tags": {}
+                  }
+                },
+                "id": "a13713ae-f207-41c6-9e29-0a848ced6039",
+                "name": null
+              },
+              "output_values": [1, 2, 3, 4, 5]
+            },
+          },
+        ]
 
+----
 
 Asking multiple questions in parallel
 =====================================
@@ -81,7 +211,7 @@ raised and no answers are returned.
 
 This method uses multithreading, allowing all the questions to be asked at once instead of one after another.
 
-Options:
+**Options**
 
 - If ``raise_errors=False`` is provided, answers are returned for all successful questions while unraised errors are
   returned for unsuccessful ones

docs/source/conf.py

@@ -51,7 +51,7 @@
 # General information about the project.
 project = "Octue SDK (Python)"
 author = "Octue Ltd"
-copyright = "2022, Octue Ltd"
+copyright = "2024, Octue Ltd"
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the