Airflow - DAGs

Dependency	Reason
Dagrun Running	Task instance's dagrun was not in the 'running' state but in the state 'success'.
Task Instance State	Task is in the 'success' state which is not a valid state for execution. The task must be cleared in order to be run.

Attribute: python_callable

def getMongoDB(**context):
    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/sheets/6396d09292869a0c9a0079dc?with_records=true&fields=",
    headers={
    "x-access-token": f"Bearer {token}",
    }
    )
    # # print('got_data')
    sheet = json.loads(response.text)
    headers = sheet['header']
    record = sheet['record']
    dataToExtract=[]
    for d in record: 
        objectToPush = {}
        for v in d['values']:
            for c in headers:
                colNameToExtract = c['colName']
                if v['colName'] == colNameToExtract:
                    # # print(v)
                    if v.get('multValue') != None:
                        if v['multValue'] == True:
                            if v['colType'] == 'Table':
                                tObjectArray = []
                                for t in v['tableValue']:
                                    tObjectToPush = {}
                                    for s in t['subValues']:
                                        tObjectToPush[s['colName']] = s.value
                                    tObjectArray.push(tObjectToPush)
                            else:
                                objectToPush[v['colName']] = v['valueArray']
                        else:
                            if v.get('value') != None:
                                if v.get('value') == 'NA':
                                    objectToPush[v['colName']] = None
                                else:
                                    objectToPush[v['colName']] = v['value']
                            else:
                                objectToPush[v['colName']] = None
                    else:
                        if v.get('value') != None:
                            if v.get('value') == 'NA':
                                objectToPush[v['colName']] = None
                            else:
                                objectToPush[v['colName']] = v['value']
                        else:
                            objectToPush[v['colName']] = None
        dataToExtract.append(objectToPush)
    # # print(len(dataToExtract))
    # # print(dataToExtract[0].keys())
    df = pd.DataFrame.from_dict(dataToExtract)
    numerics = df.select_dtypes(include="number").columns
    df=df.apply(pd.to_numeric, errors='ignore')
    df[numerics]=df[numerics].apply(lambda x: np.round(x, decimals=5))
    df['IP No.']=df['IP No.'].astype(str)
    df['Month - Year']=df['Month - Year'].apply(pd.to_datetime)
    df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')

    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"

    # #cursor Type
    # cusrsorType            = pymysql.cursors.DictCursor



    #create_engine('mysql+mysqldb://root:password@localhost:3306/mydbname', echo = False)
    conn_string = ('postgres://' +
                           dbUserName + ':' + 
                           dbUserPassword +
                           '@' + host + ':' + port +
                           '/' + database)

    # df = context.get("ti").xcom_pull(key="InsertData")
    # # print(df)
    # conn_string = 'postgres://user:password@host/data1'
    
    db = create_engine(conn_string)
    conn = db.connect()
    # # print('db connected')
    with conn as conn:
        df.to_sql('c04_finance_data', con=conn, if_exists='replace')
    conn.close()

    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/sheets/63fc9b864243400ca9af78bb?with_records=true&fields=",
    headers={
    "x-access-token": f"Bearer {token}",
    }
    )
    # # print('got_data')
    sheet = json.loads(response.text)
    headers = sheet['header']
    record = sheet['record']
    dataToExtract=[]
    for d in record: 
        objectToPush = {}
        for v in d['values']:
            for c in headers:
                colNameToExtract = c['colName']
                if v['colName'] == colNameToExtract:
                    # # print(v)
                    if v.get('multValue') != None:
                        if v['multValue'] == True:
                            if v['colType'] == 'Table':
                                tObjectArray = []
                                for t in v['tableValue']:
                                    tObjectToPush = {}
                                    for s in t['subValues']:
                                        tObjectToPush[s['colName']] = s.value
                                    tObjectArray.push(tObjectToPush)
                            else:
                                objectToPush[v['colName']] = v['valueArray']
                        else:
                            if v.get('value') != None:
                                if v.get('value') == 'NA':
                                    objectToPush[v['colName']] = None
                                else:
                                    objectToPush[v['colName']] = v['value']
                            else:
                                objectToPush[v['colName']] = None
                    else:
                        if v.get('value') != None:
                            if v.get('value') == 'NA':
                                objectToPush[v['colName']] = None
                            else:
                                objectToPush[v['colName']] = v['value']
                        else:
                            objectToPush[v['colName']] = None
        dataToExtract.append(objectToPush)
    # # print(len(dataToExtract))
    # # print(dataToExtract[0].keys())
    data= dataToExtract
    # data=[{
    #     "Month - Year": "2022-11",
    #     "Type": "Forecast of the final PWDD", 
    #     "Provisional Sum for Safety": 0,
    #     "CEs incl. Fee": 188000000,
    #     "Forecast of the final total of the Prices":2322000000,
    #     "Tendered total of the Prices":1690000000,
    #     "Price Adjustment for Inflation (PAI)":443000000,
    #     "Original Contract Value":1747000000,
    #     "Total": 2378000000
    # },
    # {
    #     "Month - Year": "2022-11",
    #     "Type": "Approved forecast total of the Prices", 
    #     "Provisional Sum for Safety":16000000,
    #     "CEs incl. Fee":177000000,
    #     "Forecast of the final total of the Prices":2322000000,
    #     "Tendered total of the Prices":1690000000,
    #     "Price Adjustment for Inflation (PAI)":265000000,
    #     "Original Contract Value":1690000000,
    #     "Total": 2148000000
    # }
    # ]
    df = pd.DataFrame.from_dict(data)
    numerics = df.select_dtypes(include="number").columns
    df=df.apply(pd.to_numeric, errors='ignore')
    df[numerics]=df[numerics].apply(lambda x: np.round(x, decimals=5))
    # df['IP No.']=df['IP No.'].astype(str)
    df['Month - Year']=df['Month - Year'].apply(pd.to_datetime)
    # df["Provisional Sum for Safety"] = df["Provisional Sum for Safety"].apply(pd.to_numeric)
    # df["CEs incl. Fee"] = df["CEs incl. Fee"].apply(pd.to_numeric)
    # df["Forecast of the final total of the Prices"] = df["Forecast of the final total of the Prices"].apply(pd.to_numeric)
    # df["Tendered total of the Prices"] = df["Tendered total of the Prices"].apply(pd.to_numeric)
    # df["Price Adjustment for Inflation (PAI)"] = df["Price Adjustment for Inflation (PAI)"].apply(pd.to_numeric)
    # df["Original Contract Value"] = df["Original Contract Value"].apply(pd.to_numeric)
    # df["Total"] = df["Total"].apply(pd.to_numeric)
    df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')
    db = create_engine(conn_string)
    conn = db.connect()
    with conn as conn:
        df.to_sql('c04_finance_status_data', con=conn, if_exists='replace')
    conn.close()

    # data=[{
    #     "Month - Year": "2022-11",
    #     "Contract": "C4",
    #     "No. of Claims": 13,
    #     "Claimed (Days)": 237.0,
    #     "Assessed (Days)": 185.0,
    #     "Awarded (Days)": 178.0,
    #     "Awarded/Assessed": 0.9622,
    #     "Unresolved (Days)": 52,
    #     "Assessed w/o EOT": 7
    # }
    # ]
    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/sheets/63fc9ef84243400ca9af7c70?with_records=true&fields=",
    headers={
    "x-access-token": f"Bearer {token}",
    }
    )
    # # print('got_data')
    sheet = json.loads(response.text)
    headers = sheet['header']
    record = sheet['record']
    dataToExtract=[]
    for d in record: 
        objectToPush = {}
        for v in d['values']:
            for c in headers:
                colNameToExtract = c['colName']
                if v['colName'] == colNameToExtract:
                    # # print(v)
                    if v.get('multValue') != None:
                        if v['multValue'] == True:
                            if v['colType'] == 'Table':
                                tObjectArray = []
                                for t in v['tableValue']:
                                    tObjectToPush = {}
                                    for s in t['subValues']:
                                        tObjectToPush[s['colName']] = s.value
                                    tObjectArray.push(tObjectToPush)
                            else:
                                objectToPush[v['colName']] = v['valueArray']
                        else:
                            if v.get('value') != None:
                                if v.get('value') == 'NA':
                                    objectToPush[v['colName']] = None
                                else:
                                    objectToPush[v['colName']] = v['value']
                            else:
                                objectToPush[v['colName']] = None
                    else:
                        if v.get('value') != None:
                            if v.get('value') == 'NA':
                                objectToPush[v['colName']] = None
                            else:
                                objectToPush[v['colName']] = v['value']
                        else:
                            objectToPush[v['colName']] = None
        dataToExtract.append(objectToPush)
    # # print(len(dataToExtract))
    # # print(dataToExtract[0].keys())
    data= dataToExtract
    
    df = pd.DataFrame.from_dict(data)
    numerics = df.select_dtypes(include="number").columns
    df=df.apply(pd.to_numeric, errors='ignore')
    df[numerics]=df[numerics].apply(lambda x: np.round(x, decimals=5))
    # df['IP No.']=df['IP No.'].astype(str)
    df['Month - Year']=df['Month - Year'].apply(pd.to_datetime)
    # df["Provisional Sum for Safety"] = df["Provisional Sum for Safety"].apply(pd.to_numeric)
    # df["CEs incl. Fee"] = df["CEs incl. Fee"].apply(pd.to_numeric)
    # df["Forecast of the final total of the Prices"] = df["Forecast of the final total of the Prices"].apply(pd.to_numeric)
    # df["Tendered total of the Prices"] = df["Tendered total of the Prices"].apply(pd.to_numeric)
    # df["Price Adjustment for Inflation (PAI)"] = df["Price Adjustment for Inflation (PAI)"].apply(pd.to_numeric)
    # df["Original Contract Value"] = df["Original Contract Value"].apply(pd.to_numeric)
    # df["Total"] = df["Total"].apply(pd.to_numeric)
    df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')
    db = create_engine(conn_string)
    conn = db.connect()
    with conn as conn:
        df.to_sql('c04_eot_data', con=conn, if_exists='replace')
    conn.close()

    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/sheets/63fc9fcc4243400ca9af7dee?with_records=true&fields=",
    headers={
    "x-access-token": f"Bearer {token}",
    }
    )
    # # print('got_data')
    sheet = json.loads(response.text)
    headers = sheet['header']
    record = sheet['record']
    dataToExtract=[]
    for d in record: 
        objectToPush = {}
        for v in d['values']:
            for c in headers:
                colNameToExtract = c['colName']
                if v['colName'] == colNameToExtract:
                    # # print(v)
                    if v.get('multValue') != None:
                        if v['multValue'] == True:
                            if v['colType'] == 'Table':
                                tObjectArray = []
                                for t in v['tableValue']:
                                    tObjectToPush = {}
                                    for s in t['subValues']:
                                        tObjectToPush[s['colName']] = s.value
                                    tObjectArray.push(tObjectToPush)
                            else:
                                objectToPush[v['colName']] = v['valueArray']
                        else:
                            if v.get('value') != None:
                                if v.get('value') == 'NA':
                                    objectToPush[v['colName']] = None
                                else:
                                    objectToPush[v['colName']] = v['value']
                            else:
                                objectToPush[v['colName']] = None
                    else:
                        if v.get('value') != None:
                            if v.get('value') == 'NA':
                                objectToPush[v['colName']] = None
                            else:
                                objectToPush[v['colName']] = v['value']
                        else:
                            objectToPush[v['colName']] = None
        dataToExtract.append(objectToPush)
    # # print(len(dataToExtract))
    # # print(dataToExtract[0].keys())
    data= dataToExtract
    # data=[
    #     {
    #     "Programme":"1st Programme",
    #     "Submission Date": "2020-11-11",
    #     "Acceptance Date": "2020-12-4",
    #     "Programme Approval Elapsed Time (days)": 23,
    #     },
    #     {
    #     "Programme":"RP01",
    #     "Submission Date": "2021-09-20",
    #     "Acceptance Date": "2021-09-30",
    #     "Programme Approval Elapsed Time (days)": 10,
    #     },
    #     {
    #     "Programme":"RP02a",
    #     "Submission Date": "2022-01-24",
    #     "Acceptance Date": "2022-02-22",
    #     "Programme Approval Elapsed Time (days)": 29,
    #     },
    #     {
    #     "Programme":"RP03",
    #     "Submission Date": "2022-01-24",
    #     "Acceptance Date": "2022-05-12",
    #     "Programme Approval Elapsed Time (days)": 108,
    #     },
    #     {
    #     "Programme":"RP04",
    #     "Submission Date": "2022-08-22",
    #     "Acceptance Date": "2022-09-19",
    #     "Programme Approval Elapsed Time (days)": 28,
    # }
    # ]
    df = pd.DataFrame.from_dict(data)
    # numerics = df.select_dtypes(include="number").columns
    # df=df.apply(pd.to_numeric, errors='ignore')
    # df[numerics]=df[numerics].apply(lambda x: np.round(x, decimals=5))
    # df['IP No.']=df['IP No.'].astype(str)
    df['Submission Date']=df['Submission Date'].apply(pd.to_datetime)
    df['Acceptance Date']=df['Acceptance Date'].apply(pd.to_datetime)
    # df["Provisional Sum for Safety"] = df["Provisional Sum for Safety"].apply(pd.to_numeric)
    # df["CEs incl. Fee"] = df["CEs incl. Fee"].apply(pd.to_numeric)
    # df["Forecast of the final total of the Prices"] = df["Forecast of the final total of the Prices"].apply(pd.to_numeric)
    # df["Tendered total of the Prices"] = df["Tendered total of the Prices"].apply(pd.to_numeric)
    # df["Price Adjustment for Inflation (PAI)"] = df["Price Adjustment for Inflation (PAI)"].apply(pd.to_numeric)
    # df["Original Contract Value"] = df["Original Contract Value"].apply(pd.to_numeric)
    # df["Total"] = df["Total"].apply(pd.to_numeric)
    df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')
    db = create_engine(conn_string)
    conn = db.connect()
    with conn as conn:
        df.to_sql('c04_programme_data', con=conn, if_exists='replace')
    conn.close()

