fix: make ResponseToFileExtractor respect configured delimiter instead of hardcoding comma

ResponseToFileExtractor was using pd.read_csv(dialect='unix') which hardcodes
a comma delimiter, ignoring the configured download_decoder's CsvDecoder
delimiter setting. This caused tab-separated (TSV) files to be parsed as a
single column.

Changes:
- Add 'delimiter' parameter to ResponseToFileExtractor (default: ',')
- Replace dialect='unix' with explicit CSV options + configurable delimiter
- Add delimiter property to declarative component schema
- Pass delimiter from model to component in factory
- Add tests for tab-delimited CSV parsing

Related: #909 (gzip decompression fix for ResponseToFileExtractor)
Co-Authored-By: Ilja Herdt <ilja.herdt@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1940,6 +1940,11 @@ definitions:
       type:
         type: string
         enum: [ResponseToFileExtractor]
+      delimiter:
+        title: Delimiter
+        description: The delimiter used to separate values in the CSV data. Defaults to comma (',').
+        type: string
+        default: ","
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/extractors/response_to_file_extractor.py

@@ -1,6 +1,7 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
+import csv
 import logging
 import os
 import uuid
@@ -30,9 +31,12 @@ class ResponseToFileExtractor(RecordExtractor):
     """
 
     parameters: InitVar[Mapping[str, Any]]
+    delimiter: str = ","
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.logger = logging.getLogger("airbyte")
+        if self.delimiter.startswith("\\"):
+            self.delimiter = self.delimiter.encode("utf-8").decode("unicode_escape")
 
     def _get_response_encoding(self, headers: Dict[str, Any]) -> str:
         """
@@ -137,7 +141,14 @@ def _read_with_chunks(
         try:
             with open(path, "r", encoding=file_encoding) as data:
                 chunks = pd.read_csv(
-                    data, chunksize=chunk_size, iterator=True, dialect="unix", dtype=object
+                    data,
+                    chunksize=chunk_size,
+                    iterator=True,
+                    dtype=object,
+                    delimiter=self.delimiter,
+                    quoting=csv.QUOTE_ALL,
+                    doublequote=True,
+                    lineterminator="\n",
                 )
                 for chunk in chunks:
                     chunk = chunk.replace({nan: None}).to_dict(orient="records")

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1,5 +1,3 @@
-# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
-
 # generated by datamodel-codegen:
 #   filename:  declarative_component_schema.yaml
 
@@ -502,6 +500,11 @@ class DpathExtractor(BaseModel):
 
 class ResponseToFileExtractor(BaseModel):
     type: Literal["ResponseToFileExtractor"]
+    delimiter: Optional[str] = Field(
+        ",",
+        description="The delimiter used to separate values in the CSV data. Defaults to comma (',').",
+        title="Delimiter",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
@@ -930,24 +933,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,
@@ -965,7 +972,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(
@@ -1469,7 +1478,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_<name>.<package>.<class_name>`.",
-        examples=["source_declarative_manifest.components.MyCustomConfigTransformation"],
+        examples=[
+            "source_declarative_manifest.components.MyCustomConfigTransformation"
+        ],
     )
     parameters: Optional[Dict[str, Any]] = Field(
         None,
@@ -1897,7 +1908,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(
@@ -2124,7 +2137,9 @@ class RecordSelector(BaseModel):
         description="Responsible for filtering records to be emitted by the Source.",
         title="Record Filter",
     )
-    schema_normalization: Optional[Union[SchemaNormalization, CustomSchemaNormalization]] = Field(
+    schema_normalization: Optional[
+        Union[SchemaNormalization, CustomSchemaNormalization]
+    ] = Field(
         None,
         description="Responsible for normalization according to the schema.",
         title="Schema Normalization",
@@ -2166,10 +2181,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",
+        )
     )
 
 
@@ -2186,10 +2203,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",
+        )
     )
 
 
@@ -2214,12 +2233,12 @@ class ConfigAddFields(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")
 
@@ -2381,9 +2400,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(
         ...,
@@ -2508,16 +2527,20 @@ 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.",
         title="Retriever",
     )
-    incremental_sync: Optional[Union[DatetimeBasedCursor, IncrementingCountCursor]] = Field(
-        None,
-        description="Component used to fetch data incrementally based on a time field in the data.",
-        title="Incremental Sync",
+    incremental_sync: Optional[Union[DatetimeBasedCursor, IncrementingCountCursor]] = (
+        Field(
+            None,
+            description="Component used to fetch data incrementally based on a time field in the data.",
+            title="Incremental Sync",
+        )
     )
     primary_key: Optional[PrimaryKey] = Field("", title="Primary Key")
     schema_loader: Optional[
@@ -2691,18 +2714,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,
@@ -2874,7 +2899,9 @@ class QueryProperties(BaseModel):
 
 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.",
@@ -2961,13 +2988,17 @@ class AsyncRetriever(BaseModel):
     status_extractor: Union[DpathExtractor, CustomRecordExtractor] = Field(
         ..., description="Responsible for fetching the actual status of the async job."
     )
-    download_target_extractor: Optional[Union[DpathExtractor, CustomRecordExtractor]] = Field(
+    download_target_extractor: Optional[
+        Union[DpathExtractor, CustomRecordExtractor]
+    ] = Field(
         None,
         description="Responsible for fetching the final result `urls` provided by the completed / finished / ready async job.",
     )
     download_extractor: Optional[
         Union[DpathExtractor, CustomRecordExtractor, ResponseToFileExtractor]
-    ] = Field(None, description="Responsible for fetching the records from provided urls.")
+    ] = 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.",