Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -136,11 +141,20 @@ 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.
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
data,
chunksize=chunk_size,
iterator=True,
dialect="unix",
dtype=object,
keep_default_na=not self.preserve_na_values,
)
for chunk in chunks:
chunk = chunk.replace({nan: 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:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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")


Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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__()

Expand All @@ -39,6 +40,45 @@ 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_when_enabled(self) -> None:
pandas_na_tokens = ["NA", "N/A", "NULL", "None", "nan", "null", "#N/A", "<NA>"]
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._preserving_extractor.extract_records(response))

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_na_string_values_are_converted_to_none_by_default(self) -> None:
pandas_na_tokens = ["NA", "N/A", "NULL", "None", "nan", "null", "#N/A", "<NA>"]
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._preserving_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()

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading