feat(csv-parser): update option to provide an array of values that should be set to None in the csv file (#581)

Author: aldogonzalez8

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3630,9 +3630,10 @@ definitions:
       delimiter:
         type: string
         default: ","
-      set_empty_cell_to_none:
-        type: boolean
-        default: false
+      set_values_to_none:
+        type: array
+        items:
+          type: string
   AsyncJobStatusMap:
     description: Matches the api job status to Async Job Status.
     type: object

airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py

@@ -9,7 +9,7 @@
 import logging
 from dataclasses import dataclass
 from io import BufferedIOBase, TextIOWrapper
-from typing import Any, Optional
+from typing import Any, List, Optional
 
 import orjson
 import requests
@@ -103,7 +103,7 @@ class CsvParser(Parser):
     # TODO: migrate implementation to re-use file-base classes
     encoding: Optional[str] = "utf-8"
     delimiter: Optional[str] = ","
-    set_empty_cell_to_none: Optional[bool] = False
+    set_values_to_none: Optional[List[str]] = None
 
     def _get_delimiter(self) -> Optional[str]:
         """
@@ -122,8 +122,8 @@ def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
         text_data = TextIOWrapper(data, encoding=self.encoding)  # type: ignore
         reader = csv.DictReader(text_data, delimiter=self._get_delimiter() or ",")
         for row in reader:
-            if self.set_empty_cell_to_none:
-                row = {k: (None if v == "" else v) for k, v in row.items()}
+            if self.set_values_to_none:
+                row = {k: (None if v in self.set_values_to_none else v) for k, v in row.items()}
             yield row
 
 

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1383,7 +1383,7 @@ class CsvDecoder(BaseModel):
     type: Literal["CsvDecoder"]
     encoding: Optional[str] = "utf-8"
     delimiter: Optional[str] = ","
-    set_empty_cell_to_none: Optional[bool] = False
+    set_values_to_none: Optional[List[str]] = None
 
 
 class AsyncJobStatusMap(BaseModel):

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2651,7 +2651,7 @@ def _get_parser(model: BaseModel, config: Config) -> Parser:
             return CsvParser(
                 encoding=model.encoding,
                 delimiter=model.delimiter,
-                set_empty_cell_to_none=model.set_empty_cell_to_none,
+                set_values_to_none=model.set_values_to_none,
             )
         elif isinstance(model, GzipDecoderModel):
             return GzipParser(

unit_tests/sources/declarative/decoders/test_composite_decoder.py

@@ -8,7 +8,7 @@
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from io import BytesIO, StringIO
 from threading import Thread
-from typing import ClassVar, Iterable
+from typing import ClassVar, Iterable, List
 from unittest.mock import Mock, patch
 
 import pytest
@@ -44,17 +44,18 @@ def generate_csv(
     encoding: str = "utf-8",
     delimiter: str = ",",
     should_compress: bool = False,
-    add_empty_strings: bool = False,
+    add_extra_column: bool = False,
+    extra_column_value: str = "",
 ) -> bytes:
     data = [
         {"id": "1", "name": "John", "age": "28"},
         {"id": "2", "name": "Alice", "age": "34"},
         {"id": "3", "name": "Bob", "age": "25"},
     ]
     fieldnames = ["id", "name", "age"]
-    if add_empty_strings:
+    if add_extra_column:
         for row in data:
-            row["gender"] = ""
+            row["gender"] = extra_column_value
         fieldnames.append("gender")
 
     output = StringIO()
@@ -266,16 +267,22 @@ def test_composite_raw_decoder_csv_parser_values(requests_mock, encoding: str, d
     assert parsed_records == expected_data
 
 
-@pytest.mark.parametrize("set_empty_cell_to_none", [True, False])
-def test_composite_raw_decoder_parse_empty_strings(requests_mock, set_empty_cell_to_none: bool):
+@pytest.mark.parametrize("set_values_to_none", [None, [""], ["--"]])
+def test_composite_raw_decoder_parse_empty_strings(
+    requests_mock, set_values_to_none: List[str] | None
+):
     requests_mock.register_uri(
         "GET",
         "https://airbyte.io/",
-        content=generate_csv(should_compress=False, add_empty_strings=True),
+        content=generate_csv(
+            should_compress=False,
+            add_extra_column=True,
+            extra_column_value=set_values_to_none[0] if set_values_to_none else "random_value",
+        ),
     )
     response = requests.get("https://airbyte.io/", stream=True)
 
-    parser = CsvParser(set_empty_cell_to_none=set_empty_cell_to_none)
+    parser = CsvParser(set_values_to_none=set_values_to_none)
     composite_raw_decoder = CompositeRawDecoder(parser=parser)
 
     expected_data = [
@@ -284,7 +291,7 @@ def test_composite_raw_decoder_parse_empty_strings(requests_mock, set_empty_cell
         {"id": "3", "name": "Bob", "age": "25"},
     ]
     for expected_record in expected_data:
-        expected_record["gender"] = None if set_empty_cell_to_none else ""
+        expected_record["gender"] = None if set_values_to_none else "random_value"
 
     parsed_records = list(composite_raw_decoder.decode(response))
     assert parsed_records == expected_data