catch pyodk error

Author: pld

odk-central-sync/src/download_results_forms.py

@@ -4,6 +4,7 @@
 import click
 import pandas as pd
 from pyodk.client import Client
+from pyodk.errors import PyODKError
 from requests.exceptions import ConnectionError, Timeout
 from rich.console import Console
 from rich.progress import Progress
@@ -173,22 +174,25 @@ def main(project_id: int, center_ids: tuple[int, ...], output_dir: Path, config:
     output_dir.mkdir(parents=True, exist_ok=True)
     output_csv = output_dir / "candidate_results.csv"
 
-    with Client(config_path=str(config)) as client:
-        if not center_ids:
-            forms = client.forms.list(project_id=project_id)
-            center_ids = tuple(
-                int(f.xmlFormId.removeprefix("results_"))
-                for f in forms
-                if f.xmlFormId.startswith("results_")
+    try:
+        with Client(config_path=str(config)) as client:
+            if not center_ids:
+                forms = client.forms.list(project_id=project_id)
+                center_ids = tuple(
+                    int(f.xmlFormId.removeprefix("results_"))
+                    for f in forms
+                    if f.xmlFormId.startswith("results_")
+                )
+                console.log(f"Discovered {len(center_ids)} results forms")
+
+            results_df = export_center_candidate_results(
+                client=client,
+                project_id=project_id,
+                center_ids=list(center_ids),
+                output_dir=output_dir,
             )
-            console.log(f"Discovered {len(center_ids)} results forms")
-
-        results_df = export_center_candidate_results(
-            client=client,
-            project_id=project_id,
-            center_ids=list(center_ids),
-            output_dir=output_dir,
-        )
+    except PyODKError as exc:
+        raise click.ClickException(str(exc)) from exc
 
     results_df.to_csv(output_csv, index=False)
     console.log(f"Saved {len(results_df)} rows to {output_csv}")

odk-central-sync/tests/test_download_results_forms.py

@@ -5,6 +5,8 @@
 import pytest
 from click.testing import CliRunner
 
+from pyodk.errors import PyODKError
+
 from src.download_results_forms import (
     export_center_candidate_results,
     export_form_submissions,
@@ -269,3 +271,19 @@ def test_auto_discovers_centers(self, output_dir):
         assert result.exit_code == 0
         mock_client.forms.list.assert_called_once_with(project_id=1)
         assert (output_dir / "candidate_results.csv").exists()
+
+    def test_pyodk_error_exits_gracefully(self, output_dir):
+        mock_client = MagicMock()
+        mock_client.__enter__ = MagicMock(return_value=mock_client)
+        mock_client.__exit__ = MagicMock(return_value=False)
+        mock_client.forms.list.side_effect = PyODKError("auth failed")
+
+        runner = CliRunner()
+        with patch("src.download_results_forms.Client", return_value=mock_client):
+            result = runner.invoke(
+                main,
+                ["--project-id=1", "--output-dir", str(output_dir)],
+            )
+
+        assert result.exit_code == 1
+        assert "auth failed" in result.output