# 63fca09d4243400ca9af7f84
    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/sheets/63fca09d4243400ca9af7f84?with_records=true&fields=",
    headers={
    "x-access-token": f"Bearer {token}",
    }
    )
    # # print('got_data')
    sheet = json.loads(response.text)
    headers = sheet['header']
    record = sheet['record']
    dataToExtract=[]
    for d in record: 
        objectToPush = {}
        for v in d['values']:
            for c in headers:
                colNameToExtract = c['colName']
                if v['colName'] == colNameToExtract:
                    # # print(v)
                    if v.get('multValue') != None:
                        if v['multValue'] == True:
                            if v['colType'] == 'Table':
                                tObjectArray = []
                                for t in v['tableValue']:
                                    tObjectToPush = {}
                                    for s in t['subValues']:
                                        tObjectToPush[s['colName']] = s.value
                                    tObjectArray.push(tObjectToPush)
                            else:
                                objectToPush[v['colName']] = v['valueArray']
                        else:
                            if v.get('value') != None:
                                if v.get('value') == 'NA':
                                    objectToPush[v['colName']] = None
                                else:
                                    objectToPush[v['colName']] = v['value']
                            else:
                                objectToPush[v['colName']] = None
                    else:
                        if v.get('value') != None:
                            if v.get('value') == 'NA':
                                objectToPush[v['colName']] = None
                            else:
                                objectToPush[v['colName']] = v['value']
                        else:
                            objectToPush[v['colName']] = None
        dataToExtract.append(objectToPush)
    # # print(len(dataToExtract))
    # # print(dataToExtract[0].keys())
    data= dataToExtract
    # data=[
    #     {
    #     "key Date":"KD1",
    #     "Planned Completion Date(PCD)": "2021-02-18",
    #     },
    #     {
    #     "key Date":"KD2",
    #     "Planned Completion Date(PCD)": "2024-06-04",
    #     },
    #     {
    #     "key Date":"KD3",
    #     "Planned Completion Date(PCD)": "2024-10-10",
    #     },
    #     {
    #     "key Date":"KD4",
    #     "Planned Completion Date(PCD)": "2025-04-11",
    #     },
    #     {
    #     "key Date":"KD5",
    #     "Planned Completion Date(PCD)": "2025-04-17",
    #     },
    #     {
    #     "key Date":"Section 1",
    #     "Planned Completion Date(PCD)": "2021-03-12",
    #     },
    #     {
    #     "key Date":"Section 2",
    #     "Planned Completion Date(PCD)": "2021-09-24",
    #     },
    #     {
    #     "key Date":"Section 3",
    #     "Planned Completion Date(PCD)": "2023-06-19",
    #     },
    #     {
    #     "key Date":"Section 4",
    #     "Planned Completion Date(PCD)": "2023-12-28",
    #     },
    #     {
    #     "key Date":"Section 5",
    #     "Planned Completion Date(PCD)": "2024-09-20",
    #     },
    #     {
    #     "key Date":"Section 6",
    #     "Planned Completion Date(PCD)": "2024-09-12",
    #     },
    #     {
    #     "key Date":"Section 7",
    #     "Planned Completion Date(PCD)": "2025-07-07",
    #     },
    #     {
    #     "key Date":"Section 8",
    #     "Planned Completion Date(PCD)": "2025-06-12",
    #     },
    #     {
    #     "key Date":"Section 9",
    #     "Planned Completion Date(PCD)": "2025-06-12",
    #     },
    #     {
    #     "key Date":"Section 10A",
    #     "Planned Completion Date(PCD)": "2024-06-18",
    #     },
    #     {
    #     "key Date":"Section 10B",
    #     "Planned Completion Date(PCD)": "2025-09-20",
    #     },
    #     {
    #     "key Date":"Section 10C",
    #     "Planned Completion Date(PCD)": "2026-06-12",
    #     },
    #     {
    #     "key Date":"Section 11",
    #     "Planned Completion Date(PCD)": "2025-07-16",
    #     }
    # ]
    df = pd.DataFrame.from_dict(data)
    # numerics = df.select_dtypes(include="number").columns
    # df=df.apply(pd.to_numeric, errors='ignore')
    # df[numerics]=df[numerics].apply(lambda x: np.round(x, decimals=5))
    # df['IP No.']=df['IP No.'].astype(str)
    df['Planned Completion Date(PCD)']=df['Planned Completion Date(PCD)'].apply(pd.to_datetime)
    # df["Provisional Sum for Safety"] = df["Provisional Sum for Safety"].apply(pd.to_numeric)
    # df["CEs incl. Fee"] = df["CEs incl. Fee"].apply(pd.to_numeric)
    # df["Forecast of the final total of the Prices"] = df["Forecast of the final total of the Prices"].apply(pd.to_numeric)
    # df["Tendered total of the Prices"] = df["Tendered total of the Prices"].apply(pd.to_numeric)
    # df["Price Adjustment for Inflation (PAI)"] = df["Price Adjustment for Inflation (PAI)"].apply(pd.to_numeric)
    # df["Original Contract Value"] = df["Original Contract Value"].apply(pd.to_numeric)
    # df["Total"] = df["Total"].apply(pd.to_numeric)
    df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')
    db = create_engine(conn_string)
    conn = db.connect()
    with conn as conn:
        df.to_sql('c04_key_date_data', con=conn, if_exists='replace')
    conn.close()

