try:

    from datetime import timedelta
    from airflow import DAG
    
    from airflow.operators.python_operator import PythonOperator
    from airflow.operators.http_operator import SimpleHttpOperator
    from datetime import datetime
    from pandas.io.json import json_normalize
    from airflow.operators.postgres_operator import PostgresOperator
    from pandas.tseries.offsets import CustomBusinessDay

    import pandas as pd
    import json
    import requests
    import numpy as np

    import psycopg2
    from sqlalchemy import create_engine
    #print("All Dag moudules are sucessfully imported")

except Exception as e:
    print("Error {} ".format(e))

dRoW_api_end_url = "https://uat2.drow.cloud"

def getDrowToken(**context):

    response = requests.post(
    url=f"{dRoW_api_end_url}/api/auth/authenticate",
    data={
    "username": "icwp2@drow.cloud",
    "password": "dGVzdDAxQHRlc3QuY29t"
    }
    ).json()
    context["ti"].xcom_push(key="token", value=response['token'])
    # return 'DLLM{}'.format(response)


def getdrowPSQLConnectionString():
    host                  = 'drowdatewarehouse.crlwwhgepgi7.ap-east-1.rds.amazonaws.com'  

    # User name of the database server
    dbUserName            = 'dRowAdmin'  

    # Password for the database user
    dbUserPassword        = 'drowsuper'  

    # Name of the database 
    database              = 'drowDateWareHouse'

    # Character set
    charSet               = "utf8mb4"  

    port                  = "5432"

    conn_string = ('postgres://' +
                           dbUserName + ':' + 
                           dbUserPassword +
                           '@' + host + ':' + port +
                           '/' + database)
    print(conn_string)
    return conn_string

