airbyte_cdk.sources.declarative.extractors

11@dataclass
12class TypeTransformer(ABC):
13    """
14    Abstract base class for implementing type transformation logic.
15
16    This class provides a blueprint for defining custom transformations
17    on data records based on a provided schema. Implementing classes
18    must override the `transform` method to specify the transformation
19    logic.
20
21    Attributes:
22        None explicitly defined, as this is a dataclass intended to be
23        subclassed.
24
25    Methods:
26        transform(record: Dict[str, Any], schema: Mapping[str, Any]) -> None:
27            Abstract method that must be implemented by subclasses.
28            It performs a transformation on a given data record based
29            on the provided schema.
30
31    Usage:
32        To use this class, create a subclass that implements the
33        `transform` method with the desired transformation logic.
34    """
35
36    @abstractmethod
37    def transform(
38        self,
39        record: Dict[str, Any],
40        schema: Mapping[str, Any],
41    ) -> None:
42        """
43        Perform a transformation on a data record based on a given schema.
44
45        Args:
46            record (Dict[str, Any]): The data record to be transformed.
47            schema (Mapping[str, Any]): The schema that dictates how
48                the record should be transformed.
49
50        Returns:
51            None
52
53        Raises:
54            NotImplementedError: If the method is not implemented
55                by a subclass.
56        """