Task Instance Attributes

Attribute	Value
dag_id	c4_financial_data
duration	12.701184
end_date	2025-04-26 08:01:19.434413+00:00
execution_date	2025-04-26T04:00:00+00:00
executor_config	{}
generate_command	<function TaskInstance.generate_command at 0x7f152f9bf320>
hostname	63fbafbc3109
is_premature	False
job_id	142649
key	('c4_financial_data', 'getMongoDB', <Pendulum [2025-04-26T04:00:00+00:00]>, 2)
log	<Logger airflow.task (INFO)>
log_filepath	/usr/local/airflow/logs/c4_financial_data/getMongoDB/2025-04-26T04:00:00+00:00.log
log_url	http://localhost:8080/admin/airflow/log?execution_date=2025-04-26T04%3A00%3A00%2B00%3A00&task_id=getMongoDB&dag_id=c4_financial_data
logger	<Logger airflow.task (INFO)>
mark_success_url	http://localhost:8080/success?task_id=getMongoDB&dag_id=c4_financial_data&execution_date=2025-04-26T04%3A00%3A00%2B00%3A00&upstream=false&downstream=false
max_tries	1
metadata	MetaData(bind=None)
next_try_number	2
operator	PythonOperator
pid	2842596
pool	default_pool
prev_attempted_tries	1
previous_execution_date_success	2025-04-26 00:00:00+00:00
previous_start_date_success	2025-04-26 04:01:42.206649+00:00
previous_ti	<TaskInstance: c4_financial_data.getMongoDB 2025-04-26 00:00:00+00:00 [success]>
previous_ti_success	<TaskInstance: c4_financial_data.getMongoDB 2025-04-26 00:00:00+00:00 [success]>
priority_weight	1
queue	default
queued_dttm	2025-04-26 08:01:04.392719+00:00
raw	False
run_as_user	None
start_date	2025-04-26 08:01:06.733229+00:00
state	success
task	<Task(PythonOperator): getMongoDB>
task_id	getMongoDB
test_mode	False
try_number	2
unixname	airflow

