From 02d5345d159df518638715ba08c43c0daeaa47e0 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 15 Oct 2026 13:21:17 +0000 Subject: [PATCH 1/7] feat(declarative): interpolate response in ConstantBackoffStrategy and add max_waiting_time_in_seconds Co-Authored-By: Daryna Ishchenko --- .../declarative_component_schema.yaml | 16 +- .../models/declarative_component_schema.py | 274 +++++++++--------- .../parsers/model_to_component_factory.py | 1 + .../constant_backoff_strategy.py | 60 +++- .../test_model_to_component_factory.py | 32 ++ .../test_constant_backoff.py | 202 +++++++++++++ .../test_default_error_handler.py | 14 + 7 files changed, 453 insertions(+), 146 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 6df783d1db..e00a84c6ff 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -709,7 +709,7 @@ definitions: enum: [ConstantBackoffStrategy] backoff_time_in_seconds: title: Backoff Time - description: Backoff time in seconds. + description: Backoff time in seconds. If the expression evaluates to an empty result, the next backoff strategy is used. anyOf: - type: number title: Number of seconds @@ -718,10 +718,24 @@ definitions: title: Interpolated Value interpolation_context: - config + - response + - headers examples: - 30 - 30.5 - "{{ config['backoff_time'] }}" + - "{{ response.retry_in }}" + max_waiting_time_in_seconds: + title: Max Waiting Time in Seconds + description: 'Stop the stream instead of waiting, when the computed backoff is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from connector config so that the bound can be changed without a connector release. A value of 0 means never wait. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.' + anyOf: + - type: number + - type: string + interpolation_context: + - config + examples: + - 3600 + - "{{ config['max_waiting_time'] * 60 }}" jitter_range_in_seconds: title: Jitter Range description: Optional additive jitter range in seconds. When set, the backoff time is uniformly distributed between backoff_time_in_seconds and backoff_time_in_seconds + (jitter_range_in_seconds * 2), so jitter only increases the base backoff. diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 0317d52d65..6c8aae953a 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -6,7 +6,7 @@ from enum import Enum from typing import Any, Dict, List, Literal, Optional, Union -from pydantic.v1 import BaseModel, Extra, Field +from pydantic.v1 import BaseModel, Extra, Field, PositiveFloat, confloat, conint from airbyte_cdk.sources.declarative.models.base_model_with_deprecations import ( BaseModelWithDeprecations, @@ -87,10 +87,9 @@ class DynamicStreamCheckConfig(BaseModel): dynamic_stream_name: str = Field( ..., description="The dynamic stream name.", title="Dynamic Stream Name" ) - stream_count: Optional[int] = Field( + stream_count: Optional[conint(ge=1)] = Field( None, description="The number of streams to attempt reading from during a check operation. If unset, all generated streams are checked. Must be a positive integer; if it exceeds the total number of available streams, all streams are checked.", - ge=1, title="Stream Count", ) @@ -134,17 +133,22 @@ class ConcurrencyLevel(BaseModel): class ConstantBackoffStrategy(BaseModel): type: Literal["ConstantBackoffStrategy"] - backoff_time_in_seconds: Union[float, str] = Field( + backoff_time_in_seconds: Union[confloat(ge=0.0), str] = Field( ..., - description="Backoff time in seconds.", - examples=[30, 30.5, "{{ config['backoff_time'] }}"], + description="Backoff time in seconds. If the expression evaluates to an empty result, the next backoff strategy is used.", + examples=[30, 30.5, "{{ config['backoff_time'] }}", "{{ response.retry_in }}"], title="Backoff Time", ) - jitter_range_in_seconds: Optional[float] = Field( + max_waiting_time_in_seconds: Optional[Union[float, str]] = Field( + None, + description="Stop the stream instead of waiting, when the computed backoff is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from connector config so that the bound can be changed without a connector release. A value of 0 means never wait. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.", + examples=[3600, "{{ config['max_waiting_time'] * 60 }}"], + title="Max Waiting Time in Seconds", + ) + jitter_range_in_seconds: Optional[confloat(ge=0.0)] = Field( None, description="Optional additive jitter range in seconds. When set, the backoff time is uniformly distributed between backoff_time_in_seconds and backoff_time_in_seconds + (jitter_range_in_seconds * 2), so jitter only increases the base backoff.", examples=[15], - ge=0, title="Jitter Range", ) parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") @@ -583,11 +587,10 @@ class ExponentialBackoffStrategy(BaseModel): examples=[5, 5.5, "10"], title="Factor", ) - jitter_range_in_seconds: Optional[float] = Field( + jitter_range_in_seconds: Optional[confloat(ge=0.0)] = Field( None, description="Optional additive jitter range in seconds. When set, the backoff time is uniformly distributed between computed_backoff and computed_backoff + (jitter_range_in_seconds * 2), so jitter only increases the computed backoff.", examples=[2], - ge=0, title="Jitter Range", ) parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") @@ -1332,39 +1335,34 @@ class ResetPolicy(Enum): class PageSizeReduction(BaseModel): type: Literal["PageSizeReduction"] - reduction_factor: Optional[float] = Field( + reduction_factor: Optional[confloat(gt=1.0)] = Field( 2, description="Divisor applied to the page size on each reduction. The new page size is floor(current page size / reduction factor).", examples=[2, 4], - gt=1, title="Reduction Factor", ) - minimum_page_size: Optional[int] = Field( + minimum_page_size: Optional[conint(ge=1)] = Field( 1, description="Page size below which the connector stops reducing and fails the sync. It must be smaller than the page size configured on the pagination strategy, otherwise no reduction could ever be applied. It is one of two bounds on the reduction and whichever is tighter wins: an unbroken run of failing pages divides the page size by reduction_factor at most max_attempts times, so reaching this floor in a single run needs max_attempts of at least log(page_size / minimum_page_size) / log(reduction_factor) - with the defaults, a page size of 1000 bottoms out at 31 records per page and a floor of 10 is never reached. Pages that succeed in between restart the max_attempts budget, so the floor is still reachable over a partition.", examples=[1, 10], - ge=1, title="Minimum Page Size", ) - max_attempts: Optional[int] = Field( + max_attempts: Optional[conint(ge=1)] = Field( 5, description="Maximum number of page size reductions made in a row without a single page succeeding, before the sync fails with a transient error. Every reduction follows a request that failed, so at most max_attempts + 1 failing requests are issued before giving up. The budget restarts after every page that succeeds, under either reset_policy, so it bounds the reductions needed to get a single page through and not the number of pages a partition may have: a stream that needs a reduction every now and then reads to the end however long it is. Under NEVER the page size also strictly decreases, so minimum_page_size bounds the reductions of the whole partition on its own. The wait between reduction attempts is the CDK's own - it grows with each attempt - and does not consult the error handler's backoff_strategies or a Retry-After header.", examples=[5, 10], - ge=1, title="Maximum Reduction Attempts", ) - backoff_seconds: Optional[float] = Field( + backoff_seconds: Optional[confloat(ge=0.0)] = Field( 0.5, description='Base number of seconds to wait before the page is re-issued, multiplied by the number of attempts made in a row, so the second attempt waits twice as long as the first. A REDUCE_PAGE_SIZE response never reaches the error handler\'s retry budget, backoff_strategies or a Retry-After header, so this is the only thing spacing those requests out. Raise it on an API whose error also means "we are briefly unwell" rather than only "your page is too big", since the default spaces the whole run of attempts over a few seconds.', examples=[0.5, 5], - ge=0, title="Backoff Seconds", ) - retries_at_minimum_page_size: Optional[int] = Field( + retries_at_minimum_page_size: Optional[conint(ge=0)] = Field( 0, description="Number of times the same page is re-issued unchanged, each after the backoff wait, once the page size cannot be shrunk any further, before the sync fails with a transient error. The default of 0 fails on the first response received at minimum_page_size. Raise it when the API returns the same error for a page that is too big and for a server-side hiccup: at the floor, reducing is no longer an option but waiting still is, and without this budget those responses end the stream on the first one. This budget is separate from max_attempts, which only counts reductions, and it restarts on every page that succeeds. It applies however the page size arrived at the floor, whether by reduction or because page_size was already there; a page size that minimum_page_size blocks from ever being reduced is still reported as a configuration error, but only once this budget is spent.", examples=[0, 3], - ge=0, title="Retries At Minimum Page Size", ) failure_message: Optional[str] = Field( @@ -2339,53 +2337,6 @@ class DefaultPaginator(BaseModel): parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") -class RecordExpander(BaseModel): - type: Literal["RecordExpander"] - expand_records_from_field: List[str] = Field( - ..., - description="Path to a nested array field within each record. Items from this array will be extracted and emitted as separate records. Supports wildcards (*) for matching multiple arrays.", - examples=[ - ["lines", "data"], - ["items"], - ["nested", "array"], - ["sections", "*", "items"], - ], - title="Expand Records From Field", - ) - remain_original_record: Optional[bool] = Field( - False, - description='If true, each expanded record will include the original parent record in an "original_record" field. Defaults to false.', - title="Remain Original Record", - ) - parent_fields: Optional[List[ParentFieldPath]] = Field( - None, - description="Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. A copied object or array is deep-copied, so a transformation that writes inside it affects only that item. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", - title="Parent Fields", - ) - merge_parent: Optional[bool] = Field( - False, - description="If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. Each item receives its own deep copy of the merged parent fields, so a transformation that writes into a nested value affects only that item. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", - title="Merge Parent", - ) - on_no_records: Optional[OnNoRecords] = Field( - OnNoRecords.skip, - description='Behavior when the expansion path is missing, not a list, or an empty list. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged.', - title="On No Records", - ) - truncation_indicator_path: Optional[List[str]] = Field( - None, - description="Path within each record to a field indicating that the embedded nested list is truncated (e.g. a `has_more` flag on the list object). When the field evaluates to a truthy value and `truncated_list_retriever` is configured, the retriever is used to fetch the complete list instead of expanding the embedded items. When the field is truthy and no retriever is configured, the embedded items are expanded as normal and a WARNING is logged once per stream so the truncation is visible instead of silent. Glob characters (`*`, `?`, `[`) are not supported in this path, nor in `expand_records_from_field` when a retriever is configured; this is enforced on the interpolated values. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", - examples=[["data", "object", "lines", "has_more"]], - title="Truncation Indicator Path", - ) - truncated_list_retriever: Optional[Union[SimpleRetriever, CustomRetriever]] = Field( - None, - description="Retriever used to fetch the complete list of items when the field at `truncation_indicator_path` is truthy on a record. The record being expanded is exposed to the retriever's interpolation context as `stream_slice['parent_record']`. One fetch is issued per truncated record, so enable `use_cache` on the requester when the same list can be fetched repeatedly. Configure a `paginator`, since without one only the first page of the complete list is read. If the retriever returns no records, the embedded items are expanded as a fallback; if it returns fewer records than the `total_count` field next to the indicator, a WARNING is logged once per stream. Request failures surface through the retriever's `error_handler` and fail the stream like any other request. `$parameters` of the enclosing stream propagate into this retriever's components (including `request_parameters` on its requester); move request-shaping parameters to the outer requester's `request_parameters` when adopting this field. `partition_router` and `pagination_reset` are not supported. In Connector Builder test reads, its requests appear as auxiliary requests and the test-read page limit applies to each fetch independently, so the fetched list may be shorter than `total_count`; the incomplete-fetch warning is not emitted in test reads when a `paginator` is configured (without one the retriever is not capped, so the warning still applies). Requires `truncation_indicator_path`. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", - title="Truncated List Retriever", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - class SessionTokenRequestApiKeyAuthenticator(BaseModel): type: Literal["ApiKey"] inject_into: RequestOption = Field( @@ -2631,47 +2582,6 @@ class Config: ) -class DpathExtractor(BaseModel): - type: Literal["DpathExtractor"] - field_path: List[str] = Field( - ..., - description='List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).', - examples=[ - ["data"], - ["data", "records"], - ["data", "{{ parameters.name }}"], - ["data", "*", "record"], - ], - title="Field Path", - ) - record_expander: Optional[RecordExpander] = Field( - None, - description="Optional component to expand records by extracting items from nested array fields.", - title="Record Expander", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - -class CombinedExtractor(BaseModel): - type: Literal["CombinedExtractor"] - extractors: List[Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor]] = Field( - ..., - description="The record extractors to combine. At least one is required. Each sub-extractor is given the same HTTP response and decodes it independently, so the response is parsed once per sub-extractor. Streaming decoders (CsvDecoder, JsonlDecoder, JsonItemsDecoder, GzipDecoder, IterableDecoder) can only read the response body once and are rejected.", - title="Extractors", - ) - mode: Optional[CombineMode] = Field( - CombineMode.union, - description='How the records of the sub-extractors are combined. "union" (default) yields every record of every sub-extractor, in the order the extractors are declared. "first_match" yields the records of the first sub-extractor that produces at least one record and skips the remaining ones; nothing is yielded if none of them produces a record. Note that a paginator which counts the records of a page counts the combined records: under "union" that is the sum over all sub-extractors, which overshoots the API page size, so "union" is rejected with an OffsetIncrement paginator because the offset would skip records. Under "first_match" the count is the count of the winning sub-extractor, which is usually the number of records the API returned for the page.', - title="Combine Mode", - ) - skip_empty_records: Optional[bool] = Field( - False, - description='Whether to drop empty records - null, {}, [], "" - yielded by a sub-extractor before the records are combined. Off by default, so the output of a sub-extractor is passed through as it is. Turn it on when the API can return nulls in the middle of a record list, which a GraphQL API does when it answers a partial response and reports the failure in a sibling error field. Under "first_match" this also changes which sub-extractor wins: a sub-extractor whose records are all empty no longer counts as a match, so the next one is tried. The dropped records still count toward the page size of an OffsetIncrement or PageIncrement paginator, so a full page holding a null does not end pagination early.', - title="Skip Empty Records", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - class ZipfileDecoder(BaseModel): class Config: extra = Extra.allow @@ -2684,27 +2594,6 @@ class Config: ) -class RecordSelector(BaseModel): - type: Literal["RecordSelector"] - extractor: Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor] - record_filter: Optional[Union[RecordFilter, CustomRecordFilter]] = Field( - None, - description="Responsible for filtering records to be emitted by the Source.", - title="Record Filter", - ) - schema_normalization: Optional[Union[SchemaNormalization, CustomSchemaNormalization]] = Field( - None, - description="Responsible for normalization according to the schema.", - title="Schema Normalization", - ) - transform_before_filtering: Optional[bool] = Field( - None, - description="If true, transformation will be applied before record filtering.", - title="Transform Before Filtering", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - class ConfigMigration(BaseModel): type: Literal["ConfigMigration"] description: Optional[str] = Field( @@ -3022,6 +2911,94 @@ class Config: parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") +class CombinedExtractor(BaseModel): + type: Literal["CombinedExtractor"] + extractors: List[Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor]] = Field( + ..., + description="The record extractors to combine. At least one is required. Each sub-extractor is given the same HTTP response and decodes it independently, so the response is parsed once per sub-extractor. Streaming decoders (CsvDecoder, JsonlDecoder, JsonItemsDecoder, GzipDecoder, IterableDecoder) can only read the response body once and are rejected.", + title="Extractors", + ) + mode: Optional[CombineMode] = Field( + CombineMode.union, + description='How the records of the sub-extractors are combined. "union" (default) yields every record of every sub-extractor, in the order the extractors are declared. "first_match" yields the records of the first sub-extractor that produces at least one record and skips the remaining ones; nothing is yielded if none of them produces a record. Note that a paginator which counts the records of a page counts the combined records: under "union" that is the sum over all sub-extractors, which overshoots the API page size, so "union" is rejected with an OffsetIncrement paginator because the offset would skip records. Under "first_match" the count is the count of the winning sub-extractor, which is usually the number of records the API returned for the page.', + title="Combine Mode", + ) + skip_empty_records: Optional[bool] = Field( + False, + description='Whether to drop empty records - null, {}, [], "" - yielded by a sub-extractor before the records are combined. Off by default, so the output of a sub-extractor is passed through as it is. Turn it on when the API can return nulls in the middle of a record list, which a GraphQL API does when it answers a partial response and reports the failure in a sibling error field. Under "first_match" this also changes which sub-extractor wins: a sub-extractor whose records are all empty no longer counts as a match, so the next one is tried. The dropped records still count toward the page size of an OffsetIncrement or PageIncrement paginator, so a full page holding a null does not end pagination early.', + title="Skip Empty Records", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + +class DpathExtractor(BaseModel): + type: Literal["DpathExtractor"] + field_path: List[str] = Field( + ..., + description='List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).', + examples=[ + ["data"], + ["data", "records"], + ["data", "{{ parameters.name }}"], + ["data", "*", "record"], + ], + title="Field Path", + ) + record_expander: Optional[RecordExpander] = Field( + None, + description="Optional component to expand records by extracting items from nested array fields.", + title="Record Expander", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + +class RecordExpander(BaseModel): + type: Literal["RecordExpander"] + expand_records_from_field: List[str] = Field( + ..., + description="Path to a nested array field within each record. Items from this array will be extracted and emitted as separate records. Supports wildcards (*) for matching multiple arrays.", + examples=[ + ["lines", "data"], + ["items"], + ["nested", "array"], + ["sections", "*", "items"], + ], + title="Expand Records From Field", + ) + remain_original_record: Optional[bool] = Field( + False, + description='If true, each expanded record will include the original parent record in an "original_record" field. Defaults to false.', + title="Remain Original Record", + ) + parent_fields: Optional[List[ParentFieldPath]] = Field( + None, + description="Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. A copied object or array is deep-copied, so a transformation that writes inside it affects only that item. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + title="Parent Fields", + ) + merge_parent: Optional[bool] = Field( + False, + description="If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. Each item receives its own deep copy of the merged parent fields, so a transformation that writes into a nested value affects only that item. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + title="Merge Parent", + ) + on_no_records: Optional[OnNoRecords] = Field( + OnNoRecords.skip, + description='Behavior when the expansion path is missing, not a list, or an empty list. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged.', + title="On No Records", + ) + truncation_indicator_path: Optional[List[str]] = Field( + None, + description="Path within each record to a field indicating that the embedded nested list is truncated (e.g. a `has_more` flag on the list object). When the field evaluates to a truthy value and `truncated_list_retriever` is configured, the retriever is used to fetch the complete list instead of expanding the embedded items. When the field is truthy and no retriever is configured, the embedded items are expanded as normal and a WARNING is logged once per stream so the truncation is visible instead of silent. Glob characters (`*`, `?`, `[`) are not supported in this path, nor in `expand_records_from_field` when a retriever is configured; this is enforced on the interpolated values. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + examples=[["data", "object", "lines", "has_more"]], + title="Truncation Indicator Path", + ) + truncated_list_retriever: Optional[Union[SimpleRetriever, CustomRetriever]] = Field( + None, + description="Retriever used to fetch the complete list of items when the field at `truncation_indicator_path` is truthy on a record. The record being expanded is exposed to the retriever's interpolation context as `stream_slice['parent_record']`. One fetch is issued per truncated record, so enable `use_cache` on the requester when the same list can be fetched repeatedly. Configure a `paginator`, since without one only the first page of the complete list is read. If the retriever returns no records, the embedded items are expanded as a fallback; if it returns fewer records than the `total_count` field next to the indicator, a WARNING is logged once per stream. Request failures surface through the retriever's `error_handler` and fail the stream like any other request. `$parameters` of the enclosing stream propagate into this retriever's components (including `request_parameters` on its requester); move request-shaping parameters to the outer requester's `request_parameters` when adopting this field. `partition_router` and `pagination_reset` are not supported. In Connector Builder test reads, its requests appear as auxiliary requests and the test-read page limit applies to each fetch independently, so the fetched list may be shorter than `total_count`; the incomplete-fetch warning is not emitted in test reads when a `paginator` is configured (without one the retriever is not capped, so the warning still applies). Requires `truncation_indicator_path`. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + title="Truncated List Retriever", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class SessionTokenAuthenticator(BaseModel): type: Literal["SessionTokenAuthenticator"] login_requester: HttpRequester = Field( @@ -3206,18 +3183,16 @@ class HttpRequester(BaseModelWithDeprecations): description="Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records).", title="Use Cache", ) - connect_timeout_in_seconds: Optional[float] = Field( + connect_timeout_in_seconds: Optional[PositiveFloat] = Field( None, description="Maximum time to establish the connection (TCP + TLS handshake). When unset, no connect timeout is applied.", examples=[30], - gt=0, title="Connect Timeout (seconds)", ) - read_timeout_in_seconds: Optional[float] = Field( + read_timeout_in_seconds: Optional[PositiveFloat] = Field( None, description="The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes. Each timed-out attempt counts against the error handler's retry window (600 s for manifest connectors), so a value near 300 allows only a few retries.", examples=[300], - gt=0, title="Read Timeout (seconds)", ) use_tcp_keepalive: Optional[bool] = Field( @@ -3342,6 +3317,27 @@ class QueryProperties(BaseModel): parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") +class RecordSelector(BaseModel): + type: Literal["RecordSelector"] + extractor: Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor] + record_filter: Optional[Union[RecordFilter, CustomRecordFilter]] = Field( + None, + description="Responsible for filtering records to be emitted by the Source.", + title="Record Filter", + ) + schema_normalization: Optional[Union[SchemaNormalization, CustomSchemaNormalization]] = Field( + None, + description="Responsible for normalization according to the schema.", + title="Schema Normalization", + ) + transform_before_filtering: Optional[bool] = Field( + None, + description="If true, transformation will be applied before record filtering.", + title="Transform Before Filtering", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class StateDelegatingStream(BaseModel): type: Literal["StateDelegatingStream"] name: str = Field(..., description="The stream name.", example=["Users"], title="Name") @@ -3455,7 +3451,12 @@ class AsyncRetriever(BaseModel): description="Responsible for fetching the final result `urls` provided by the completed / finished / ready async job.", ) download_extractor: Optional[ - Union[DpathExtractor, CustomRecordExtractor, ResponseToFileExtractor, CombinedExtractor] + Union[ + DpathExtractor, + CustomRecordExtractor, + ResponseToFileExtractor, + CombinedExtractor, + ] ] = Field(None, description="Responsible for fetching the records from provided urls.") creation_requester: Union[HttpRequester, CustomRequester] = Field( ..., @@ -3469,10 +3470,9 @@ class AsyncRetriever(BaseModel): None, description="The time in minutes after which the single Async Job should be considered as Timed Out.", ) - failed_retry_wait_time_in_seconds: Optional[Union[int, str]] = Field( + failed_retry_wait_time_in_seconds: Optional[Union[conint(ge=1), str]] = Field( None, description="Time in seconds to wait before retrying a failed async job. Only applies to jobs that ran on the API side and reported a FAILED status (e.g. report generation failed due to a cooldown). Creation failures (HTTP errors when starting a job, such as 429s) and TIMED_OUT jobs are retried immediately and are not affected by this setting. When set, the orchestrator defers retry of real failed jobs until the wait time has elapsed, without blocking other jobs.", - ge=1, ) download_target_requester: Optional[Union[HttpRequester, CustomRequester]] = Field( None, @@ -3659,13 +3659,13 @@ class DynamicDeclarativeStream(BaseModel): ComplexFieldType.update_forward_refs() GzipDecoder.update_forward_refs() CompositeErrorHandler.update_forward_refs() -CombinedExtractor.update_forward_refs() DeclarativeSource1.update_forward_refs() DeclarativeSource2.update_forward_refs() SelectiveAuthenticator.update_forward_refs() ConditionalStreams.update_forward_refs() FileUploader.update_forward_refs() DeclarativeStream.update_forward_refs() +CombinedExtractor.update_forward_refs() DpathExtractor.update_forward_refs() RecordExpander.update_forward_refs() SessionTokenAuthenticator.update_forward_refs() 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 219e0f70ad..9b3b71a74f 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -1855,6 +1855,7 @@ def create_constant_backoff_strategy( return ConstantBackoffStrategy( backoff_time_in_seconds=model.backoff_time_in_seconds, jitter_range_in_seconds=model.jitter_range_in_seconds, + max_waiting_time_in_seconds=model.max_waiting_time_in_seconds, config=config, parameters=model.parameters or {}, ) diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py index 931183b2ef..67812192fe 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py @@ -8,24 +8,35 @@ import requests +from airbyte_cdk.models import FailureType from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString +from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.max_waiting_time_helper import ( + evaluate_max_waiting_time, + interpolated_max_waiting_time, +) from airbyte_cdk.sources.streams.http.error_handlers import BackoffStrategy from airbyte_cdk.sources.types import Config +from airbyte_cdk.utils import AirbyteTracedException @dataclass class ConstantBackoffStrategy(BackoffStrategy): - """ - Backoff strategy with a constant backoff interval + """Compute a constant backoff from config, response JSON, and response headers. + + An empty interpolated result defers to the next strategy. The optional cap is evaluated from + config and checked against the final backoff after jitter. Attributes: - backoff_time_in_seconds (float): time to backoff before retrying a retryable request. + backoff_time_in_seconds: Backoff time in seconds. + jitter_range_in_seconds: Optional additive jitter range that only increases the wait. + max_waiting_time_in_seconds: Optional inclusive cap on the computed wait after jitter. """ backoff_time_in_seconds: Union[float, InterpolatedString, str] parameters: InitVar[Mapping[str, Any]] config: Config jitter_range_in_seconds: Optional[float] = None + max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None def __post_init__(self, parameters: Mapping[str, Any]) -> None: if not isinstance(self.backoff_time_in_seconds, InterpolatedString): @@ -38,16 +49,49 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None: self.backoff_time_in_seconds = InterpolatedString.create( self.backoff_time_in_seconds, parameters=parameters ) + self._max_waiting_time_in_seconds = interpolated_max_waiting_time( + self.max_waiting_time_in_seconds, parameters + ) + evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) def backoff_time( self, response_or_exception: Optional[Union[requests.Response, requests.RequestException]], attempt_count: int, ) -> Optional[float]: - backoff_time = float( - cast(InterpolatedString, self.backoff_time_in_seconds).eval(self.config) + response: Any = {} + headers: Mapping[str, Any] = {} + if isinstance(response_or_exception, requests.Response): + try: + response = response_or_exception.json() + except requests.exceptions.JSONDecodeError: + pass + headers = response_or_exception.headers + + evaluated_backoff_time = cast(InterpolatedString, self.backoff_time_in_seconds).eval( + self.config, response=response, headers=headers ) - if self.jitter_range_in_seconds is None: - return backoff_time + if evaluated_backoff_time is None or ( + isinstance(evaluated_backoff_time, str) and not evaluated_backoff_time.strip() + ): + return None + + backoff_time = float(evaluated_backoff_time) + if self.jitter_range_in_seconds is not None: + backoff_time = random.uniform( + backoff_time, backoff_time + (self.jitter_range_in_seconds * 2) + ) + + max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) + if max_waiting_time is not None and backoff_time and backoff_time >= max_waiting_time: + raise AirbyteTracedException( + internal_message=( + f"Rate limit wait time {backoff_time}s is greater than or equal to the " + f"maximum of {max_waiting_time}s this stream is allowed to wait. " + f"Stopping the stream..." + ), + message="The rate limit wait time is longer than the connector is allowed to wait.", + failure_type=FailureType.transient_error, + ) - return random.uniform(backoff_time, backoff_time + (self.jitter_range_in_seconds * 2)) + return backoff_time 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 e54f23e8e8..b19a9810de 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 @@ -2786,6 +2786,38 @@ def test_create_requester_with_backoff_jitter( assert backoff_strategy.jitter_range_in_seconds == expected_jitter_range +def test_create_requester_with_constant_backoff_max_waiting_time(): + content = """ +requester: + type: HttpRequester + path: "/v3/marketing/lists" + url_base: "https://api.sendgrid.com" + error_handler: + backoff_strategies: + - type: ConstantBackoffStrategy + backoff_time_in_seconds: 5 + max_waiting_time_in_seconds: 30 + """ + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + requester_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["requester"], {} + ) + + requester = factory.create_component( + model_type=HttpRequesterModel, + component_definition=requester_manifest, + config=input_config, + name="name", + decoder=None, + ) + + assert isinstance(requester.error_handler, DefaultErrorHandler) + backoff_strategy = requester.error_handler.backoff_strategies[0] + assert isinstance(backoff_strategy, ConstantBackoffStrategy) + assert backoff_strategy.max_waiting_time_in_seconds == 30 + + @pytest.mark.parametrize( "backoff_strategy_model, backoff_strategy_arguments", [ diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py index 7fae93fa15..86c24af334 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py @@ -2,17 +2,23 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # +import json from unittest.mock import MagicMock import pytest +import requests +from freezegun import freeze_time +from airbyte_cdk.models import FailureType from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy import ( ConstantBackoffStrategy, ) +from airbyte_cdk.utils import AirbyteTracedException BACKOFF_TIME = 10 PARAMETERS_BACKOFF_TIME = 20 CONFIG_BACKOFF_TIME = 30 +GMAIL_BACKOFF_EXPRESSION = r"{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response.get('error', {}).get('message', '') | regex_search('Retry after (\\S+Z)')) else '' }}" @pytest.mark.parametrize( @@ -80,3 +86,199 @@ def test_constant_backoff_with_jitter_bounds( assert all(expected_lower_bound <= backoff <= expected_upper_bound for backoff in backoff_times) assert len(set(backoff_times)) > 1 + + +def test_constant_backoff_interpolates_response_json(): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds="{{ response.retry_in }}", parameters={}, config={} + ) + + assert backoff_strategy.backoff_time(_response({"retry_in": "12.5"}), 1) == 12.5 + + +def test_constant_backoff_interpolates_headers(): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds="{{ headers['Retry-After'] }}", parameters={}, config={} + ) + + assert backoff_strategy.backoff_time(_response({}, headers={"Retry-After": "15"}), 1) == 15.0 + + +@pytest.mark.parametrize("empty_value", ["", " "]) +def test_constant_backoff_returns_none_for_empty_interpolation(empty_value): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} + ) + + assert backoff_strategy.backoff_time(_response({"retry_in": empty_value}), 1) is None + + +def test_constant_backoff_returns_none_when_json_key_is_missing(): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} + ) + + assert backoff_strategy.backoff_time(_response({}), 1) is None + + +def test_constant_backoff_returns_none_for_non_json_body(): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} + ) + + assert backoff_strategy.backoff_time(_response_body(b"not JSON"), 1) is None + + +@pytest.mark.parametrize( + "response_or_exception", + [None, requests.RequestException()], + ids=["none", "exception"], +) +def test_constant_backoff_returns_none_for_non_response_input(response_or_exception): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} + ) + + assert backoff_strategy.backoff_time(response_or_exception, 1) is None + + +def test_constant_backoff_does_not_apply_jitter_to_empty_interpolation(mocker): + uniform = mocker.patch( + "airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy.random.uniform" + ) + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds="{{ response.get('retry_in', '') }}", + parameters={}, + config={}, + jitter_range_in_seconds=5, + ) + + assert backoff_strategy.backoff_time(_response({}), 1) is None + uniform.assert_not_called() + + +@pytest.mark.parametrize("wait_time", [10, 11], ids=["equal_to_cap", "above_cap"]) +def test_constant_backoff_wait_at_or_above_cap_raises_transient_error(wait_time): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds="{{ response.retry_in }}", + parameters={}, + config={}, + max_waiting_time_in_seconds=10, + ) + + with pytest.raises(AirbyteTracedException) as exception: + backoff_strategy.backoff_time(_response({"retry_in": wait_time}), 1) + + assert exception.value.failure_type == FailureType.transient_error + assert ( + exception.value.message + == "The rate limit wait time is longer than the connector is allowed to wait." + ) + assert "greater than or equal to the maximum of 10.0s" in exception.value.internal_message + + +def test_constant_backoff_wait_below_cap_is_returned(): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds=9, + parameters={}, + config={}, + max_waiting_time_in_seconds=10, + ) + + assert backoff_strategy.backoff_time(None, 1) == 9.0 + + +def test_constant_backoff_cap_is_interpolated_from_config(): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds="{{ response.retry_in }}", + parameters={}, + config={"max_waiting_time": 20}, + max_waiting_time_in_seconds="{{ config['max_waiting_time'] }}", + ) + + assert backoff_strategy.backoff_time(_response({"retry_in": 10}), 1) == 10.0 + + +def test_constant_backoff_invalid_cap_fails_at_construction(): + with pytest.raises(AirbyteTracedException) as exception: + ConstantBackoffStrategy( + backoff_time_in_seconds=10, + parameters={}, + config={"max_waiting_time": "not a number"}, + max_waiting_time_in_seconds="{{ config['max_waiting_time'] }}", + ) + + assert exception.value.failure_type == FailureType.system_error + + +def test_constant_backoff_zero_wait_is_not_refused_by_zero_cap(): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds=0, + parameters={}, + config={}, + max_waiting_time_in_seconds=0, + ) + + assert backoff_strategy.backoff_time(None, 1) == 0 + + +def test_constant_backoff_cap_is_checked_after_jitter(mocker): + mocker.patch( + "airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy.random.uniform", + return_value=11, + ) + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds=10, + parameters={}, + config={}, + jitter_range_in_seconds=1, + max_waiting_time_in_seconds=11, + ) + + with pytest.raises(AirbyteTracedException) as exception: + backoff_strategy.backoff_time(None, 1) + + assert exception.value.failure_type == FailureType.transient_error + + +@freeze_time("2026-10-15T13:00:00Z") +@pytest.mark.parametrize( + "message, expected_backoff", + [ + pytest.param( + "User-rate limit exceeded. Retry after 2026-10-15T13:15:00Z", + 900, + id="future_timestamp", + ), + pytest.param( + "User-rate limit exceeded. Retry after 2026-10-15T12:59:59Z", + 1, + id="past_timestamp", + ), + pytest.param("User-rate limit exceeded.", None, id="missing_timestamp"), + ], +) +def test_constant_backoff_gmail_retry_after_expression(message, expected_backoff): + backoff_strategy = ConstantBackoffStrategy( + backoff_time_in_seconds=GMAIL_BACKOFF_EXPRESSION, parameters={}, config={} + ) + + actual_backoff = backoff_strategy.backoff_time(_response({"error": {"message": message}}), 1) + + if expected_backoff is None: + assert actual_backoff is None + else: + assert actual_backoff == pytest.approx(expected_backoff) + + +def _response(json_body, headers=None): + return _response_body(json.dumps(json_body).encode(), headers=headers) + + +def _response_body(body, headers=None): + response = requests.Response() + response.status_code = 429 + response.headers = headers or {} + response._content = body + response.encoding = "utf-8" + return response diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py index 526e5760fc..713f2de699 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py @@ -242,6 +242,20 @@ def test_default_error_handler_with_constant_backoff_strategy(): ) +def test_default_error_handler_falls_back_when_constant_backoff_is_empty(): + response_mock = create_response(429, json_body={"retry_in": ""}) + error_handler = DefaultErrorHandler( + config={}, + parameters={}, + backoff_strategies=[ + ConstantBackoffStrategy("{{ response.get('retry_in', '') }}", config={}, parameters={}), + ExponentialBackoffStrategy(factor=5, config={}, parameters={}), + ], + ) + + assert error_handler.backoff_time(response_mock, attempt_count=2) == 20 + + @pytest.mark.parametrize( "attempt_count", [ From f8787485fa4c26108f87a261ba46960d86adcb8d Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 15 Oct 2026 13:23:10 +0000 Subject: [PATCH 2/7] fix(declarative): preserve response backoff model typing Co-Authored-By: Daryna Ishchenko --- .../models/declarative_component_schema.py | 264 +++++++++--------- 1 file changed, 135 insertions(+), 129 deletions(-) diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 6c8aae953a..ae8fb976c7 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -6,7 +6,7 @@ from enum import Enum from typing import Any, Dict, List, Literal, Optional, Union -from pydantic.v1 import BaseModel, Extra, Field, PositiveFloat, confloat, conint +from pydantic.v1 import BaseModel, Extra, Field from airbyte_cdk.sources.declarative.models.base_model_with_deprecations import ( BaseModelWithDeprecations, @@ -87,9 +87,10 @@ class DynamicStreamCheckConfig(BaseModel): dynamic_stream_name: str = Field( ..., description="The dynamic stream name.", title="Dynamic Stream Name" ) - stream_count: Optional[conint(ge=1)] = Field( + stream_count: Optional[int] = Field( None, description="The number of streams to attempt reading from during a check operation. If unset, all generated streams are checked. Must be a positive integer; if it exceeds the total number of available streams, all streams are checked.", + ge=1, title="Stream Count", ) @@ -133,7 +134,7 @@ class ConcurrencyLevel(BaseModel): class ConstantBackoffStrategy(BaseModel): type: Literal["ConstantBackoffStrategy"] - backoff_time_in_seconds: Union[confloat(ge=0.0), str] = Field( + backoff_time_in_seconds: Union[float, str] = Field( ..., description="Backoff time in seconds. If the expression evaluates to an empty result, the next backoff strategy is used.", examples=[30, 30.5, "{{ config['backoff_time'] }}", "{{ response.retry_in }}"], @@ -145,10 +146,11 @@ class ConstantBackoffStrategy(BaseModel): examples=[3600, "{{ config['max_waiting_time'] * 60 }}"], title="Max Waiting Time in Seconds", ) - jitter_range_in_seconds: Optional[confloat(ge=0.0)] = Field( + jitter_range_in_seconds: Optional[float] = Field( None, description="Optional additive jitter range in seconds. When set, the backoff time is uniformly distributed between backoff_time_in_seconds and backoff_time_in_seconds + (jitter_range_in_seconds * 2), so jitter only increases the base backoff.", examples=[15], + ge=0, title="Jitter Range", ) parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") @@ -587,10 +589,11 @@ class ExponentialBackoffStrategy(BaseModel): examples=[5, 5.5, "10"], title="Factor", ) - jitter_range_in_seconds: Optional[confloat(ge=0.0)] = Field( + jitter_range_in_seconds: Optional[float] = Field( None, description="Optional additive jitter range in seconds. When set, the backoff time is uniformly distributed between computed_backoff and computed_backoff + (jitter_range_in_seconds * 2), so jitter only increases the computed backoff.", examples=[2], + ge=0, title="Jitter Range", ) parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") @@ -1335,34 +1338,39 @@ class ResetPolicy(Enum): class PageSizeReduction(BaseModel): type: Literal["PageSizeReduction"] - reduction_factor: Optional[confloat(gt=1.0)] = Field( + reduction_factor: Optional[float] = Field( 2, description="Divisor applied to the page size on each reduction. The new page size is floor(current page size / reduction factor).", examples=[2, 4], + gt=1, title="Reduction Factor", ) - minimum_page_size: Optional[conint(ge=1)] = Field( + minimum_page_size: Optional[int] = Field( 1, description="Page size below which the connector stops reducing and fails the sync. It must be smaller than the page size configured on the pagination strategy, otherwise no reduction could ever be applied. It is one of two bounds on the reduction and whichever is tighter wins: an unbroken run of failing pages divides the page size by reduction_factor at most max_attempts times, so reaching this floor in a single run needs max_attempts of at least log(page_size / minimum_page_size) / log(reduction_factor) - with the defaults, a page size of 1000 bottoms out at 31 records per page and a floor of 10 is never reached. Pages that succeed in between restart the max_attempts budget, so the floor is still reachable over a partition.", examples=[1, 10], + ge=1, title="Minimum Page Size", ) - max_attempts: Optional[conint(ge=1)] = Field( + max_attempts: Optional[int] = Field( 5, description="Maximum number of page size reductions made in a row without a single page succeeding, before the sync fails with a transient error. Every reduction follows a request that failed, so at most max_attempts + 1 failing requests are issued before giving up. The budget restarts after every page that succeeds, under either reset_policy, so it bounds the reductions needed to get a single page through and not the number of pages a partition may have: a stream that needs a reduction every now and then reads to the end however long it is. Under NEVER the page size also strictly decreases, so minimum_page_size bounds the reductions of the whole partition on its own. The wait between reduction attempts is the CDK's own - it grows with each attempt - and does not consult the error handler's backoff_strategies or a Retry-After header.", examples=[5, 10], + ge=1, title="Maximum Reduction Attempts", ) - backoff_seconds: Optional[confloat(ge=0.0)] = Field( + backoff_seconds: Optional[float] = Field( 0.5, description='Base number of seconds to wait before the page is re-issued, multiplied by the number of attempts made in a row, so the second attempt waits twice as long as the first. A REDUCE_PAGE_SIZE response never reaches the error handler\'s retry budget, backoff_strategies or a Retry-After header, so this is the only thing spacing those requests out. Raise it on an API whose error also means "we are briefly unwell" rather than only "your page is too big", since the default spaces the whole run of attempts over a few seconds.', examples=[0.5, 5], + ge=0, title="Backoff Seconds", ) - retries_at_minimum_page_size: Optional[conint(ge=0)] = Field( + retries_at_minimum_page_size: Optional[int] = Field( 0, description="Number of times the same page is re-issued unchanged, each after the backoff wait, once the page size cannot be shrunk any further, before the sync fails with a transient error. The default of 0 fails on the first response received at minimum_page_size. Raise it when the API returns the same error for a page that is too big and for a server-side hiccup: at the floor, reducing is no longer an option but waiting still is, and without this budget those responses end the stream on the first one. This budget is separate from max_attempts, which only counts reductions, and it restarts on every page that succeeds. It applies however the page size arrived at the floor, whether by reduction or because page_size was already there; a page size that minimum_page_size blocks from ever being reduced is still reported as a configuration error, but only once this budget is spent.", examples=[0, 3], + ge=0, title="Retries At Minimum Page Size", ) failure_message: Optional[str] = Field( @@ -2337,6 +2345,53 @@ class DefaultPaginator(BaseModel): parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") +class RecordExpander(BaseModel): + type: Literal["RecordExpander"] + expand_records_from_field: List[str] = Field( + ..., + description="Path to a nested array field within each record. Items from this array will be extracted and emitted as separate records. Supports wildcards (*) for matching multiple arrays.", + examples=[ + ["lines", "data"], + ["items"], + ["nested", "array"], + ["sections", "*", "items"], + ], + title="Expand Records From Field", + ) + remain_original_record: Optional[bool] = Field( + False, + description='If true, each expanded record will include the original parent record in an "original_record" field. Defaults to false.', + title="Remain Original Record", + ) + parent_fields: Optional[List[ParentFieldPath]] = Field( + None, + description="Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. A copied object or array is deep-copied, so a transformation that writes inside it affects only that item. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + title="Parent Fields", + ) + merge_parent: Optional[bool] = Field( + False, + description="If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. Each item receives its own deep copy of the merged parent fields, so a transformation that writes into a nested value affects only that item. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + title="Merge Parent", + ) + on_no_records: Optional[OnNoRecords] = Field( + OnNoRecords.skip, + description='Behavior when the expansion path is missing, not a list, or an empty list. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged.', + title="On No Records", + ) + truncation_indicator_path: Optional[List[str]] = Field( + None, + description="Path within each record to a field indicating that the embedded nested list is truncated (e.g. a `has_more` flag on the list object). When the field evaluates to a truthy value and `truncated_list_retriever` is configured, the retriever is used to fetch the complete list instead of expanding the embedded items. When the field is truthy and no retriever is configured, the embedded items are expanded as normal and a WARNING is logged once per stream so the truncation is visible instead of silent. Glob characters (`*`, `?`, `[`) are not supported in this path, nor in `expand_records_from_field` when a retriever is configured; this is enforced on the interpolated values. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + examples=[["data", "object", "lines", "has_more"]], + title="Truncation Indicator Path", + ) + truncated_list_retriever: Optional[Union[SimpleRetriever, CustomRetriever]] = Field( + None, + description="Retriever used to fetch the complete list of items when the field at `truncation_indicator_path` is truthy on a record. The record being expanded is exposed to the retriever's interpolation context as `stream_slice['parent_record']`. One fetch is issued per truncated record, so enable `use_cache` on the requester when the same list can be fetched repeatedly. Configure a `paginator`, since without one only the first page of the complete list is read. If the retriever returns no records, the embedded items are expanded as a fallback; if it returns fewer records than the `total_count` field next to the indicator, a WARNING is logged once per stream. Request failures surface through the retriever's `error_handler` and fail the stream like any other request. `$parameters` of the enclosing stream propagate into this retriever's components (including `request_parameters` on its requester); move request-shaping parameters to the outer requester's `request_parameters` when adopting this field. `partition_router` and `pagination_reset` are not supported. In Connector Builder test reads, its requests appear as auxiliary requests and the test-read page limit applies to each fetch independently, so the fetched list may be shorter than `total_count`; the incomplete-fetch warning is not emitted in test reads when a `paginator` is configured (without one the retriever is not capped, so the warning still applies). Requires `truncation_indicator_path`. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + title="Truncated List Retriever", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class SessionTokenRequestApiKeyAuthenticator(BaseModel): type: Literal["ApiKey"] inject_into: RequestOption = Field( @@ -2582,6 +2637,47 @@ class Config: ) +class DpathExtractor(BaseModel): + type: Literal["DpathExtractor"] + field_path: List[str] = Field( + ..., + description='List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).', + examples=[ + ["data"], + ["data", "records"], + ["data", "{{ parameters.name }}"], + ["data", "*", "record"], + ], + title="Field Path", + ) + record_expander: Optional[RecordExpander] = Field( + None, + description="Optional component to expand records by extracting items from nested array fields.", + title="Record Expander", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + +class CombinedExtractor(BaseModel): + type: Literal["CombinedExtractor"] + extractors: List[Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor]] = Field( + ..., + description="The record extractors to combine. At least one is required. Each sub-extractor is given the same HTTP response and decodes it independently, so the response is parsed once per sub-extractor. Streaming decoders (CsvDecoder, JsonlDecoder, JsonItemsDecoder, GzipDecoder, IterableDecoder) can only read the response body once and are rejected.", + title="Extractors", + ) + mode: Optional[CombineMode] = Field( + CombineMode.union, + description='How the records of the sub-extractors are combined. "union" (default) yields every record of every sub-extractor, in the order the extractors are declared. "first_match" yields the records of the first sub-extractor that produces at least one record and skips the remaining ones; nothing is yielded if none of them produces a record. Note that a paginator which counts the records of a page counts the combined records: under "union" that is the sum over all sub-extractors, which overshoots the API page size, so "union" is rejected with an OffsetIncrement paginator because the offset would skip records. Under "first_match" the count is the count of the winning sub-extractor, which is usually the number of records the API returned for the page.', + title="Combine Mode", + ) + skip_empty_records: Optional[bool] = Field( + False, + description='Whether to drop empty records - null, {}, [], "" - yielded by a sub-extractor before the records are combined. Off by default, so the output of a sub-extractor is passed through as it is. Turn it on when the API can return nulls in the middle of a record list, which a GraphQL API does when it answers a partial response and reports the failure in a sibling error field. Under "first_match" this also changes which sub-extractor wins: a sub-extractor whose records are all empty no longer counts as a match, so the next one is tried. The dropped records still count toward the page size of an OffsetIncrement or PageIncrement paginator, so a full page holding a null does not end pagination early.', + title="Skip Empty Records", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class ZipfileDecoder(BaseModel): class Config: extra = Extra.allow @@ -2594,6 +2690,27 @@ class Config: ) +class RecordSelector(BaseModel): + type: Literal["RecordSelector"] + extractor: Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor] + record_filter: Optional[Union[RecordFilter, CustomRecordFilter]] = Field( + None, + description="Responsible for filtering records to be emitted by the Source.", + title="Record Filter", + ) + schema_normalization: Optional[Union[SchemaNormalization, CustomSchemaNormalization]] = Field( + None, + description="Responsible for normalization according to the schema.", + title="Schema Normalization", + ) + transform_before_filtering: Optional[bool] = Field( + None, + description="If true, transformation will be applied before record filtering.", + title="Transform Before Filtering", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class ConfigMigration(BaseModel): type: Literal["ConfigMigration"] description: Optional[str] = Field( @@ -2911,94 +3028,6 @@ class Config: parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") -class CombinedExtractor(BaseModel): - type: Literal["CombinedExtractor"] - extractors: List[Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor]] = Field( - ..., - description="The record extractors to combine. At least one is required. Each sub-extractor is given the same HTTP response and decodes it independently, so the response is parsed once per sub-extractor. Streaming decoders (CsvDecoder, JsonlDecoder, JsonItemsDecoder, GzipDecoder, IterableDecoder) can only read the response body once and are rejected.", - title="Extractors", - ) - mode: Optional[CombineMode] = Field( - CombineMode.union, - description='How the records of the sub-extractors are combined. "union" (default) yields every record of every sub-extractor, in the order the extractors are declared. "first_match" yields the records of the first sub-extractor that produces at least one record and skips the remaining ones; nothing is yielded if none of them produces a record. Note that a paginator which counts the records of a page counts the combined records: under "union" that is the sum over all sub-extractors, which overshoots the API page size, so "union" is rejected with an OffsetIncrement paginator because the offset would skip records. Under "first_match" the count is the count of the winning sub-extractor, which is usually the number of records the API returned for the page.', - title="Combine Mode", - ) - skip_empty_records: Optional[bool] = Field( - False, - description='Whether to drop empty records - null, {}, [], "" - yielded by a sub-extractor before the records are combined. Off by default, so the output of a sub-extractor is passed through as it is. Turn it on when the API can return nulls in the middle of a record list, which a GraphQL API does when it answers a partial response and reports the failure in a sibling error field. Under "first_match" this also changes which sub-extractor wins: a sub-extractor whose records are all empty no longer counts as a match, so the next one is tried. The dropped records still count toward the page size of an OffsetIncrement or PageIncrement paginator, so a full page holding a null does not end pagination early.', - title="Skip Empty Records", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - -class DpathExtractor(BaseModel): - type: Literal["DpathExtractor"] - field_path: List[str] = Field( - ..., - description='List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).', - examples=[ - ["data"], - ["data", "records"], - ["data", "{{ parameters.name }}"], - ["data", "*", "record"], - ], - title="Field Path", - ) - record_expander: Optional[RecordExpander] = Field( - None, - description="Optional component to expand records by extracting items from nested array fields.", - title="Record Expander", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - -class RecordExpander(BaseModel): - type: Literal["RecordExpander"] - expand_records_from_field: List[str] = Field( - ..., - description="Path to a nested array field within each record. Items from this array will be extracted and emitted as separate records. Supports wildcards (*) for matching multiple arrays.", - examples=[ - ["lines", "data"], - ["items"], - ["nested", "array"], - ["sections", "*", "items"], - ], - title="Expand Records From Field", - ) - remain_original_record: Optional[bool] = Field( - False, - description='If true, each expanded record will include the original parent record in an "original_record" field. Defaults to false.', - title="Remain Original Record", - ) - parent_fields: Optional[List[ParentFieldPath]] = Field( - None, - description="Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. A copied object or array is deep-copied, so a transformation that writes inside it affects only that item. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", - title="Parent Fields", - ) - merge_parent: Optional[bool] = Field( - False, - description="If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. Each item receives its own deep copy of the merged parent fields, so a transformation that writes into a nested value affects only that item. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", - title="Merge Parent", - ) - on_no_records: Optional[OnNoRecords] = Field( - OnNoRecords.skip, - description='Behavior when the expansion path is missing, not a list, or an empty list. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged.', - title="On No Records", - ) - truncation_indicator_path: Optional[List[str]] = Field( - None, - description="Path within each record to a field indicating that the embedded nested list is truncated (e.g. a `has_more` flag on the list object). When the field evaluates to a truthy value and `truncated_list_retriever` is configured, the retriever is used to fetch the complete list instead of expanding the embedded items. When the field is truthy and no retriever is configured, the embedded items are expanded as normal and a WARNING is logged once per stream so the truncation is visible instead of silent. Glob characters (`*`, `?`, `[`) are not supported in this path, nor in `expand_records_from_field` when a retriever is configured; this is enforced on the interpolated values. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", - examples=[["data", "object", "lines", "has_more"]], - title="Truncation Indicator Path", - ) - truncated_list_retriever: Optional[Union[SimpleRetriever, CustomRetriever]] = Field( - None, - description="Retriever used to fetch the complete list of items when the field at `truncation_indicator_path` is truthy on a record. The record being expanded is exposed to the retriever's interpolation context as `stream_slice['parent_record']`. One fetch is issued per truncated record, so enable `use_cache` on the requester when the same list can be fetched repeatedly. Configure a `paginator`, since without one only the first page of the complete list is read. If the retriever returns no records, the embedded items are expanded as a fallback; if it returns fewer records than the `total_count` field next to the indicator, a WARNING is logged once per stream. Request failures surface through the retriever's `error_handler` and fail the stream like any other request. `$parameters` of the enclosing stream propagate into this retriever's components (including `request_parameters` on its requester); move request-shaping parameters to the outer requester's `request_parameters` when adopting this field. `partition_router` and `pagination_reset` are not supported. In Connector Builder test reads, its requests appear as auxiliary requests and the test-read page limit applies to each fetch independently, so the fetched list may be shorter than `total_count`; the incomplete-fetch warning is not emitted in test reads when a `paginator` is configured (without one the retriever is not capped, so the warning still applies). Requires `truncation_indicator_path`. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", - title="Truncated List Retriever", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - class SessionTokenAuthenticator(BaseModel): type: Literal["SessionTokenAuthenticator"] login_requester: HttpRequester = Field( @@ -3183,16 +3212,18 @@ class HttpRequester(BaseModelWithDeprecations): description="Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records).", title="Use Cache", ) - connect_timeout_in_seconds: Optional[PositiveFloat] = Field( + connect_timeout_in_seconds: Optional[float] = Field( None, description="Maximum time to establish the connection (TCP + TLS handshake). When unset, no connect timeout is applied.", examples=[30], + gt=0, title="Connect Timeout (seconds)", ) - read_timeout_in_seconds: Optional[PositiveFloat] = Field( + read_timeout_in_seconds: Optional[float] = Field( None, description="The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes. Each timed-out attempt counts against the error handler's retry window (600 s for manifest connectors), so a value near 300 allows only a few retries.", examples=[300], + gt=0, title="Read Timeout (seconds)", ) use_tcp_keepalive: Optional[bool] = Field( @@ -3317,27 +3348,6 @@ class QueryProperties(BaseModel): parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") -class RecordSelector(BaseModel): - type: Literal["RecordSelector"] - extractor: Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor] - record_filter: Optional[Union[RecordFilter, CustomRecordFilter]] = Field( - None, - description="Responsible for filtering records to be emitted by the Source.", - title="Record Filter", - ) - schema_normalization: Optional[Union[SchemaNormalization, CustomSchemaNormalization]] = Field( - None, - description="Responsible for normalization according to the schema.", - title="Schema Normalization", - ) - transform_before_filtering: Optional[bool] = Field( - None, - description="If true, transformation will be applied before record filtering.", - title="Transform Before Filtering", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - class StateDelegatingStream(BaseModel): type: Literal["StateDelegatingStream"] name: str = Field(..., description="The stream name.", example=["Users"], title="Name") @@ -3451,12 +3461,7 @@ class AsyncRetriever(BaseModel): description="Responsible for fetching the final result `urls` provided by the completed / finished / ready async job.", ) download_extractor: Optional[ - Union[ - DpathExtractor, - CustomRecordExtractor, - ResponseToFileExtractor, - CombinedExtractor, - ] + Union[DpathExtractor, CustomRecordExtractor, ResponseToFileExtractor, CombinedExtractor] ] = Field(None, description="Responsible for fetching the records from provided urls.") creation_requester: Union[HttpRequester, CustomRequester] = Field( ..., @@ -3470,9 +3475,10 @@ class AsyncRetriever(BaseModel): None, description="The time in minutes after which the single Async Job should be considered as Timed Out.", ) - failed_retry_wait_time_in_seconds: Optional[Union[conint(ge=1), str]] = Field( + failed_retry_wait_time_in_seconds: Optional[Union[int, str]] = Field( None, description="Time in seconds to wait before retrying a failed async job. Only applies to jobs that ran on the API side and reported a FAILED status (e.g. report generation failed due to a cooldown). Creation failures (HTTP errors when starting a job, such as 429s) and TIMED_OUT jobs are retried immediately and are not affected by this setting. When set, the orchestrator defers retry of real failed jobs until the wait time has elapsed, without blocking other jobs.", + ge=1, ) download_target_requester: Optional[Union[HttpRequester, CustomRequester]] = Field( None, @@ -3659,13 +3665,13 @@ class DynamicDeclarativeStream(BaseModel): ComplexFieldType.update_forward_refs() GzipDecoder.update_forward_refs() CompositeErrorHandler.update_forward_refs() +CombinedExtractor.update_forward_refs() DeclarativeSource1.update_forward_refs() DeclarativeSource2.update_forward_refs() SelectiveAuthenticator.update_forward_refs() ConditionalStreams.update_forward_refs() FileUploader.update_forward_refs() DeclarativeStream.update_forward_refs() -CombinedExtractor.update_forward_refs() DpathExtractor.update_forward_refs() RecordExpander.update_forward_refs() SessionTokenAuthenticator.update_forward_refs() From eeceed3597f02a6456f32cbd9383c98f8f5968b4 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 15 Oct 2026 13:26:25 +0000 Subject: [PATCH 3/7] chore: explain non-JSON fallback in ConstantBackoffStrategy Co-Authored-By: Daryna Ishchenko --- .../backoff_strategies/constant_backoff_strategy.py | 1 + 1 file changed, 1 insertion(+) diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py index 67812192fe..710d5a791b 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py @@ -65,6 +65,7 @@ def backoff_time( try: response = response_or_exception.json() except requests.exceptions.JSONDecodeError: + # Non-JSON body: interpolate with an empty response. pass headers = response_or_exception.headers From 8b8c451640eb60d7af0c27477196dfb35c9adddf Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 15 Oct 2026 13:36:34 +0000 Subject: [PATCH 4/7] feat(declarative): add WaitTimeFromResponse backoff strategy Co-Authored-By: Daryna Ishchenko --- .../declarative_component_schema.yaml | 53 +++-- .../models/declarative_component_schema.py | 31 ++- .../parsers/model_to_component_factory.py | 17 +- .../backoff_strategies/__init__.py | 4 + .../constant_backoff_strategy.py | 61 +----- ...ait_time_from_response_backoff_strategy.py | 81 +++++++ .../test_model_to_component_factory.py | 9 +- .../test_constant_backoff.py | 202 ------------------ .../test_wait_time_from_response.py | 179 ++++++++++++++++ .../test_default_error_handler.py | 33 +-- 10 files changed, 373 insertions(+), 297 deletions(-) create mode 100644 airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py create mode 100644 unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index e00a84c6ff..83b1aca169 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -709,7 +709,7 @@ definitions: enum: [ConstantBackoffStrategy] backoff_time_in_seconds: title: Backoff Time - description: Backoff time in seconds. If the expression evaluates to an empty result, the next backoff strategy is used. + description: Backoff time in seconds. anyOf: - type: number title: Number of seconds @@ -718,24 +718,10 @@ definitions: title: Interpolated Value interpolation_context: - config - - response - - headers examples: - 30 - 30.5 - "{{ config['backoff_time'] }}" - - "{{ response.retry_in }}" - max_waiting_time_in_seconds: - title: Max Waiting Time in Seconds - description: 'Stop the stream instead of waiting, when the computed backoff is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from connector config so that the bound can be changed without a connector release. A value of 0 means never wait. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.' - anyOf: - - type: number - - type: string - interpolation_context: - - config - examples: - - 3600 - - "{{ config['max_waiting_time'] * 60 }}" jitter_range_in_seconds: title: Jitter Range description: Optional additive jitter range in seconds. When set, the backoff time is uniformly distributed between backoff_time_in_seconds and backoff_time_in_seconds + (jitter_range_in_seconds * 2), so jitter only increases the base backoff. @@ -2148,6 +2134,7 @@ definitions: - "$ref": "#/definitions/ConstantBackoffStrategy" - "$ref": "#/definitions/ExponentialBackoffStrategy" - "$ref": "#/definitions/WaitTimeFromHeader" + - "$ref": "#/definitions/WaitTimeFromResponse" - "$ref": "#/definitions/WaitUntilTimeFromHeader" - "$ref": "#/definitions/CustomBackoffStrategy" max_retries: @@ -5131,6 +5118,42 @@ definitions: $parameters: type: object additionalProperties: true + WaitTimeFromResponse: + title: Wait Time Extracted From Response + description: Compute wait time from the response body or headers. If the expression evaluates to an empty result, the next backoff strategy is used. + type: object + required: + - type + - wait_time_in_seconds + properties: + type: + type: string + enum: [WaitTimeFromResponse] + wait_time_in_seconds: + title: Wait Time in Seconds + description: The expression defining how long to wait before retrying. An empty result defers to the next backoff strategy. + type: string + interpolation_context: + - config + - response + - headers + examples: + - "{{ response.retry_after_seconds }}" + - "{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response.get('error', {}).get('message', '') | regex_search('Retry after (\\S+Z)')) else '' }}" + max_waiting_time_in_seconds: + title: Max Waiting Time in Seconds + description: 'Stop the stream instead of waiting, when the computed wait is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.' + anyOf: + - type: number + - type: string + interpolation_context: + - config + examples: + - 3600 + - "{{ config['max_waiting_time'] * 60 }}" + $parameters: + type: object + additionalProperties: true WaitUntilTimeFromHeader: title: Wait Until Time Defined In Response Header description: Extract time at which we can retry the request from response header and wait for the difference between now and that time. diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index ae8fb976c7..1b12c002eb 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -136,16 +136,10 @@ class ConstantBackoffStrategy(BaseModel): type: Literal["ConstantBackoffStrategy"] backoff_time_in_seconds: Union[float, str] = Field( ..., - description="Backoff time in seconds. If the expression evaluates to an empty result, the next backoff strategy is used.", - examples=[30, 30.5, "{{ config['backoff_time'] }}", "{{ response.retry_in }}"], + description="Backoff time in seconds.", + examples=[30, 30.5, "{{ config['backoff_time'] }}"], title="Backoff Time", ) - max_waiting_time_in_seconds: Optional[Union[float, str]] = Field( - None, - description="Stop the stream instead of waiting, when the computed backoff is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from connector config so that the bound can be changed without a connector release. A value of 0 means never wait. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.", - examples=[3600, "{{ config['max_waiting_time'] * 60 }}"], - title="Max Waiting Time in Seconds", - ) jitter_range_in_seconds: Optional[float] = Field( None, description="Optional additive jitter range in seconds. When set, the backoff time is uniformly distributed between backoff_time_in_seconds and backoff_time_in_seconds + (jitter_range_in_seconds * 2), so jitter only increases the base backoff.", @@ -1461,6 +1455,26 @@ class WaitTimeFromHeader(BaseModel): parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") +class WaitTimeFromResponse(BaseModel): + type: Literal["WaitTimeFromResponse"] + wait_time_in_seconds: str = Field( + ..., + description="The expression defining how long to wait before retrying. An empty result defers to the next backoff strategy.", + examples=[ + "{{ response.retry_after_seconds }}", + "{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response.get('error', {}).get('message', '') | regex_search('Retry after (\\S+Z)')) else '' }}", + ], + title="Wait Time in Seconds", + ) + max_waiting_time_in_seconds: Optional[Union[float, str]] = Field( + None, + description="Stop the stream instead of waiting, when the computed wait is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.", + examples=[3600, "{{ config['max_waiting_time'] * 60 }}"], + title="Max Waiting Time in Seconds", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class WaitUntilTimeFromHeader(BaseModel): type: Literal["WaitUntilTimeFromHeader"] header: str = Field( @@ -2302,6 +2316,7 @@ class DefaultErrorHandler(BaseModel): ConstantBackoffStrategy, ExponentialBackoffStrategy, WaitTimeFromHeader, + WaitTimeFromResponse, WaitUntilTimeFromHeader, CustomBackoffStrategy, ] 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 9b3b71a74f..86d696af74 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -495,6 +495,9 @@ from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( WaitTimeFromHeader as WaitTimeFromHeaderModel, ) +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + WaitTimeFromResponse as WaitTimeFromResponseModel, +) from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( WaitUntilTimeFromHeader as WaitUntilTimeFromHeaderModel, ) @@ -538,6 +541,7 @@ ConstantBackoffStrategy, ExponentialBackoffStrategy, WaitTimeFromHeaderBackoffStrategy, + WaitTimeFromResponseBackoffStrategy, WaitUntilTimeFromHeaderBackoffStrategy, ) from airbyte_cdk.sources.declarative.requesters.http_job_repository import AsyncHttpJobRepository @@ -885,6 +889,7 @@ def _init_mappings(self) -> None: SubstreamPartitionRouterModel: self.create_substream_partition_router, ValidateAdheresToSchemaModel: self.create_validate_adheres_to_schema, WaitTimeFromHeaderModel: self.create_wait_time_from_header, + WaitTimeFromResponseModel: self.create_wait_time_from_response, WaitUntilTimeFromHeaderModel: self.create_wait_until_time_from_header, AsyncRetrieverModel: self.create_async_retriever, HttpComponentsResolverModel: self.create_http_components_resolver, @@ -1855,7 +1860,6 @@ def create_constant_backoff_strategy( return ConstantBackoffStrategy( backoff_time_in_seconds=model.backoff_time_in_seconds, jitter_range_in_seconds=model.jitter_range_in_seconds, - max_waiting_time_in_seconds=model.max_waiting_time_in_seconds, config=config, parameters=model.parameters or {}, ) @@ -5189,6 +5193,17 @@ def create_wait_time_from_header( max_waiting_time_in_seconds=model.max_waiting_time_in_seconds, ) + @staticmethod + def create_wait_time_from_response( + model: WaitTimeFromResponseModel, config: Config, **kwargs: Any + ) -> WaitTimeFromResponseBackoffStrategy: + return WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds=model.wait_time_in_seconds, + parameters=model.parameters or {}, + config=config, + max_waiting_time_in_seconds=model.max_waiting_time_in_seconds, + ) + @staticmethod def create_wait_until_time_from_header( model: WaitUntilTimeFromHeaderModel, config: Config, **kwargs: Any diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/__init__.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/__init__.py index 26ecafbde7..4900813817 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/__init__.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/__init__.py @@ -11,6 +11,9 @@ from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.wait_time_from_header_backoff_strategy import ( WaitTimeFromHeaderBackoffStrategy, ) +from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.wait_time_from_response_backoff_strategy import ( + WaitTimeFromResponseBackoffStrategy, +) from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.wait_until_time_from_header_backoff_strategy import ( WaitUntilTimeFromHeaderBackoffStrategy, ) @@ -19,5 +22,6 @@ "ConstantBackoffStrategy", "ExponentialBackoffStrategy", "WaitTimeFromHeaderBackoffStrategy", + "WaitTimeFromResponseBackoffStrategy", "WaitUntilTimeFromHeaderBackoffStrategy", ] diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py index 710d5a791b..931183b2ef 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py @@ -8,35 +8,24 @@ import requests -from airbyte_cdk.models import FailureType from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString -from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.max_waiting_time_helper import ( - evaluate_max_waiting_time, - interpolated_max_waiting_time, -) from airbyte_cdk.sources.streams.http.error_handlers import BackoffStrategy from airbyte_cdk.sources.types import Config -from airbyte_cdk.utils import AirbyteTracedException @dataclass class ConstantBackoffStrategy(BackoffStrategy): - """Compute a constant backoff from config, response JSON, and response headers. - - An empty interpolated result defers to the next strategy. The optional cap is evaluated from - config and checked against the final backoff after jitter. + """ + Backoff strategy with a constant backoff interval Attributes: - backoff_time_in_seconds: Backoff time in seconds. - jitter_range_in_seconds: Optional additive jitter range that only increases the wait. - max_waiting_time_in_seconds: Optional inclusive cap on the computed wait after jitter. + backoff_time_in_seconds (float): time to backoff before retrying a retryable request. """ backoff_time_in_seconds: Union[float, InterpolatedString, str] parameters: InitVar[Mapping[str, Any]] config: Config jitter_range_in_seconds: Optional[float] = None - max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None def __post_init__(self, parameters: Mapping[str, Any]) -> None: if not isinstance(self.backoff_time_in_seconds, InterpolatedString): @@ -49,50 +38,16 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None: self.backoff_time_in_seconds = InterpolatedString.create( self.backoff_time_in_seconds, parameters=parameters ) - self._max_waiting_time_in_seconds = interpolated_max_waiting_time( - self.max_waiting_time_in_seconds, parameters - ) - evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) def backoff_time( self, response_or_exception: Optional[Union[requests.Response, requests.RequestException]], attempt_count: int, ) -> Optional[float]: - response: Any = {} - headers: Mapping[str, Any] = {} - if isinstance(response_or_exception, requests.Response): - try: - response = response_or_exception.json() - except requests.exceptions.JSONDecodeError: - # Non-JSON body: interpolate with an empty response. - pass - headers = response_or_exception.headers - - evaluated_backoff_time = cast(InterpolatedString, self.backoff_time_in_seconds).eval( - self.config, response=response, headers=headers + backoff_time = float( + cast(InterpolatedString, self.backoff_time_in_seconds).eval(self.config) ) - if evaluated_backoff_time is None or ( - isinstance(evaluated_backoff_time, str) and not evaluated_backoff_time.strip() - ): - return None - - backoff_time = float(evaluated_backoff_time) - if self.jitter_range_in_seconds is not None: - backoff_time = random.uniform( - backoff_time, backoff_time + (self.jitter_range_in_seconds * 2) - ) - - max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) - if max_waiting_time is not None and backoff_time and backoff_time >= max_waiting_time: - raise AirbyteTracedException( - internal_message=( - f"Rate limit wait time {backoff_time}s is greater than or equal to the " - f"maximum of {max_waiting_time}s this stream is allowed to wait. " - f"Stopping the stream..." - ), - message="The rate limit wait time is longer than the connector is allowed to wait.", - failure_type=FailureType.transient_error, - ) + if self.jitter_range_in_seconds is None: + return backoff_time - return backoff_time + return random.uniform(backoff_time, backoff_time + (self.jitter_range_in_seconds * 2)) diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py new file mode 100644 index 0000000000..36785a8469 --- /dev/null +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py @@ -0,0 +1,81 @@ +# +# Copyright (c) 2023 Airbyte, Inc., all rights reserved. +# + +from dataclasses import InitVar, dataclass +from typing import Any, Mapping, Optional, Union, cast + +import requests + +from airbyte_cdk.models import FailureType +from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString +from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.max_waiting_time_helper import ( + evaluate_max_waiting_time, + interpolated_max_waiting_time, +) +from airbyte_cdk.sources.streams.http.error_handlers import BackoffStrategy +from airbyte_cdk.sources.types import Config +from airbyte_cdk.utils import AirbyteTracedException + + +@dataclass +class WaitTimeFromResponseBackoffStrategy(BackoffStrategy): + """ + Extract wait time from the response body or headers. + + Attributes: + wait_time_in_seconds: Expression defining how long to wait before retrying. + max_waiting_time_in_seconds: Optional inclusive cap on the computed wait. + """ + + wait_time_in_seconds: Union[InterpolatedString, str] + parameters: InitVar[Mapping[str, Any]] + config: Config + max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None + + def __post_init__(self, parameters: Mapping[str, Any]) -> None: + self.wait_time_in_seconds = InterpolatedString.create( + self.wait_time_in_seconds, parameters=parameters + ) + self._max_waiting_time_in_seconds = interpolated_max_waiting_time( + self.max_waiting_time_in_seconds, parameters + ) + evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) + + def backoff_time( + self, + response_or_exception: Optional[Union[requests.Response, requests.RequestException]], + attempt_count: int, + ) -> Optional[float]: + response: Any = {} + headers: Mapping[str, Any] = {} + if isinstance(response_or_exception, requests.Response): + try: + response = response_or_exception.json() + except requests.exceptions.JSONDecodeError: + # Non-JSON body: interpolate with an empty response. + pass + headers = response_or_exception.headers + + evaluated_wait_time = cast(InterpolatedString, self.wait_time_in_seconds).eval( + self.config, response=response, headers=headers + ) + if evaluated_wait_time is None or ( + isinstance(evaluated_wait_time, str) and not evaluated_wait_time.strip() + ): + return None + + wait_time = float(evaluated_wait_time) + max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) + if max_waiting_time is not None and wait_time and wait_time >= max_waiting_time: + raise AirbyteTracedException( + internal_message=( + f"Rate limit wait time {wait_time}s is greater than or equal to the " + f"maximum of {max_waiting_time}s this stream is allowed to wait. " + f"Stopping the stream..." + ), + message="The rate limit wait time is longer than the connector is allowed to wait.", + failure_type=FailureType.transient_error, + ) + + return wait_time 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 b19a9810de..0f993c2f14 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 @@ -189,6 +189,7 @@ ConstantBackoffStrategy, ExponentialBackoffStrategy, WaitTimeFromHeaderBackoffStrategy, + WaitTimeFromResponseBackoffStrategy, WaitUntilTimeFromHeaderBackoffStrategy, ) from airbyte_cdk.sources.declarative.requesters.http_job_repository import AsyncHttpJobRepository @@ -2786,7 +2787,7 @@ def test_create_requester_with_backoff_jitter( assert backoff_strategy.jitter_range_in_seconds == expected_jitter_range -def test_create_requester_with_constant_backoff_max_waiting_time(): +def test_create_requester_with_wait_time_from_response_strategy(): content = """ requester: type: HttpRequester @@ -2794,8 +2795,8 @@ def test_create_requester_with_constant_backoff_max_waiting_time(): url_base: "https://api.sendgrid.com" error_handler: backoff_strategies: - - type: ConstantBackoffStrategy - backoff_time_in_seconds: 5 + - type: WaitTimeFromResponse + wait_time_in_seconds: "{{ response.retry_after_seconds }}" max_waiting_time_in_seconds: 30 """ parsed_manifest = YamlDeclarativeSource._parse(content) @@ -2814,7 +2815,7 @@ def test_create_requester_with_constant_backoff_max_waiting_time(): assert isinstance(requester.error_handler, DefaultErrorHandler) backoff_strategy = requester.error_handler.backoff_strategies[0] - assert isinstance(backoff_strategy, ConstantBackoffStrategy) + assert isinstance(backoff_strategy, WaitTimeFromResponseBackoffStrategy) assert backoff_strategy.max_waiting_time_in_seconds == 30 diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py index 86c24af334..7fae93fa15 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py @@ -2,23 +2,17 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # -import json from unittest.mock import MagicMock import pytest -import requests -from freezegun import freeze_time -from airbyte_cdk.models import FailureType from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy import ( ConstantBackoffStrategy, ) -from airbyte_cdk.utils import AirbyteTracedException BACKOFF_TIME = 10 PARAMETERS_BACKOFF_TIME = 20 CONFIG_BACKOFF_TIME = 30 -GMAIL_BACKOFF_EXPRESSION = r"{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response.get('error', {}).get('message', '') | regex_search('Retry after (\\S+Z)')) else '' }}" @pytest.mark.parametrize( @@ -86,199 +80,3 @@ def test_constant_backoff_with_jitter_bounds( assert all(expected_lower_bound <= backoff <= expected_upper_bound for backoff in backoff_times) assert len(set(backoff_times)) > 1 - - -def test_constant_backoff_interpolates_response_json(): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds="{{ response.retry_in }}", parameters={}, config={} - ) - - assert backoff_strategy.backoff_time(_response({"retry_in": "12.5"}), 1) == 12.5 - - -def test_constant_backoff_interpolates_headers(): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds="{{ headers['Retry-After'] }}", parameters={}, config={} - ) - - assert backoff_strategy.backoff_time(_response({}, headers={"Retry-After": "15"}), 1) == 15.0 - - -@pytest.mark.parametrize("empty_value", ["", " "]) -def test_constant_backoff_returns_none_for_empty_interpolation(empty_value): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} - ) - - assert backoff_strategy.backoff_time(_response({"retry_in": empty_value}), 1) is None - - -def test_constant_backoff_returns_none_when_json_key_is_missing(): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} - ) - - assert backoff_strategy.backoff_time(_response({}), 1) is None - - -def test_constant_backoff_returns_none_for_non_json_body(): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} - ) - - assert backoff_strategy.backoff_time(_response_body(b"not JSON"), 1) is None - - -@pytest.mark.parametrize( - "response_or_exception", - [None, requests.RequestException()], - ids=["none", "exception"], -) -def test_constant_backoff_returns_none_for_non_response_input(response_or_exception): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} - ) - - assert backoff_strategy.backoff_time(response_or_exception, 1) is None - - -def test_constant_backoff_does_not_apply_jitter_to_empty_interpolation(mocker): - uniform = mocker.patch( - "airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy.random.uniform" - ) - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds="{{ response.get('retry_in', '') }}", - parameters={}, - config={}, - jitter_range_in_seconds=5, - ) - - assert backoff_strategy.backoff_time(_response({}), 1) is None - uniform.assert_not_called() - - -@pytest.mark.parametrize("wait_time", [10, 11], ids=["equal_to_cap", "above_cap"]) -def test_constant_backoff_wait_at_or_above_cap_raises_transient_error(wait_time): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds="{{ response.retry_in }}", - parameters={}, - config={}, - max_waiting_time_in_seconds=10, - ) - - with pytest.raises(AirbyteTracedException) as exception: - backoff_strategy.backoff_time(_response({"retry_in": wait_time}), 1) - - assert exception.value.failure_type == FailureType.transient_error - assert ( - exception.value.message - == "The rate limit wait time is longer than the connector is allowed to wait." - ) - assert "greater than or equal to the maximum of 10.0s" in exception.value.internal_message - - -def test_constant_backoff_wait_below_cap_is_returned(): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds=9, - parameters={}, - config={}, - max_waiting_time_in_seconds=10, - ) - - assert backoff_strategy.backoff_time(None, 1) == 9.0 - - -def test_constant_backoff_cap_is_interpolated_from_config(): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds="{{ response.retry_in }}", - parameters={}, - config={"max_waiting_time": 20}, - max_waiting_time_in_seconds="{{ config['max_waiting_time'] }}", - ) - - assert backoff_strategy.backoff_time(_response({"retry_in": 10}), 1) == 10.0 - - -def test_constant_backoff_invalid_cap_fails_at_construction(): - with pytest.raises(AirbyteTracedException) as exception: - ConstantBackoffStrategy( - backoff_time_in_seconds=10, - parameters={}, - config={"max_waiting_time": "not a number"}, - max_waiting_time_in_seconds="{{ config['max_waiting_time'] }}", - ) - - assert exception.value.failure_type == FailureType.system_error - - -def test_constant_backoff_zero_wait_is_not_refused_by_zero_cap(): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds=0, - parameters={}, - config={}, - max_waiting_time_in_seconds=0, - ) - - assert backoff_strategy.backoff_time(None, 1) == 0 - - -def test_constant_backoff_cap_is_checked_after_jitter(mocker): - mocker.patch( - "airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy.random.uniform", - return_value=11, - ) - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds=10, - parameters={}, - config={}, - jitter_range_in_seconds=1, - max_waiting_time_in_seconds=11, - ) - - with pytest.raises(AirbyteTracedException) as exception: - backoff_strategy.backoff_time(None, 1) - - assert exception.value.failure_type == FailureType.transient_error - - -@freeze_time("2026-10-15T13:00:00Z") -@pytest.mark.parametrize( - "message, expected_backoff", - [ - pytest.param( - "User-rate limit exceeded. Retry after 2026-10-15T13:15:00Z", - 900, - id="future_timestamp", - ), - pytest.param( - "User-rate limit exceeded. Retry after 2026-10-15T12:59:59Z", - 1, - id="past_timestamp", - ), - pytest.param("User-rate limit exceeded.", None, id="missing_timestamp"), - ], -) -def test_constant_backoff_gmail_retry_after_expression(message, expected_backoff): - backoff_strategy = ConstantBackoffStrategy( - backoff_time_in_seconds=GMAIL_BACKOFF_EXPRESSION, parameters={}, config={} - ) - - actual_backoff = backoff_strategy.backoff_time(_response({"error": {"message": message}}), 1) - - if expected_backoff is None: - assert actual_backoff is None - else: - assert actual_backoff == pytest.approx(expected_backoff) - - -def _response(json_body, headers=None): - return _response_body(json.dumps(json_body).encode(), headers=headers) - - -def _response_body(body, headers=None): - response = requests.Response() - response.status_code = 429 - response.headers = headers or {} - response._content = body - response.encoding = "utf-8" - return response diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py new file mode 100644 index 0000000000..7ad1cddac7 --- /dev/null +++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py @@ -0,0 +1,179 @@ +# +# Copyright (c) 2023 Airbyte, Inc., all rights reserved. +# + +import json + +import pytest +import requests +from freezegun import freeze_time + +from airbyte_cdk.models import FailureType +from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.wait_time_from_response_backoff_strategy import ( + WaitTimeFromResponseBackoffStrategy, +) +from airbyte_cdk.utils import AirbyteTracedException + +GMAIL_WAIT_TIME_EXPRESSION = r"{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response.get('error', {}).get('message', '') | regex_search('Retry after (\\S+Z)')) else '' }}" + + +def test_wait_time_from_response_interpolates_json_body(): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.retry_in }}", parameters={}, config={} + ) + + assert strategy.backoff_time(_response({"retry_in": "12.5"}), 1) == 12.5 + + +def test_wait_time_from_response_interpolates_headers(): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ headers['Retry-After'] }}", parameters={}, config={} + ) + + assert strategy.backoff_time(_response({}, headers={"Retry-After": "15"}), 1) == 15.0 + + +@pytest.mark.parametrize("empty_value", ["", " "]) +def test_wait_time_from_response_returns_none_for_empty_result(empty_value): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} + ) + + assert strategy.backoff_time(_response({"retry_in": empty_value}), 1) is None + + +def test_wait_time_from_response_returns_none_for_missing_key(): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} + ) + + assert strategy.backoff_time(_response({}), 1) is None + + +def test_wait_time_from_response_returns_none_for_non_json_body(): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} + ) + + assert strategy.backoff_time(_response_body(b"not JSON"), 1) is None + + +@pytest.mark.parametrize( + "response_or_exception", + [None, requests.RequestException()], + ids=["none", "exception"], +) +def test_wait_time_from_response_returns_none_for_non_response_input(response_or_exception): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.get('retry_in', '') }}", parameters={}, config={} + ) + + assert strategy.backoff_time(response_or_exception, 1) is None + + +@pytest.mark.parametrize("wait_time", [10, 11], ids=["equal_to_cap", "above_cap"]) +def test_wait_time_from_response_at_or_above_cap_raises_transient_error(wait_time): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.retry_in }}", + parameters={}, + config={}, + max_waiting_time_in_seconds=10, + ) + + with pytest.raises(AirbyteTracedException) as exception: + strategy.backoff_time(_response({"retry_in": wait_time}), 1) + + assert exception.value.failure_type == FailureType.transient_error + assert ( + exception.value.message + == "The rate limit wait time is longer than the connector is allowed to wait." + ) + assert "greater than or equal to the maximum of 10.0s" in exception.value.internal_message + + +def test_wait_time_from_response_below_cap_is_returned(): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.retry_in }}", + parameters={}, + config={}, + max_waiting_time_in_seconds=10, + ) + + assert strategy.backoff_time(_response({"retry_in": 9}), 1) == 9.0 + + +def test_wait_time_from_response_cap_is_interpolated_from_config(): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.retry_in }}", + parameters={}, + config={"max_waiting_time": 20}, + max_waiting_time_in_seconds="{{ config['max_waiting_time'] }}", + ) + + assert strategy.backoff_time(_response({"retry_in": 10}), 1) == 10.0 + + +def test_wait_time_from_response_invalid_cap_fails_at_construction(): + with pytest.raises(AirbyteTracedException) as exception: + WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="10", + parameters={}, + config={"max_waiting_time": "not a number"}, + max_waiting_time_in_seconds="{{ config['max_waiting_time'] }}", + ) + + assert exception.value.failure_type == FailureType.system_error + + +def test_wait_time_from_response_zero_wait_is_not_refused_by_zero_cap(): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.retry_in }}", + parameters={}, + config={}, + max_waiting_time_in_seconds=0, + ) + + assert strategy.backoff_time(_response({"retry_in": 0}), 1) == 0 + + +@freeze_time("2026-10-15T13:00:00Z") +@pytest.mark.parametrize( + "message, expected_wait", + [ + pytest.param( + "User-rate limit exceeded. Retry after 2026-10-15T13:15:00Z", + 900, + id="future_timestamp", + ), + pytest.param( + "User-rate limit exceeded. Retry after 2026-10-15T12:59:59Z", + 1, + id="past_timestamp", + ), + pytest.param("User-rate limit exceeded.", None, id="missing_timestamp"), + ], +) +def test_wait_time_from_response_gmail_expression(message, expected_wait): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds=GMAIL_WAIT_TIME_EXPRESSION, parameters={}, config={} + ) + + actual_wait = strategy.backoff_time(_response({"error": {"message": message}}), 1) + + if expected_wait is None: + assert actual_wait is None + else: + assert actual_wait == pytest.approx(expected_wait) + + +def _response(json_body, headers=None): + return _response_body(json.dumps(json_body).encode(), headers=headers) + + +def _response_body(body, headers=None): + response = requests.Response() + response.status_code = 429 + response.headers = headers or {} + response._content = body + response.encoding = "utf-8" + return response diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py index 713f2de699..09d25f5a47 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py @@ -13,6 +13,9 @@ from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.exponential_backoff_strategy import ( ExponentialBackoffStrategy, ) +from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.wait_time_from_response_backoff_strategy import ( + WaitTimeFromResponseBackoffStrategy, +) from airbyte_cdk.sources.declarative.requesters.error_handlers.default_error_handler import ( DefaultErrorHandler, HttpResponseFilter, @@ -29,6 +32,22 @@ SOME_BACKOFF_TIME = 60 +def test_default_error_handler_falls_back_when_response_backoff_is_empty(): + response_mock = create_response(429, json_body={"retry_in": ""}) + error_handler = DefaultErrorHandler( + config={}, + parameters={}, + backoff_strategies=[ + WaitTimeFromResponseBackoffStrategy( + "{{ response.get('retry_in', '') }}", config={}, parameters={} + ), + ExponentialBackoffStrategy(factor=5, config={}, parameters={}), + ], + ) + + assert error_handler.backoff_time(response_mock, attempt_count=2) == 20 + + @pytest.mark.parametrize( "test_name, http_status_code, expected_error_resolution", [ @@ -242,20 +261,6 @@ def test_default_error_handler_with_constant_backoff_strategy(): ) -def test_default_error_handler_falls_back_when_constant_backoff_is_empty(): - response_mock = create_response(429, json_body={"retry_in": ""}) - error_handler = DefaultErrorHandler( - config={}, - parameters={}, - backoff_strategies=[ - ConstantBackoffStrategy("{{ response.get('retry_in', '') }}", config={}, parameters={}), - ExponentialBackoffStrategy(factor=5, config={}, parameters={}), - ], - ) - - assert error_handler.backoff_time(response_mock, attempt_count=2) == 20 - - @pytest.mark.parametrize( "attempt_count", [ From 9f6721d901c5d9e21598f8f9bfddc0f1bcb9e903 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:06:10 +0300 Subject: [PATCH 5/7] fix(declarative): defer to the next backoff strategy on a negative or non-numeric WaitTimeFromResponse result A wait computed as `retry_at - now` goes negative once that time has passed. Returned as-is, `HttpClient` kept it as a user-defined backoff and `time.sleep(wait + 1)` failed the stream with "sleep length must be non-negative". A non-numeric result raised a bare `ValueError` from `float()`, and NaN slipped past the cap into `sleep`. - Return `None` for a negative or non-finite wait, as `WaitUntilTimeFromHeader` does for a past timestamp. A zero wait is still returned as 0, like `WaitTimeFromHeader`. - Return `None` for a non-numeric result with a warning naming the strategy and the result's type (never its value), as the header strategies do. - Say so in the `wait_time_in_seconds` description (schema and generated model, which still matches `datamodel-codegen==0.26.3` output byte for byte). - Tests for negative, NaN, infinite and non-numeric results, and for the floor-less `timestamp(response.retry_at) - now_utc().timestamp()` expression with a past and a future time. Copyright year on the two new files. Co-Authored-By: Claude Fable 5.1 --- .../declarative_component_schema.yaml | 2 +- .../models/declarative_component_schema.py | 2 +- ...ait_time_from_response_backoff_strategy.py | 27 ++++++++++-- .../test_wait_time_from_response.py | 43 ++++++++++++++++++- 4 files changed, 68 insertions(+), 6 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 83b1aca169..aa5ffca7f2 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -5131,7 +5131,7 @@ definitions: enum: [WaitTimeFromResponse] wait_time_in_seconds: title: Wait Time in Seconds - description: The expression defining how long to wait before retrying. An empty result defers to the next backoff strategy. + description: The expression defining how long to wait before retrying. An empty, negative or non-numeric result defers to the next backoff strategy. type: string interpolation_context: - config diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 1b12c002eb..1739bca55c 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -1459,7 +1459,7 @@ class WaitTimeFromResponse(BaseModel): type: Literal["WaitTimeFromResponse"] wait_time_in_seconds: str = Field( ..., - description="The expression defining how long to wait before retrying. An empty result defers to the next backoff strategy.", + description="The expression defining how long to wait before retrying. An empty, negative or non-numeric result defers to the next backoff strategy.", examples=[ "{{ response.retry_after_seconds }}", "{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response.get('error', {}).get('message', '') | regex_search('Retry after (\\S+Z)')) else '' }}", diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py index 36785a8469..53b67f4248 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py @@ -1,7 +1,9 @@ # -# Copyright (c) 2023 Airbyte, Inc., all rights reserved. +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. # +import logging +import math from dataclasses import InitVar, dataclass from typing import Any, Mapping, Optional, Union, cast @@ -17,6 +19,8 @@ from airbyte_cdk.sources.types import Config from airbyte_cdk.utils import AirbyteTracedException +LOGGER = logging.getLogger("airbyte") + @dataclass class WaitTimeFromResponseBackoffStrategy(BackoffStrategy): @@ -24,7 +28,8 @@ class WaitTimeFromResponseBackoffStrategy(BackoffStrategy): Extract wait time from the response body or headers. Attributes: - wait_time_in_seconds: Expression defining how long to wait before retrying. + wait_time_in_seconds: Expression defining how long to wait before retrying. An empty, + negative, non-finite or non-numeric result defers to the next backoff strategy. max_waiting_time_in_seconds: Optional inclusive cap on the computed wait. """ @@ -65,7 +70,23 @@ def backoff_time( ): return None - wait_time = float(evaluated_wait_time) + try: + wait_time = float(evaluated_wait_time) + except (TypeError, ValueError): + # Mirrors the header strategies, which return None for a value they cannot read, so + # the next backoff strategy decides. The value is not logged: it comes from the + # response body and may hold anything. + LOGGER.warning( + "WaitTimeFromResponse: `wait_time_in_seconds` resolved to a non-numeric " + f"{type(evaluated_wait_time).__name__}; deferring to the next backoff strategy." + ) + return None + if not math.isfinite(wait_time) or wait_time < 0: + # A retry time that has already passed gives a negative wait, which is not a wait to + # take: the next strategy decides, as `WaitUntilTimeFromHeader` does for a past + # timestamp. Returned as-is, a negative value would reach `time.sleep` and fail the + # stream. A zero wait is returned as 0, like `WaitTimeFromHeader` for a `0` header. + return None max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) if max_waiting_time is not None and wait_time and wait_time >= max_waiting_time: raise AirbyteTracedException( diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py index 7ad1cddac7..3828f4890d 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py @@ -1,5 +1,5 @@ # -# Copyright (c) 2023 Airbyte, Inc., all rights reserved. +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. # import json @@ -166,6 +166,47 @@ def test_wait_time_from_response_gmail_expression(message, expected_wait): assert actual_wait == pytest.approx(expected_wait) +@pytest.mark.parametrize( + "retry_in", + [ + pytest.param(-0.5, id="negative_fraction"), + pytest.param(-30, id="negative"), + pytest.param("nan", id="not_a_number"), + pytest.param("inf", id="infinite"), + ], +) +def test_wait_time_from_response_returns_none_for_negative_or_non_finite_result(retry_in): + """A negative wait would reach `time.sleep` and fail the stream; the next strategy decides.""" + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.retry_in }}", parameters={}, config={} + ) + + assert strategy.backoff_time(_response({"retry_in": retry_in}), 1) is None + + +def test_wait_time_from_response_returns_none_for_non_numeric_result(caplog): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.retry_in }}", parameters={}, config={} + ) + + assert strategy.backoff_time(_response({"retry_in": "soon"}), 1) is None + assert any("non-numeric str" in message for message in caplog.messages) + assert all("soon" not in message for message in caplog.messages) + + +@freeze_time("2026-10-15T13:00:00Z") +def test_wait_time_from_response_past_timestamp_without_a_floor_defers(): + """`retry_at - now` is negative once the time has passed, without the `max(1, ...)` floor.""" + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ timestamp(response.retry_at) - now_utc().timestamp() }}", + parameters={}, + config={}, + ) + + assert strategy.backoff_time(_response({"retry_at": "2026-10-15T12:59:30Z"}), 1) is None + assert strategy.backoff_time(_response({"retry_at": "2026-10-15T13:00:30Z"}), 1) == 30 + + def _response(json_body, headers=None): return _response_body(json.dumps(json_body).encode(), headers=headers) From d30f8348acf775d1822d281138b2d31bae6d32c7 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:31:38 +0300 Subject: [PATCH 6/7] fix(declarative): defer when the WaitTimeFromResponse expression does not fit the response body The strategy runs on every retryable response, not only the rate limit its expression was written for. The documented Gmail example called `.get` on `response` and on `response.error`, so a 5xx with a JSON list body or `{"error": "..."}` raised `UndefinedError` out of `backoff_time` and failed the stream instead of falling back to the next strategy. - Catch `UndefinedError`, `TypeError` and `ValueError` from the evaluation, log a warning with the error type (never the body), and return `None`. - Make the example shape-safe (`response is mapping and response.get('error') is mapping and ...`) in the schema, the generated model and the test constant. The model still matches `datamodel-codegen==0.26.3` output byte for byte. - Tests: the example on a list body, a string `error`, an `error` without `message`, a null `message` and a scalar body; and an expression that raises on the body. Co-Authored-By: Claude Fable 5.1 --- .../declarative_component_schema.yaml | 2 +- .../models/declarative_component_schema.py | 2 +- ...ait_time_from_response_backoff_strategy.py | 21 ++++++++--- .../test_wait_time_from_response.py | 36 ++++++++++++++++++- 4 files changed, 54 insertions(+), 7 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index aa5ffca7f2..f67851dfe1 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -5139,7 +5139,7 @@ definitions: - headers examples: - "{{ response.retry_after_seconds }}" - - "{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response.get('error', {}).get('message', '') | regex_search('Retry after (\\S+Z)')) else '' }}" + - "{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response is mapping and response.get('error') is mapping and ((response.error.message or '') | string | regex_search('Retry after (\\S+Z)'))) else '' }}" max_waiting_time_in_seconds: title: Max Waiting Time in Seconds description: 'Stop the stream instead of waiting, when the computed wait is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.' diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 1739bca55c..6fccbd39a0 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -1462,7 +1462,7 @@ class WaitTimeFromResponse(BaseModel): description="The expression defining how long to wait before retrying. An empty, negative or non-numeric result defers to the next backoff strategy.", examples=[ "{{ response.retry_after_seconds }}", - "{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response.get('error', {}).get('message', '') | regex_search('Retry after (\\S+Z)')) else '' }}", + "{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response is mapping and response.get('error') is mapping and ((response.error.message or '') | string | regex_search('Retry after (\\S+Z)'))) else '' }}", ], title="Wait Time in Seconds", ) diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py index 53b67f4248..39a756600b 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py @@ -8,6 +8,7 @@ from typing import Any, Mapping, Optional, Union, cast import requests +from jinja2.exceptions import UndefinedError from airbyte_cdk.models import FailureType from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString @@ -29,7 +30,8 @@ class WaitTimeFromResponseBackoffStrategy(BackoffStrategy): Attributes: wait_time_in_seconds: Expression defining how long to wait before retrying. An empty, - negative, non-finite or non-numeric result defers to the next backoff strategy. + negative, non-finite or non-numeric result, or an expression that cannot be evaluated + on the response body, defers to the next backoff strategy. max_waiting_time_in_seconds: Optional inclusive cap on the computed wait. """ @@ -62,9 +64,20 @@ def backoff_time( pass headers = response_or_exception.headers - evaluated_wait_time = cast(InterpolatedString, self.wait_time_in_seconds).eval( - self.config, response=response, headers=headers - ) + try: + evaluated_wait_time = cast(InterpolatedString, self.wait_time_in_seconds).eval( + self.config, response=response, headers=headers + ) + except (UndefinedError, TypeError, ValueError) as error: + # The strategy runs on every retryable response, not only the one the expression was + # written for: a body of another shape (a list, or a string where an object was + # expected) must defer to the next strategy, not fail the stream. The body is not + # logged; the error type is enough to point at the expression. + LOGGER.warning( + "WaitTimeFromResponse: `wait_time_in_seconds` could not be evaluated on this " + f"response ({type(error).__name__}); deferring to the next backoff strategy." + ) + return None if evaluated_wait_time is None or ( isinstance(evaluated_wait_time, str) and not evaluated_wait_time.strip() ): diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py index 3828f4890d..6f7e4b509a 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py @@ -14,7 +14,7 @@ ) from airbyte_cdk.utils import AirbyteTracedException -GMAIL_WAIT_TIME_EXPRESSION = r"{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response.get('error', {}).get('message', '') | regex_search('Retry after (\\S+Z)')) else '' }}" +GMAIL_WAIT_TIME_EXPRESSION = r"{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response is mapping and response.get('error') is mapping and ((response.error.message or '') | string | regex_search('Retry after (\\S+Z)'))) else '' }}" def test_wait_time_from_response_interpolates_json_body(): @@ -207,6 +207,40 @@ def test_wait_time_from_response_past_timestamp_without_a_floor_defers(): assert strategy.backoff_time(_response({"retry_at": "2026-10-15T13:00:30Z"}), 1) == 30 +@pytest.mark.parametrize( + "body", + [ + pytest.param([{"error": {"message": "Retry after 2026-10-15T13:15:00Z"}}], id="list_body"), + pytest.param({"error": "rate limited"}, id="error_is_a_string"), + pytest.param({"error": {}}, id="error_without_message"), + pytest.param({"error": {"message": None}}, id="message_is_null"), + pytest.param(True, id="scalar_body"), + ], +) +def test_wait_time_from_response_gmail_expression_defers_on_other_body_shapes(body): + """The documented example runs on every retryable response, not only Gmail's rate limit.""" + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds=GMAIL_WAIT_TIME_EXPRESSION, parameters={}, config={} + ) + + assert strategy.backoff_time(_response(body), 1) is None + + +def test_wait_time_from_response_returns_none_when_the_expression_fails_on_the_body(caplog): + """An expression that does not fit the body defers to the next strategy instead of failing. + + `.get` on a list raises `UndefinedError` from Jinja, which `InterpolatedString.eval` re-raises + when it evaluates the default (the expression itself) a second time. + """ + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.get('retry') }}", parameters={}, config={} + ) + + assert strategy.backoff_time(_response([{"retry": "soon"}]), 1) is None + assert any("could not be evaluated" in message for message in caplog.messages) + assert all("soon" not in message for message in caplog.messages) + + def _response(json_body, headers=None): return _response_body(json.dumps(json_body).encode(), headers=headers) From 19c24ff3a70fdd525b8ea1cb4d624733eb2421d9 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Fri, 16 Oct 2026 13:09:15 +0300 Subject: [PATCH 7/7] fix(declarative): keep WaitTimeFromResponse working after an unexpected response body Addresses review feedback on the strategy's error handling and docs. - Evaluate with `JinjaInterpolation` directly. `InterpolatedString` cached the template as plain text when a render `TypeError` returned it unchanged, so one odd body disabled the strategy for the rest of the sync. A result equal to the raw template now defers. - Fail at construction when the expression uses a variable other than `config`, `response`, `headers` or `parameters`, instead of silently deferring on every response. - Catch `ArithmeticError` (division by zero, float overflow) during evaluation and conversion. - Fall back to an empty `response` on any unreadable body (`ValueError`, `RequestException`, `RecursionError`), not only `JSONDecodeError`. - Descriptions: relative seconds, JSON body only, zero/unevaluable results defer, cap strongly recommended. Fix the code comment on zero waits. - Tests: instance reuse after a failed render, arithmetic errors, unreadable body, undeclared variable, `$parameters`, factory-built `backoff_time`, and `HttpClient` end to end (sleep and cap-exceeded trace). Co-Authored-By: Claude Opus 5.5 --- .../declarative_component_schema.yaml | 4 +- .../models/declarative_component_schema.py | 2 +- ...ait_time_from_response_backoff_strategy.py | 69 +++++++++-- .../test_model_to_component_factory.py | 8 +- .../test_wait_time_from_response.py | 113 ++++++++++++++++++ 5 files changed, 179 insertions(+), 17 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index f67851dfe1..d8f2a29676 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -5120,7 +5120,7 @@ definitions: additionalProperties: true WaitTimeFromResponse: title: Wait Time Extracted From Response - description: Compute wait time from the response body or headers. If the expression evaluates to an empty result, the next backoff strategy is used. + description: Compute the wait time, in relative seconds, from the JSON response body or the response headers. A non-JSON body is exposed as an empty `response`. If the expression evaluates to an empty, zero, negative or non-numeric result, or cannot be evaluated on the response, the next backoff strategy is used. type: object required: - type @@ -5131,7 +5131,7 @@ definitions: enum: [WaitTimeFromResponse] wait_time_in_seconds: title: Wait Time in Seconds - description: The expression defining how long to wait before retrying. An empty, negative or non-numeric result defers to the next backoff strategy. + description: The expression defining how long to wait before retrying, in seconds relative to now (convert an absolute timestamp with `timestamp(...) - now_utc().timestamp()`). An empty, zero, negative or non-numeric result, or an expression that cannot be evaluated on the response, defers to the next backoff strategy. Nothing bounds the wait unless `max_waiting_time_in_seconds` is set, so setting it is strongly recommended. type: string interpolation_context: - config diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 6fccbd39a0..b79a35b777 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -1459,7 +1459,7 @@ class WaitTimeFromResponse(BaseModel): type: Literal["WaitTimeFromResponse"] wait_time_in_seconds: str = Field( ..., - description="The expression defining how long to wait before retrying. An empty, negative or non-numeric result defers to the next backoff strategy.", + description="The expression defining how long to wait before retrying, in seconds relative to now (convert an absolute timestamp with `timestamp(...) - now_utc().timestamp()`). An empty, zero, negative or non-numeric result, or an expression that cannot be evaluated on the response, defers to the next backoff strategy. Nothing bounds the wait unless `max_waiting_time_in_seconds` is set, so setting it is strongly recommended.", examples=[ "{{ response.retry_after_seconds }}", "{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response is mapping and response.get('error') is mapping and ((response.error.message or '') | string | regex_search('Retry after (\\S+Z)'))) else '' }}", diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py index 39a756600b..4409da9395 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_response_backoff_strategy.py @@ -5,13 +5,14 @@ import logging import math from dataclasses import InitVar, dataclass -from typing import Any, Mapping, Optional, Union, cast +from typing import Any, Mapping, Optional, Union import requests from jinja2.exceptions import UndefinedError from airbyte_cdk.models import FailureType from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString +from airbyte_cdk.sources.declarative.interpolation.jinja import JinjaInterpolation from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.max_waiting_time_helper import ( evaluate_max_waiting_time, interpolated_max_waiting_time, @@ -22,16 +23,18 @@ LOGGER = logging.getLogger("airbyte") +_INTERPOLATION_CONTEXT = frozenset({"config", "response", "headers", "parameters"}) + @dataclass class WaitTimeFromResponseBackoffStrategy(BackoffStrategy): """ - Extract wait time from the response body or headers. + Extract wait time, in relative seconds, from the JSON response body or the response headers. Attributes: wait_time_in_seconds: Expression defining how long to wait before retrying. An empty, - negative, non-finite or non-numeric result, or an expression that cannot be evaluated - on the response body, defers to the next backoff strategy. + zero, negative, non-finite or non-numeric result, or an expression that cannot be + evaluated on the response body, defers to the next backoff strategy. max_waiting_time_in_seconds: Optional inclusive cap on the computed wait. """ @@ -41,9 +44,36 @@ class WaitTimeFromResponseBackoffStrategy(BackoffStrategy): max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None def __post_init__(self, parameters: Mapping[str, Any]) -> None: - self.wait_time_in_seconds = InterpolatedString.create( - self.wait_time_in_seconds, parameters=parameters + # Evaluated with `JinjaInterpolation` directly rather than `InterpolatedString`: the latter + # caches a template as plain text when its first evaluation returns the template unchanged, + # which is what `JinjaInterpolation` does when rendering raises `TypeError`. One response + # body of an unexpected shape would then turn the strategy off for the rest of the sync. + self._wait_time_template = ( + self.wait_time_in_seconds.string + if isinstance(self.wait_time_in_seconds, InterpolatedString) + else self.wait_time_in_seconds + ) + self._interpolation = JinjaInterpolation() + self._parameters = parameters + undeclared = ( + JinjaInterpolation._find_undeclared_variables(self._wait_time_template) + - _INTERPOLATION_CONTEXT ) + if undeclared: + # A typo such as `resposne` would otherwise fail on every response and leave the + # strategy silently inert; it is a manifest bug, so it fails at startup. + raise AirbyteTracedException( + internal_message=( + f"WaitTimeFromResponse `wait_time_in_seconds` {self._wait_time_template!r} " + f"uses undeclared variables {sorted(undeclared)}; only " + f"{sorted(_INTERPOLATION_CONTEXT)} are available." + ), + message=( + "The connector's rate limit wait time expression is invalid. This is a " + "problem with the connector rather than with your configuration." + ), + failure_type=FailureType.system_error, + ) self._max_waiting_time_in_seconds = interpolated_max_waiting_time( self.max_waiting_time_in_seconds, parameters ) @@ -59,16 +89,28 @@ def backoff_time( if isinstance(response_or_exception, requests.Response): try: response = response_or_exception.json() - except requests.exceptions.JSONDecodeError: - # Non-JSON body: interpolate with an empty response. + except (ValueError, requests.RequestException, RecursionError): + # Unreadable body (not JSON, too deeply nested, or a broken stream): interpolate + # with an empty response so an expression reading only `headers` still works. pass headers = response_or_exception.headers try: - evaluated_wait_time = cast(InterpolatedString, self.wait_time_in_seconds).eval( - self.config, response=response, headers=headers + evaluated_wait_time = self._interpolation.eval( + self._wait_time_template, + self.config, + # The template as its own default re-raises an `UndefinedError` from the first + # render instead of returning None, so it is reported below. + self._wait_time_template, + parameters=self._parameters, + response=response, + headers=headers, ) - except (UndefinedError, TypeError, ValueError) as error: + if evaluated_wait_time == self._wait_time_template: + # `JinjaInterpolation._eval` returns the raw template when rendering raises + # `TypeError`, e.g. arithmetic on a string from the body. + raise TypeError("the expression could not be rendered") + except (UndefinedError, TypeError, ValueError, ArithmeticError) as error: # The strategy runs on every retryable response, not only the one the expression was # written for: a body of another shape (a list, or a string where an object was # expected) must defer to the next strategy, not fail the stream. The body is not @@ -85,7 +127,7 @@ def backoff_time( try: wait_time = float(evaluated_wait_time) - except (TypeError, ValueError): + except (TypeError, ValueError, ArithmeticError): # Mirrors the header strategies, which return None for a value they cannot read, so # the next backoff strategy decides. The value is not logged: it comes from the # response body and may hold anything. @@ -98,7 +140,8 @@ def backoff_time( # A retry time that has already passed gives a negative wait, which is not a wait to # take: the next strategy decides, as `WaitUntilTimeFromHeader` does for a past # timestamp. Returned as-is, a negative value would reach `time.sleep` and fail the - # stream. A zero wait is returned as 0, like `WaitTimeFromHeader` for a `0` header. + # stream. A zero wait is returned as 0, but `DefaultErrorHandler` treats it as no + # answer and moves on to the next strategy, as it does for `WaitTimeFromHeader`. return None max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) if max_waiting_time is not None and wait_time and wait_time >= max_waiting_time: 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 0f993c2f14..d9caa54459 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 @@ -2796,8 +2796,10 @@ def test_create_requester_with_wait_time_from_response_strategy(): error_handler: backoff_strategies: - type: WaitTimeFromResponse - wait_time_in_seconds: "{{ response.retry_after_seconds }}" + wait_time_in_seconds: "{{ response[parameters.wait_field] }}" max_waiting_time_in_seconds: 30 + $parameters: + wait_field: retry_after_seconds """ parsed_manifest = YamlDeclarativeSource._parse(content) resolved_manifest = resolver.preprocess_manifest(parsed_manifest) @@ -2817,6 +2819,10 @@ def test_create_requester_with_wait_time_from_response_strategy(): backoff_strategy = requester.error_handler.backoff_strategies[0] assert isinstance(backoff_strategy, WaitTimeFromResponseBackoffStrategy) assert backoff_strategy.max_waiting_time_in_seconds == 30 + rate_limited = requests.Response() + rate_limited.status_code = 429 + rate_limited._content = b'{"retry_after_seconds": 12}' + assert backoff_strategy.backoff_time(rate_limited, 1) == 12 @pytest.mark.parametrize( diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py index 6f7e4b509a..51829a4d85 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_response.py @@ -3,6 +3,8 @@ # import json +import logging +from unittest.mock import patch import pytest import requests @@ -12,6 +14,10 @@ from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.wait_time_from_response_backoff_strategy import ( WaitTimeFromResponseBackoffStrategy, ) +from airbyte_cdk.sources.declarative.requesters.error_handlers.default_error_handler import ( + DefaultErrorHandler, +) +from airbyte_cdk.sources.streams.http.http_client import HttpClient from airbyte_cdk.utils import AirbyteTracedException GMAIL_WAIT_TIME_EXPRESSION = r"{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response is mapping and response.get('error') is mapping and ((response.error.message or '') | string | regex_search('Retry after (\\S+Z)'))) else '' }}" @@ -241,6 +247,113 @@ def test_wait_time_from_response_returns_none_when_the_expression_fails_on_the_b assert all("soon" not in message for message in caplog.messages) +def test_wait_time_from_response_a_failed_render_does_not_disable_the_strategy(caplog): + """A `TypeError` while rendering one body must not turn the strategy off for later bodies.""" + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.reset_in - 1 }}", parameters={}, config={} + ) + + assert strategy.backoff_time(_response({"reset_in": "abc"}), 1) is None + assert any("could not be evaluated" in message for message in caplog.messages) + assert strategy.backoff_time(_response({"reset_in": 30}), 1) == 29 + + +@pytest.mark.parametrize( + "expression, body", + [ + pytest.param("{{ 10 / response.n }}", b'{"n": 0}', id="division_by_zero"), + pytest.param("{{ response.n }}", b'{"n": 1' + b"0" * 400 + b"}", id="float_overflow"), + ], +) +def test_wait_time_from_response_returns_none_for_arithmetic_errors(expression, body): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds=expression, parameters={}, config={} + ) + + assert strategy.backoff_time(_response_body(body), 1) is None + + +def test_wait_time_from_response_unreadable_body_falls_back_to_headers(): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ headers['Retry-After'] }}", parameters={}, config={} + ) + + deeply_nested_body = b"[" * 100_000 + assert strategy.backoff_time(_response_body(deeply_nested_body, {"Retry-After": "5"}), 1) == 5 + + +def test_wait_time_from_response_undeclared_variable_fails_at_construction(): + with pytest.raises(AirbyteTracedException) as error: + WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ resposne.retry_in }}", parameters={}, config={} + ) + + assert error.value.failure_type == FailureType.system_error + assert "resposne" in error.value.internal_message + + +def test_wait_time_from_response_interpolates_parameters(): + strategy = WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response[parameters.wait_field] }}", + parameters={"wait_field": "retry_in"}, + config={}, + ) + + assert strategy.backoff_time(_response({"retry_in": 7}), 1) == 7 + + +def test_wait_time_from_response_wait_reaches_the_sleep_in_http_client(requests_mock): + requests_mock.get( + "https://api.example.com/data", + [{"status_code": 429, "json": {"retry_in": 7}}, {"status_code": 200, "json": {}}], + ) + http_client = _http_client( + WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.retry_in }}", parameters={}, config={} + ) + ) + + sleeps = [] + with patch("time.sleep", side_effect=sleeps.append): + _, response = http_client.send_request( + http_method="GET", url="https://api.example.com/data", request_kwargs={} + ) + + assert response.status_code == 200 + # `user_defined_backoff_handler` adds one second to cover fractions of a second; the `backoff` + # library's own wait is 0. + assert max(sleeps) == 8 + + +def test_wait_time_from_response_wait_above_cap_stops_the_request_in_http_client(requests_mock): + requests_mock.get("https://api.example.com/data", status_code=429, json={"retry_in": 3600}) + http_client = _http_client( + WaitTimeFromResponseBackoffStrategy( + wait_time_in_seconds="{{ response.retry_in }}", + parameters={}, + config={}, + max_waiting_time_in_seconds=60, + ) + ) + + with patch("time.sleep") as sleep, pytest.raises(AirbyteTracedException) as error: + http_client.send_request( + http_method="GET", url="https://api.example.com/data", request_kwargs={} + ) + + assert error.value.failure_type == FailureType.transient_error + sleep.assert_not_called() + + +def _http_client(backoff_strategy): + return HttpClient( + name="test", + logger=logging.getLogger("test"), + error_handler=DefaultErrorHandler(config={}, parameters={}), + backoff_strategy=[backoff_strategy], + ) + + def _response(json_body, headers=None): return _response_body(json.dumps(json_body).encode(), headers=headers)