From 6f82c9f59d6dd207b39d8e2c5d93ce9ece8d35aa Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 22 Jun 2026 19:05:10 +0000 Subject: [PATCH 1/6] fix: preserve 'NA' string values in CSV parsing instead of converting to None pandas.read_csv() by default treats strings like 'NA', 'N/A', 'NaN' as NaN values (via keep_default_na=True). This caused legitimate string values like 'NA' (North America) in Salesforce fields to be incorrectly converted to None/NULL during bulk API CSV parsing. Adding keep_default_na=False prevents pandas from interpreting these strings as missing values. Empty CSV fields are still correctly converted to None via explicit empty string replacement. Co-Authored-By: lucas.leadbetter@airbyte.io <5595530+lleadbet@users.noreply.github.com> --- .../extractors/response_to_file_extractor.py | 9 ++++++-- .../test_response_to_file_extractor.py | 22 +++++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py b/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py index eea4b80b2a..d476bb701a 100644 --- a/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py +++ b/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py @@ -137,10 +137,15 @@ def _read_with_chunks( try: with open(path, "r", encoding=file_encoding) as data: chunks = pd.read_csv( - data, chunksize=chunk_size, iterator=True, dialect="unix", dtype=object + data, + chunksize=chunk_size, + iterator=True, + dialect="unix", + dtype=object, + keep_default_na=False, ) for chunk in chunks: - chunk = chunk.replace({nan: None}).to_dict(orient="records") + chunk = chunk.replace({nan: None, "": None}).to_dict(orient="records") for row in chunk: yield row except pd.errors.EmptyDataError as e: diff --git a/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py b/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py index eeefe08174..94b5ab2045 100644 --- a/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py +++ b/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py @@ -39,6 +39,28 @@ def test_text_response_with_null_bytes(self) -> None: assert extracted_records == [{"FIRST_NAME": "a first name", "LAST_NAME": "a last name"}] + def test_na_string_values_are_preserved(self) -> None: + csv_with_na = '"region","name"\n"NA","North America"\n"EMEA","Europe"\n' + response = self._mock_streamed_response(BytesIO(csv_with_na.encode("utf-8"))) + + extracted_records = list(self._extractor.extract_records(response)) + + assert extracted_records == [ + {"region": "NA", "name": "North America"}, + {"region": "EMEA", "name": "Europe"}, + ] + + def test_empty_fields_are_converted_to_none(self) -> None: + csv_with_empty = '"region","name"\n"NA","North America"\n,""\n' + response = self._mock_streamed_response(BytesIO(csv_with_empty.encode("utf-8"))) + + extracted_records = list(self._extractor.extract_records(response)) + + assert extracted_records == [ + {"region": "NA", "name": "North America"}, + {"region": None, "name": None}, + ] + def _test_folder_path(self) -> Path: return Path(__file__).parent.resolve() From 6e4989f50919cb1b07472bb482464d1922ec9c67 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 23 Jun 2026 10:41:06 +0000 Subject: [PATCH 2/6] test: extend NA preservation test to cover all pandas default NA tokens Covers NA, N/A, NULL, None, nan, null, #N/A, per reviewer feedback. Co-Authored-By: lucas.leadbetter@airbyte.io <5595530+lleadbet@users.noreply.github.com> --- .../extractors/test_response_to_file_extractor.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py b/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py index 94b5ab2045..3b424bf64e 100644 --- a/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py +++ b/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py @@ -40,15 +40,17 @@ def test_text_response_with_null_bytes(self) -> None: assert extracted_records == [{"FIRST_NAME": "a first name", "LAST_NAME": "a last name"}] def test_na_string_values_are_preserved(self) -> None: - csv_with_na = '"region","name"\n"NA","North America"\n"EMEA","Europe"\n' + pandas_na_tokens = ["NA", "N/A", "NULL", "None", "nan", "null", "#N/A", ""] + csv_rows = "\n".join(f'"{token}","{token} value"' for token in pandas_na_tokens) + csv_with_na = f'"region","name"\n{csv_rows}\n' response = self._mock_streamed_response(BytesIO(csv_with_na.encode("utf-8"))) extracted_records = list(self._extractor.extract_records(response)) - assert extracted_records == [ - {"region": "NA", "name": "North America"}, - {"region": "EMEA", "name": "Europe"}, - ] + assert len(extracted_records) == len(pandas_na_tokens) + for record, token in zip(extracted_records, pandas_na_tokens): + assert record["region"] == token, f"Expected '{token}' to be preserved but got {record['region']}" + assert record["name"] == f"{token} value" def test_empty_fields_are_converted_to_none(self) -> None: csv_with_empty = '"region","name"\n"NA","North America"\n,""\n' From edc81bf0ab921f238318f6524ebad5ce9c3b9335 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 23 Jun 2026 10:41:58 +0000 Subject: [PATCH 3/6] style: fix ruff formatting Co-Authored-By: lucas.leadbetter@airbyte.io <5595530+lleadbet@users.noreply.github.com> --- .../declarative/extractors/test_response_to_file_extractor.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py b/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py index 3b424bf64e..ccdf417d49 100644 --- a/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py +++ b/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py @@ -49,7 +49,9 @@ def test_na_string_values_are_preserved(self) -> None: assert len(extracted_records) == len(pandas_na_tokens) for record, token in zip(extracted_records, pandas_na_tokens): - assert record["region"] == token, f"Expected '{token}' to be preserved but got {record['region']}" + assert record["region"] == token, ( + f"Expected '{token}' to be preserved but got {record['region']}" + ) assert record["name"] == f"{token} value" def test_empty_fields_are_converted_to_none(self) -> None: From 5d73b802548593a4215eb720185f4ebe733e8e6d Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Tue, 7 Jul 2026 18:51:20 +0300 Subject: [PATCH 4/6] fix: gate 'NA' string preservation behind opt-in preserve_na_values flag ResponseToFileExtractor previously always passed keep_default_na=False, which changed CSV parsing for every connector using it. Make the behavior opt-in via a new preserve_na_values field (default False), preserving the historical behavior (pandas NA tokens -> null) unless explicitly enabled. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../extractors/response_to_file_extractor.py | 14 ++++++++++-- .../test_response_to_file_extractor.py | 22 +++++++++++++++---- 2 files changed, 30 insertions(+), 6 deletions(-) diff --git a/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py b/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py index d476bb701a..a92b08ddeb 100644 --- a/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py +++ b/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py @@ -27,9 +27,14 @@ class ResponseToFileExtractor(RecordExtractor): Eventually, we want to support multiple file type by re-using the file based CDK parsers if possible. However, the lift is too high for a first iteration so we will only support CSV parsing using pandas as salesforce and sendgrid were doing. + + By default, pandas interprets strings such as "NA", "N/A", "null" and "NaN" as missing values and converts them to null. When + ``preserve_na_values`` is True, these strings are kept as-is (only empty cells are treated as null). This is opt-in to avoid changing + behavior for existing connectors. """ parameters: InitVar[Mapping[str, Any]] + preserve_na_values: bool = False def __post_init__(self, parameters: Mapping[str, Any]) -> None: self.logger = logging.getLogger("airbyte") @@ -136,16 +141,21 @@ def _read_with_chunks( try: with open(path, "r", encoding=file_encoding) as data: + # When preserving NA values, disable pandas' default NA parsing so strings like "NA"/"N/A" are kept as-is, + # and explicitly map empty cells to None. Otherwise, keep the historical behavior where pandas converts its + # default NA tokens (including empty cells) to NaN, which we then map to None. + read_kwargs = {"keep_default_na": False} if self.preserve_na_values else {} + na_replacements = {nan: None, "": None} if self.preserve_na_values else {nan: None} chunks = pd.read_csv( data, chunksize=chunk_size, iterator=True, dialect="unix", dtype=object, - keep_default_na=False, + **read_kwargs, ) for chunk in chunks: - chunk = chunk.replace({nan: None, "": None}).to_dict(orient="records") + chunk = chunk.replace(na_replacements).to_dict(orient="records") for row in chunk: yield row except pd.errors.EmptyDataError as e: diff --git a/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py b/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py index ccdf417d49..6294db49d8 100644 --- a/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py +++ b/unit_tests/sources/declarative/extractors/test_response_to_file_extractor.py @@ -15,6 +15,7 @@ class ResponseToFileExtractorTest(TestCase): def setUp(self) -> None: self._extractor = ResponseToFileExtractor({}) + self._preserving_extractor = ResponseToFileExtractor({}, preserve_na_values=True) self._http_mocker = requests_mock.Mocker() self._http_mocker.__enter__() @@ -39,13 +40,13 @@ def test_text_response_with_null_bytes(self) -> None: assert extracted_records == [{"FIRST_NAME": "a first name", "LAST_NAME": "a last name"}] - def test_na_string_values_are_preserved(self) -> None: + def test_na_string_values_are_preserved_when_enabled(self) -> None: pandas_na_tokens = ["NA", "N/A", "NULL", "None", "nan", "null", "#N/A", ""] csv_rows = "\n".join(f'"{token}","{token} value"' for token in pandas_na_tokens) csv_with_na = f'"region","name"\n{csv_rows}\n' response = self._mock_streamed_response(BytesIO(csv_with_na.encode("utf-8"))) - extracted_records = list(self._extractor.extract_records(response)) + extracted_records = list(self._preserving_extractor.extract_records(response)) assert len(extracted_records) == len(pandas_na_tokens) for record, token in zip(extracted_records, pandas_na_tokens): @@ -54,11 +55,24 @@ def test_na_string_values_are_preserved(self) -> None: ) assert record["name"] == f"{token} value" - def test_empty_fields_are_converted_to_none(self) -> None: + def test_na_string_values_are_converted_to_none_by_default(self) -> None: + pandas_na_tokens = ["NA", "N/A", "NULL", "None", "nan", "null", "#N/A", ""] + csv_rows = "\n".join(f'"{token}","preserved"' for token in pandas_na_tokens) + csv_with_na = f'"region","name"\n{csv_rows}\n' + response = self._mock_streamed_response(BytesIO(csv_with_na.encode("utf-8"))) + + extracted_records = list(self._extractor.extract_records(response)) + + assert len(extracted_records) == len(pandas_na_tokens) + for record in extracted_records: + assert record["region"] is None + assert record["name"] == "preserved" + + def test_empty_fields_are_converted_to_none_when_preserving(self) -> None: csv_with_empty = '"region","name"\n"NA","North America"\n,""\n' response = self._mock_streamed_response(BytesIO(csv_with_empty.encode("utf-8"))) - extracted_records = list(self._extractor.extract_records(response)) + extracted_records = list(self._preserving_extractor.extract_records(response)) assert extracted_records == [ {"region": "NA", "name": "North America"}, From cc11aaef1cef9e917295eab5a9d7f112878cec18 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Tue, 7 Jul 2026 19:06:05 +0300 Subject: [PATCH 5/6] fix: pass keep_default_na as direct bool to satisfy read_csv mypy overload Unpacking a dict[str, bool] into pd.read_csv broke mypy overload resolution. Pass keep_default_na=not self.preserve_na_values directly instead. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../declarative/extractors/response_to_file_extractor.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py b/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py index a92b08ddeb..f9a655db0c 100644 --- a/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py +++ b/airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py @@ -144,7 +144,6 @@ def _read_with_chunks( # When preserving NA values, disable pandas' default NA parsing so strings like "NA"/"N/A" are kept as-is, # and explicitly map empty cells to None. Otherwise, keep the historical behavior where pandas converts its # default NA tokens (including empty cells) to NaN, which we then map to None. - read_kwargs = {"keep_default_na": False} if self.preserve_na_values else {} na_replacements = {nan: None, "": None} if self.preserve_na_values else {nan: None} chunks = pd.read_csv( data, @@ -152,7 +151,7 @@ def _read_with_chunks( iterator=True, dialect="unix", dtype=object, - **read_kwargs, + keep_default_na=not self.preserve_na_values, ) for chunk in chunks: chunk = chunk.replace(na_replacements).to_dict(orient="records") From aec87f051d4aaac26b93a5f54a392b5d2a43f3ba Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Tue, 7 Jul 2026 23:46:45 +0300 Subject: [PATCH 6/6] feat: expose preserve_na_values in ResponseToFileExtractor manifest schema Add preserve_na_values to the declarative schema, generated model, and factory so low-code/manifest connectors can enable it via YAML (default false). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../declarative_component_schema.yaml | 5 +++++ .../models/declarative_component_schema.py | 5 +++++ .../parsers/model_to_component_factory.py | 5 ++++- .../test_model_to_component_factory.py | 20 +++++++++++++++++++ 4 files changed, 34 insertions(+), 1 deletion(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 7d99a01881..8a46a9e9be 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -1982,6 +1982,11 @@ definitions: type: type: string enum: [ResponseToFileExtractor] + preserve_na_values: + title: Preserve NA Values + description: When enabled, string values such as "NA", "N/A", "NULL", "None" and "NaN" are kept as-is instead of being interpreted as missing and converted to null. Empty cells are still treated as null. Defaults to false to preserve historical behavior. + type: boolean + default: false $parameters: type: object additionalProperties: true diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 3bee30ca68..d7fead317d 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -503,6 +503,11 @@ class Config: class ResponseToFileExtractor(BaseModel): type: Literal["ResponseToFileExtractor"] + preserve_na_values: Optional[bool] = Field( + False, + description='When enabled, string values such as "NA", "N/A", "NULL", "None" and "NaN" are kept as-is instead of being interpreted as missing and converted to null. Empty cells are still treated as null. Defaults to false to preserve historical behavior.', + title="Preserve NA Values", + ) parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 1e5801bb06..5a75912b36 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -2438,7 +2438,10 @@ def create_response_to_file_extractor( model: ResponseToFileExtractorModel, **kwargs: Any, ) -> ResponseToFileExtractor: - return ResponseToFileExtractor(parameters=model.parameters or {}) + return ResponseToFileExtractor( + parameters=model.parameters or {}, + preserve_na_values=model.preserve_na_values or False, + ) @staticmethod def create_exponential_backoff_strategy( diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index 5b39b069e5..5149f46d2b 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -5561,3 +5561,23 @@ def get_retriever(stream: Union[DeclarativeStream, DefaultStream]): if isinstance(stream, DeclarativeStream) else stream._stream_partition_generator._partition_factory._retriever ) + + +def test_create_response_to_file_extractor_preserve_na_values(): + from airbyte_cdk.sources.declarative.extractors import ResponseToFileExtractor + from airbyte_cdk.sources.declarative.models import ( + ResponseToFileExtractor as ResponseToFileExtractorModel, + ) + + factory = ModelToComponentFactory() + + default_component = factory.create_response_to_file_extractor( + ResponseToFileExtractorModel(type="ResponseToFileExtractor") + ) + assert isinstance(default_component, ResponseToFileExtractor) + assert default_component.preserve_na_values is False + + enabled_component = factory.create_response_to_file_extractor( + ResponseToFileExtractorModel(type="ResponseToFileExtractor", preserve_na_values=True) + ) + assert enabled_component.preserve_na_values is True