Task Attributes

Attribute	Value
dag	<DAG: c4_financial_data>
dag_id	c4_financial_data
depends_on_past	False
deps	{<TIDep(Not In Retry Period)>, <TIDep(Trigger Rule)>, <TIDep(Previous Dagrun State)>}
do_xcom_push	True
downstream_list	[]
downstream_task_ids	set()
email	None
email_on_failure	True
email_on_retry	True
end_date	None
execution_timeout	None
executor_config	{}
extra_links	[]
global_operator_extra_link_dict	{}
inlets	[]
lineage_data	None
log	<Logger airflow.task.operators (INFO)>
logger	<Logger airflow.task.operators (INFO)>
max_retry_delay	None
on_failure_callback	None
on_retry_callback	None
on_success_callback	None
op_args	[]
op_kwargs	{'name': 'Dylan'}
operator_extra_link_dict	{}
operator_extra_links	()
outlets	[]
owner	airflow
params	{}
pool	default_pool
priority_weight	1
priority_weight_total	1
provide_context	True
queue	default
resources	None
retries	1
retry_delay	0:05:00
retry_exponential_backoff	False
run_as_user	None
schedule_interval	0 0,4,8,11,16 * * *
shallow_copy_attrs	('python_callable', 'op_kwargs')
sla	None
start_date	2022-10-24T00:00:00+00:00
subdag	None
task_concurrency	None
task_id	getMongoDB
task_type	PythonOperator
template_ext	[]
template_fields	('templates_dict', 'op_args', 'op_kwargs')
templates_dict	None
trigger_rule	all_success
ui_color	#ffefeb
ui_fgcolor	#000
upstream_list	[<Task(PythonOperator): getDrowToken>]
upstream_task_ids	{'getDrowToken'}
wait_for_downstream	False
weight_rule	downstream

DAG: c4_financial_data

schedule: 0 0,4,8,11,16 * * *

Task Instance: getMongoDB

Task Instance Details

Dependencies Blocking Task From Getting Scheduled

Attribute: python_callable

Task Instance Attributes

Task Attributes