from datetime import datetime, timedelta from textwrap import dedent import pendulum from airflow.operators.bash import BashOperator from airflow import DAG from data.gasan_data import gasanData # import cj-gasan data # set timezone local_tz = pendulum.timezone("Asia/Seoul") with DAG( 'batch_test', default_args={ 'depends_on_past': False, 'email': 'kim-jy@lotte.net', }, description='Test Batch Job', schedule_interval='*/1 * * * *', start_date=datetime(2022, 5, 13, tzinfo=local_tz), tags=['test'], catchup=False, ) as dag: # gasan 작업 병렬처리 post_gasan_tasks = [] for i, data in enumerate(gasanData): post_gasan_task = BashOperator( task_id='post_gasan'+str(i+1), bash_command="curl -X \'POST\' \'http://10.231.238.224:30999/api/v1/camera/writeimage\' -H \'Content-Type: application/json\' -d \'{\"id\": \"test\", \"pw\": \"test\", \"ip\": \"%s\", \"serialNum\": \"%s\", \"camName\": \"%s\"}\'" %(data["ip"], data["serialNum"], data["camName"]), ) post_gasan_tasks.append(post_gasan_task) # 작업 순서 정의 post_gasan_tasks