CDAM implementation

ARIA_DABs/resources/workflows/case_linking_databricks_job.yml

@@ -18,5 +18,5 @@ resources:
             - task_key: DLT_creation_of_case_linking
           job_cluster_key: case_linking_databricks_job
           notebook_task:
-            notebook_path: "/Workspace/live/ACTIVE/MVP/Active_CCD_Publish_Case_Linking"
+            notebook_path: "/Workspace/live/ACTIVE/MVP/CaseLinking/Active_CCD_Publish_Case_Linking"
             source: WORKSPACE

ARIA_DABs/resources/workflows/case_linking_results_databricks_job.yml

@@ -12,7 +12,7 @@ resources:
         - task_key: active_ccd_results_case_linking
           job_cluster_key: ccd_case_linking_result_databricks_job
           notebook_task:
-            notebook_path: "/Workspace/live/ACTIVE/MVP/Active_CCD_Results_Case_Linking"
+            notebook_path: "/Workspace/live/ACTIVE/MVP/CaseLinking/Active_CCD_Results_Case_Linking"
             source: WORKSPACE
       queue:
         enabled: true

AzureFunctions/ACTIVE/active_caselink_ccd/cl_ccdFunctions.py

@@ -138,7 +138,7 @@ def submit_case_event(ccd_base_url, uid, jid, ctid, cid, etid, event_token, payl
 @retry_on_result(
     max_retries=2,
     base_delay=30,
-    max_delay=120,
+    max_delay=60,
     retry_on=lambda r: isinstance(r, dict) and r.get("Status") == "ERROR",
 )
 def process_event(env, ccdReference, runId, caseLinkPayload, PR_REFERENCE, overwrite=False):

AzureFunctions/ACTIVE/active_caselink_ccd/function_app.py

@@ -1,8 +1,6 @@
 import asyncio
 import azure.functions as func
 import logging
-import logging.handlers
-import queue as _queue
 import json
 import os
 

AzureFunctions/ACTIVE/active_caselink_ccd/host.json

@@ -15,16 +15,5 @@
   "extensionBundle": {
     "id": "Microsoft.Azure.Functions.ExtensionBundle",
     "version": "[4.*, 5.0.0)"
-  },
-  // "scale": {
-  //   "minInstances": 1,
-  //   "maxInstances": 5,
-  //   "maxBurst": 1
-  // },
-  "retry": {
-    "strategy": "exponentialBackoff",
-    "maxRetryCount": 3,
-    "minimumInterval": "00:00:02",
-    "maximumInterval": "00:00:10"
   }
 }

AzureFunctions/ACTIVE/active_ccd/ccdFunctions.py

@@ -2,6 +2,10 @@
 import logging
 import requests
 from datetime import datetime, timezone
+try:
+    from .retry_decorator import retry_on_result
+except ImportError:
+    from retry_decorator import retry_on_result
 
 # tokenManager lives in the same package. When this module is imported by the
 # Functions host the package root will be `AzureFunctions.ACTIVE.active_ccd`.
@@ -124,6 +128,12 @@ def submit_case(ccd_base_url, event_token, payloadData, jid, ctid, idam_token, u
 
 
 # caseNo = event.key, payloadData = event.value
+@retry_on_result(
+    max_retries=2,
+    base_delay=30,
+    max_delay=60,
+    retry_on=lambda r: isinstance(r, dict) and r.get("Status") == "ERROR",
+)
 def process_case(env, caseNo, payloadData, runId, state, PR_REFERENCE):
     print(f"Starting processing case for {caseNo}")
 

AzureFunctions/ACTIVE/active_ccd/host.json

@@ -12,11 +12,5 @@
   "extensionBundle": {
     "id": "Microsoft.Azure.Functions.ExtensionBundle",
     "version": "[4.*, 5.0.0)"
-  },
-  "retry": {
-    "strategy": "exponentialBackoff",
-    "maxRetryCount": 3,
-    "minimumInterval": "00:00:02",
-    "maximumInterval": "00:00:10"
   }
 }

AzureFunctions/ACTIVE/active_ccd/retry_decorator.py

@@ -0,0 +1,70 @@
+import time
+import random
+from functools import wraps
+
+
+def retry_on_result(
+    max_retries: int = 3,
+    base_delay: float = 1.0,
+    max_delay: float = 60.0,
+    jitter: bool = True,
+    retry_on=None,
+):
+    """
+    Retry decorator with exponential backoff for sync functions intended
+    for use with asyncio.to_thread. Retries only when retry_on
+    predicate returns True — exceptions are not caught.
+
+    Args:
+        max_retries:      Number of retry attempts after the first failure.
+        base_delay:       Initial delay in seconds (doubles each attempt).
+        max_delay:        Upper cap on delay in seconds.
+        jitter:           Randomise delay to 50-100% of computed value.
+        retry_on:         Optional callable(result) -> bool. If provided, a
+                          return value for which this returns True is treated as
+                          a retryable failure (the result is returned as-is
+                          after all attempts are exhausted).
+
+    Usage:
+        @retry_on_result(max_retries=3, retry_on=lambda r: r.get("Status") == "ERROR")
+        def my_blocking_call():
+            ...
+
+        result = await asyncio.to_thread(my_blocking_call)
+    """
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            last_result = None
+
+            for attempt in range(max_retries + 1):
+                result = func(*args, **kwargs)
+
+                if retry_on is not None and retry_on(result):
+                    last_result = result
+
+                    if attempt == max_retries:
+                        print(
+                            f"{func.__name__} returned a retryable result after "
+                            f"{max_retries + 1} attempts. Returning last result."
+                        )
+                        return last_result
+
+                    delay = min(base_delay * (2 ** attempt), max_delay)
+                    if jitter:
+                        delay *= 0.5 + random.random() * 0.5
+
+                    print(
+                        f"{func.__name__} returned a retryable result "
+                        f"(attempt {attempt + 1}/{max_retries + 1}). "
+                        f"Retrying in {delay:.2f}s..."
+                    )
+                    time.sleep(delay)
+                    continue
+
+                return result
+
+            return last_result
+
+        return wrapper
+    return decorator

AzureFunctions/ACTIVE/active_cdam/.funcignore

@@ -0,0 +1,8 @@
+.git*
+.vscode
+__azurite_db*__.json
+__blobstorage__
+__queuestorage__
+local.settings.json
+test
+.venv

AzureFunctions/ACTIVE/active_cdam/.gitignore

@@ -0,0 +1,135 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don’t work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# Azure Functions artifacts
+bin
+obj
+appsettings.json
+local.settings.json
+
+# Azurite artifacts
+__blobstorage__
+__queuestorage__
+__azurite_db*__.json
+.python_packages