[API] Passing dynamic parameters to the jobs via REST API or via Airflow

[ankushsharma92]

Hey there team!

I have a use case for which I am evaluating whether Airbyte would be a good use fit. Within our company, we use Airflow as an orchestrator to schedule jobs. We have use cases where we ingest data from a REST api endpoint, and that data is dumped as it is(no transformations) into S3. The way it is dumped into S3 is that we pass the bucket_name, object_key and a date parameter, where this date parameter is the date when this ingestion is supposed to happen. Moreover, the endpoints have to be triggered daily via Airflow, wherein these endpoints take dynamic query parameters fromDate and toDate.

So to summarize:
Source : https://company-gateway.com/api/endpoint?fromDate=<from_date>&endDate=<end_date>

Destination :s3://bucket-name/object_key/<ingestion_date>/data.json (where data.json is the data returned by the API)

I was trying the lowcode UI builder, and cannot find a way to parameterize the source query parameters(like fromDate and endDate) as well as destination. Moreover, I looked at the public API and it seems there is no way to pass external parameters via the Rest API to trigger a job. Can somebody help me with this and if this can be accomplished somehow via Airbyte?

TL;DR: Is there a way to inject dynamic parameters such as query params from outside of Airbyte(such as via rest api or airflow) to a job? The injection needs to happen in such a way that the the job would replace the source and destination parameters dynamically at runtime.

[marcosmarxm]

You can use the /source or /connection update API endpoint to change the value as you need and after trigger the sync after.
Airbyte connectors doesn't have can't receive dynamic parameters, there are a few exceptions like S3 where you can create macro to dynamically render some parameters, but if you want to inject from somewhere else you need to update them manually.
AFAIK the AirbyteOperator in Airflow is very basic and only send a request to /sync endpoint. I know today you can manage all the Airflow DAG operations using the HTTPOperator + sensor so it's doable to inject this changes but not using the official operator.

[ankushsharma92]

I see! Thanks a lot for your reply, I appreciate it!

Just to be on safer side, based on what you said, my interpretation is :

1. Use the low level cdk to build a source - this way the source has a source_id
2. Have it published to the UI
3. Create an S3 destination with a valid bucket path - but the path here does not really matter initially
4. When the DAG runs, first perform an update against the source (and completely overwrite the URL), then perform an update against the dest (fully updating the bucket path and passing the <ingestion_date> via this update rest api call)
5. Now the source and dest object are configured for my DAG run
6. Do the job sync (via the available operator)

And then each DAG run would overwrite always the source and destination. Can you please let me know if my interpretation is correct?

Best
Ankush