14class HttpSelector:
15    """
16    Responsible for translating an HTTP response into a list of records by extracting records from the response and optionally filtering
17    records based on a heuristic.
18    """
19
20    @abstractmethod
21    def select_records(
22        self,
23        response: requests.Response,
24        stream_state: StreamState,
25        records_schema: Mapping[str, Any],
26        stream_slice: Optional[StreamSlice] = None,
27        next_page_token: Optional[Mapping[str, Any]] = None,
28    ) -> Iterable[Record]:
29        """
30        Selects records from the response
31        :param response: The response to select the records from
32        :param stream_state: The stream state
33        :param records_schema: json schema of records to return
34        :param stream_slice: The stream slice
35        :param next_page_token: The paginator token
36        :return: List of Records selected from the response
37        """
38        pass

 46@dataclass
 47class CombinedExtractor(RecordExtractor):
 48    """Combines the output of several record extractors into a single stream of records.
 49
 50    A single `DpathExtractor` can only describe one path into the response. Several connectors need
 51    more than that and reach for a custom `components.py` today, which keeps them out of the
 52    Connector Builder. This component covers the two shapes those connectors implement:
 53
 54    - `union` (default) — yield every record of every sub-extractor, in sub-extractor order. Use it
 55      when one response carries records under several paths, for example a GraphQL document that
 56      returns `issues.nodes` and `pullRequests.nodes` side by side.
 57    - `first_match` — yield the records of the first sub-extractor that produces at least one
 58      record, and skip the remaining sub-extractors. If no sub-extractor produces a record, nothing
 59      is yielded. Use it when an API answers with one of several alternative shapes. The winning
 60      sub-extractor is only peeked at, never restarted: the peeked record is chained back in front
 61      of the remaining ones, so large responses are still streamed lazily and no sub-extractor is
 62      ever materialized into a list.
 63
 64    Examples of instantiating this component:
 65    ```
 66      extractor:
 67        type: CombinedExtractor
 68        mode: union
 69        extractors:
 70          - type: DpathExtractor
 71            field_path: ["data", "repository", "issues", "nodes"]
 72          - type: DpathExtractor
 73            field_path: ["data", "repository", "pullRequests", "nodes"]
 74    ```
 75
 76    ```
 77      extractor:
 78        type: CombinedExtractor
 79        mode: first_match
 80        extractors:
 81          - type: DpathExtractor
 82            field_path: ["data", "boards", "*", "items_page", "items", "*"]
 83          - type: DpathExtractor
 84            field_path: ["data", "next_items_page", "items"]
 85    ```
 86
 87    ## Streaming decoders are rejected
 88
 89    Every sub-extractor is handed the same `requests.Response`, so the response body must be
 90    readable more than once. That holds for the decoders that buffer the whole body, which is the
 91    common case: `JsonDecoder` and any `CompositeRawDecoder` built with `stream_response=False`
 92    read `response.content`, `XmlDecoder` reads `response.text` and `ZipfileDecoder` reads
 93    `response.content` — `requests` caches all of those, so every sub-extractor sees the full body.
 94
 95    It does NOT hold for streaming decoders, i.e. a `CompositeRawDecoder` with
 96    `stream_response=True`, which is what `CsvDecoder`, `JsonlDecoder`, `JsonItemsDecoder` and
 97    `GzipDecoder` resolve to outside the Connector Builder, nor for `IterableDecoder`. Those read
 98    `response.raw` and close it afterwards, so the first sub-extractor drains the stream and the
 99    later ones read a dead body. That loss is SILENT rather than loud: `requests` puts a
100    `urllib3.HTTPResponse` in `response.raw`, and reading a closed `urllib3.HTTPResponse` returns
101    an empty body instead of raising, so `union` emits only the first sub-extractor's records and
102    `first_match` emits nothing at all when the first path misses.
103
104    Because of that, `ModelToComponentFactory.create_combined_extractor` refuses to build this
105    component over a streaming decoder and raises a configuration error, naming the decoder in the
106    internal message. The Connector Builder forces those same decoders to `stream_response=False`,
107    so the rejection is applied there too rather than letting a manifest test-read correctly and
108    lose records once published.
109
110    ## Empty records
111
112    `skip_empty_records` drops falsy records — `None`, `{}`, `[]`, `""` — before any of the modes
113    sees them. It is off by default, so a sub-extractor's output is passed through verbatim.
114
115    Turn it on when an API can answer with nulls in the middle of a record list. A GraphQL API
116    that returns a partial response puts `null` in the `data` array for the fields it failed to
117    resolve and describes the failure in a sibling `errors` field; without this flag those nulls
118    reach the record stream and fail schema validation downstream.
119
120    Under `first_match` the flag also changes which sub-extractor wins: a sub-extractor whose
121    records are all empty no longer counts as a match, so the next sub-extractor is tried. That is
122    the behaviour of `source-monday`'s `MondayIncrementalItemsExtractor`, where a page of nulls
123    must fall through to the pagination path rather than lock in the primary path.
124
125    The dropped records still count toward the size of the page, see "Record-counting paginators"
126    below.
127
128    ## Cost
129
130    Each sub-extractor decodes the response independently — `CompositeRawDecoder` re-parses the
131    cached body on every `decode()` call — so a response is parsed once per sub-extractor.
132    Combining three extractors over a large response costs roughly three times the decode time of
133    a single one.
134
135    ## Record-counting paginators
136
137    The count `OffsetIncrement` and `PageIncrement` obtain is the combined count, not the API's
138    page size:
139
140    - `union` returns the SUM over all sub-extractors. An `OffsetIncrement` advances the offset by
141      that sum, so the next page starts past the records that were never read — two sub-extractors
142      returning two records each with `page_size: 2` request offsets 0, 4, 8 instead of 0, 2, 4 and
143      silently drop two thirds of the records. `ModelToComponentFactory.create_offset_increment`
144      therefore rejects a `union` `CombinedExtractor` (nested ones included) with a configuration
145      error. A `PageIncrement` only compares the count against `page_size` to decide whether to
146      stop, so it loses nothing, but it issues one extra request whenever the summed count of the
147      last page happens to equal `page_size`.
148    - `first_match` returns the count of the winning sub-extractor, which does not inflate the
149      count, so it works with either paginator as long as the path the count comes from is the one
150      the API paginates over.
151
152    Empty records dropped by `skip_empty_records` are part of that count. The API returned them as
153    slots of the page, so a full page holding a null must still read as full: counting only the
154    surviving records would make both paginators stop at the first page with a null, and would
155    make an `OffsetIncrement` re-request records it already read. The factory sets
156    `count_dropped_empty_records` on the extractor of a `RecordSelector`, which makes it yield a
157    placeholder for every empty record it drops; the `RecordSelector` counts the placeholders and
158    discards them. The winner under `first_match` is still chosen by the surviving records alone,
159    so the placeholders never change which records are emitted.
160
161    Attributes:
162        extractors (List[RecordExtractor]): The sub-extractors to combine. At least one is required.
163        mode (CombineMode): How the sub-extractor outputs are combined. Defaults to `union`.
164        skip_empty_records (bool): Whether falsy records are dropped before they are combined.
165            Defaults to `False`.
166        count_dropped_empty_records (bool): Set by the factory on the extractor of a
167            `RecordSelector`, which discards the placeholders: yield a placeholder in place of
168            every dropped empty record so it counts toward the page size. Defaults to `False`.
169    """
170
171    extractors: List[RecordExtractor]
172    parameters: InitVar[Mapping[str, Any]]
173    mode: CombineMode = CombineMode.union
174    skip_empty_records: bool = False
175    count_dropped_empty_records: bool = False
176
177    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
178        if not self.extractors:
179            raise ValueError(
180                "CombinedExtractor requires at least one extractor in its `extractors` field."
181            )
182        if not isinstance(self.mode, CombineMode):
183            self.mode = CombineMode(self.mode)
184
185    def extract_records(self, response: requests.Response) -> Iterable[Mapping[str, Any]]:
186        if self.mode == CombineMode.first_match:
187            yield from self._extract_first_match(response)
188        else:
189            yield from self._extract_union(response)
190
191    def _extract_union(self, response: requests.Response) -> Iterable[Mapping[str, Any]]:
192        for index in range(len(self.extractors)):
193            yield from self._records_of(index, response)
194
195    def _extract_first_match(self, response: requests.Response) -> Iterable[Mapping[str, Any]]:
196        for index in range(len(self.extractors)):
197            records: Iterator[Mapping[str, Any]] = iter(self._records_of(index, response))
198            # Placeholders for dropped empty records do not make a sub-extractor win, but they are
199            # kept so the winner's count still includes the ones that came before its first record.
200            dropped_placeholders: List[Mapping[str, Any]] = []
201            first_record: Union[Mapping[str, Any], _NoRecord] = _NO_RECORD
202            for record in records:
203                if isinstance(record, _DroppedEmptyRecord):
204                    dropped_placeholders.append(record)
205                    continue
206                first_record = record
207                break
208            if isinstance(first_record, _NoRecord):
209                continue
210            # Chain the peeked record back instead of restarting the extractor: the rest of the
211            # records stay lazy and the response is never read twice.
212            yield from chain(dropped_placeholders, [first_record], records)
213            return
214
215    def _records_of(
216        self, extractor_index: int, response: requests.Response
217    ) -> Iterable[Mapping[str, Any]]:
218        records = self.extractors[extractor_index].extract_records(response)
219        if not self.skip_empty_records:
220            return records
221        return self._without_empty_records(records, extractor_index)
222
223    def _without_empty_records(
224        self, records: Iterable[Mapping[str, Any]], extractor_index: int
225    ) -> Iterator[Mapping[str, Any]]:
226        dropped = 0
227        for record in records:
228            if not record:
229                dropped += 1
230                if self.count_dropped_empty_records:
231                    yield _DROPPED_EMPTY_RECORD  # type: ignore[misc]  # only ever counted, never emitted
232                continue
233            yield record
234        if dropped:
235            logger.warning(
236                "CombinedExtractor dropped %s empty record(s) yielded by sub-extractor %s (%s) "
237                "because `skip_empty_records` is enabled. An API that returns empty records "
238                "alongside real ones usually reports the reason in an error field of the "
239                "response body.",
240                dropped,
241                extractor_index,
242                type(self.extractors[extractor_index]).__name__,
243            )

19class CombineMode(Enum):
20    """
21    How a `CombinedExtractor` combines the output of its sub-extractors.
22    """
23
24    union = "union"
25    first_match = "first_match"

 19@dataclass
 20class DpathExtractor(RecordExtractor):
 21    """
 22    Record extractor that searches a decoded response over a path defined as an array of fields.
 23
 24    If the field path points to an array, that array is returned.
 25    If the field path points to an object, that object is returned wrapped as an array.
 26    If the field path points to an empty object, an empty array is returned.
 27    If the field path points to a non-existing path, an empty array is returned.
 28
 29    Optionally, records can be expanded by providing a RecordExpander component.
 30    When record_expander is configured, each extracted record is passed through the
 31    expander which extracts items from nested array fields and emits each item as a
 32    separate record.
 33
 34    Examples of instantiating this transform:
 35    ```
 36      extractor:
 37        type: DpathExtractor
 38        field_path:
 39          - "root"
 40          - "data"
 41    ```
 42
 43    ```
 44      extractor:
 45        type: DpathExtractor
 46        field_path:
 47          - "root"
 48          - "{{ parameters['field'] }}"
 49    ```
 50
 51    ```
 52      extractor:
 53        type: DpathExtractor
 54        field_path: []
 55    ```
 56
 57    ```
 58      extractor:
 59        type: DpathExtractor
 60        field_path:
 61          - "data"
 62          - "object"
 63        record_expander:
 64          type: RecordExpander
 65          expand_records_from_field:
 66            - "lines"
 67            - "data"
 68          remain_original_record: true
 69    ```
 70
 71    Attributes:
 72        field_path (Union[InterpolatedString, str]): Path to the field that should be extracted
 73        config (Config): The user-provided configuration as specified by the source's spec
 74        decoder (Decoder): The decoder responsible to transfom the response in a Mapping
 75        record_expander (Optional[RecordExpander]): Optional component to expand records by extracting items from nested array fields
 76    """
 77
 78    field_path: List[Union[InterpolatedString, str]]
 79    config: Config
 80    parameters: InitVar[Mapping[str, Any]]
 81    decoder: Decoder = field(default_factory=lambda: JsonDecoder(parameters={}))
 82    record_expander: Optional[RecordExpander] = None
 83
 84    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 85        self._field_path = [
 86            InterpolatedString.create(path, parameters=parameters) for path in self.field_path
 87        ]
 88        for path_index in range(len(self.field_path)):
 89            if isinstance(self.field_path[path_index], str):
 90                self._field_path[path_index] = InterpolatedString.create(
 91                    self.field_path[path_index], parameters=parameters
 92                )
 93
 94    def extract_records(self, response: requests.Response) -> Iterable[MutableMapping[Any, Any]]:
 95        for body in self.decoder.decode(response):
 96            if len(self._field_path) == 0:
 97                extracted = body
 98            else:
 99                path = [path.eval(self.config) for path in self._field_path]
