ENH: Add helper function to send breadcrumb on interp termination

.github/workflows/build-test.yml

@@ -13,17 +13,20 @@ on:
     # 7am EST / 8am EDT Mondays
     - cron: '0 12 * * 1'
 
+defaults:
+  run:
+    shell: bash
+
 concurrency:
-  group: python-${{ github.ref }}
+  group: ${{ github.workflow }}-${{ github.ref }}
   cancel-in-progress: true
 
 
 jobs:
   build:
-    if: "!contains(github.event.head_commit.message, '[skip ci]')"
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
       with:
         fetch-depth: 0
     - name: Set up Python 3
@@ -47,7 +50,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
+        python-version: ['3.8', '3.9', '3.10', '3.11', '3.12']
         install: [repo]
         include:
           - python-version: '3.11'
@@ -59,7 +62,7 @@ jobs:
     env:
       INSTALL_TYPE: ${{ matrix.install }}
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
       if: matrix.install == 'repo' || matrix.install == 'editable'
       with:
         fetch-depth: 0
@@ -110,7 +113,7 @@ jobs:
     needs: [build, test]
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
       with:
         fetch-depth: 0
     - name: Set up Python 3
@@ -130,7 +133,7 @@ jobs:
         user: __token__
         password: ${{ secrets.PYPI_API_TOKEN }}
     - name: Create GitHub release from annotated tag
-      uses: spenserblack/actions-tag-to-release@v1.1.0
+      uses: spenserblack/actions-tag-to-release@v3
       with:
         prerelease: auto
         prerelease-pattern: '*rc*'

.github/workflows/test-docker.yml

@@ -8,14 +8,18 @@ on:
     branches:
       - main
 
+defaults:
+  run:
+    shell: bash
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
 
 jobs:
   docker:
     if: "contains(github.event.commits[0].message, '[docker test]')"
     name: Docker tests
-    defaults:
-      run:
-        shell: bash --noprofile --norc -exo pipefail {0}
     strategy:
       fail-fast: false
       matrix:
@@ -37,7 +41,7 @@ jobs:
             docker_version: "20.10"
     runs-on: ${{ matrix.os }}
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: setup Docker
         uses: docker-practice/actions-setup-docker@master
         with:

migas/__init__.py

@@ -3,6 +3,7 @@
 __version__ = _version.get_versions()['version']
 
 from .config import print_config, setup
+from .helpers import track_exit
 from .operations import add_project, get_usage
 
 __all__ = (
@@ -11,4 +12,6 @@
     "get_usage",
     "print_config",
     "setup",
+    "track_exit",
 )
+

migas/helpers.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import atexit
+import sys
+
+from migas.operations import add_project
+
+
+def track_exit(project: str, version: str, error_funcs: dict | None = None) -> None:
+    atexit.register(_final_breadcrumb, project, version, error_funcs)
+
+def _final_breadcrumb(project: str, version: str, error_funcs: dict | None = None) -> dict:
+    kwargs = _inspect_error(error_funcs)
+    return add_project(project, version, **kwargs)
+
+
+def _inspect_error(error_funcs: dict | None) -> dict:
+    etype, evalue, etb = None, None, None
+
+    # Python 3.12, new method
+    # MG: Cannot reproduce behavior while testing with 3.12.0
+    # if hasattr(sys, 'last_exc'):
+    #     etype, evalue, etb = sys.last_exc
+
+    # < 3.11
+    if hasattr(sys, 'last_type'):
+        etype = sys.last_type
+        evalue = sys.last_value
+        etb = sys.last_traceback
+
+    if etype:
+        ename = etype.__name__
+
+        if isinstance(error_funcs, dict) and ename in error_funcs:
+            func = error_funcs[ename]
+            kwargs = func(etype, evalue, etb)
+
+        elif ename in ('KeyboardInterrupt', 'BdbQuit'):
+            kwargs = {
+                'status': 'S',
+                'status_desc': 'Suspended',
+            }
+
+        else:
+            kwargs = {
+                'status': 'F',
+                'status_desc': 'Errored',
+                'error_type': ename,
+                'error_desc': evalue,
+            }
+    else:
+        kwargs = {
+            'status': 'C',
+            'status_desc': 'Completed',
+        }
+
+    return kwargs