def getMongoDB(**context):
    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/module/document-export/airflow/workflow/6052af4637fb7a08f96ecd0b?export_type=0",
    headers={
    "x-access-token": f"Bearer {token}",
    "ICWPxAccessKey": "5WSD21ICWP_[1AG:4UdI){n=b~"
    }
    )
    print('got_data')
    RISC_Data = json.loads(response.text)
    # Request Submission A03 - Proposed Inspection Date Time minus 24hrs.
    Mapping= {
    "B01 - Request Received Date Time" : "b01_request_received_date_time",
    'C02 - Inspect on Date Time' : 'c02_inspect_or_survey_on_date_time',
    "A03 - Proposed Inspection Date Time" : "a05_proposed_inspection_or_survey_date_time",
    "E01 - Received on behalf of Contractor on Date Time" : "e01_received_on_behalf_of_contractor_on_date_time",
    'A01a - Request No. Revision': "a01a_request_no_revision",
    "A01 - Request No.": "a01_request_no",
    "C08 Time Pass to IOW" : 'c12_time_pass_to_senior_or_contractor',
    "D01 - Countersigned on Date Time" : 'd01_countersigned_on_date_time',
    'C03 - Approval given?' : "c03_approval_given",
    "Work Category" : "a01b_work_category",
    "A04b6 - Location in TWD": "a03_location_of_work",
    "A10 - Request Date Time": "a10_request_submission_date_time_",
    "C10 - Pass on Date Time": "c10_pass_on_date_time"
    # 'C12 Time Pass to Senior or Contractor' : 'Sign time by manager'
    #     "B01 - Request Received Date Time" : "b01_request_received_date_time",
    # 'C02 - Inspect on Date Time' : 'c02_inspect_or_survey_on_date_time',
    # "A03 - Proposed Inspection Date Time" : "a05_proposed_inspection_or_survey_date_time",
    # "E01 - Received on behalf of Contractor on Date Time" : "e01_received_on_behalf_of_contractor_on_date_time",
    # 'A01a - Request No. Revision': "a01a_request_no_revision",
    # "A01 - Request No.": "a01_request_no",
    # "C08 Time Pass to IOW" : 'c12_time_pass_to_senior_or_contractor',
    # "D01 - Countersigned on Date Time" : 'd01_countersigned_on_date_time',
    # 'C03 - Approval given?' : "c03_approval_given",
    # "Work Category" : "a01b_work_category",
    # "A04b6 - Location in TWD": "a03_location_of_work",
    # "A10 - Request Date Time": "a10_request_submission_date_time_",
    # "C10 - Pass on Date Time": "c10_pass_on_date_time"
    }
    
    conn_string = getdrowPSQLConnectionString()
    db = create_engine(conn_string)
    conn = db.connect()
    print('db connected')
    with conn as conn:
        df = pd.DataFrame()
        for x in RISC_Data:
            create_date = datetime.fromisoformat(x['Create Date'].rstrip('Z'))
            # Check if the 'Create Date' is later than the threshold
            if create_date < datetime(2023, 1, 1):
                continue
            df_nested_list = json_normalize(x['data'])
            df2 = df_nested_list.reindex(columns=Mapping.keys())
            df2["request_no"] = df2["A01 - Request No."].astype(str) + df2["A01a - Request No. Revision"]
            df2["alphabetic_part"] = df2["A01a - Request No. Revision"].str.extract('([A-Za-z]+)', expand=False)

            if(df2["request_no"].isnull().any()):
                continue
            if df2['Work Category'].isnull().any() or (df2['Work Category'] == "").any():
                df2['Work Category']='Other'
            if len(x['ApproveLogSummary']) > 0:
                # request_date = df2["A03 - Proposed Inspection Date Time"]
                # request_data = [data for data in x['ApproveLogSummary'] if data.get('statusName')=="B1 : SRE/RE Assign"]
                # if len(request_data) > 0:
                if not df2['A10 - Request Date Time'].isnull().any():
                    _request_date = df2['A10 - Request Date Time'].iloc[0]
                else:
                    _request_date = None
            else:
                _request_date = None
            df2["a10_request_submission_date_time"] = _request_date
            # if not df2["a10_request_submission_date_time"].isnull().bool():
            #     df2["a10_request_submission_date_time"]=df2["a10_request_submission_date_time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            # if (not df2["A03 - Proposed Inspection Date Time"].isnull().bool()):
            #     df2["A03 - Proposed Inspection Date Time"] = df2["A03 - Proposed Inspection Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            # if (not df2["C02 - Inspect on Date Time"].isnull().bool()):
            #     df2["C02 - Inspect on Date Time"] = df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["C08 Time Pass to IOW"].isnull().bool()):
                df2["C08 Time Pass to IOW"] = df2["C08 Time Pass to IOW"].astype('datetime64[ns]') + pd.Timedelta(8, unit='h')

            # if(not df2["C02 - Inspect on Date Time"].isnull().bool() and not df2["a10_request_submission_date_time"].isnull().bool() and ((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool())):
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))
            # else:
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))

            if (not df2["C02 - Inspect on Date Time"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool()):
                df2['nc_report'] = True
            else:
                df2['nc_report'] = False

            if (not df2["A03 - Proposed Inspection Date Time"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (((df2["A03 - Proposed Inspection Date Time"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) < pd.Timedelta(24, unit='h')).bool() and (((df2["A03 - Proposed Inspection Date Time"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) > pd.Timedelta(0, unit='h')).bool()):
                df2['urgent_report'] = True
            else:
                df2['urgent_report'] = False

            if (not df2['E01 - Received on behalf of Contractor on Date Time'].isnull().bool() and not df2['a10_request_submission_date_time'].isnull().bool()):
                # Define custom business days, excluding weekends (Saturday and Sunday)
                custom_business_day = CustomBusinessDay(weekmask='Mon Tue Wed Thu Fri')
                if _request_date!=None and x['data']['E01 - Received on behalf of Contractor on Date Time']!=None:
                    start_date = np.datetime64(_request_date, 'ns').astype('datetime64[D]')
                    end_date = np.datetime64(x['data']['E01 - Received on behalf of Contractor on Date Time'], 'ns').astype('datetime64[D]')
                    # Calculate the elapsed time in business days
                    df2['elapsed_time'] = (end_date - start_date)/np.timedelta64(1,"D")
                    # df2['elapsed_time'] = np.busday_count(start_date, end_date, weekmask=custom_business_day.weekmask)
                    # df2['elapsed_time'] = np.busday_count(_request_date,
                    #                       x['E01 - Received on behalf of Contractor on Date Time'],
                    #                       weekmask=custom_business_day.weekmask)
                    # Round the elapsed time to two decimal places
                    df2['elapsed_time'] = df2['elapsed_time'].round(2)
                else:
                    df2['elapsed_time'] = 0
            else:
                df2['elapsed_time'] = 0
            
            if (not df2["C02 - Inspect on Date Time"].isnull().bool() and not df2["C08 Time Pass to IOW"].isnull().bool() and (((df2["C08 Time Pass to IOW"].astype('datetime64[ns]') - df2["C02 - Inspect on Date Time"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['overdue_report'] = True
            else:
                df2['overdue_report'] = False
            if (not df2["C10 - Pass on Date Time"].isnull().bool() and not df2["C08 Time Pass to IOW"].isnull().bool() and (((df2["C10 - Pass on Date Time"].astype('datetime64[ns]') - df2["C08 Time Pass to IOW"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['delayed_approval_report'] = True
            else:
                df2['delayed_approval_report'] = False

            if (((df2['A01a - Request No. Revision']=="").bool() or df2['A01a - Request No. Revision'].isnull().bool()) and (df2['C03 - Approval given?']=="No").bool()):
                df2['fail_in_first_inspection'] = True
            else:
                df2['fail_in_first_inspection'] = False
            
            if (not df2["C08 Time Pass to IOW"].isnull().bool() and not df2["C02 - Inspect on Date Time"].isnull().bool() and not df2["E01 - Received on behalf of Contractor on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'complete'
            elif ((df2["C08 Time Pass to IOW"].isnull().bool() or df2["E01 - Received on behalf of Contractor on Date Time"].isnull().bool()) and not df2["C02 - Inspect on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'in-complete'
            elif ((not df2["C08 Time Pass to IOW"].isnull().bool() or df2["E01 - Received on behalf of Contractor on Date Time"].isnull().bool()) and df2["C02 - Inspect on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'outstanding'
            else:
                df2['complete_incomplete_outstanding_report'] = 'outstanding'            

            #print('process 2')
            df2.rename(columns=Mapping, inplace=True)
            df = df.append(df2)
        # Sort dataframe by Request No. and Rev No. in descending order
        df.sort_values(['a01_request_no', 'a01a_request_no_revision'], ascending=[True, False], inplace=True)

        # Create a new column 'is_latest' and set the initial value to False
        df['is_latest'] = False

        # Assign a unique identifier to None values
        none_identifier = np.nan

        # Replace None values with the unique identifier
        df['a01_request_no'].fillna(none_identifier, inplace=True)

        # Group by Request No. and update the 'is_latest' column for the latest version
        df['is_latest'] = df.groupby('a01_request_no')['a01a_request_no_revision'].transform(lambda x: x.fillna('') == x.fillna('').max())

        # Reset the unique identifier to None
        df['a01_request_no'].replace(none_identifier, None, inplace=True)
        df.to_sql('risc_dc201911', con=conn, if_exists='append', index= False)
        conn.close()
        
def getMongoDB2(**context):
    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/module/document-export/airflow/workflow/5fc2211f84354175cbf5fd35?export_type=0",
    headers={
    "x-access-token": f"Bearer {token}",
    "ICWPxAccessKey": "5WSD21ICWP_[1AG:4UdI){n=b~"
    }
    )
    print('got_data')
    RISC_Data = json.loads(response.text)
    # Request Submission A03 - Proposed Inspection Date Time minus 24hrs.
    Mapping= {
    "B01 - Request Received Date Time" : "b01_request_received_date_time",
    'C02 - Check Date Time' : 'c02_inspect_or_survey_on_date_time',
    "A04 - Proposed Checking Date Time" : "a05_proposed_inspection_or_survey_date_time",
    "E01 - Received on behalf of Contractor on Date Time" : "e01_received_on_behalf_of_contractor_on_date_time",
    'A01b - Revision': "a01a_request_no_revision",
    "A01a - Request No.": "a01_request_no",
    "C08 - Time Pass to SSOE" : 'c12_time_pass_to_senior_or_contractor',
    "D01 - Countersigned on Date Time" : 'd01_countersigned_on_date_time',
    'C03 - Approval given?' : "c03_approval_given",
    # "Work Category" : "a01b_work_category",
    "A02b2 - Location in KCC": "a03_location_of_work",
    "A11 - Submission Date Time": "a10_request_submission_date_time_",
    "Date Time Pass to Next Status": "c10_pass_on_date_time"
    # 'C12 Time Pass to Senior or Contractor' : 'Sign time by manager'
    }
    
    conn_string = getdrowPSQLConnectionString()
    db = create_engine(conn_string)
    conn = db.connect()
    print('db connected')
    with conn as conn:
        df = pd.DataFrame()
        for x in RISC_Data:
            create_date = datetime.fromisoformat(x['Create Date'].rstrip('Z'))
            # Check if the 'Create Date' is later than the threshold
            if create_date < datetime(2023, 1, 1):
                continue
            #print(x)
            df_nested_list = json_normalize(x['data'])
            df2 = df_nested_list.reindex(columns=Mapping.keys())
            df2["request_no"] = df2["A01a - Request No."].astype(str) + df2["A01b - Revision"]
            if(df2["request_no"].isnull().any()):
                continue
            # if(df2['Work Category'].isnull().bool()):
            df2['a01b_work_category']='Survey'
            if len(x['ApproveLogSummary']) > 0:
                # request_date = df2['A04 - Proposed Checking Date Time']
                # request_data = [data for data in x['ApproveLogSummary'] if data.get('statusName')=="B1 : SRE/RE Assign"]
                if not df2['A11 - Submission Date Time'].isnull().any():
                    request_date = df2['A11 - Submission Date Time'].iloc[0]
                else:
                    request_date = None
            else:
                request_date = None
            df2["a10_request_submission_date_time"] = request_date
            # if not df2["a10_request_submission_date_time"].isnull().bool():
            #     df2["a10_request_submission_date_time"]=df2["a10_request_submission_date_time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            # # df2["a10_request_submission_date_time"] = _request_date
            # if (not df2["C02 - Check Date Time"].isnull().bool()):
            #     df2["C02 - Check Date Time"] = df2["C02 - Check Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            # if (not df2["A04 - Proposed Checking Date Time"].isnull().bool()):
            #     df2["A04 - Proposed Checking Date Time"] = df2["A04 - Proposed Checking Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["C08 - Time Pass to SSOE"].isnull().bool()):
                df2["C08 - Time Pass to SSOE"] = df2["C08 - Time Pass to SSOE"].astype('datetime64[ns]') + pd.Timedelta(8, unit='h')

            
            # if(not df2["C02 - Inspect on Date Time"].isnull().bool() and not df2["a10_request_submission_date_time"].isnull().bool() and ((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool())):
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))
            # else:
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))

            if (not df2["C02 - Check Date Time"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (df2["C02 - Check Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool()):
                df2['nc_report'] = True
            else:
                df2['nc_report'] = False

            if (not df2["A04 - Proposed Checking Date Time"].isnull().bool() and not df2["a10_request_submission_date_time"].isnull().bool() and (((df2["A04 - Proposed Checking Date Time"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) < pd.Timedelta(24, unit='h')).bool() and (((df2["A04 - Proposed Checking Date Time"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) > pd.Timedelta(0, unit='h')).bool()):
                df2['urgent_report'] = True
            else:
                df2['urgent_report'] = False

            if (not df2['E01 - Received on behalf of Contractor on Date Time'].isnull().bool() and not df2['a10_request_submission_date_time'].isnull().bool()):
                custom_business_day = CustomBusinessDay(weekmask='Mon Tue Wed Thu Fri')
                if request_date!=None and x['data']['E01 - Received on behalf of Contractor on Date Time']!=None:
                    start_date = np.datetime64(request_date, 'ns').astype('datetime64[D]')
                    end_date = np.datetime64(x['data']['E01 - Received on behalf of Contractor on Date Time'], 'ns').astype('datetime64[D]')
                    # Calculate the elapsed time in business days
                    df2['elapsed_time'] = (end_date - start_date)/np.timedelta64(1,"D")
                    # df2['elapsed_time'] = np.busday_count(start_date, end_date, weekmask=custom_business_day.weekmask)
                    # df2['elapsed_time'] = np.busday_count(_request_date,
                    #                       x['E01 - Received on behalf of Contractor on Date Time'],
                    #                       weekmask=custom_business_day.weekmask)
                    # Round the elapsed time to two decimal places
                    df2['elapsed_time'] = df2['elapsed_time'].round(2)
            else:
                df2['elapsed_time'] = 0
            
            if (not df2["C02 - Check Date Time"].isnull().bool() and not df2["C08 - Time Pass to SSOE"].isnull().bool() and (((df2["C08 - Time Pass to SSOE"].astype('datetime64[ns]') - df2["C02 - Check Date Time"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['overdue_report'] = True
            else:
                df2['overdue_report'] = False
            print(x['data']['A01a - Request No.'])
            if (not df2["Date Time Pass to Next Status"].isnull().bool() and not df2["C08 - Time Pass to SSOE"].isnull().bool() and (((df2["Date Time Pass to Next Status"].astype('datetime64[ns]') - df2["C08 - Time Pass to SSOE"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['delayed_approval_report'] = True
            else:
                df2['delayed_approval_report'] = False

            if (((df2['A01b - Revision']=="").bool() or df2['A01b - Revision'].isnull().bool()) and (df2['C03 - Approval given?']=="No").bool()):
                df2['fail_in_first_inspection'] = True
            else:
                df2['fail_in_first_inspection'] = False
            
            if (not df2["C08 - Time Pass to SSOE"].isnull().bool() and not df2["C02 - Check Date Time"].isnull().bool() and not df2["E01 - Received on behalf of Contractor on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'complete'
            elif ((df2["C08 - Time Pass to SSOE"].isnull().bool() or df2["E01 - Received on behalf of Contractor on Date Time"].isnull().bool()) and not df2["C02 - Check Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'in-complete'
            elif ((not df2["C08 - Time Pass to SSOE"].isnull().bool() or df2["E01 - Received on behalf of Contractor on Date Time"].isnull().bool()) and df2["C02 - Check Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'outstanding'
            else:
                df2['complete_incomplete_outstanding_report'] = 'outstanding'            

            #print('process 2')
            df2.rename(columns=Mapping, inplace=True)
            df = df.append(df2)
        # Sort dataframe by Request No. and Rev No. in descending order
        df.sort_values(['a01_request_no', 'a01a_request_no_revision'], ascending=[True, False], inplace=True)

        # Create a new column 'is_latest' and set the initial value to False
        df['is_latest'] = False

        # Assign a unique identifier to None values
        none_identifier = np.nan

        # Replace None values with the unique identifier
        df['a01_request_no'].fillna(none_identifier, inplace=True)

        # Group by Request No. and update the 'is_latest' column for the latest version
        df['is_latest'] = df.groupby('a01_request_no')['a01a_request_no_revision'].transform(lambda x: x.fillna('') == x.fillna('').max())

        # Reset the unique identifier to None
        df['a01_request_no'].replace(none_identifier, None, inplace=True)
        df.to_sql('risc_dc201911', con=conn, if_exists='append', index= False)
        conn.close()


delete_table_sql_query = """
DROP TABLE IF EXISTS risc_dc201911
"""

# id SERIAL
create_table_sql_query = """ 
    CREATE TABLE IF NOT EXISTS risc_dc201911 (
    a10_request_submission_date_time TIMESTAMP, 
    b01_request_received_date_time TIMESTAMP,
    c02_inspect_or_survey_on_date_time TIMESTAMP,
    a05_proposed_inspection_or_survey_date_time TIMESTAMP,
    e01_received_on_behalf_of_contractor_on_date_time TIMESTAMP,
    a01a_request_no_revision VARCHAR (150),
    a01_request_no VARCHAR (150),
    c12_time_pass_to_senior_or_contractor TIMESTAMP,
    d01_countersigned_on_date_time TIMESTAMP,
    c10_pass_on_date_time TIMESTAMP,
    c03_approval_given VARCHAR (150),
    request_no VARCHAR (150),
    alphabetic_part VARCHAR (150),
    a01b_work_category VARCHAR (150),
    nc_report BOOLEAN,
    urgent_report BOOLEAN,
    elapsed_time NUMERIC(10,2),
    overdue_report BOOLEAN,
    delayed_approval_report BOOLEAN,
    fail_in_first_inspection BOOLEAN,
    complete_incomplete_outstanding_report VARCHAR(250),
    a03_location_of_work VARCHAR(250),
    is_latest BOOLEAN,
    a10_request_submission_date_time_ TIMESTAMP
    );
    """

# */2 * * * * Execute every two minute 
with DAG(
        dag_id="dc201911_risc_icwp_inspection",
        schedule_interval="0 15 * * *",
        default_args={
            "owner": "airflow",
            "retries": 1,
            "retry_delay": timedelta(minutes=5),
            "start_date": datetime(2023, 1, 17)
        },
        catchup=False) as f:
    
    getMongoDB = PythonOperator(
        task_id="getMongoDB",
        python_callable=getMongoDB,
        op_kwargs={"name": "Dylan"},
        provide_context=True,
    )
    
    getMongoDB2 = PythonOperator(
        task_id="getMongoDB2",
        python_callable=getMongoDB2,
        op_kwargs={"name": "Dylan"},
        provide_context=True,
    )

    # reformData = PythonOperator(
    #     task_id="reformData",
    #     python_callable=reformData,
    #     provide_context=True,
    #     # op_kwargs={"name": "Dylan"}
    # )

    getDrowToken = PythonOperator(
        task_id="getDrowToken",
        python_callable=getDrowToken,
        provide_context=True,
        # op_kwargs={"name": "Dylan"}
    )

    # insertData = PythonOperator(
    #     task_id="insetDateToPG",
    #     python_callable=insertData,
    #     provide_context=True,
    #     # op_kwargs={"name": "Dylan"}
    # )

    create_table = PostgresOperator(
        sql = create_table_sql_query,
        task_id = "create_table_task",
        postgres_conn_id = "postgres_rds",
    )

    delete_table = PostgresOperator(
        sql = delete_table_sql_query,
        task_id = "delete_table_task",
        postgres_conn_id = "postgres_rds",
    )

    # insert_data = PostgresOperator(
    #     sql = insert_data_sql_query,
    #     task_id = "insertData_sql_query_task",
    #     postgres_conn_id = "postgres_rds",
    # )

# getDrowToken >> getMongoDB >> getMongoDB2
delete_table >> create_table >> getDrowToken >> getMongoDB >> getMongoDB2