From 644fd538857987e500d178645d38237ae0280069 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:38:59 +0000 Subject: [PATCH 1/3] Add opaque token reset recovery Co-Authored-By: bot_apk --- .../declarative_component_schema.yaml | 2 + .../incremental/opaque_token_cursor.py | 24 + .../models/declarative_component_schema.py | 494 ++++++++++-------- .../parsers/model_to_component_factory.py | 3 + .../error_handlers/composite_error_handler.py | 1 + .../retrievers/simple_retriever.py | 18 + .../http/error_handlers/response_models.py | 1 + .../sources/streams/http/http_client.py | 6 + .../http/opaque_token_reset_exception.py | 28 + .../test_opaque_token_cursor_integration.py | 315 +++++++++++ .../test_composite_error_handler.py | 18 + .../sources/streams/http/test_http_client.py | 20 + 12 files changed, 698 insertions(+), 232 deletions(-) create mode 100644 airbyte_cdk/sources/streams/http/opaque_token_reset_exception.py diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index eadd756458..66e518084f 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -2884,6 +2884,7 @@ definitions: - RETRY - IGNORE - RESET_PAGINATION + - RESET_OPAQUE_TOKEN - RATE_LIMITED - REFRESH_TOKEN_THEN_RETRY - REDUCE_PAGE_SIZE @@ -2894,6 +2895,7 @@ definitions: - RETRY - IGNORE - RESET_PAGINATION + - RESET_OPAQUE_TOKEN - RATE_LIMITED - REFRESH_TOKEN_THEN_RETRY - REDUCE_PAGE_SIZE diff --git a/airbyte_cdk/sources/declarative/incremental/opaque_token_cursor.py b/airbyte_cdk/sources/declarative/incremental/opaque_token_cursor.py index cd75943f7e..3bd1c56f8c 100644 --- a/airbyte_cdk/sources/declarative/incremental/opaque_token_cursor.py +++ b/airbyte_cdk/sources/declarative/incremental/opaque_token_cursor.py @@ -197,6 +197,30 @@ def on_page_completed(self, page: PageCompleted) -> None: .replace("+00:00", "Z"), } + def restart_without_token( + self, stream_slice: StreamSlice, reason: Optional[str] = None + ) -> StreamSlice: + if "token" not in stream_slice.cursor_slice: + message = ( + f"Stream {self._stream_name}: the API rejected a read without a change token. " + "Check the RESET_OPAQUE_TOKEN response filter." + ) + raise AirbyteTracedException( + internal_message=message, + message=message, + failure_type=FailureType.system_error, + ) + + partition_key = self._serializer.to_partition_key(stream_slice.partition or {}) + with self._lock: + self._pending.pop(partition_key, None) + + return StreamSlice( + partition=stream_slice.partition, + cursor_slice={}, + extra_fields=stream_slice.extra_fields, + ) + def _coerce_token(self, value: Any, path: list[str]) -> Optional[str]: if value is None or value == "": return None diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 01ed2f3bd8..0bded1bba8 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,16 @@ 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'] }}"], title="Backoff Time", ) - 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 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 +581,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") @@ -614,6 +611,7 @@ class Action(Enum): RETRY = "RETRY" IGNORE = "IGNORE" RESET_PAGINATION = "RESET_PAGINATION" + RESET_OPAQUE_TOKEN = "RESET_OPAQUE_TOKEN" RATE_LIMITED = "RATE_LIMITED" REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY" REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE" @@ -637,6 +635,7 @@ class HttpResponseFilter(BaseModel): "RETRY", "IGNORE", "RESET_PAGINATION", + "RESET_OPAQUE_TOKEN", "RATE_LIMITED", "REFRESH_TOKEN_THEN_RETRY", "REDUCE_PAGE_SIZE", @@ -916,14 +915,18 @@ class Scope(BaseModel): class Config: extra = Extra.allow - scope: str = Field(..., description="The OAuth scope string to request from the provider.") + scope: str = Field( + ..., description="The OAuth scope string to request from the provider." + ) class OptionalScope(BaseModel): class Config: extra = Extra.allow - scope: str = Field(..., description="The OAuth scope string to request from the provider.") + scope: str = Field( + ..., description="The OAuth scope string to request from the provider." + ) class ScopesJoinStrategy(Enum): @@ -1060,24 +1063,28 @@ class OAuthConfigSpecification(BaseModel): class Config: extra = Extra.allow - oauth_user_input_from_connector_config_specification: Optional[Dict[str, Any]] = Field( - None, - description="OAuth specific blob. This is a Json Schema used to validate Json configurations used as input to OAuth.\nMust be a valid non-nested JSON that refers to properties from ConnectorSpecification.connectionSpecification\nusing special annotation 'path_in_connector_config'.\nThese are input values the user is entering through the UI to authenticate to the connector, that might also shared\nas inputs for syncing data via the connector.\nExamples:\nif no connector values is shared during oauth flow, oauth_user_input_from_connector_config_specification=[]\nif connector values such as 'app_id' inside the top level are used to generate the API url for the oauth flow,\n oauth_user_input_from_connector_config_specification={\n app_id: {\n type: string\n path_in_connector_config: ['app_id']\n }\n }\nif connector values such as 'info.app_id' nested inside another object are used to generate the API url for the oauth flow,\n oauth_user_input_from_connector_config_specification={\n app_id: {\n type: string\n path_in_connector_config: ['info', 'app_id']\n }\n }", - examples=[ - {"app_id": {"type": "string", "path_in_connector_config": ["app_id"]}}, - { - "app_id": { - "type": "string", - "path_in_connector_config": ["info", "app_id"], - } - }, - ], - title="OAuth user input", + oauth_user_input_from_connector_config_specification: Optional[Dict[str, Any]] = ( + Field( + None, + description="OAuth specific blob. This is a Json Schema used to validate Json configurations used as input to OAuth.\nMust be a valid non-nested JSON that refers to properties from ConnectorSpecification.connectionSpecification\nusing special annotation 'path_in_connector_config'.\nThese are input values the user is entering through the UI to authenticate to the connector, that might also shared\nas inputs for syncing data via the connector.\nExamples:\nif no connector values is shared during oauth flow, oauth_user_input_from_connector_config_specification=[]\nif connector values such as 'app_id' inside the top level are used to generate the API url for the oauth flow,\n oauth_user_input_from_connector_config_specification={\n app_id: {\n type: string\n path_in_connector_config: ['app_id']\n }\n }\nif connector values such as 'info.app_id' nested inside another object are used to generate the API url for the oauth flow,\n oauth_user_input_from_connector_config_specification={\n app_id: {\n type: string\n path_in_connector_config: ['info', 'app_id']\n }\n }", + examples=[ + {"app_id": {"type": "string", "path_in_connector_config": ["app_id"]}}, + { + "app_id": { + "type": "string", + "path_in_connector_config": ["info", "app_id"], + } + }, + ], + title="OAuth user input", + ) ) - oauth_connector_input_specification: Optional[OauthConnectorInputSpecification] = Field( - None, - description='The DeclarativeOAuth specific blob.\nPertains to the fields defined by the connector relating to the OAuth flow.\n\nInterpolation capabilities:\n- The variables placeholders are declared as `{{my_var}}`.\n- The nested resolution variables like `{{ {{my_nested_var}} }}` is allowed as well.\n\n- The allowed interpolation context is:\n + base64Encoder - encode to `base64`, {{ {{my_var_a}}:{{my_var_b}} | base64Encoder }}\n + base64Decorer - decode from `base64` encoded string, {{ {{my_string_variable_or_string_value}} | base64Decoder }}\n + urlEncoder - encode the input string to URL-like format, {{ https://test.host.com/endpoint | urlEncoder}}\n + urlDecorer - decode the input url-encoded string into text format, {{ urlDecoder:https%3A%2F%2Fairbyte.io | urlDecoder}}\n + codeChallengeS256 - get the `codeChallenge` encoded value to provide additional data-provider specific authorisation values, {{ {{state_value}} | codeChallengeS256 }}\n\nExamples:\n - The TikTok Marketing DeclarativeOAuth spec:\n {\n "oauth_connector_input_specification": {\n "type": "object",\n "additionalProperties": false,\n "properties": {\n "consent_url": "https://ads.tiktok.com/marketing_api/auth?{{client_id_key}}={{client_id_value}}&{{redirect_uri_key}}={{ {{redirect_uri_value}} | urlEncoder}}&{{state_key}}={{state_value}}",\n "access_token_url": "https://business-api.tiktok.com/open_api/v1.3/oauth2/access_token/",\n "access_token_params": {\n "{{ auth_code_key }}": "{{ auth_code_value }}",\n "{{ client_id_key }}": "{{ client_id_value }}",\n "{{ client_secret_key }}": "{{ client_secret_value }}"\n },\n "access_token_headers": {\n "Content-Type": "application/json",\n "Accept": "application/json"\n },\n "extract_output": ["data.access_token"],\n "client_id_key": "app_id",\n "client_secret_key": "secret",\n "auth_code_key": "auth_code"\n }\n }\n }', - title="DeclarativeOAuth Connector Specification", + oauth_connector_input_specification: Optional[OauthConnectorInputSpecification] = ( + Field( + None, + description='The DeclarativeOAuth specific blob.\nPertains to the fields defined by the connector relating to the OAuth flow.\n\nInterpolation capabilities:\n- The variables placeholders are declared as `{{my_var}}`.\n- The nested resolution variables like `{{ {{my_nested_var}} }}` is allowed as well.\n\n- The allowed interpolation context is:\n + base64Encoder - encode to `base64`, {{ {{my_var_a}}:{{my_var_b}} | base64Encoder }}\n + base64Decorer - decode from `base64` encoded string, {{ {{my_string_variable_or_string_value}} | base64Decoder }}\n + urlEncoder - encode the input string to URL-like format, {{ https://test.host.com/endpoint | urlEncoder}}\n + urlDecorer - decode the input url-encoded string into text format, {{ urlDecoder:https%3A%2F%2Fairbyte.io | urlDecoder}}\n + codeChallengeS256 - get the `codeChallenge` encoded value to provide additional data-provider specific authorisation values, {{ {{state_value}} | codeChallengeS256 }}\n\nExamples:\n - The TikTok Marketing DeclarativeOAuth spec:\n {\n "oauth_connector_input_specification": {\n "type": "object",\n "additionalProperties": false,\n "properties": {\n "consent_url": "https://ads.tiktok.com/marketing_api/auth?{{client_id_key}}={{client_id_value}}&{{redirect_uri_key}}={{ {{redirect_uri_value}} | urlEncoder}}&{{state_key}}={{state_value}}",\n "access_token_url": "https://business-api.tiktok.com/open_api/v1.3/oauth2/access_token/",\n "access_token_params": {\n "{{ auth_code_key }}": "{{ auth_code_value }}",\n "{{ client_id_key }}": "{{ client_id_value }}",\n "{{ client_secret_key }}": "{{ client_secret_value }}"\n },\n "access_token_headers": {\n "Content-Type": "application/json",\n "Accept": "application/json"\n },\n "extract_output": ["data.access_token"],\n "client_id_key": "app_id",\n "client_secret_key": "secret",\n "auth_code_key": "auth_code"\n }\n }\n }', + title="DeclarativeOAuth Connector Specification", + ) ) complete_oauth_output_specification: Optional[Dict[str, Any]] = Field( None, @@ -1095,7 +1102,9 @@ class Config: complete_oauth_server_input_specification: Optional[Dict[str, Any]] = Field( None, description="OAuth specific blob. This is a Json Schema used to validate Json configurations persisted as Airbyte Server configurations.\nMust be a valid non-nested JSON describing additional fields configured by the Airbyte Instance or Workspace Admins to be used by the\nserver when completing an OAuth flow (typically exchanging an auth code for refresh token).\nExamples:\n complete_oauth_server_input_specification={\n client_id: {\n type: string\n },\n client_secret: {\n type: string\n }\n }", - examples=[{"client_id": {"type": "string"}, "client_secret": {"type": "string"}}], + examples=[ + {"client_id": {"type": "string"}, "client_secret": {"type": "string"}} + ], title="OAuth input specification", ) complete_oauth_server_output_specification: Optional[Dict[str, Any]] = Field( @@ -1332,39 +1341,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( @@ -1690,7 +1694,9 @@ class CustomConfigTransformation(BaseModel): class_name: str = Field( ..., description="Fully-qualified name of the class that will be implementing the custom config transformation. The format is `source_..`.", - examples=["source_declarative_manifest.components.MyCustomConfigTransformation"], + examples=[ + "source_declarative_manifest.components.MyCustomConfigTransformation" + ], ) parameters: Optional[Dict[str, Any]] = Field( None, @@ -2008,6 +2014,42 @@ class DatetimeBasedCursor(BaseModel): parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") +class OpaqueTokenCursor(BaseModel): + class Config: + extra = Extra.allow + + type: Literal["OpaqueTokenCursor"] + token_path: List[str] = Field( + ..., + description="Path to the next token in the decoded response that ends a pagination pass. The value must be a string or an integer, and is stored and sent as a string. A missing, null or empty value means the response carried no token.", + examples=[["syncToken"], ["nextSyncToken"], ["meta", "sync_token"]], + title="Token Path", + ) + token_option: RequestOption = Field( + ..., + description="Where to send the stored token on each request of a pass: the JSON body, the query string, a header or the form body.", + title="Send Token In", + ) + completion_condition: Optional[str] = Field( + None, + description="Must be true on the last response of a pass for its token to be committed. Defaults to a token being present at token_path. Set it when the API repeats the old token on intermediate pages or returns a position on every page.", + examples=["{{ not response.moreDataAvailable }}"], + title="Completion Condition", + ) + initial_token: Optional[str] = Field( + None, + description='Value to send when no token is stored, for an API that needs one to start a full pass, such as Todoist\'s "*". When unset, nothing is sent.', + examples=["*"], + title="Initial Token", + ) + cursor_field: Optional[str] = Field( + None, + description="Optional top-level record field advertised as the stream's cursor, so destinations can order versions of a record when deduplicating. It is never used to filter records or to build state.", + title="Destination Cursor Field", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class JwtAuthenticator(BaseModel): type: Literal["JwtAuthenticator"] secret_key: str = Field( @@ -2188,7 +2230,9 @@ class OAuthAuthenticator(BaseModel): scopes: Optional[List[str]] = Field( None, description="List of scopes that should be granted to the access token.", - examples=[["crm.list.read", "crm.objects.contacts.read", "crm.schema.contacts.read"]], + examples=[ + ["crm.list.read", "crm.objects.contacts.read", "crm.schema.contacts.read"] + ], title="Scopes", ) token_expiry_date: Optional[str] = Field( @@ -2339,53 +2383,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( @@ -2483,10 +2480,12 @@ class DpathValidator(BaseModel): ], title="Field Path", ) - validation_strategy: Union[ValidateAdheresToSchema, CustomValidationStrategy] = Field( - ..., - description="The condition that the specified config value will be evaluated against", - title="Validation Strategy", + validation_strategy: Union[ValidateAdheresToSchema, CustomValidationStrategy] = ( + Field( + ..., + description="The condition that the specified config value will be evaluated against", + title="Validation Strategy", + ) ) @@ -2503,10 +2502,12 @@ class PredicateValidator(BaseModel): ], title="Value", ) - validation_strategy: Union[ValidateAdheresToSchema, CustomValidationStrategy] = Field( - ..., - description="The validation strategy to apply to the value.", - title="Validation Strategy", + validation_strategy: Union[ValidateAdheresToSchema, CustomValidationStrategy] = ( + Field( + ..., + description="The validation strategy to apply to the value.", + title="Validation Strategy", + ) ) @@ -2588,12 +2589,12 @@ class RateLimitedMultipleTokenAuthenticator(BaseModel): class CompositeErrorHandler(BaseModel): type: Literal["CompositeErrorHandler"] - error_handlers: List[Union[CompositeErrorHandler, DefaultErrorHandler, CustomErrorHandler]] = ( - Field( - ..., - description="List of error handlers to iterate on to determine how to handle a failed response.", - title="Error Handlers", - ) + error_handlers: List[ + Union[CompositeErrorHandler, DefaultErrorHandler, CustomErrorHandler] + ] = Field( + ..., + description="List of error handlers to iterate on to determine how to handle a failed response.", + title="Error Handlers", ) parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") @@ -2631,80 +2632,20 @@ 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 type: Literal["ZipfileDecoder"] - decoder: Union[CsvDecoder, GzipDecoder, JsonDecoder, JsonItemsDecoder, JsonlDecoder] = Field( + decoder: Union[ + CsvDecoder, GzipDecoder, JsonDecoder, JsonItemsDecoder, JsonlDecoder + ] = Field( ..., description="Parser to parse the decompressed data from the zipfile(s).", title="Parser", ) -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( @@ -2822,9 +2763,9 @@ class Config: type: Literal["DeclarativeSource"] check: Union[CheckStream, CheckDynamicStream] - streams: Optional[List[Union[ConditionalStreams, DeclarativeStream, StateDelegatingStream]]] = ( - None - ) + streams: Optional[ + List[Union[ConditionalStreams, DeclarativeStream, StateDelegatingStream]] + ] = None dynamic_streams: List[DynamicDeclarativeStream] version: str = Field( ..., @@ -2931,11 +2872,11 @@ class FileUploader(BaseModel): ..., description="Requester component that describes how to prepare HTTP requests to send to the source API.", ) - download_target_extractor: Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor] = ( - Field( - ..., - description="Responsible for fetching the url where the file is located. This is applied on each records and not on the HTTP response", - ) + download_target_extractor: Union[ + DpathExtractor, CustomRecordExtractor, CombinedExtractor + ] = Field( + ..., + description="Responsible for fetching the url where the file is located. This is applied on each records and not on the HTTP response", ) file_extractor: Optional[Union[DpathExtractor, CustomRecordExtractor]] = Field( None, @@ -2952,48 +2893,14 @@ class FileUploader(BaseModel): parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") -class OpaqueTokenCursor(BaseModel): - class Config: - extra = Extra.allow - - type: Literal["OpaqueTokenCursor"] - token_path: List[str] = Field( - ..., - description="Path to the next token in the decoded response that ends a pagination pass. The value must be a string or an integer, and is stored and sent as a string. A missing, null or empty value means the response carried no token.", - examples=[["syncToken"], ["nextSyncToken"], ["meta", "sync_token"]], - title="Token Path", - ) - token_option: RequestOption = Field( - ..., - description="Where to send the stored token on each request of a pass: the JSON body, the query string, a header or the form body.", - title="Send Token In", - ) - completion_condition: Optional[str] = Field( - None, - description="Must be true on the last response of a pass for its token to be committed. Defaults to a token being present at token_path. Set it when the API repeats the old token on intermediate pages or returns a position on every page.", - examples=["{{ not response.moreDataAvailable }}"], - title="Completion Condition", - ) - initial_token: Optional[str] = Field( - None, - description='Value to send when no token is stored, for an API that needs one to start a full pass, such as Todoist\'s "*". When unset, nothing is sent.', - examples=["*"], - title="Initial Token", - ) - cursor_field: Optional[str] = Field( - None, - description="Optional top-level record field advertised as the stream's cursor, so destinations can order versions of a record when deduplicating. It is never used to filter records or to build state.", - title="Destination Cursor Field", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - class DeclarativeStream(BaseModel): class Config: extra = Extra.allow type: Literal["DeclarativeStream"] - name: Optional[str] = Field("", description="The stream name.", example=["Users"], title="Name") + name: Optional[str] = Field( + "", description="The stream name.", example=["Users"], title="Name" + ) retriever: Union[SimpleRetriever, AsyncRetriever, CustomRetriever] = Field( ..., description="Component used to coordinate how records are extracted across stream slices and request pages.", @@ -3060,6 +2967,96 @@ 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( @@ -3179,18 +3176,20 @@ class HttpRequester(BaseModelWithDeprecations): description="For APIs that require explicit specification of the properties to query for, this component will take a static or dynamic set of properties (which can be optionally split into chunks) and allow them to be injected into an outbound request by accessing stream_partition.extra_fields.", title="Query Properties", ) - request_parameters: Optional[Union[Dict[str, Union[str, QueryProperties]], str]] = Field( - None, - description="Specifies the query parameters that should be set on an outgoing HTTP request given the inputs.", - examples=[ - {"unit": "day"}, - { - "query": 'last_event_time BETWEEN TIMESTAMP "{{ stream_interval.start_time }}" AND TIMESTAMP "{{ stream_interval.end_time }}"' - }, - {"searchIn": "{{ ','.join(config.get('search_in', [])) }}"}, - {"sort_by[asc]": "updated_at"}, - ], - title="Query Parameters", + request_parameters: Optional[Union[Dict[str, Union[str, QueryProperties]], str]] = ( + Field( + None, + description="Specifies the query parameters that should be set on an outgoing HTTP request given the inputs.", + examples=[ + {"unit": "day"}, + { + "query": 'last_event_time BETWEEN TIMESTAMP "{{ stream_interval.start_time }}" AND TIMESTAMP "{{ stream_interval.end_time }}"' + }, + {"searchIn": "{{ ','.join(config.get('search_in', [])) }}"}, + {"sort_by[asc]": "updated_at"}, + ], + title="Query Parameters", + ) ) request_headers: Optional[Union[Dict[str, str], str]] = Field( None, @@ -3244,18 +3243,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( @@ -3380,9 +3377,34 @@ 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") + name: str = Field( + ..., description="The stream name.", example=["Users"], title="Name" + ) full_refresh_stream: DeclarativeStream = Field( ..., description="Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.", @@ -3483,7 +3505,9 @@ class AsyncRetriever(BaseModel): status_mapping: AsyncJobStatusMap = Field( ..., description="Async Job Status to Airbyte CDK Async Job Status mapping." ) - status_extractor: Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor] = Field( + status_extractor: Union[ + DpathExtractor, CustomRecordExtractor, CombinedExtractor + ] = Field( ..., description="Responsible for fetching the actual status of the async job." ) download_target_extractor: Optional[ @@ -3493,8 +3517,15 @@ 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] - ] = Field(None, description="Responsible for fetching the records from provided urls.") + Union[ + DpathExtractor, + CustomRecordExtractor, + ResponseToFileExtractor, + CombinedExtractor, + ] + ] = Field( + None, description="Responsible for fetching the records from provided urls." + ) creation_requester: Union[HttpRequester, CustomRequester] = Field( ..., description="Requester component that describes how to prepare HTTP requests to send to the source API to create the async server-side job.", @@ -3507,10 +3538,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, @@ -3697,13 +3727,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 23902e1484..7d8d04eb7a 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -4182,6 +4182,9 @@ def _get_url(req: Requester) -> str: page_observer=( cursor.on_page_completed if isinstance(cursor, OpaqueTokenCursor) else None ), + token_reset_handler=( + cursor.restart_without_token if isinstance(cursor, OpaqueTokenCursor) else None + ), parameters=model.parameters or {}, ) diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py index 2ef35a06ff..06560593af 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py @@ -71,6 +71,7 @@ def interpret_response( ResponseAction.RETRY, ResponseAction.IGNORE, ResponseAction.RESET_PAGINATION, + ResponseAction.RESET_OPAQUE_TOKEN, ResponseAction.REDUCE_PAGE_SIZE, ResponseAction.SPLIT_REQUEST_WINDOW, ]: diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index d7b63e85fc..6adac54946 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -57,6 +57,10 @@ from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer from airbyte_cdk.sources.source import ExperimentalClassWarning from airbyte_cdk.sources.streams.core import StreamData +from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import ( + OpaqueTokenResetNotSupportedException, + OpaqueTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionNotSupportedException, PageSizeReductionRequiredException, @@ -154,6 +158,7 @@ class SimpleRetriever(Retriever): Callable[[StreamSlice, Optional[datetime.timedelta]], Optional[List[StreamSlice]]] ] = None page_observer: Optional[Callable[["PageCompleted"], None]] = None + token_reset_handler: Optional[Callable[[StreamSlice, Optional[str]], StreamSlice]] = None def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._paginator = self.paginator or NoPagination(parameters=parameters) @@ -703,6 +708,19 @@ def _read_records_or_split_request_window( yield from self._read_records_or_split_request_window( records_schema, child, original_slice, depth + 1 ) + except OpaqueTokenResetRequiredException as exception: + if self.token_reset_handler is None: + raise OpaqueTokenResetNotSupportedException(stream_name=self.name) from exception + + reset_slice = self.token_reset_handler(stream_slice, exception.error_message) + reason = exception.error_message or "the API rejected the stored change token" + LOGGER.warning( + f"Stream {self.name}: {reason}; reading this stream in full. " + f"{emitted_count} records read earlier in this attempt may be emitted again." + ) + yield from self._read_records_or_split_request_window( + records_schema, reset_slice, original_slice, depth + 1 + ) def _with_request_window_failure_message(self, message: str) -> str: """ diff --git a/airbyte_cdk/sources/streams/http/error_handlers/response_models.py b/airbyte_cdk/sources/streams/http/error_handlers/response_models.py index 4cda91ffb8..129e086d49 100644 --- a/airbyte_cdk/sources/streams/http/error_handlers/response_models.py +++ b/airbyte_cdk/sources/streams/http/error_handlers/response_models.py @@ -17,6 +17,7 @@ class ResponseAction(Enum): FAIL = "FAIL" IGNORE = "IGNORE" RESET_PAGINATION = "RESET_PAGINATION" + RESET_OPAQUE_TOKEN = "RESET_OPAQUE_TOKEN" RATE_LIMITED = "RATE_LIMITED" REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY" REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE" diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 8da499aa32..97cf6e125d 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -43,6 +43,9 @@ RequestBodyException, UserDefinedBackoffException, ) +from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import ( + OpaqueTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionRequiredException, ) @@ -620,6 +623,9 @@ def _handle_error_resolution( if error_resolution.response_action == ResponseAction.RESET_PAGINATION: raise PaginationResetRequiredException() + if error_resolution.response_action == ResponseAction.RESET_OPAQUE_TOKEN: + raise OpaqueTokenResetRequiredException(error_resolution.error_message) + if error_resolution.response_action == ResponseAction.REDUCE_PAGE_SIZE: raise PageSizeReductionRequiredException( stream_name=self._name, error_message=error_resolution.error_message diff --git a/airbyte_cdk/sources/streams/http/opaque_token_reset_exception.py b/airbyte_cdk/sources/streams/http/opaque_token_reset_exception.py new file mode 100644 index 0000000000..b03e36d4be --- /dev/null +++ b/airbyte_cdk/sources/streams/http/opaque_token_reset_exception.py @@ -0,0 +1,28 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# + +from typing import Optional + +from airbyte_cdk.models import FailureType +from airbyte_cdk.utils.traced_exception import AirbyteTracedException + + +class OpaqueTokenResetRequiredException(Exception): + def __init__(self, error_message: Optional[str] = None) -> None: + self.error_message = error_message + super().__init__(error_message) + + +class OpaqueTokenResetNotSupportedException(AirbyteTracedException): + def __init__(self, stream_name: Optional[str] = None) -> None: + stream = f"Stream {stream_name}" if stream_name else "The stream" + message = ( + f"{stream} resolves to a token reset but does not use an OpaqueTokenCursor. " + "Remove the RESET_OPAQUE_TOKEN action from its error handler." + ) + super().__init__( + internal_message=message, + message=message, + failure_type=FailureType.config_error, + ) diff --git a/unit_tests/sources/declarative/incremental/test_opaque_token_cursor_integration.py b/unit_tests/sources/declarative/incremental/test_opaque_token_cursor_integration.py index 652d45f851..17939dbf33 100644 --- a/unit_tests/sources/declarative/incremental/test_opaque_token_cursor_integration.py +++ b/unit_tests/sources/declarative/incremental/test_opaque_token_cursor_integration.py @@ -13,6 +13,7 @@ ConfiguredAirbyteCatalog, ConfiguredAirbyteStream, DestinationSyncMode, + FailureType, SyncMode, Type, ) @@ -215,6 +216,43 @@ def _incoming_state(stream_state): return [state_manager.create_state_message("events", None).state] +def _reset_error_handler(*, reduce_page_size=False, pagination_reset=False): + response_filters = [] + if reduce_page_size: + response_filters.append( + { + "type": "HttpResponseFilter", + "action": "REDUCE_PAGE_SIZE", + "http_codes": [413], + "error_message": "page too large", + } + ) + response_filters.append( + { + "type": "HttpResponseFilter", + "action": "RESET_OPAQUE_TOKEN", + "predicate": "{{ response.get('success') == false }}", + "error_message": "Ashby rejected the stored change token (sync_token_expired)", + } + ) + if pagination_reset: + response_filters.append( + { + "type": "HttpResponseFilter", + "action": "RESET_PAGINATION", + "http_codes": [400], + "failure_type": "system_error", + "error_message": "The page cursor expired", + } + ) + return { + "type": "DefaultErrorHandler", + "max_retries": 0, + "backoff_strategies": [], + "response_filters": response_filters, + } + + def test_t_boot_commits_after_three_pages_and_records(): with HttpMocker() as http_mocker: http_mocker.post( @@ -311,6 +349,283 @@ def test_t_fail_preserves_committed_token_and_emits_incomplete_status(): ) +def test_t_reset_restarts_without_token_and_commits_terminal_token(caplog): + state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler()) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + HttpResponse(json.dumps({"items": [{"id": "old-1"}], "nextCursor": "old-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2"}), + HttpResponse(json.dumps({"items": [{"id": "old-2"}], "nextCursor": "old-p3"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p3"}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + http_mocker.post( + HttpRequest(_URL, body=None), + HttpResponse(json.dumps({"items": [{"id": "new-1"}], "nextCursor": "new-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "new-p2"}), + HttpResponse(json.dumps({"items": [{"id": "new-2"}], "syncToken": "T2"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read(source, catalog, state) + + assert [message.record.data["id"] for message in messages if message.type == Type.RECORD] == [ + "old-1", + "old-2", + "new-1", + "new-2", + ] + assert len(_state(messages)) == 1 + assert _state(messages)[0].stream.stream_state.__dict__["token"] == "T2" + assert ( + "Ashby rejected the stored change token (sync_token_expired); reading this stream in full." + in caplog.text + ) + assert "2 records read earlier in this attempt may be emitted again." in caplog.text + + +def test_t_reset_context_is_fresh_after_page_size_reduction(caplog): + state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler(reduce_page_size=True)) + stream = manifest["streams"][0] + stream["retriever"]["paginator"]["page_size_option"] = { + "type": "RequestOption", + "field_name": "pageSize", + "inject_into": "body_json", + } + stream["retriever"]["paginator"]["pagination_strategy"]["page_size"] = 8 + stream["retriever"]["page_size_reduction"] = { + "type": "PageSizeReduction", + "minimum_page_size": 1, + "backoff_seconds": 0, + } + stream["retriever"]["record_selector"]["record_filter"] = { + "type": "RecordFilter", + "condition": "{{ record['expected_token'] == stream_interval.get('token', 'none') }}", + } + stream["transformations"] = [ + { + "type": "AddFields", + "fields": [ + { + "path": ["slice_token"], + "value": "{{ stream_interval.get('token', 'none') }}", + } + ], + } + ] + stream["schema_loader"]["schema"]["properties"].update( + {"expected_token": {"type": "string"}, "slice_token": {"type": "string"}} + ) + + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "pageSize": 8}), + HttpResponse("{}", status_code=413), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "pageSize": 4}), + HttpResponse( + json.dumps( + { + "items": [{"id": "old", "expected_token": "T1"}], + "nextCursor": "old-p2", + } + ) + ), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2", "pageSize": 4}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"pageSize": 8}), + HttpResponse( + json.dumps( + { + "items": [{"id": "new", "expected_token": "none"}], + "nextCursor": "new-p2", + } + ) + ), + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "new-p2", "pageSize": 8}), + HttpResponse(json.dumps({"items": [], "syncToken": "T2"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read(source, catalog, state) + + records = [message.record.data for message in messages if message.type == Type.RECORD] + assert records == [ + {"id": "old", "expected_token": "T1", "slice_token": "T1"}, + {"id": "new", "expected_token": "none", "slice_token": "none"}, + ] + assert _state(messages)[0].stream.stream_state.__dict__["token"] == "T2" + assert "sync_token_expired" in caplog.text + + +def test_t_reset_twice_errors_and_preserves_committed_state(): + state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler()) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + HttpResponse(json.dumps({"items": [{"id": "old"}], "nextCursor": "old-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2"}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + http_mocker.post( + HttpRequest(_URL, body=None), + HttpResponse(json.dumps({"items": [{"id": "new"}], "nextCursor": "new-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "new-p2"}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read_until_error(source, catalog, state) + + assert any(message.type == Type.RECORD for message in messages) + assert _state(messages)[-1].stream.stream_state.__dict__["token"] == "T1" + assert any( + message.type == Type.TRACE + and message.trace.error + and message.trace.error.failure_type == FailureType.system_error + and message.trace.error.message + == "Stream events: the API rejected a read without a change token. " + "Check the RESET_OPAQUE_TOKEN response filter." + for message in messages + ) + + +def test_t_rpag_reset_replays_resumed_page_one_with_the_same_token(): + state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler(pagination_reset=True)) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + [ + HttpResponse(json.dumps({"items": [{"id": "page-1"}], "nextCursor": "old-p2"})), + HttpResponse( + json.dumps({"items": [{"id": "replayed-page-1"}], "nextCursor": "new-p2"}) + ), + ], + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2"}), + HttpResponse("{}", status_code=400), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "new-p2"}), + HttpResponse(json.dumps({"items": [{"id": "replayed-page-2"}], "syncToken": "T2"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read(source, catalog, state) + + assert [message.record.data["id"] for message in messages if message.type == Type.RECORD] == [ + "page-1", + "replayed-page-1", + "replayed-page-2", + ] + assert _state(messages)[-1].stream.stream_state.__dict__["token"] == "T2" + + +def test_t_rpag_reset_after_token_reset_replays_tokenless_page_one(): + state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler(pagination_reset=True)) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + HttpResponse(json.dumps({"items": [{"id": "old-page-1"}], "nextCursor": "old-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2"}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + http_mocker.post( + HttpRequest(_URL, body=None), + [ + HttpResponse(json.dumps({"items": [{"id": "page-1"}], "nextCursor": "p2"})), + HttpResponse( + json.dumps({"items": [{"id": "replayed-page-1"}], "nextCursor": "new-p2"}) + ), + ], + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "p2"}), + HttpResponse("{}", status_code=400), + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "new-p2"}), + HttpResponse(json.dumps({"items": [{"id": "replayed-page-2"}], "syncToken": "T2"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read(source, catalog, state) + + assert [message.record.data["id"] for message in messages if message.type == Type.RECORD] == [ + "old-page-1", + "page-1", + "replayed-page-1", + "replayed-page-2", + ] + assert _state(messages)[-1].stream.stream_state.__dict__["token"] == "T2" + + +def test_t_rpag_second_pagination_reset_is_system_error(): + state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler(pagination_reset=True)) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + [ + HttpResponse(json.dumps({"items": [{"id": "page-1"}], "nextCursor": "p2"})), + HttpResponse( + json.dumps({"items": [{"id": "replayed-page-1"}], "nextCursor": "p2"}) + ), + ], + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "p2"}), + [HttpResponse("{}", status_code=400), HttpResponse("{}", status_code=400)], + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read_until_error(source, catalog, state) + + assert any( + message.type == Type.TRACE + and message.trace.error + and message.trace.error.failure_type == FailureType.system_error + for message in messages + ) + assert _state(messages)[-1].stream.stream_state.__dict__["token"] == "T1" + + +def test_reset_action_on_non_token_stream_is_not_supported(): + manifest = _manifest(error_handler=_reset_error_handler()) + del manifest["streams"][0]["incremental_sync"] + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body=None), + HttpResponse(json.dumps({"success": False, "error": "unexpected_reset"})), + ) + source, catalog = _source_and_catalog(manifest, sync_mode=SyncMode.full_refresh) + with pytest.raises(AirbyteTracedException) as exc_info: + _read(source, catalog) + + assert exc_info.value.failure_type.value == "config_error" + assert "RESET_OPAQUE_TOKEN" in str(exc_info.value) + + def test_t_builder_page_limit_and_max_records_do_not_commit_tokens(): with HttpMocker() as http_mocker: http_mocker.post( diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py b/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py index 78d3b7ee8a..4bfbdfc5fa 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py @@ -132,6 +132,24 @@ def test_given_unmatched_response_or_exception_then_return_default_error_resolut assert error_resolution.failure_type == FailureType.system_error +def test_composite_error_handler_preserves_reset_opaque_token(): + reset_handler = MagicMock() + reset_resolution = ErrorResolution( + response_action=ResponseAction.RESET_OPAQUE_TOKEN, + error_message="sync_token_expired", + ) + reset_handler.interpret_response.return_value = reset_resolution + success_handler = MagicMock() + success_handler.interpret_response.return_value = ErrorResolution( + response_action=ResponseAction.SUCCESS + ) + composite_error_handler = CompositeErrorHandler( + error_handlers=[reset_handler, success_handler], parameters={} + ) + + assert composite_error_handler.interpret_response(MagicMock()) == reset_resolution + + def test_composite_error_handler_no_handlers(): try: CompositeErrorHandler(error_handlers=[], parameters={}) diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index 31d4ff9a1c..abc7a03596 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -35,6 +35,9 @@ UserDefinedBackoffException, ) from airbyte_cdk.sources.streams.http.http_client import MessageRepresentationAirbyteTracedErrors +from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import ( + OpaqueTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionRequiredException, ) @@ -244,6 +247,23 @@ def test_send_raises_airbyte_traced_exception_with_fail_response_action(): assert http_client._session.send.call_count == 1 +def test_reset_opaque_token_action_raises_exception_with_error_message(): + http_client = test_http_client() + error_message = "sync_token_expired" + with pytest.raises(OpaqueTokenResetRequiredException) as exc_info: + http_client._handle_error_resolution( + response=None, + exc=None, + request=requests.PreparedRequest(), + error_resolution=ErrorResolution( + response_action=ResponseAction.RESET_OPAQUE_TOKEN, + error_message=error_message, + ), + ) + + assert exc_info.value.error_message == error_message + + def test_send_ignores_with_ignore_reponse_action_and_returns_response(): mocked_session = MagicMock(spec=requests.Session) mocked_response = MagicMock(spec=requests.Response) From 6bca0ac2b68127965f0fb68bb6d98fa2288bb1ae Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:40:58 +0000 Subject: [PATCH 2/3] Trim generated schema model diff Co-Authored-By: bot_apk --- .../models/declarative_component_schema.py | 492 +++++++++--------- 1 file changed, 232 insertions(+), 260 deletions(-) diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 0bded1bba8..a7d79fc62d 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,16 +134,17 @@ 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.", examples=[30, 30.5, "{{ config['backoff_time'] }}"], title="Backoff Time", ) - 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") @@ -581,10 +583,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") @@ -915,18 +918,14 @@ class Scope(BaseModel): class Config: extra = Extra.allow - scope: str = Field( - ..., description="The OAuth scope string to request from the provider." - ) + scope: str = Field(..., description="The OAuth scope string to request from the provider.") class OptionalScope(BaseModel): class Config: extra = Extra.allow - scope: str = Field( - ..., description="The OAuth scope string to request from the provider." - ) + scope: str = Field(..., description="The OAuth scope string to request from the provider.") class ScopesJoinStrategy(Enum): @@ -1063,28 +1062,24 @@ class OAuthConfigSpecification(BaseModel): class Config: extra = Extra.allow - oauth_user_input_from_connector_config_specification: Optional[Dict[str, Any]] = ( - Field( - None, - description="OAuth specific blob. This is a Json Schema used to validate Json configurations used as input to OAuth.\nMust be a valid non-nested JSON that refers to properties from ConnectorSpecification.connectionSpecification\nusing special annotation 'path_in_connector_config'.\nThese are input values the user is entering through the UI to authenticate to the connector, that might also shared\nas inputs for syncing data via the connector.\nExamples:\nif no connector values is shared during oauth flow, oauth_user_input_from_connector_config_specification=[]\nif connector values such as 'app_id' inside the top level are used to generate the API url for the oauth flow,\n oauth_user_input_from_connector_config_specification={\n app_id: {\n type: string\n path_in_connector_config: ['app_id']\n }\n }\nif connector values such as 'info.app_id' nested inside another object are used to generate the API url for the oauth flow,\n oauth_user_input_from_connector_config_specification={\n app_id: {\n type: string\n path_in_connector_config: ['info', 'app_id']\n }\n }", - examples=[ - {"app_id": {"type": "string", "path_in_connector_config": ["app_id"]}}, - { - "app_id": { - "type": "string", - "path_in_connector_config": ["info", "app_id"], - } - }, - ], - title="OAuth user input", - ) + oauth_user_input_from_connector_config_specification: Optional[Dict[str, Any]] = Field( + None, + description="OAuth specific blob. This is a Json Schema used to validate Json configurations used as input to OAuth.\nMust be a valid non-nested JSON that refers to properties from ConnectorSpecification.connectionSpecification\nusing special annotation 'path_in_connector_config'.\nThese are input values the user is entering through the UI to authenticate to the connector, that might also shared\nas inputs for syncing data via the connector.\nExamples:\nif no connector values is shared during oauth flow, oauth_user_input_from_connector_config_specification=[]\nif connector values such as 'app_id' inside the top level are used to generate the API url for the oauth flow,\n oauth_user_input_from_connector_config_specification={\n app_id: {\n type: string\n path_in_connector_config: ['app_id']\n }\n }\nif connector values such as 'info.app_id' nested inside another object are used to generate the API url for the oauth flow,\n oauth_user_input_from_connector_config_specification={\n app_id: {\n type: string\n path_in_connector_config: ['info', 'app_id']\n }\n }", + examples=[ + {"app_id": {"type": "string", "path_in_connector_config": ["app_id"]}}, + { + "app_id": { + "type": "string", + "path_in_connector_config": ["info", "app_id"], + } + }, + ], + title="OAuth user input", ) - oauth_connector_input_specification: Optional[OauthConnectorInputSpecification] = ( - Field( - None, - description='The DeclarativeOAuth specific blob.\nPertains to the fields defined by the connector relating to the OAuth flow.\n\nInterpolation capabilities:\n- The variables placeholders are declared as `{{my_var}}`.\n- The nested resolution variables like `{{ {{my_nested_var}} }}` is allowed as well.\n\n- The allowed interpolation context is:\n + base64Encoder - encode to `base64`, {{ {{my_var_a}}:{{my_var_b}} | base64Encoder }}\n + base64Decorer - decode from `base64` encoded string, {{ {{my_string_variable_or_string_value}} | base64Decoder }}\n + urlEncoder - encode the input string to URL-like format, {{ https://test.host.com/endpoint | urlEncoder}}\n + urlDecorer - decode the input url-encoded string into text format, {{ urlDecoder:https%3A%2F%2Fairbyte.io | urlDecoder}}\n + codeChallengeS256 - get the `codeChallenge` encoded value to provide additional data-provider specific authorisation values, {{ {{state_value}} | codeChallengeS256 }}\n\nExamples:\n - The TikTok Marketing DeclarativeOAuth spec:\n {\n "oauth_connector_input_specification": {\n "type": "object",\n "additionalProperties": false,\n "properties": {\n "consent_url": "https://ads.tiktok.com/marketing_api/auth?{{client_id_key}}={{client_id_value}}&{{redirect_uri_key}}={{ {{redirect_uri_value}} | urlEncoder}}&{{state_key}}={{state_value}}",\n "access_token_url": "https://business-api.tiktok.com/open_api/v1.3/oauth2/access_token/",\n "access_token_params": {\n "{{ auth_code_key }}": "{{ auth_code_value }}",\n "{{ client_id_key }}": "{{ client_id_value }}",\n "{{ client_secret_key }}": "{{ client_secret_value }}"\n },\n "access_token_headers": {\n "Content-Type": "application/json",\n "Accept": "application/json"\n },\n "extract_output": ["data.access_token"],\n "client_id_key": "app_id",\n "client_secret_key": "secret",\n "auth_code_key": "auth_code"\n }\n }\n }', - title="DeclarativeOAuth Connector Specification", - ) + oauth_connector_input_specification: Optional[OauthConnectorInputSpecification] = Field( + None, + description='The DeclarativeOAuth specific blob.\nPertains to the fields defined by the connector relating to the OAuth flow.\n\nInterpolation capabilities:\n- The variables placeholders are declared as `{{my_var}}`.\n- The nested resolution variables like `{{ {{my_nested_var}} }}` is allowed as well.\n\n- The allowed interpolation context is:\n + base64Encoder - encode to `base64`, {{ {{my_var_a}}:{{my_var_b}} | base64Encoder }}\n + base64Decorer - decode from `base64` encoded string, {{ {{my_string_variable_or_string_value}} | base64Decoder }}\n + urlEncoder - encode the input string to URL-like format, {{ https://test.host.com/endpoint | urlEncoder}}\n + urlDecorer - decode the input url-encoded string into text format, {{ urlDecoder:https%3A%2F%2Fairbyte.io | urlDecoder}}\n + codeChallengeS256 - get the `codeChallenge` encoded value to provide additional data-provider specific authorisation values, {{ {{state_value}} | codeChallengeS256 }}\n\nExamples:\n - The TikTok Marketing DeclarativeOAuth spec:\n {\n "oauth_connector_input_specification": {\n "type": "object",\n "additionalProperties": false,\n "properties": {\n "consent_url": "https://ads.tiktok.com/marketing_api/auth?{{client_id_key}}={{client_id_value}}&{{redirect_uri_key}}={{ {{redirect_uri_value}} | urlEncoder}}&{{state_key}}={{state_value}}",\n "access_token_url": "https://business-api.tiktok.com/open_api/v1.3/oauth2/access_token/",\n "access_token_params": {\n "{{ auth_code_key }}": "{{ auth_code_value }}",\n "{{ client_id_key }}": "{{ client_id_value }}",\n "{{ client_secret_key }}": "{{ client_secret_value }}"\n },\n "access_token_headers": {\n "Content-Type": "application/json",\n "Accept": "application/json"\n },\n "extract_output": ["data.access_token"],\n "client_id_key": "app_id",\n "client_secret_key": "secret",\n "auth_code_key": "auth_code"\n }\n }\n }', + title="DeclarativeOAuth Connector Specification", ) complete_oauth_output_specification: Optional[Dict[str, Any]] = Field( None, @@ -1102,9 +1097,7 @@ class Config: complete_oauth_server_input_specification: Optional[Dict[str, Any]] = Field( None, description="OAuth specific blob. This is a Json Schema used to validate Json configurations persisted as Airbyte Server configurations.\nMust be a valid non-nested JSON describing additional fields configured by the Airbyte Instance or Workspace Admins to be used by the\nserver when completing an OAuth flow (typically exchanging an auth code for refresh token).\nExamples:\n complete_oauth_server_input_specification={\n client_id: {\n type: string\n },\n client_secret: {\n type: string\n }\n }", - examples=[ - {"client_id": {"type": "string"}, "client_secret": {"type": "string"}} - ], + examples=[{"client_id": {"type": "string"}, "client_secret": {"type": "string"}}], title="OAuth input specification", ) complete_oauth_server_output_specification: Optional[Dict[str, Any]] = Field( @@ -1341,34 +1334,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( @@ -1694,9 +1692,7 @@ class CustomConfigTransformation(BaseModel): class_name: str = Field( ..., description="Fully-qualified name of the class that will be implementing the custom config transformation. The format is `source_..`.", - examples=[ - "source_declarative_manifest.components.MyCustomConfigTransformation" - ], + examples=["source_declarative_manifest.components.MyCustomConfigTransformation"], ) parameters: Optional[Dict[str, Any]] = Field( None, @@ -2014,42 +2010,6 @@ class DatetimeBasedCursor(BaseModel): parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") -class OpaqueTokenCursor(BaseModel): - class Config: - extra = Extra.allow - - type: Literal["OpaqueTokenCursor"] - token_path: List[str] = Field( - ..., - description="Path to the next token in the decoded response that ends a pagination pass. The value must be a string or an integer, and is stored and sent as a string. A missing, null or empty value means the response carried no token.", - examples=[["syncToken"], ["nextSyncToken"], ["meta", "sync_token"]], - title="Token Path", - ) - token_option: RequestOption = Field( - ..., - description="Where to send the stored token on each request of a pass: the JSON body, the query string, a header or the form body.", - title="Send Token In", - ) - completion_condition: Optional[str] = Field( - None, - description="Must be true on the last response of a pass for its token to be committed. Defaults to a token being present at token_path. Set it when the API repeats the old token on intermediate pages or returns a position on every page.", - examples=["{{ not response.moreDataAvailable }}"], - title="Completion Condition", - ) - initial_token: Optional[str] = Field( - None, - description='Value to send when no token is stored, for an API that needs one to start a full pass, such as Todoist\'s "*". When unset, nothing is sent.', - examples=["*"], - title="Initial Token", - ) - cursor_field: Optional[str] = Field( - None, - description="Optional top-level record field advertised as the stream's cursor, so destinations can order versions of a record when deduplicating. It is never used to filter records or to build state.", - title="Destination Cursor Field", - ) - parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") - - class JwtAuthenticator(BaseModel): type: Literal["JwtAuthenticator"] secret_key: str = Field( @@ -2230,9 +2190,7 @@ class OAuthAuthenticator(BaseModel): scopes: Optional[List[str]] = Field( None, description="List of scopes that should be granted to the access token.", - examples=[ - ["crm.list.read", "crm.objects.contacts.read", "crm.schema.contacts.read"] - ], + examples=[["crm.list.read", "crm.objects.contacts.read", "crm.schema.contacts.read"]], title="Scopes", ) token_expiry_date: Optional[str] = Field( @@ -2383,6 +2341,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( @@ -2480,12 +2485,10 @@ class DpathValidator(BaseModel): ], title="Field Path", ) - validation_strategy: Union[ValidateAdheresToSchema, CustomValidationStrategy] = ( - Field( - ..., - description="The condition that the specified config value will be evaluated against", - title="Validation Strategy", - ) + validation_strategy: Union[ValidateAdheresToSchema, CustomValidationStrategy] = Field( + ..., + description="The condition that the specified config value will be evaluated against", + title="Validation Strategy", ) @@ -2502,12 +2505,10 @@ class PredicateValidator(BaseModel): ], title="Value", ) - validation_strategy: Union[ValidateAdheresToSchema, CustomValidationStrategy] = ( - Field( - ..., - description="The validation strategy to apply to the value.", - title="Validation Strategy", - ) + validation_strategy: Union[ValidateAdheresToSchema, CustomValidationStrategy] = Field( + ..., + description="The validation strategy to apply to the value.", + title="Validation Strategy", ) @@ -2589,12 +2590,12 @@ class RateLimitedMultipleTokenAuthenticator(BaseModel): class CompositeErrorHandler(BaseModel): type: Literal["CompositeErrorHandler"] - error_handlers: List[ - Union[CompositeErrorHandler, DefaultErrorHandler, CustomErrorHandler] - ] = Field( - ..., - description="List of error handlers to iterate on to determine how to handle a failed response.", - title="Error Handlers", + error_handlers: List[Union[CompositeErrorHandler, DefaultErrorHandler, CustomErrorHandler]] = ( + Field( + ..., + description="List of error handlers to iterate on to determine how to handle a failed response.", + title="Error Handlers", + ) ) parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") @@ -2632,20 +2633,80 @@ 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 type: Literal["ZipfileDecoder"] - decoder: Union[ - CsvDecoder, GzipDecoder, JsonDecoder, JsonItemsDecoder, JsonlDecoder - ] = Field( + decoder: Union[CsvDecoder, GzipDecoder, JsonDecoder, JsonItemsDecoder, JsonlDecoder] = Field( ..., description="Parser to parse the decompressed data from the zipfile(s).", title="Parser", ) +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( @@ -2763,9 +2824,9 @@ class Config: type: Literal["DeclarativeSource"] check: Union[CheckStream, CheckDynamicStream] - streams: Optional[ - List[Union[ConditionalStreams, DeclarativeStream, StateDelegatingStream]] - ] = None + streams: Optional[List[Union[ConditionalStreams, DeclarativeStream, StateDelegatingStream]]] = ( + None + ) dynamic_streams: List[DynamicDeclarativeStream] version: str = Field( ..., @@ -2872,11 +2933,11 @@ class FileUploader(BaseModel): ..., description="Requester component that describes how to prepare HTTP requests to send to the source API.", ) - download_target_extractor: Union[ - DpathExtractor, CustomRecordExtractor, CombinedExtractor - ] = Field( - ..., - description="Responsible for fetching the url where the file is located. This is applied on each records and not on the HTTP response", + download_target_extractor: Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor] = ( + Field( + ..., + description="Responsible for fetching the url where the file is located. This is applied on each records and not on the HTTP response", + ) ) file_extractor: Optional[Union[DpathExtractor, CustomRecordExtractor]] = Field( None, @@ -2893,14 +2954,48 @@ class FileUploader(BaseModel): parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") +class OpaqueTokenCursor(BaseModel): + class Config: + extra = Extra.allow + + type: Literal["OpaqueTokenCursor"] + token_path: List[str] = Field( + ..., + description="Path to the next token in the decoded response that ends a pagination pass. The value must be a string or an integer, and is stored and sent as a string. A missing, null or empty value means the response carried no token.", + examples=[["syncToken"], ["nextSyncToken"], ["meta", "sync_token"]], + title="Token Path", + ) + token_option: RequestOption = Field( + ..., + description="Where to send the stored token on each request of a pass: the JSON body, the query string, a header or the form body.", + title="Send Token In", + ) + completion_condition: Optional[str] = Field( + None, + description="Must be true on the last response of a pass for its token to be committed. Defaults to a token being present at token_path. Set it when the API repeats the old token on intermediate pages or returns a position on every page.", + examples=["{{ not response.moreDataAvailable }}"], + title="Completion Condition", + ) + initial_token: Optional[str] = Field( + None, + description='Value to send when no token is stored, for an API that needs one to start a full pass, such as Todoist\'s "*". When unset, nothing is sent.', + examples=["*"], + title="Initial Token", + ) + cursor_field: Optional[str] = Field( + None, + description="Optional top-level record field advertised as the stream's cursor, so destinations can order versions of a record when deduplicating. It is never used to filter records or to build state.", + title="Destination Cursor Field", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class DeclarativeStream(BaseModel): class Config: extra = Extra.allow type: Literal["DeclarativeStream"] - name: Optional[str] = Field( - "", description="The stream name.", example=["Users"], title="Name" - ) + name: Optional[str] = Field("", description="The stream name.", example=["Users"], title="Name") retriever: Union[SimpleRetriever, AsyncRetriever, CustomRetriever] = Field( ..., description="Component used to coordinate how records are extracted across stream slices and request pages.", @@ -2967,96 +3062,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( @@ -3176,20 +3181,18 @@ class HttpRequester(BaseModelWithDeprecations): description="For APIs that require explicit specification of the properties to query for, this component will take a static or dynamic set of properties (which can be optionally split into chunks) and allow them to be injected into an outbound request by accessing stream_partition.extra_fields.", title="Query Properties", ) - request_parameters: Optional[Union[Dict[str, Union[str, QueryProperties]], str]] = ( - Field( - None, - description="Specifies the query parameters that should be set on an outgoing HTTP request given the inputs.", - examples=[ - {"unit": "day"}, - { - "query": 'last_event_time BETWEEN TIMESTAMP "{{ stream_interval.start_time }}" AND TIMESTAMP "{{ stream_interval.end_time }}"' - }, - {"searchIn": "{{ ','.join(config.get('search_in', [])) }}"}, - {"sort_by[asc]": "updated_at"}, - ], - title="Query Parameters", - ) + request_parameters: Optional[Union[Dict[str, Union[str, QueryProperties]], str]] = Field( + None, + description="Specifies the query parameters that should be set on an outgoing HTTP request given the inputs.", + examples=[ + {"unit": "day"}, + { + "query": 'last_event_time BETWEEN TIMESTAMP "{{ stream_interval.start_time }}" AND TIMESTAMP "{{ stream_interval.end_time }}"' + }, + {"searchIn": "{{ ','.join(config.get('search_in', [])) }}"}, + {"sort_by[asc]": "updated_at"}, + ], + title="Query Parameters", ) request_headers: Optional[Union[Dict[str, str], str]] = Field( None, @@ -3243,16 +3246,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( @@ -3377,34 +3382,9 @@ 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" - ) + name: str = Field(..., description="The stream name.", example=["Users"], title="Name") full_refresh_stream: DeclarativeStream = Field( ..., description="Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.", @@ -3505,9 +3485,7 @@ class AsyncRetriever(BaseModel): status_mapping: AsyncJobStatusMap = Field( ..., description="Async Job Status to Airbyte CDK Async Job Status mapping." ) - status_extractor: Union[ - DpathExtractor, CustomRecordExtractor, CombinedExtractor - ] = Field( + status_extractor: Union[DpathExtractor, CustomRecordExtractor, CombinedExtractor] = Field( ..., description="Responsible for fetching the actual status of the async job." ) download_target_extractor: Optional[ @@ -3517,15 +3495,8 @@ 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, - ] - ] = Field( - None, description="Responsible for fetching the records from provided urls." - ) + Union[DpathExtractor, CustomRecordExtractor, ResponseToFileExtractor, CombinedExtractor] + ] = Field(None, description="Responsible for fetching the records from provided urls.") creation_requester: Union[HttpRequester, CustomRequester] = Field( ..., description="Requester component that describes how to prepare HTTP requests to send to the source API to create the async server-side job.", @@ -3538,9 +3509,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, @@ -3727,13 +3699,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 4ced6454cba98c2f18ffc271fe31594c6bd2663d Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 20 Oct 2026 04:41:13 +0000 Subject: [PATCH 3/3] refactor(declarative): rename RESET_OPAQUE_TOKEN to RESET_SYNC_TOKEN Co-Authored-By: bot_apk --- .../declarative_component_schema.yaml | 4 ++-- .../incremental/sync_token_cursor.py | 2 +- .../models/declarative_component_schema.py | 4 ++-- .../parsers/model_to_component_factory.py | 2 +- .../error_handlers/composite_error_handler.py | 2 +- .../declarative/retrievers/simple_retriever.py | 12 ++++++------ .../http/error_handlers/response_models.py | 2 +- .../sources/streams/http/http_client.py | 10 +++++----- ...eption.py => sync_token_reset_exception.py} | 8 ++++---- .../test_sync_token_cursor_integration.py | 18 +++++++++--------- .../test_composite_error_handler.py | 4 ++-- .../sources/streams/http/test_http_client.py | 12 ++++++------ 12 files changed, 40 insertions(+), 40 deletions(-) rename airbyte_cdk/sources/streams/http/{opaque_token_reset_exception.py => sync_token_reset_exception.py} (77%) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 5469bc2383..f2e4d376ee 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -2884,7 +2884,7 @@ definitions: - RETRY - IGNORE - RESET_PAGINATION - - RESET_OPAQUE_TOKEN + - RESET_SYNC_TOKEN - RATE_LIMITED - REFRESH_TOKEN_THEN_RETRY - REDUCE_PAGE_SIZE @@ -2895,7 +2895,7 @@ definitions: - RETRY - IGNORE - RESET_PAGINATION - - RESET_OPAQUE_TOKEN + - RESET_SYNC_TOKEN - RATE_LIMITED - REFRESH_TOKEN_THEN_RETRY - REDUCE_PAGE_SIZE diff --git a/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py b/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py index d124560d29..caf76f85d4 100644 --- a/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py +++ b/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py @@ -203,7 +203,7 @@ def restart_without_token( if "token" not in stream_slice.cursor_slice: message = ( f"Stream {self._stream_name}: the API rejected a read without a change token. " - "Check the RESET_OPAQUE_TOKEN response filter." + "Check the RESET_SYNC_TOKEN response filter." ) raise AirbyteTracedException( internal_message=message, diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index a63e42c4e1..abca5f724d 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -614,7 +614,7 @@ class Action(Enum): RETRY = "RETRY" IGNORE = "IGNORE" RESET_PAGINATION = "RESET_PAGINATION" - RESET_OPAQUE_TOKEN = "RESET_OPAQUE_TOKEN" + RESET_SYNC_TOKEN = "RESET_SYNC_TOKEN" RATE_LIMITED = "RATE_LIMITED" REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY" REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE" @@ -638,7 +638,7 @@ class HttpResponseFilter(BaseModel): "RETRY", "IGNORE", "RESET_PAGINATION", - "RESET_OPAQUE_TOKEN", + "RESET_SYNC_TOKEN", "RATE_LIMITED", "REFRESH_TOKEN_THEN_RETRY", "REDUCE_PAGE_SIZE", 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 fc6ce54306..121c573696 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -4182,7 +4182,7 @@ def _get_url(req: Requester) -> str: cursor.on_page_completed if isinstance(cursor, SyncTokenCursor) else None ), token_reset_handler=( - cursor.restart_without_token if isinstance(cursor, OpaqueTokenCursor) else None + cursor.restart_without_token if isinstance(cursor, SyncTokenCursor) else None ), parameters=model.parameters or {}, ) diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py index 06560593af..0d416adaaf 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py @@ -71,7 +71,7 @@ def interpret_response( ResponseAction.RETRY, ResponseAction.IGNORE, ResponseAction.RESET_PAGINATION, - ResponseAction.RESET_OPAQUE_TOKEN, + ResponseAction.RESET_SYNC_TOKEN, ResponseAction.REDUCE_PAGE_SIZE, ResponseAction.SPLIT_REQUEST_WINDOW, ]: diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 6adac54946..2cde0f8adb 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -57,10 +57,6 @@ from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer from airbyte_cdk.sources.source import ExperimentalClassWarning from airbyte_cdk.sources.streams.core import StreamData -from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import ( - OpaqueTokenResetNotSupportedException, - OpaqueTokenResetRequiredException, -) from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionNotSupportedException, PageSizeReductionRequiredException, @@ -72,6 +68,10 @@ RequestWindowSplitNotSupportedException, RequestWindowSplitRequiredException, ) +from airbyte_cdk.sources.streams.http.sync_token_reset_exception import ( + SyncTokenResetNotSupportedException, + SyncTokenResetRequiredException, +) from airbyte_cdk.sources.types import Config, Record, StreamSlice from airbyte_cdk.utils.mapping_helpers import combine_mappings from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -708,9 +708,9 @@ def _read_records_or_split_request_window( yield from self._read_records_or_split_request_window( records_schema, child, original_slice, depth + 1 ) - except OpaqueTokenResetRequiredException as exception: + except SyncTokenResetRequiredException as exception: if self.token_reset_handler is None: - raise OpaqueTokenResetNotSupportedException(stream_name=self.name) from exception + raise SyncTokenResetNotSupportedException(stream_name=self.name) from exception reset_slice = self.token_reset_handler(stream_slice, exception.error_message) reason = exception.error_message or "the API rejected the stored change token" diff --git a/airbyte_cdk/sources/streams/http/error_handlers/response_models.py b/airbyte_cdk/sources/streams/http/error_handlers/response_models.py index 129e086d49..6b53582494 100644 --- a/airbyte_cdk/sources/streams/http/error_handlers/response_models.py +++ b/airbyte_cdk/sources/streams/http/error_handlers/response_models.py @@ -17,7 +17,7 @@ class ResponseAction(Enum): FAIL = "FAIL" IGNORE = "IGNORE" RESET_PAGINATION = "RESET_PAGINATION" - RESET_OPAQUE_TOKEN = "RESET_OPAQUE_TOKEN" + RESET_SYNC_TOKEN = "RESET_SYNC_TOKEN" RATE_LIMITED = "RATE_LIMITED" REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY" REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE" diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 97cf6e125d..aae206966a 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -43,9 +43,6 @@ RequestBodyException, UserDefinedBackoffException, ) -from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import ( - OpaqueTokenResetRequiredException, -) from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionRequiredException, ) @@ -68,6 +65,9 @@ ResponseAwareAuthenticator, TokenRotatingAuthenticator, ) +from airbyte_cdk.sources.streams.http.sync_token_reset_exception import ( + SyncTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter from airbyte_cdk.sources.utils.types import JsonType from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets @@ -623,8 +623,8 @@ def _handle_error_resolution( if error_resolution.response_action == ResponseAction.RESET_PAGINATION: raise PaginationResetRequiredException() - if error_resolution.response_action == ResponseAction.RESET_OPAQUE_TOKEN: - raise OpaqueTokenResetRequiredException(error_resolution.error_message) + if error_resolution.response_action == ResponseAction.RESET_SYNC_TOKEN: + raise SyncTokenResetRequiredException(error_resolution.error_message) if error_resolution.response_action == ResponseAction.REDUCE_PAGE_SIZE: raise PageSizeReductionRequiredException( diff --git a/airbyte_cdk/sources/streams/http/opaque_token_reset_exception.py b/airbyte_cdk/sources/streams/http/sync_token_reset_exception.py similarity index 77% rename from airbyte_cdk/sources/streams/http/opaque_token_reset_exception.py rename to airbyte_cdk/sources/streams/http/sync_token_reset_exception.py index b03e36d4be..b8ec4cfd64 100644 --- a/airbyte_cdk/sources/streams/http/opaque_token_reset_exception.py +++ b/airbyte_cdk/sources/streams/http/sync_token_reset_exception.py @@ -8,18 +8,18 @@ from airbyte_cdk.utils.traced_exception import AirbyteTracedException -class OpaqueTokenResetRequiredException(Exception): +class SyncTokenResetRequiredException(Exception): def __init__(self, error_message: Optional[str] = None) -> None: self.error_message = error_message super().__init__(error_message) -class OpaqueTokenResetNotSupportedException(AirbyteTracedException): +class SyncTokenResetNotSupportedException(AirbyteTracedException): def __init__(self, stream_name: Optional[str] = None) -> None: stream = f"Stream {stream_name}" if stream_name else "The stream" message = ( - f"{stream} resolves to a token reset but does not use an OpaqueTokenCursor. " - "Remove the RESET_OPAQUE_TOKEN action from its error handler." + f"{stream} resolves to a token reset but does not use a SyncTokenCursor. " + "Remove the RESET_SYNC_TOKEN action from its error handler." ) super().__init__( internal_message=message, diff --git a/unit_tests/sources/declarative/incremental/test_sync_token_cursor_integration.py b/unit_tests/sources/declarative/incremental/test_sync_token_cursor_integration.py index e2dc46087f..91f368f76d 100644 --- a/unit_tests/sources/declarative/incremental/test_sync_token_cursor_integration.py +++ b/unit_tests/sources/declarative/incremental/test_sync_token_cursor_integration.py @@ -230,7 +230,7 @@ def _reset_error_handler(*, reduce_page_size=False, pagination_reset=False): response_filters.append( { "type": "HttpResponseFilter", - "action": "RESET_OPAQUE_TOKEN", + "action": "RESET_SYNC_TOKEN", "predicate": "{{ response.get('success') == false }}", "error_message": "Ashby rejected the stored change token (sync_token_expired)", } @@ -350,7 +350,7 @@ def test_t_fail_preserves_committed_token_and_emits_incomplete_status(): def test_t_reset_restarts_without_token_and_commits_terminal_token(caplog): - state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) manifest = _manifest(error_handler=_reset_error_handler()) with HttpMocker() as http_mocker: http_mocker.post( @@ -392,7 +392,7 @@ def test_t_reset_restarts_without_token_and_commits_terminal_token(caplog): def test_t_reset_context_is_fresh_after_page_size_reduction(caplog): - state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) manifest = _manifest(error_handler=_reset_error_handler(reduce_page_size=True)) stream = manifest["streams"][0] stream["retriever"]["paginator"]["page_size_option"] = { @@ -473,7 +473,7 @@ def test_t_reset_context_is_fresh_after_page_size_reduction(caplog): def test_t_reset_twice_errors_and_preserves_committed_state(): - state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) manifest = _manifest(error_handler=_reset_error_handler()) with HttpMocker() as http_mocker: http_mocker.post( @@ -503,13 +503,13 @@ def test_t_reset_twice_errors_and_preserves_committed_state(): and message.trace.error.failure_type == FailureType.system_error and message.trace.error.message == "Stream events: the API rejected a read without a change token. " - "Check the RESET_OPAQUE_TOKEN response filter." + "Check the RESET_SYNC_TOKEN response filter." for message in messages ) def test_t_rpag_reset_replays_resumed_page_one_with_the_same_token(): - state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) manifest = _manifest(error_handler=_reset_error_handler(pagination_reset=True)) with HttpMocker() as http_mocker: http_mocker.post( @@ -541,7 +541,7 @@ def test_t_rpag_reset_replays_resumed_page_one_with_the_same_token(): def test_t_rpag_reset_after_token_reset_replays_tokenless_page_one(): - state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) manifest = _manifest(error_handler=_reset_error_handler(pagination_reset=True)) with HttpMocker() as http_mocker: http_mocker.post( @@ -582,7 +582,7 @@ def test_t_rpag_reset_after_token_reset_replays_tokenless_page_one(): def test_t_rpag_second_pagination_reset_is_system_error(): - state = _incoming_state({"state_type": "opaque_token", "token": "T1"}) + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) manifest = _manifest(error_handler=_reset_error_handler(pagination_reset=True)) with HttpMocker() as http_mocker: http_mocker.post( @@ -623,7 +623,7 @@ def test_reset_action_on_non_token_stream_is_not_supported(): _read(source, catalog) assert exc_info.value.failure_type.value == "config_error" - assert "RESET_OPAQUE_TOKEN" in str(exc_info.value) + assert "RESET_SYNC_TOKEN" in str(exc_info.value) def test_t_builder_page_limit_and_max_records_do_not_commit_tokens(): diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py b/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py index 4bfbdfc5fa..6181cd1b72 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py @@ -132,10 +132,10 @@ def test_given_unmatched_response_or_exception_then_return_default_error_resolut assert error_resolution.failure_type == FailureType.system_error -def test_composite_error_handler_preserves_reset_opaque_token(): +def test_composite_error_handler_preserves_reset_sync_token(): reset_handler = MagicMock() reset_resolution = ErrorResolution( - response_action=ResponseAction.RESET_OPAQUE_TOKEN, + response_action=ResponseAction.RESET_SYNC_TOKEN, error_message="sync_token_expired", ) reset_handler.interpret_response.return_value = reset_resolution diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index abc7a03596..0b8bb28d1f 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -35,9 +35,6 @@ UserDefinedBackoffException, ) from airbyte_cdk.sources.streams.http.http_client import MessageRepresentationAirbyteTracedErrors -from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import ( - OpaqueTokenResetRequiredException, -) from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionRequiredException, ) @@ -48,6 +45,9 @@ Oauth2Authenticator, TokenAuthenticator, ) +from airbyte_cdk.sources.streams.http.sync_token_reset_exception import ( + SyncTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter from airbyte_cdk.utils.constants import ENV_REQUEST_CACHE_PATH from airbyte_cdk.utils.datetime_helpers import ab_datetime_now @@ -247,16 +247,16 @@ def test_send_raises_airbyte_traced_exception_with_fail_response_action(): assert http_client._session.send.call_count == 1 -def test_reset_opaque_token_action_raises_exception_with_error_message(): +def test_reset_sync_token_action_raises_exception_with_error_message(): http_client = test_http_client() error_message = "sync_token_expired" - with pytest.raises(OpaqueTokenResetRequiredException) as exc_info: + with pytest.raises(SyncTokenResetRequiredException) as exc_info: http_client._handle_error_resolution( response=None, exc=None, request=requests.PreparedRequest(), error_resolution=ErrorResolution( - response_action=ResponseAction.RESET_OPAQUE_TOKEN, + response_action=ResponseAction.RESET_SYNC_TOKEN, error_message=error_message, ), )