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: # (Task1) 헬스체크 health_check = BashOperator( task_id='health_check', bash_command="curl -X GET -v http://10.231.238.224:30999/api/v1/core/health \'-H accept: application/json\'" ) # (Task2) 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 health_check