https://linen.dev logo
Join Slack
Powered by
<@U04S8PLJWMT> if I am building a custom airbyte s...
# replication-ask-ai
y
@kapa.ai if I am building a custom airbyte source that pulls from an API but the response is in csv and not json. How would I account for this in my source definition?
k
Airbyte can handle CSV responses, but how you do it depends on whether you’re using: - Connector Builder / config-based (low-code), or - Custom code (Python CDK). Below is what’s documented. --- ### 1. Using Connector Builder (no-/low-code) In the Connector Builder UI, you can declare that the HTTP response is CSV so Airbyte will decode it into JSON records for you. For a normal (non-async) HTTP stream: - In the Record processing / Response Decoding section, set the HTTP Response Format (or equivalent) to CSV. - Airbyte’s CSV decoder will: - Use the first row as headers. - Convert each subsequent row into a JSON object keyed by those headers. - Allow you to change the delimiter if it’s not a comma. [[Record processing CSV](https://docs.airbyte.com/platform/connector-development/connector-builder-ui/record-processing#csv)] For asynchronous job streams (Builder “Async streams”): - In the Download tab, set: - Download HTTP Response Format (or HTTP Response Format) to CSV. - Leave the Record Selector / Field Path empty if the entire CSV is the dataset (like the SendGrid example). [[Async Download tab](https://docs.airbyte.com/platform/connector-development/connector-builder-ui/async-streams#download-tab); [Record processing CSV](https://docs.airbyte.com/platform/connector-development/connector-builder-ui/record-processing#csv)] This is fully supported in recent platform docs (1.6+ / 2.0). --- ### 2. Using the config-based low-code framework (YAML) Older guidance (before built‑in CSV decoding in Builder) says: - The low-code framework is primarily JSON-oriented. - For CSV API responses, you can: - Override the extractor with a custom CSV extractor component instead of the default 
DpathExtractor
. - In YAML, that looks like:
Copy code
yaml     record_selector:       type: RecordSelector       extractor:         type: YourCustomCSVExtractor       field_path: []
- Your custom extractor must parse the CSV and return a list of records (dicts). [[CSV API response config](https://discuss.airbyte.io/t/5237); [Configure CSV in builder](https://discuss.airbyte.io/t/5341)] This requires some Python coding and is not purely low-code. --- ### 3. Using the Python CDK The knowledge sources only give a pattern, not full official docs, but they show: - Implement 
parse_response
to: - Read 
response.text
(CSV), - Use 
csv.DictReader
to turn rows into dicts, - 
yield
each row as a record. [[Plan to build CSV parsers](https://discuss.airbyte.io/t/5236)] Example from the forum:
Copy code
python def parse_response(     self,     response: requests.Response,     stream_state: Mapping[str, Any],     stream_slice: Mapping[str, Any] = None,     next_page_token: Mapping[str, Any] = None, ) -> Iterable[Mapping]:     csv_rows = csv.DictReader(response.text.splitlines())     for row in csv_rows:         yield {             "timestamp": row["timestamp"],             "country": row["country"],             "spend": row["spend"],         }
--- Summary - Connector Builder: set the HTTP response format to **CSV**; Airbyte will decode rows into JSON records. [[Record processing CSV](https://docs.airbyte.com/platform/connector-development/connector-builder-ui/record-processing#csv)] - Config-based YAML: replace 
DpathExtractor
with a custom CSV extractor. [[CSV API response config](https://discuss.airbyte.io/t/5237)] - Python CDK: implement 
parse_response
to parse CSV and yield dicts. [[Plan to build CSV parsers](https://discuss.airbyte.io/t/5236)] If you tell me which approach you’re using (Builder vs YAML vs Python CDK), I can map this to the exact fields you need to set.
y
@kapa.ai in the newer versions of airbyte-cdk it looks like a csv decoder is available for use but cannot see any docs on how this is defined in the manifest
k
You’re right that a CSV decoder exists in the config-based (manifest) framework; it’s exposed as 
CsvDecoder
in the YAML schema. The docs that do exist are the YAML reference entries for that component. From the YAML reference: > CsvDecoder object > Select 'CSV' for response data that is formatted as CSV (comma-separated values). Can specify an encoding (default: 'utf-8') and a delimiter (default: ','). [CsvDecoder ref] Properties: • 
encoding
(string) • 
delimiter
(string) • 
set_values_to_none
(array) [CsvDecoder ref] However, the knowledge sources do not show a full manifest snippet wiring 
CsvDecoder
into a stream (e.g., under 
retriever
→ 
record_selector
→ 
decoder
or similar), nor do they show how to combine it with 
GzipDecoder
or other components. [CsvDecoder ref; GzipDecoder ref] So, based only on the available docs: • Yes, 
CsvDecoder
is a first-class component in the manifest schema. • You can configure at least 
encoding
, 
delimiter
, and 
set_values_to_none
. • The docs do not explicitly document the exact manifest location/structure where 
CsvDecoder
should be referenced (e.g., which field name to use in a stream definition). If you need the precise manifest shape (the exact keys/section where 
CsvDecoder
is plugged in), that detail is not present in these knowledge sources, so I can’t reliably give you a concrete YAML example without guessing. 👍 0 button 👎 0 button ℹ️ Tip: Mention @kapa.ai in the thread for follow-ups or DM the app for 1-to-1 help.
y
@kapa.ai i am not able to get the substreampartitionrouter to inject the path parameter. When running in debug mode I see the url still as 
health/{{ stream_partition.healthModelId }}/team-health-historical
and when I run a check i get a 500. Error: Internal server error
k
The knowledge sources don’t show your exact manifest, but they do clarify a couple of common pitfalls with 
SubstreamPartitionRouter
and path injection. Key points from the docs: 1. Use 
stream_partition
, not 
stream_slice
, in the path For substreams, the value coming from the parent is exposed as `stream_partition.`partition_field. In the forum example, the fix was exactly this:
Copy code
path: "/event/webinars/{{ stream_partition.event }}"
   ...
   partition_router:
     - type: SubstreamPartitionRouter
       parent_stream_configs:
         - type: ParentStreamConfig
           parent_key: id
           partition_field: event
           stream:
             $ref: '#/definitions/events_stream'
The user originally used 
stream_slice
and saw the literal template in the URL; switching to 
stream_partition
made it work. [Forum fix] You’re already using 
{{ stream_partition.healthModelId }}
, which is correct if your 
partition_field
is 
healthModelId
. 2. 
partition_field
must match what you use in the template In the config-based docs, the 
partition_field
is the identifier you later reference in the template:
Copy code
partition_router:
     type: SubstreamPartitionRouter
     parent_stream_configs:
       - stream: "#/repositories_stream"
         parent_key: "id"
         partition_field: "repository"
   ...
   path: "/repositories/{{ stream_slice.repository }}/commits"
(Note: this example uses 
stream_slice
, but the forum thread shows that for the current framework you should use 
stream_partition
in the path.) [Substream router; Forum fix] So, if your 
partition_field
is 
healthModelId
, the path must use exactly that name:
Copy code
partition_field: "healthModelId"
   ...
   path: "health/{{ stream_partition.healthModelId }}/team-health-historical"
If the 
partition_field
is different (e.g. 
health_model_id
), then the template must match that:
Copy code
partition_field: "health_model_id"
   path: "health/{{ stream_partition.health_model_id }}/team-health-historical"
3. Where the value is injected When the partition value is part of the URL path, you typically do not configure a 
request_option
on the `ParentStreamConfig`; instead, you just reference `stream_partition.`partition_field in the 
path
. The docs show this pattern for nested resources: [Substream router]
Copy code
requester:
     path: "/repositories/{{ stream_slice.repository }}/commits"
   partition_router:
     type: SubstreamPartitionRouter
     parent_stream_configs:
       - stream: "#/repositories_stream"
         parent_key: "id"
         partition_field: "repository"
In your case, that translates to:
Copy code
requester:
     path: "health/{{ stream_partition.healthModelId }}/team-health-historical"
   partition_router:
     type: SubstreamPartitionRouter
     parent_stream_configs:
       - type: ParentStreamConfig
         stream: "#/definitions/<your_parent_stream>"
         parent_key: "<parent_field_with_id>"
         partition_field: "healthModelId"
4. Why you still see the raw template in debug The knowledge sources don’t explicitly describe the debug output format, but the forum example shows that when the wrong variable (
stream_slice
instead of 
stream_partition
) is used, Airbyte logs the URL with the literal template and the request goes out without substitution, causing errors. [Forum fix] Given that you see 
health/{{ stream_partition.healthModelId }}/team-health-historical
in debug, the most likely issues (based on the docs and forum thread) are: ◦ 
partition_field
name doesn’t match 
healthModelId
, or ◦ The 
SubstreamPartitionRouter
/ 
parent_stream_configs
block is not actually attached to this stream (e.g., wrong nesting or 
$parameters
placement), so 
stream_partition
is never populated. Unfortunately, the knowledge sources do not show a complete, version-specific manifest example with 
SubstreamPartitionRouter
and path injection beyond the generic patterns above, so I can’t verify the exact key names or nesting for your version. To move forward, check these against your manifest: • Under the child stream, ensure you have:
Copy code
$parameters:
      partition_router:
        - type: SubstreamPartitionRouter
          parent_stream_configs:
            - type: ParentStreamConfig
              stream: "#/definitions/<parent_stream_def>"
              parent_key: "<id_field_in_parent>"
              partition_field: "healthModelId"
and
Copy code
path: "health/{{ stream_partition.healthModelId }}/team-health-historical"
• Confirm that the parent stream definition referenced by 
stream
actually emits a field with the ID you expect in 
parent_key
. If you can share the relevant YAML snippet (parent stream + child stream), I can map it directly to the patterns from the docs and the working forum example. 👍 0 button 👎 0 button ℹ️ Tip: Mention @kapa.ai in the thread for follow-ups or DM the app for 1-to-1 help.
y
does the partition_router fall under parameters or is it on the same level as parameters?
@kapa.ai does the partition_router fall under parameters or is it on the same level as parameters?
k
It goes on the same level as 
$parameters
, not inside it. In the config-based YAML schema, 
partition_router
is defined directly under the retriever (or stream) object, alongside other top-level properties; 
$parameters
is a separate property used to pass inherited parameters. [Partition router schema; Substream router ref] So structurally it should look like:
Copy code
retriever:
  requester:
    path: "health/{{ stream_partition.healthModelId }}/team-health-historical"
  partition_router:
    type: SubstreamPartitionRouter
    parent_stream_configs:
      - stream: "#/definitions/parent_stream"
        parent_key: "id"
        partition_field: "healthModelId"
  $parameters:
    some_param: some_value
partition_router
and 
$parameters
are siblings in the manifest, not nested within one another. 👍 0 button 👎 0 button ℹ️ Tip: Mention @kapa.ai in the thread for follow-ups or DM the app for 1-to-1 help.
2 Views
Open in Slack
PreviousNext
Powered by