datascan/airflow/dq_fail_scan_if_not_passed_dag.py [54:95]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
)

get_job_result = DataplexGetDataQualityScanResultOperator(
            task_id="get_data_scan_job_result",
            project_id=PROJECT_ID,
            region=REGION,
            dag=dag,
            data_scan_id=DATA_SCAN_ID)

def process_data_from_data_scan_job(**kwargs):
    ti = kwargs['ti']
    job_data = ti.xcom_pull(task_ids='get_data_scan_job_result')
    if "dataQualityResult" not in job_data:
        return "failed_job"
    print(f"data quality job result: {job_data.get('dataQualityResult')}")
    if "passed" in job_data["dataQualityResult"]:
        return "passed_job"
    return "failed_job"

def pass_job(**kwargs):
    print("data quality job passed")

def fail_job(**kwargs):
    print("data quality job failed")

passed_job = PythonOperator(
    task_id = "passed_job",
    python_callable=pass_job,
    dag=dag
    )

failed_job = PythonOperator(
    task_id = "failed_job",
    python_callable=fail_job,
    dag=dag
    )

process_data = BranchPythonOperator(
      task_id='process_data_from_data_scan_job',
      python_callable=process_data_from_data_scan_job,
      dag=dag
      )
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



datascan/airflow/dq_runAsync_evaluate_dag.py [63:104]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
    )

get_job_result = DataplexGetDataQualityScanResultOperator(
            task_id="get_data_scan_job_result",
            project_id=PROJECT_ID,
            region=REGION,
            dag=dag,
            data_scan_id=DATA_SCAN_ID)

def process_data_from_data_scan_job(**kwargs):
    ti = kwargs['ti']
    job_data = ti.xcom_pull(task_ids='get_data_scan_job_result')
    if "dataQualityResult" not in job_data:
        return "failed_job"
    print(f"data quality job result: {job_data.get('dataQualityResult')}")
    if "passed" in job_data["dataQualityResult"]:
        return "passed_job"
    return "failed_job"

def pass_job(**kwargs):
    print("data quality job passed")

def fail_job(**kwargs):
    print("data quality job failed")

passed_job = PythonOperator(
    task_id = "passed_job",
    python_callable=pass_job,
    dag=dag
    )

failed_job = PythonOperator(
    task_id = "failed_job",
    python_callable=fail_job,
    dag=dag
    )

process_data = BranchPythonOperator(
      task_id='process_data_from_data_scan_job',
      python_callable=process_data_from_data_scan_job,
      dag=dag
      )
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -