airbyte_cdk.sources.declarative.expanders

85class OnNoRecords(Enum):
86    """
87    Behavior when record expansion produces no records.
88    """
89
90    skip = "skip"
91    emit_parent = "emit_parent"

 94@dataclass
 95class ParentFieldPath:
 96    """One field to copy from the record being expanded onto each expanded item.
 97
 98    `parent_path` locates the value on the parent, `record_path` says where to put it on the
 99    child. Both are lists, so a value nested inside the parent can be copied into a nested
100    position on the child. Glob metacharacters are rejected: each path must identify a single
101    field.
102    """
103
104    parent_path: Sequence[str]
105    record_path: Sequence[str]
106    config: Config
107    parameters: InitVar[Mapping[str, Any]]
108
109    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
110        if not self.parent_path:
111            raise ValueError("`parent_path` cannot be empty.")
112        if not self.record_path:
113            raise ValueError("`record_path` cannot be empty.")
114        self._parent_path: list[InterpolatedString] = [
115            InterpolatedString.create(path, parameters=parameters) for path in self.parent_path
116        ]
117        self._record_path: list[InterpolatedString] = [
118            InterpolatedString.create(path, parameters=parameters) for path in self.record_path
119        ]
120        # The paths only interpolate `config`, so they are evaluated once instead of per item.
121        self._evaluated_parent_path = self.evaluated_parent_path()
122        self._evaluated_record_path = self.evaluated_record_path()
123        RecordExpander._reject_globs(self._evaluated_parent_path, "parent_path")
124        RecordExpander._reject_globs(self._evaluated_record_path, "record_path")
125
126    def evaluated_parent_path(self) -> list[Any]:
127        return [segment.eval(self.config) for segment in self._parent_path]
128
129    def evaluated_record_path(self) -> list[Any]:
130        return [segment.eval(self.config) for segment in self._record_path]
131
132    def copy_onto(
133        self, parent_record: Mapping[str, Any], child_record: MutableMapping[str, Any]
134    ) -> None:
135        """Copy the parent value onto the child, overwriting whatever was there.
136
137        A `parent_path` that the parent does not have copies `None`, which is what the custom
138        connector classes this replaces do (`parent.get(field)`). Distinguishing "absent" from
139        "present and null" would need a third option and no connector needs one.
140
141        A container value is deep-copied, so that a downstream transformation writing inside it
142        cannot reach the parent record or the items expanded from it alongside this one. Both the
143        lookup and the copy cost proportionally to the named value, not to the whole parent.
144
145        A `record_path` running through a value the item holds as a scalar or a list cannot be
146        written and fails the sync as a config error naming the path.
147        """
148        value = _lookup(parent_record, self._evaluated_parent_path)
149        copied = None if value is _MISSING else _detached(value)
150        try:
151            dpath.new(child_record, self._evaluated_record_path, copied)
152        except (dpath.exceptions.PathNotFound, TypeError) as error:
153            message = (
154                f"Cannot write parent field to `record_path` {self._evaluated_record_path}: "
155                "the expanded item already holds a non-object value on that path."
156            )
157            raise AirbyteTracedException(
158                message=message,
159                internal_message=f"{message} {error!r}",
160                failure_type=FailureType.config_error,
161            ) from error

164@dataclass
165class RecordExpander:
166    """Expands records by extracting items from a nested array field.
167
168    When configured, this component extracts items from a specified nested array path
169    within each record and emits each item as a separate record. Set `remain_original_record: true`
170    to embed the full parent record under `original_record` in each expanded item when you need
171    downstream transformations to access parent context.
172
173    When only a few parent fields are needed, prefer `parent_fields` over
174    `remain_original_record`: it copies the named values onto each expanded item instead of
175    deep-copying the whole parent once per item, which matters when the parent is large and the
176    nested list is long.
177
178    Set `merge_parent: true` to flatten the parent into each item instead: the parent's fields,
179    minus the expanded list, are shallow-merged underneath the item's own fields, so the item wins
180    on any key both have.
181
182    The expand_records_from_field path supports wildcards (*) for matching multiple arrays.
183    When wildcards are used, items from all matched arrays are extracted and emitted.
184
185    Examples of instantiating this component:
186    ```
187      record_expander:
188        type: RecordExpander
189        expand_records_from_field:
190          - "lines"
191          - "data"
192        remain_original_record: true
193    ```
194
195    ```
196      record_expander:
197        type: RecordExpander
198        expand_records_from_field:
199          - "reviews"
200          - "nodes"
201        parent_fields:
202          - parent_path: ["url"]
203            record_path: ["pull_request_url"]
204    ```
205
206    ```
207      record_expander:
208        type: RecordExpander
209        expand_records_from_field:
210          - "activity"
211        merge_parent: true
212    ```
213
214    ```
215      record_expander:
216        type: RecordExpander
217        expand_records_from_field:
218          - "sections"
219          - "*"
220          - "items"
221        on_no_records: emit_parent
222    ```
223
224    Attributes:
225        expand_records_from_field: Path to a nested array field within each record.
226            Items from this array will be extracted and emitted as separate records.
227            Supports wildcards (*).
228        remain_original_record: If True, each expanded record will include the original
229            parent record in an "original_record" field. Defaults to False.
230        parent_fields: Named values to copy from the record being expanded onto each expanded
231            item. Each entry has a `parent_path` and a `record_path`; an existing value at
232            `record_path` is overwritten, and a `parent_path` the parent does not have copies
233            `None`. A copied container is deep-copied, so writing into it downstream does not
234            reach the parent or the sibling items. Independent of `remain_original_record` -
235            both may be set, though copying named fields is the cheaper way to carry parent
236            context. Applies to items fetched through `truncated_list_retriever` as well as to
237            embedded ones. Glob metacharacters are rejected in both paths.
238        merge_parent: If True, each expanded item is the parent record shallow-merged with the
239            item, the item's own keys winning on collision, and the expanded list removed from
240            the parent's copy. Only the value at `expand_records_from_field` is removed: for a
241            multi-segment path the top-level key stays with its other fields. Each item gets its
242            own deep copy of the merged parent, so a downstream transformation writing into a
243            nested value affects only that item. The merge happens before `parent_fields` are
244            copied and before `original_record` is embedded, so both can overwrite merged
245            values. Applies to items fetched through `truncated_list_retriever` as well as to
246            embedded ones. Defaults to False.
247        on_no_records: Behavior when expansion produces no records. "skip" (default)
248            emits nothing. "emit_parent" emits the original parent record unchanged.
249        truncation_indicator_path: Path within each record to a field indicating that the
250            embedded nested list is truncated (e.g. a `has_more` flag on the list object).
251            When the indicator is truthy and no `truncated_list_retriever` is configured, the
252            embedded items are expanded as normal and a WARNING is logged (once per stream
253            instance) describing the expansion path and the embedded item count, so that the
254            data loss is visible instead of silent. Glob metacharacters (`*`, `?`, `[`) are
255            rejected in this path, and in `expand_records_from_field` when a retriever is
256            configured; the check runs on the interpolated values.
257        truncated_list_retriever: Retriever used to fetch the complete list of items when
258            the field at `truncation_indicator_path` is truthy. The record being expanded is
259            exposed to the retriever's interpolation context as `stream_slice['parent_record']`.
260            One fetch is issued per truncated parent record; enable `use_cache` on its requester
261            when the same list can be fetched repeatedly. Without a `paginator` only the first
262            page is read. If the retriever returns no records, the embedded items are expanded
263            as a fallback; if it returns fewer records than the `total_count` field next to the
264            indicator, a WARNING is logged once per stream instance. Request failures surface
265            through the retriever's error handler and fail the stream like any other request.
266            `$parameters` of the enclosing stream propagate into this retriever's components.
267            In Connector Builder test reads the page limit applies to each fetch independently,
268            so the fetched list may be shorter than `total_count`; no incomplete-fetch warning
269            is emitted there.
270        message_repository: Optional repository through which the truncation warnings are emitted
271            as Airbyte LOG messages so they are visible in the Connector Builder. When it is not
272            set, the warnings go to the `airbyte` logger instead.
273        suppress_incomplete_fetch_warning: Skip the incomplete-fetch WARNING. Set by the factory
274            for Connector Builder test reads when the page limit caps the retriever's paginator,
275            so a shortfall against `total_count` is expected. The truncated-without-retriever
276            warning is not affected.
277        config: The user-provided configuration as specified by the source's spec.
278    """
279
280    expand_records_from_field: Sequence[str]
281    config: Config
282    parameters: InitVar[Mapping[str, Any]]
283    remain_original_record: bool = False
284    on_no_records: OnNoRecords = OnNoRecords.skip
285    truncation_indicator_path: Optional[Sequence[str]] = None
286    truncated_list_retriever: Optional["Retriever"] = None
287    message_repository: Optional[MessageRepository] = None
288    suppress_incomplete_fetch_warning: bool = False
289    # Appended after the pre-existing fields so positional construction keeps its meaning.
290    parent_fields: Optional[Sequence[ParentFieldPath]] = None
291    merge_parent: bool = False
292
293    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
294        self._expand_path: list[InterpolatedString] = [
295            InterpolatedString.create(path, parameters=parameters)
296            for path in self.expand_records_from_field
297        ]
298        if self.truncated_list_retriever and not self.truncation_indicator_path:
299            raise ValueError(
300                "`truncation_indicator_path` is required when `truncated_list_retriever` is configured."
301            )
302        self._truncation_indicator_path: list[InterpolatedString] = [
303            InterpolatedString.create(path, parameters=parameters)
304            for path in (self.truncation_indicator_path or [])
305        ]
306        # The paths only interpolate `config`, so their evaluated values are validated up front.
307        self._reject_globs(self._evaluated_indicator_path(), "truncation_indicator_path")
308        if self.truncated_list_retriever:
309            self._reject_globs(self._evaluated_expand_path(), "expand_records_from_field")
310        self._warning_lock = threading.Lock()
311        self._warned_truncation_without_retriever = False
312        self._warned_incomplete_fetch = False
313
314    @staticmethod
315    def _reject_globs(path: Sequence[Any], field_name: str) -> None:
316        if any(
317            isinstance(segment, str) and any(char in segment for char in _GLOB_METACHARACTERS)
318            for segment in path
319        ):
320            raise ValueError(
321                f"Glob characters {_GLOB_METACHARACTERS} are not supported in `{field_name}`: "
322                "the path must identify a single field."
323            )
324
325    def _evaluated_indicator_path(self) -> list[Any]:
326        return [segment.eval(self.config) for segment in self._truncation_indicator_path]
327
328    def _evaluated_expand_path(self) -> list[Any]:
329        return [segment.eval(self.config) for segment in self._expand_path]
330
331    def expand_record(self, record: MutableMapping[Any, Any]) -> Iterable[MutableMapping[Any, Any]]:
332        """Expand a record by extracting items from a nested array field."""
333        if not isinstance(record, Mapping):
334            # If the input isn't a mapping, expansion can't proceed; yield as-is.
335            yield record
336            return
337
338        if not self._expand_path:
339            yield record
340            return
341
342        parent_record = record
343
344        expand_path = self._evaluated_expand_path()
345        truncated = bool(self._truncation_indicator_path) and self._is_truncated(parent_record)
346        # Built once per parent, not once per item: the walk that finds the expanded list costs
347        # as much as extracting it does.
348        merge_base = (
349            self._without_expanded_list(parent_record, expand_path) if self.merge_parent else None
350        )
351
352        if truncated and self.truncated_list_retriever:
353            # Streamed, so the shortfall is only known once the retriever is exhausted. If the
354            # consumer stops early the fetch was cut short by it, not by the API, and no warning
355            # would be accurate anyway.
356            fetched_count = 0
357            for fetched in self._fetch_complete_list(parent_record, merge_base):
358                fetched_count += 1
359                yield fetched
360            self._warn_if_fetch_incomplete(parent_record, expand_path, fetched_count)
361            if fetched_count > 0:
362                return
363
364        try:
365            extracted_values = dpath.values(parent_record, expand_path)
366        except KeyError:
367            extracted_values = []
368
369        embedded_lists = [
370            extracted for extracted in extracted_values if isinstance(extracted, list)
371        ]
372        embedded_count = sum(len(items) for items in embedded_lists)
373
374        if truncated and not self.truncated_list_retriever:
375            self._warn_truncated_without_retriever(parent_record, expand_path, embedded_count)
376
377        for items in embedded_lists:
378            for item in items:
379                if isinstance(item, dict):
380                    yield self._apply_parent_context(parent_record, dict(item), merge_base)
381                elif self._carries_parent_context():
382                    yield self._apply_parent_context(parent_record, {"value": item}, merge_base)
383                else:
384                    yield item
385
386        if embedded_count == 0 and self.on_no_records == OnNoRecords.emit_parent:
387            yield parent_record
388
389    def _warn_truncated_without_retriever(
390        self, parent_record: Mapping[str, Any], expand_path: list[Any], embedded_count: int
391    ) -> None:
392        with self._warning_lock:
393            if self._warned_truncation_without_retriever:
394                return
395            self._warned_truncation_without_retriever = True
396
397        indicator_path = self._evaluated_indicator_path()
398        total_count = self._get_sibling_total_count(parent_record, indicator_path)
399        total_fragment = f" of {total_count} total" if total_count is not None else ""
400        self._emit_warning(
401            f"The nested list at {expand_path} is marked as truncated (the field at {indicator_path} "
402            f"is truthy) but no `truncated_list_retriever` is configured, so only the "
403            f"{embedded_count} embedded item(s){total_fragment} were expanded and the remaining "
404            "items are not emitted. Configure `truncated_list_retriever` to fetch the complete list "
405            "if the API provides an endpoint for it. This warning is emitted once per stream; other "
406            "records may be truncated as well."
407        )
408
409    def _warn_if_fetch_incomplete(
410        self, parent_record: Mapping[str, Any], expand_path: list[Any], fetched_count: int
411    ) -> None:
412        if self.suppress_incomplete_fetch_warning:
413            return
414        indicator_path = self._evaluated_indicator_path()
415        total_count = self._get_sibling_total_count(parent_record, indicator_path)
416        if total_count is None or fetched_count >= total_count:
417            return
418        with self._warning_lock:
419            if self._warned_incomplete_fetch:
420                return
421            self._warned_incomplete_fetch = True
422        fallback_fragment = (
423            " The embedded items were expanded as a fallback." if fetched_count == 0 else ""
424        )
425        self._emit_warning(
426            f"The `truncated_list_retriever` for the nested list at {expand_path} returned "
427            f"{fetched_count} record(s) but the `total_count` field next to {indicator_path} reports "
428            f"{total_count}, so the fetched list is still incomplete.{fallback_fragment} Check that "
429            "the retriever has a `paginator` configured and that its request matches the "
430            "complete-list endpoint. This warning is emitted once per stream; other records may be "
431            "affected as well."
432        )
433
434    def _emit_warning(self, message: str) -> None:
435        if self.message_repository:
436            self.message_repository.emit_message(
437                AirbyteMessage(
438                    type=MessageType.LOG,
439                    log=AirbyteLogMessage(level=Level.WARN, message=message),
440                )
441            )
442        else:
443            logger.warning(message)
444
445    def _get_sibling_total_count(
446        self, parent_record: Mapping[str, Any], indicator_path: list[Any]
447    ) -> Optional[int]:
448        """Best-effort lookup of a `total_count` field next to the truncation indicator."""
449        if not indicator_path:
450            return None
451        try:
452            total = dpath.get(dict(parent_record), [*indicator_path[:-1], "total_count"])
453        except (KeyError, ValueError):
454            return None
455        return total if isinstance(total, int) and not isinstance(total, bool) else None
456
457    def _is_truncated(self, parent_record: MutableMapping[Any, Any]) -> bool:
458        indicator_path = self._evaluated_indicator_path()
459        try:
460            return bool(dpath.get(parent_record, indicator_path))
461        except (KeyError, ValueError):
462            return False
463
464    def _fetch_complete_list(
465        self, parent_record: Mapping[str, Any], merge_base: Optional[Mapping[str, Any]]
466    ) -> Iterable[Any]:
467        if not self.truncated_list_retriever:
468            return
469        stream_slice = StreamSlice(partition={"parent_record": parent_record}, cursor_slice={})
470        for item in self.truncated_list_retriever.read_records(
471            records_schema={}, stream_slice=stream_slice
472        ):
473            if isinstance(item, AirbyteMessage):
474                if item.type != MessageType.RECORD or item.record is None:
475                    continue
476                data: Any = item.record.data
477            elif isinstance(item, Record):
478                data = item.data
479            elif isinstance(item, _BARE_PROTOCOL_MESSAGES):
480                continue
481            else:
482                data = item
483            if isinstance(data, Mapping):
484                yield self._apply_parent_context(parent_record, dict(data), merge_base)
485            elif self._carries_parent_context():
486                yield self._apply_parent_context(parent_record, {"value": data}, merge_base)
487            else:
488                yield data
489
490    def _apply_parent_context(
491        self,
492        parent_record: Mapping[str, Any],
493        child_record: MutableMapping[str, Any],
494        merge_base: Optional[Mapping[str, Any]],
495    ) -> MutableMapping[str, Any]:
496        """Return the child record carrying the configured parent context.
497
498        The order is fixed: the parent is merged underneath the child first, then `parent_fields`
499        are copied on top, then `original_record` is embedded. A named copy can therefore
500        overwrite a merged value.
501        """
502        if merge_base is not None:
503            # Deep-copied per item: a shallow merge would alias every nested container of the
504            # parent into every item, so a downstream transformation writing into one of them
505            # would reach the parent record and the items already yielded. The expanded list is
506            # already stripped from `merge_base`, so the copy costs less than
507            # `remain_original_record` does.
508            child_record = {**copy.deepcopy(merge_base), **child_record}
509        for parent_field in self.parent_fields or []:
510            parent_field.copy_onto(parent_record, child_record)
511        if self.remain_original_record:
512            child_record["original_record"] = copy.deepcopy(parent_record)
513        return child_record
514
515    def _carries_parent_context(self) -> bool:
516        return bool(self.remain_original_record or self.parent_fields or self.merge_parent)
517
518    @classmethod
519    def _without_expanded_list(
520        cls, parent_record: Mapping[str, Any], expand_path: list[Any]
521    ) -> Mapping[str, Any]:
522        """Copy of the parent with only the value at `expand_path` removed.
523
524        The containers on the way to the list are copied and the rest of the parent is shared,
525        so siblings of the list are kept and the parent itself is not mutated here. This base is
526        built once per parent record and then deep-copied per item in `_apply_parent_context`,
527        which is what keeps the parent and the sibling items isolated from one another. Every
528        list matched by the path is removed, which covers glob paths that select several lists;
529        a match that is not a list is kept, since nothing is expanded from it.
530        """
531        matched_paths = [
532            path
533            for path, value in dpath.segments.walk(parent_record)  # type: ignore[no-untyped-call]
534            if isinstance(value, list) and dpath.segments.match(path, expand_path)
535        ]
536        stripped: Any = parent_record
537        # Deepest and right-most first, so removing a list element never shifts the index of a
538        # match still to be removed, and a match nested inside another match goes first.
539        for path in sorted(matched_paths, key=cls._path_sort_key, reverse=True):
540            stripped = cls._without_path(stripped, path)
541        return dict(stripped)
542
543    @staticmethod
544    def _path_sort_key(path: Sequence[Any]) -> list[tuple[int, Any]]:
545        # Integers and strings are not mutually comparable, so tag each segment by kind.
546        return [(0, segment) if isinstance(segment, int) else (1, str(segment)) for segment in path]
547
548    @classmethod
549    def _without_path(cls, container: Any, path: Sequence[Any]) -> Any:
550        head, rest = path[0], path[1:]
551        if isinstance(container, Mapping):
552            copied: Any = dict(container)
553        elif isinstance(container, list):
554            copied = list(container)
555        else:
556            return container
557        if rest:
558            copied[head] = cls._without_path(copied[head], rest)
559        else:
560            del copied[head]
561        return copied