100                if "*" in path:
101                    extracted = dpath.values(body, path)
102                else:
103                    extracted = dpath.get(body, path, default=[])  # type: ignore # extracted will be a MutableMapping, given input data structure
104            if isinstance(extracted, list):
105                if not self.record_expander:
106                    yield from extracted
107                else:
108                    for record in extracted:
109                        yield from self.record_expander.expand_record(record)
110            elif extracted:
111                if self.record_expander:
112                    yield from self.record_expander.expand_record(extracted)
113                else:
114                    yield extracted
115            else:
116                yield from []

13@dataclass
14class RecordFilter:
15    """
16    Filter applied on a list of Records
17
18    config (Config): The user-provided configuration as specified by the source's spec
19    condition (str): The string representing the predicate to filter a record. Records will be removed if evaluated to False
20    """
21
22    parameters: InitVar[Mapping[str, Any]]
23    config: Config
24    condition: str = ""
25
26    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
27        self._filter_interpolator = InterpolatedBoolean(
28            condition=self.condition, parameters=parameters
29        )
30
31    def filter_records(
32        self,
33        records: Iterable[Mapping[str, Any]],
34        stream_state: StreamState,
35        stream_slice: Optional[StreamSlice] = None,
36        next_page_token: Optional[Mapping[str, Any]] = None,
37    ) -> Iterable[Mapping[str, Any]]:
38        kwargs = {
39            "stream_state": stream_state,
40            "stream_slice": stream_slice,
41            "next_page_token": next_page_token,
42            "stream_slice.extra_fields": stream_slice.extra_fields if stream_slice else {},
43        }
44        for record in records:
45            if self._filter_interpolator.eval(self.config, record=record, **kwargs):
46                yield record

 39@dataclass
 40class RecordSelector(HttpSelector):
 41    """
 42    Responsible for translating an HTTP response into a list of records by extracting records from the response and optionally filtering
 43    records based on a heuristic.
 44
 45    Attributes:
 46        extractor (RecordExtractor): The record extractor responsible for extracting records from a response
 47        schema_normalization (TypeTransformer): The record normalizer responsible for casting record values to stream schema types
 48        record_filter (RecordFilter): The record filter responsible for filtering extracted records
 49        transformations (List[RecordTransformation]): The transformations to be done on the records
 50    """
 51
 52    extractor: RecordExtractor
 53    config: Config
 54    parameters: InitVar[Mapping[str, Any]]
 55    schema_normalization: Union[TypeTransformer, DeclarativeTypeTransformer]
 56    name: str
 57    _name: Union[InterpolatedString, str] = field(init=False, repr=False, default="")
 58    record_filter: Optional[RecordFilter] = None
 59    transformations: List[RecordTransformation] = field(default_factory=lambda: [])
 60    transform_before_filtering: bool = False
 61    file_uploader: Optional[DefaultFileUploader] = None
 62
 63    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 64        self._parameters = parameters
 65        self._name = (
 66            InterpolatedString(self._name, parameters=parameters)
 67            if isinstance(self._name, str)
 68            else self._name
 69        )
 70
 71    @property  # type: ignore
 72    def name(self) -> str:
 73        """
 74        :return: Stream name
 75        """
 76        return (
 77            str(self._name.eval(self.config))
 78            if isinstance(self._name, InterpolatedString)
 79            else self._name
 80        )
 81
 82    @name.setter
 83    def name(self, value: str) -> None:
 84        if not isinstance(value, property):
 85            self._name = value
 86
 87    def select_records(
 88        self,
 89        response: requests.Response,
 90        stream_state: StreamState,
 91        records_schema: Mapping[str, Any],
 92        stream_slice: Optional[StreamSlice] = None,
 93        next_page_token: Optional[Mapping[str, Any]] = None,
 94    ) -> Iterable[Record]:
 95        """
 96        Selects records from the response
 97        :param response: The response to select the records from
 98        :param stream_state: The stream state
 99        :param records_schema: json schema of records to return
100        :param stream_slice: The stream slice
101        :param next_page_token: The paginator token
102        :return: List of Records selected from the response
103        """
104        all_data: Iterable[Mapping[str, Any]] = self._extract_and_count(response)
105        yield from self.filter_and_transform(
106            all_data, stream_state, records_schema, stream_slice, next_page_token
107        )
108        # A custom record filter can stop reading before the end of the page, which would leave the
109        # page uncounted.
110        for _ in all_data:
111            pass
112
113    def _extract_and_count(self, response: requests.Response) -> Iterable[Mapping[str, Any]]:
114        count = 0
115        for record in self.extractor.extract_records(response):
116            count += 1
117            if not isinstance(record, _DroppedEmptyRecord):
118                yield record
119        # Kept on the response, not on the selector: a stream has one retriever, and so one
120        # selector, reading all of its partitions concurrently.
121        setattr(response, _EXTRACTED_RECORD_COUNT, count)
122
123    def filter_and_transform(
124        self,
125        all_data: Iterable[Mapping[str, Any]],
126        stream_state: StreamState,
127        records_schema: Mapping[str, Any],
128        stream_slice: Optional[StreamSlice] = None,
129        next_page_token: Optional[Mapping[str, Any]] = None,
130    ) -> Iterable[Record]:
131        """
132        There is an issue with the selector as of 2024-08-30: it does technology-agnostic processing like filtering, transformation and
133        normalization with an API that is technology-specific (as requests.Response is only for HTTP communication using the requests
134        library).
135
136        Until we decide to move this logic away from the selector, we made this method public so that users like AsyncJobRetriever could
137        share the logic of doing transformations on a set of records.
138        """
139        if self.transform_before_filtering:
140            transformed_data = self._transform(all_data, stream_state, stream_slice)
141            transformed_filtered_data = self._filter(
142                transformed_data, stream_state, stream_slice, next_page_token
143            )
144        else:
145            filtered_data = self._filter(all_data, stream_state, stream_slice, next_page_token)
146            transformed_filtered_data = self._transform(filtered_data, stream_state, stream_slice)
147        normalized_data = self._normalize_by_schema(
148            transformed_filtered_data, schema=records_schema
149        )
150        for data in normalized_data:
151            record = Record(data=data, stream_name=self.name, associated_slice=stream_slice)
152            if self.file_uploader:
153                self.file_uploader.upload(record)
154            yield record
155
156    def _normalize_by_schema(
157        self, records: Iterable[Mapping[str, Any]], schema: Optional[Mapping[str, Any]]
158    ) -> Iterable[Mapping[str, Any]]:
159        if schema:
160            # record has type Mapping[str, Any], but dict[str, Any] expected
161            for record in records:
162                normalized_record = dict(record)
163                self.schema_normalization.transform(normalized_record, schema)
164                yield normalized_record
165        else:
166            yield from records
167
168    def _filter(
169        self,
170        records: Iterable[Mapping[str, Any]],
171        stream_state: StreamState,
172        stream_slice: Optional[StreamSlice],
173        next_page_token: Optional[Mapping[str, Any]],
174    ) -> Iterable[Mapping[str, Any]]:
175        if self.record_filter:
176            yield from self.record_filter.filter_records(
177                records,
178                stream_state=stream_state,
179                stream_slice=stream_slice,
180                next_page_token=next_page_token,
181            )
182        else:
183            yield from records
184
185    def _transform(
186        self,
187        records: Iterable[Mapping[str, Any]],
188        stream_state: StreamState,
189        stream_slice: Optional[StreamSlice] = None,
190    ) -> Iterable[Mapping[str, Any]]:
191        for record in records:
192            for transformation in self.transformations:
193                transformation.transform(
194                    record,  # type: ignore  # record has type Mapping[str, Any], but Dict[str, Any] expected
195                    config=self.config,
196                    stream_state=stream_state,
197                    stream_slice=stream_slice,
198                )
199            yield record

 23@dataclass
 24class ResponseToFileExtractor(RecordExtractor):
 25    """
 26    This class is used when having very big HTTP responses (usually streamed) which would require too much memory so we use disk space as
 27    a tradeoff.
 28
 29    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
 30    a first iteration so we will only support CSV parsing using pandas as salesforce and sendgrid were doing.
 31
 32    By default, pandas interprets strings such as "NA", "N/A", "null" and "NaN" as missing values and converts them to null. When
 33    ``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
 34    behavior for existing connectors.
 35    """
 36
 37    parameters: InitVar[Mapping[str, Any]]
 38    preserve_na_values: bool = False
 39
 40    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 41        self.logger = logging.getLogger("airbyte")
 42
 43    def _get_response_encoding(self, headers: Dict[str, Any]) -> str:
 44        """
 45        Get the encoding of the response based on the provided headers. This method is heavily inspired by the requests library
 46        implementation.
 47
 48        Args:
 49            headers (Dict[str, Any]): The headers of the response.
 50        Returns:
 51            str: The encoding of the response.
 52        """
 53
 54        content_type = headers.get("content-type")
 55
 56        if not content_type:
 57            return DEFAULT_ENCODING
 58
 59        content_type, params = requests.utils.parse_header_links(content_type)
 60
 61        if "charset" in params:
 62            return params["charset"].strip("'\"")  # type: ignore  # we assume headers are returned as str
 63
 64        return DEFAULT_ENCODING
 65
 66    def _filter_null_bytes(self, b: bytes) -> bytes:
 67        """
 68        Filter out null bytes from a bytes object.
 69
 70        Args:
 71            b (bytes): The input bytes object.
 72        Returns:
 73            bytes: The filtered bytes object with null bytes removed.
 74
 75        Referenced Issue:
 76            https://github.com/airbytehq/airbyte/issues/8300
 77        """
 78
 79        res = b.replace(b"\x00", b"")
 80        if len(res) < len(b):
 81            self.logger.warning(
 82                "Filter 'null' bytes from string, size reduced %d -> %d chars", len(b), len(res)
 83            )
 84        return res
 85
 86    def _save_to_file(self, response: requests.Response) -> Tuple[str, str]:
 87        """
 88        Saves the binary data from the given response to a temporary file and returns the filepath and response encoding.
 89
 90        Args:
 91            response (Optional[requests.Response]): The response object containing the binary data. Defaults to None.
 92
 93        Returns:
 94            Tuple[str, str]: A tuple containing the filepath of the temporary file and the response encoding.
 95
 96        Raises:
 97            ValueError: If the temporary file does not exist after saving the binary data.
 98        """
 99        # set filepath for binary data from response
