SK-3118: Close flowvault coverage gaps and code-review cleanups

Address the code-review P0/P1 findings for flowvault:

- Cover the remaining error-handling/edge branches to reach 100% on flowvault
  product code: batching .env fallback + zero/negative concurrency, response
  parsing None/[None] cases, get_vault_url validation + get_metrics fallbacks
  (new test__utils.py), full validate_bulk_insert/detokenize_request coverage,
  update ApiError body variants, async bulk error paths, and the bulk response
  __str__/empty-retry helpers.
- Exclude setup.py and the `if __name__` boilerplate from coverage via
  .coveragerc.
- requirements.txt: bump pydantic floor 1.9.2 -> 2.0.0 (setup.py and the Fern
  client require pydantic v2).
- Remove dead GetRecordRequest import and MAX_BULK_DATA_SIZE constant; extract
  the repeated 'additional_headers'/'Unknown error' literals into named
  constants; drop explanatory comments from _validations.py.
- get_vault_url: reject an invalid env with `not isinstance(env, Env)` instead
  of `env not in Env`, which raised TypeError (and left the guard unreachable)
  for a non-member on Python < 3.12.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: saileshwar-skyflow

flowvault/.coveragerc

@@ -2,3 +2,9 @@
 omit =
     */generated/*
     generated/*
+    setup.py
+
+[report]
+exclude_lines =
+    pragma: no cover
+    if __name__ == .__main__.:

flowvault/requirements.txt

@@ -1,5 +1,5 @@
 httpx>=0.21.2
-pydantic>= 1.9.2
+pydantic>= 2.0.0
 pydantic-core>=2.18.2
 typing_extensions>= 4.0.0
 coverage >= 7.8.0

flowvault/skyflow_flowvault/utils/_batching.py

@@ -6,8 +6,6 @@
 from common.utils.logger import log_warn
 from skyflow_flowvault.utils._skyflow_messages import SkyflowMessages
 
-MAX_BULK_DATA_SIZE = 10000
-
 DEFAULT_BATCH_SIZE = 50
 MAX_BATCH_SIZE = 1000
 DEFAULT_CONCURRENCY = 1

flowvault/skyflow_flowvault/utils/_utils.py

@@ -18,7 +18,7 @@ def get_vault_url(cluster_id, env, vault_id, logger=None):
     if not cluster_id or not isinstance(cluster_id, str) or not cluster_id.strip():
         raise SkyflowError(CommonMessages.Error.INVALID_CLUSTER_ID.value.format(vault_id), invalid_input_error_code)
 
-    if env not in Env:
+    if not isinstance(env, Env):
         raise SkyflowError(CommonMessages.Error.INVALID_ENV.value.format(vault_id), invalid_input_error_code)
 
     base_url = EnvUrls[env.name].value

flowvault/skyflow_flowvault/utils/validations/_validations.py

@@ -15,10 +15,6 @@
 
 invalid_input_error_code = CommonMessages.ErrorCodes.INVALID_INPUT.value
 
-# validate_vault_config/validate_update_vault_config/validate_credentials are re-exported
-# directly from common.utils.validations -- flowvault's own logic here was field-for-field
-# identical to v2's, confirmed, so both variants now share one implementation.
-
 
 def _validate_upsert(logger, upsert):
     if upsert is None:
@@ -33,7 +29,7 @@ def _validate_upsert(logger, upsert):
         raise SkyflowError(SkyflowMessages.Error.INVALID_UPSERT_UPDATE_TYPE_IN_INSERT.value, invalid_input_error_code)
 
 
-MAX_INSERT_RECORDS = 10000  # matches Java's v3 Validations.validateInsertRequest (hardcoded, not configurable)
+MAX_INSERT_RECORDS = 10000
 
 
 def validate_insert_request(logger, request):
@@ -46,15 +42,10 @@ def validate_insert_request(logger, request):
     if len(request.records) > MAX_INSERT_RECORDS:
         raise SkyflowError(SkyflowMessages.Error.TOO_MANY_RECORDS_IN_INSERT.value, invalid_input_error_code)
 
-    # record.table format and record.data emptiness/key/value validity are checked by the controller
-    # via BaseVaultController._validate_table_name_if_present() / _validate_field_values().
-
     _validate_upsert(logger, request.upsert)
     for record in request.records:
         _validate_upsert(logger, record.upsert)
 
-    # table must be set in exactly one place -- request-level (every record) or per-record (no
-    # partial mix) -- and upsert must live at that same place (mirrors Java's v3 Validations).
     table_at_request_level = request.table_name is not None
 
     if table_at_request_level:
@@ -75,7 +66,6 @@ def validate_insert_request(logger, request):
 
 
 def validate_get_request(logger, request):
-    # Two mutually exclusive modes: multi-table batch (request.records) vs single-table.
     if request.records is not None:
         single_table_fields_set = (
             request.table or request.ids or request.unique_values or request.columns

flowvault/skyflow_flowvault/vault/controller/_vault.py

@@ -41,7 +41,6 @@
     InsertRequest,
     InsertResponse,
     GetRequest,
-    GetRecordRequest,
     GetResponse,
     UpdateRequest,
     UpdateResponse,
@@ -60,6 +59,8 @@
 )
 
 REQUEST_ID_HEADER = "x-request-id"
+ADDITIONAL_HEADERS_KEY = "additional_headers"
+UNKNOWN_ERROR_MESSAGE = "Unknown error"
 
 
 class VaultController(BaseVaultController):
@@ -98,7 +99,7 @@ def insert(self, request: InsertRequest) -> InsertResponse:
                 vault_id=self._vault_client.get_vault_id(),
                 table_name=request.table_name,
                 records=wire_records,
-                request_options={'additional_headers': headers},
+                request_options={ADDITIONAL_HEADERS_KEY: headers},
                 **upsert_kwargs,
             )
             records = [self.__record_row(record, include_data=False) for record in (raw_response.data.records or [])]
@@ -137,7 +138,7 @@ def get(self, request: GetRequest) -> GetResponse:
             log_info(SkyflowMessages.Info.GET_TRIGGERED.value, self._vault_client.get_logger())
             raw_response = records_api.with_raw_response.get_records(
                 vault_id=self._vault_client.get_vault_id(),
-                request_options={'additional_headers': self.__build_headers()},
+                request_options={ADDITIONAL_HEADERS_KEY: self.__build_headers()},
                 **call_kwargs,
             )
             records = [self.__record_row(record, include_data=True) for record in (raw_response.data.records or [])]
@@ -174,7 +175,7 @@ def update(self, request: UpdateRequest) -> UpdateResponse:
                 vault_id=self._vault_client.get_vault_id(),
                 table_name=request.table_name,
                 records=wire_records,
-                request_options={'additional_headers': self.__build_headers()},
+                request_options={ADDITIONAL_HEADERS_KEY: self.__build_headers()},
             )
             request_id = self.__extract_request_id(raw_response.headers)
             records, errors = self.__split_success_and_errors(
@@ -204,7 +205,7 @@ def delete(self, request: DeleteRequest) -> DeleteResponse:
                 table_name=request.table,
                 skyflow_i_ds=request.ids,
                 unique_values=self.__to_unique_values(request.unique_values),
-                request_options={'additional_headers': self.__build_headers()},
+                request_options={ADDITIONAL_HEADERS_KEY: self.__build_headers()},
             )
             records = [self.__delete_row(record) for record in (raw_response.data.records or [])]
         except Exception as e:
@@ -227,7 +228,7 @@ def query(self, request: QueryRequest) -> QueryResponse:
             raw_response = query_api.with_raw_response.execute_query(
                 vault_id=self._vault_client.get_vault_id(),
                 query=request.query,
-                request_options={'additional_headers': self.__build_headers()},
+                request_options={ADDITIONAL_HEADERS_KEY: self.__build_headers()},
             )
             records = [{'data': getattr(record, 'data', None)} for record in (raw_response.data.records or [])]
             metadata = self.__query_metadata(raw_response.data)
@@ -253,7 +254,7 @@ def detokenize(self, request: DetokenizeRequest) -> DetokenizeResponse:
                 vault_id=self._vault_client.get_vault_id(),
                 tokens=request.tokens,
                 token_group_redactions=self.__to_token_group_redactions(request.token_group_redactions),
-                request_options={'additional_headers': self.__build_headers()},
+                request_options={ADDITIONAL_HEADERS_KEY: self.__build_headers()},
             )
             records = [self.__detokenize_row(resp) for resp in (raw_response.data.response or [])]
         except Exception as e:
@@ -275,7 +276,7 @@ def call_batch(batch, start_index):
                     vault_id=self._vault_client.get_vault_id(),
                     table_name=request.table,
                     records=batch,
-                    request_options={'additional_headers': self.__build_headers()},
+                    request_options={ADDITIONAL_HEADERS_KEY: self.__build_headers()},
                     **top_kwargs,
                 )
                 return self.__format_bulk_insert_batch(raw_response.data.records or [], start_index, raw_response.headers)
@@ -299,7 +300,7 @@ async def call_batch(batch, start_index):
                     vault_id=self._vault_client.get_vault_id(),
                     table_name=request.table,
                     records=batch,
-                    request_options={'additional_headers': self.__build_headers()},
+                    request_options={ADDITIONAL_HEADERS_KEY: self.__build_headers()},
                     **top_kwargs,
                 )
                 return self.__format_bulk_insert_batch(raw_response.data.records or [], start_index, raw_response.headers)
@@ -323,7 +324,7 @@ def call_batch(batch, start_index):
                     vault_id=self._vault_client.get_vault_id(),
                     tokens=batch,
                     token_group_redactions=redactions,
-                    request_options={'additional_headers': self.__build_headers()},
+                    request_options={ADDITIONAL_HEADERS_KEY: self.__build_headers()},
                 )
                 return self.__format_bulk_detokenize_batch(raw_response.data.response or [], start_index, raw_response.headers)
             except Exception as e:
@@ -346,7 +347,7 @@ async def call_batch(batch, start_index):
                     vault_id=self._vault_client.get_vault_id(),
                     tokens=batch,
                     token_group_redactions=redactions,
-                    request_options={'additional_headers': self.__build_headers()},
+                    request_options={ADDITIONAL_HEADERS_KEY: self.__build_headers()},
                 )
                 return self.__format_bulk_detokenize_batch(raw_response.data.response or [], start_index, raw_response.headers)
             except Exception as e:
@@ -476,7 +477,7 @@ def __bulk_batch_error_tuples(self, e, count, start_index):
             if body and isinstance(body.get('records'), list) and body['records']:
                 tuples = [
                     (start_index + offset, request_id,
-                     record.get('error', record.get('message', 'Unknown error')),
+                     record.get('error', record.get('message', UNKNOWN_ERROR_MESSAGE)),
                      record.get('http_code', record.get('httpCode', record.get('statusCode', status))))
                     for offset, record in enumerate(body['records']) if isinstance(record, dict)
                 ]
@@ -710,5 +711,5 @@ def __errors_from_exception(self, e, records, start_index):
 
     def __error_dict_from_record_map(self, record_map, request_index, request_id):
         code = record_map.get('http_code', record_map.get('httpCode', record_map.get('statusCode')))
-        message = record_map.get('error', record_map.get('message', 'Unknown error'))
+        message = record_map.get('error', record_map.get('message', UNKNOWN_ERROR_MESSAGE))
         return {'request_index': request_index, 'error': message, 'code': code, 'request_id': request_id}

flowvault/tests/utils/test__batching.py

@@ -63,12 +63,29 @@ def test_invalid_concurrency_falls_back(self):
             _, concurrency = resolve_batch_config(INSERT_BATCH_SIZE_KEY, INSERT_CONCURRENCY_LIMIT_KEY, 500)
         self.assertEqual(concurrency, 1)
 
+    def test_zero_or_negative_concurrency_falls_back(self):
+        for raw in ("0", "-3"):
+            with _with_settings({INSERT_CONCURRENCY_LIMIT_KEY: raw}):
+                _, concurrency = resolve_batch_config(INSERT_BATCH_SIZE_KEY, INSERT_CONCURRENCY_LIMIT_KEY, 500)
+            self.assertEqual(concurrency, 1)
+
 
 class TestResolveSetting(unittest.TestCase):
     def test_reads_process_env_first(self):
         with patch.dict(os.environ, {INSERT_BATCH_SIZE_KEY: "77"}):
             self.assertEqual(_batching._resolve_setting(INSERT_BATCH_SIZE_KEY), "77")
 
+    def test_falls_back_to_dotenv_when_env_absent(self):
+        with patch.dict(os.environ, {}, clear=True), \
+                patch.object(_batching, "find_dotenv", return_value="/tmp/.env"), \
+                patch.object(_batching, "dotenv_values", return_value={INSERT_BATCH_SIZE_KEY: "88"}):
+            self.assertEqual(_batching._resolve_setting(INSERT_BATCH_SIZE_KEY), "88")
+
+    def test_dotenv_lookup_error_returns_none(self):
+        with patch.dict(os.environ, {}, clear=True), \
+                patch.object(_batching, "find_dotenv", side_effect=Exception("boom")):
+            self.assertIsNone(_batching._resolve_setting(INSERT_BATCH_SIZE_KEY))
+
 
 if __name__ == "__main__":
     unittest.main()

flowvault/tests/utils/test__response_parsing.py

@@ -29,6 +29,12 @@ def test_bare_value_becomes_a_token(self):
     def test_none_returns_none(self):
         self.assertIsNone(parse_tokens(None))
 
+    def test_none_column_value_is_skipped(self):
+        self.assertEqual(parse_tokens({"ssn": None}), {})
+
+    def test_none_entry_in_list_is_dropped(self):
+        self.assertEqual(parse_tokens({"ssn": [None]}), {"ssn": []})
+
 
 class TestParseHashedData(unittest.TestCase):
     def test_list_of_hash_entries(self):
@@ -44,6 +50,9 @@ def test_bare_value_wrapped(self):
     def test_none_returns_none(self):
         self.assertIsNone(parse_hashed_data(None))
 
+    def test_none_entry_in_list_is_dropped(self):
+        self.assertEqual(parse_hashed_data({"ssn": [None]}), {"ssn": []})
+
 
 class TestParseMetadata(unittest.TestCase):
     def test_reads_both_casings(self):

flowvault/tests/utils/test__utils.py

@@ -0,0 +1,70 @@
+import unittest
+from unittest.mock import patch
+
+from common.errors import SkyflowError
+from common.utils.enums import Env
+from common.utils.constants import SdkMetricsKey
+from skyflow_flowvault.utils import _utils
+from skyflow_flowvault.utils._utils import get_vault_url, get_metrics
+
+
+class TestGetVaultUrl(unittest.TestCase):
+    def test_valid(self):
+        url = get_vault_url("cluster123", Env.PROD, "v1")
+        self.assertTrue(url.startswith("https://cluster123."))
+
+    def test_invalid_cluster_id_raises(self):
+        for bad in ("", "   ", None, 123):
+            with self.assertRaises(SkyflowError):
+                get_vault_url(bad, Env.PROD, "v1")
+
+    def test_invalid_env_raises(self):
+        with self.assertRaises(SkyflowError):
+            get_vault_url("cluster123", "not-an-env", "v1")
+
+
+class _SysPlatformRaises:
+    version = "v"
+
+    @property
+    def platform(self):
+        raise RuntimeError("boom")
+
+
+class _SysVersionRaises:
+    platform = "linux"
+
+    @property
+    def version(self):
+        raise RuntimeError("boom")
+
+
+class TestGetMetrics(unittest.TestCase):
+    def setUp(self):
+        _utils._CACHED_METRICS.clear()
+
+    def tearDown(self):
+        _utils._CACHED_METRICS.clear()
+
+    def test_caches_after_first_call(self):
+        first = get_metrics()
+        self.assertIs(get_metrics(), first)
+
+    def test_device_model_error_falls_back(self):
+        with patch.object(_utils.platform, "node", side_effect=RuntimeError("boom")):
+            metrics = get_metrics()
+        self.assertEqual(metrics[SdkMetricsKey.SDK_CLIENT_DEVICE_MODEL], "")
+
+    def test_os_details_error_falls_back(self):
+        with patch.object(_utils, "sys", _SysPlatformRaises()):
+            metrics = get_metrics()
+        self.assertEqual(metrics[SdkMetricsKey.SDK_CLIENT_OS_DETAILS], "")
+
+    def test_runtime_error_falls_back(self):
+        with patch.object(_utils, "sys", _SysVersionRaises()):
+            metrics = get_metrics()
+        self.assertEqual(metrics[SdkMetricsKey.SDK_RUNTIME_DETAILS], _utils.SdkPrefix.PYTHON_RUNTIME + "")
+
+
+if __name__ == "__main__":
+    unittest.main()