Airflow loop through pagination for URL

user_93684 0
Viewed 429

I have created a DAG which to get JSON data from an API. After getting the data from the API it uploads the data into S3. The data from the API is paginated and I can only get 100 records at a time. Depending on the data created I might have to loop a couple of times getting the data. I only know how many pages to loop after picking the first data from API.

So my tasks are

initialize_env = DummyOperator(task_id='initialize_env')

get_data = SimpleHttpOperator(task_id='get_data',
                              url=url,
                              endpoint='users?page={page_id}') #this task loops number of times
                              
upload_to_s3 = MyOperator(task_id='upload_to_s3') #this task loops number of times

read_data = PythonOperator(task_id='upload_to_s3',
                            python_callable=function_reading_data) #this task should stop the loop
                            
send_email = DummyOperator(task_id='send_email') #send email at the end of looping

I would like to find out how I can do looping in Airflow so that I go through all the pages?

airflow python
1 Answers

Looping tasks isn't a good practice in Airflow (if it's even possible).

I'd recommend having a Python operator that does the looping for you. Inside that python operator create your loop to hit the API and upload to S3. Alternatively, you could loop through and create a list of pages in get_data, pass that list through XCOM to upload, then pass that same list on again to read the data.

Depends how much you want to put in python and how many tasks you want to use.

Theoretically you could put everything in 1 python operator, but that's not great practice either.

user_182997500
Related