100        decompressor = zlib.decompressobj(zlib.MAX_WBITS | 32)
101        needs_decompression = True  # we will assume at first that the response is compressed and change the flag if not
102
103        tmp_file = str(uuid.uuid4())
104        with closing(response) as response, open(tmp_file, "wb") as data_file:
105            response_encoding = self._get_response_encoding(dict(response.headers or {}))
106            for chunk in response.iter_content(chunk_size=DOWNLOAD_CHUNK_SIZE):
107                try:
108                    if needs_decompression:
109                        data_file.write(decompressor.decompress(chunk))
110                        needs_decompression = True
111                    else:
112                        data_file.write(self._filter_null_bytes(chunk))
113                except zlib.error:
114                    data_file.write(self._filter_null_bytes(chunk))
115                    needs_decompression = False
116
117        # check the file exists
118        if os.path.isfile(tmp_file):
119            return tmp_file, response_encoding
120        else:
121            raise ValueError(
122                f"The IO/Error occured while verifying binary data. Tmp file {tmp_file} doesn't exist."
123            )
124
125    def _read_with_chunks(
126        self, path: str, file_encoding: str, chunk_size: int = 100
127    ) -> Iterable[Mapping[str, Any]]:
128        """
129        Reads data from a file in chunks and yields each row as a dictionary.
130
131        Args:
132            path (str): The path to the file to be read.
133            file_encoding (str): The encoding of the file.
134            chunk_size (int, optional): The size of each chunk to be read. Defaults to 100.
135
136        Yields:
137            Mapping[str, Any]: A dictionary representing each row of data.
138
139        Raises:
140            ValueError: If an IO/Error occurs while reading the temporary data.
141        """
142
143        try:
144            with open(path, "r", encoding=file_encoding) as data:
145                # When preserving NA values, disable pandas' default NA parsing so strings like "NA"/"N/A" are kept as-is,
146                # and explicitly map empty cells to None. Otherwise, keep the historical behavior where pandas converts its
147                # default NA tokens (including empty cells) to NaN, which we then map to None.
148                na_replacements = {nan: None, "": None} if self.preserve_na_values else {nan: None}
149                chunks = pd.read_csv(
150                    data,
151                    chunksize=chunk_size,
152                    iterator=True,
153                    dialect="unix",
154                    dtype=object,
155                    keep_default_na=not self.preserve_na_values,
156                )
157                for chunk in chunks:
158                    chunk = chunk.replace(na_replacements).to_dict(orient="records")
159                    for row in chunk:
160                        yield row
161        except pd.errors.EmptyDataError as e:
162            self.logger.info(f"Empty data received. {e}")
163            yield from []
164        except IOError as ioe:
165            raise ValueError(f"The IO/Error occured while reading tmp data. Called: {path}", ioe)
166        finally:
167            # remove binary tmp file, after data is read
168            os.remove(path)
169
170    def extract_records(
171        self, response: Optional[requests.Response] = None
172    ) -> Iterable[Mapping[str, Any]]:
173        """
174        Extracts records from the given response by:
175            1) Saving the result to a tmp file
176            2) Reading from saved file by chunks to avoid OOM
177
178        Args:
179            response (Optional[requests.Response]): The response object containing the data. Defaults to None.
180
181        Yields:
182            Iterable[Mapping[str, Any]]: An iterable of mappings representing the extracted records.
183
184        Returns:
185            None
186        """
187        if response:
188            file_path, encoding = self._save_to_file(response)
189            yield from self._read_with_chunks(file_path, encoding)
190        else:
191            yield from []