migas/operations.py

@@ -1,23 +1,18 @@
 """
 Create queries and mutations to be sent to the graphql endpoint.
 """
-import sys
-import typing
+from __future__ import annotations
+
+import typing as ty
 
 from migas.config import Config, logger, telemetry_enabled
 from migas.request import request
 
-if sys.version_info[:2] >= (3, 8):
-    from typing import TypedDict
-else:
-    # TODO: 3.8 - Remove backport
-    from typing_extensions import TypedDict
-
 
 DEFAULT_ERROR = '[migas-py] An error occurred.'
 
 
-class OperationTemplate(TypedDict):
+class OperationTemplate(ty.TypedDict):
     operation: str
     args: dict
     response: dict
@@ -135,7 +130,7 @@ def add_project(
     return res
 
 
-def _introspec(func: typing.Callable, func_locals: dict) -> dict:
+def _introspec(func: ty.Callable, func_locals: dict) -> dict:
     """Inspect a function and return all parameters (not defaults)."""
     import inspect
 
@@ -164,7 +159,7 @@ def _formulate_query(params: dict, template: OperationTemplate) -> str:
     return query
 
 
-def _filter_response(response: typing.Union[dict, str], operation: str, fallback: dict):
+def _filter_response(response: dict | str, operation: str, fallback: dict):
     if isinstance(response, dict):
         res = response.get("data")
         # success

migas/tests/conftest.py

@@ -0,0 +1,15 @@
+import pytest
+
+import migas
+
+TEST_ROOT = "http://localhost:8080/"
+TEST_ENDPOINT = f"{TEST_ROOT}graphql"
+
+
+
+@pytest.fixture(scope='module')
+def setup_migas():
+    """Ensure migas is configured to communicate with the staging app."""
+    migas.setup(endpoint=TEST_ENDPOINT)
+
+    assert migas.config.Config._is_setup

migas/tests/test_helpers.py

@@ -0,0 +1,42 @@
+import sys
+
+import pytest
+
+import migas
+
+
+class CustomException(Exception):
+    ...
+
+
+def sample_error_func(etype: Exception, evalue: str, etb: str):
+    ename = etype.__name__
+    if ename == "CustomException":
+        return {
+            'status': 'F',
+            'status_desc': 'Custom Error!',
+            'error_type': ename,
+            'error_desc': 'Custom Error!',
+        }
+
+
+@pytest.mark.parametrize('error_funcs,error,status,error_desc', [
+    (None, None, 'C', None),
+    (None, KeyboardInterrupt, 'S', None),
+    (None, KeyError, 'F', 'KeyError: \'foo\''),
+    ({'CustomException': sample_error_func}, CustomException, 'F', 'Custom Error!'),
+])
+def test_inspect_error(monkeypatch, error_funcs, error, status, error_desc):
+
+    # do not actually call the server
+    if error is not None:
+        monkeypatch.setattr(sys, 'last_type', error, raising=False)
+        monkeypatch.setattr(sys, 'last_value', error_desc, raising=False)
+        monkeypatch.setattr(sys, 'last_traceback', 'Traceback...', raising=False)
+
+    from migas.helpers import _inspect_error
+    res = _inspect_error(error_funcs)
+
+    assert res.get('status') == status
+    if error_desc is not None:
+        assert res.get('error_desc') == error_desc

migas/tests/test_operations.py

@@ -5,7 +5,7 @@
 
 import pytest
 
-from migas import __version__, setup
+from migas import __version__
 from migas.operations import add_project, get_usage
 
 from .utils import do_server_tests
@@ -19,13 +19,7 @@
 today = today.strftime('%Y-%m-%d')
 
 
-@pytest.fixture(scope='module', autouse=True)
-def setup_migas(endpoint):
-    """Ensure migas is configured to communicate with the staging app."""
-    setup(endpoint=endpoint)
-
-
-def test_operations():
+def test_operations(setup_migas):
     _test_add_project()
     # add delay to ensure server has updated
     time.sleep(2)

migas/tests/utils.py

@@ -12,5 +12,4 @@ def _check_server_available() -> bool:
         return False
     return True
 
-
 do_server_tests = _check_server_available()