fix(water-level-import): handle savepoint initialization failure gracefully

- Move `begin_nested` savepoint call inside the try block for proper error handling.
- Ensure session expiration when savepoint creation fails.
- Add test coverage for savepoint creation failure scenarios.

Author: ksmuczynski

services/water_level_csv.py

@@ -351,8 +351,9 @@ def _create_records(
     errors: list[str] = []
 
     for row in rows:
-        savepoint = session.begin_nested()
+        savepoint = None
         try:
+            savepoint = session.begin_nested()
             sample_name = _build_sample_name(row)
             sample = _find_existing_imported_sample(session, row, sample_name)
 
@@ -406,7 +407,7 @@ def _create_records(
                 }
             )
         except Exception as exc:  # pragma: no cover - exercised via DB tests
-            if savepoint.is_active:
+            if savepoint is not None and savepoint.is_active:
                 savepoint.rollback()
             else:
                 session.expire_all()

tests/test_water_level_csv_service.py

@@ -8,6 +8,7 @@
 from tests import get_parameter_id
 from services.water_level_csv import (
     _build_sample_name,
+    _create_records,
     _resolve_measuring_point_height,
     _validate_depth_to_water_against_well,
     bulk_upload_water_levels,
@@ -116,6 +117,30 @@ def test_build_sample_name_uses_deterministic_well_inventory_style_format():
     assert _build_sample_name(row) == "AR0001-WL-202502151030"
 
 
+def test_create_records_reports_savepoint_creation_failure_as_row_error():
+    class BrokenSession:
+        def __init__(self):
+            self.expire_all_called = False
+
+        def begin_nested(self):
+            raise RuntimeError("savepoint failed")
+
+        def expire_all(self):
+            self.expire_all_called = True
+
+    session = BrokenSession()
+
+    created, errors = _create_records(
+        session,
+        parameter_id=1,
+        rows=[SimpleNamespace(row_index=7)],
+    )
+
+    assert created == []
+    assert errors == ["Row 7: savepoint failed"]
+    assert session.expire_all_called is True
+
+
 def test_bulk_upload_water_levels_is_idempotent(water_well_thing):
     csv_content = "\n".join(
         [