def generate_update_doc_registry_job_params_fn()

in components/dpu-workflow/src/docs_processing_orchestrator.py [0:0]

• 13 lines of code
• 1 McCabe index (conditional complexity)

def generate_update_doc_registry_job_params_fn(**context):
    bq_table = context["ti"].xcom_pull(key="bigquery_table")
    input_bq_table = (
        f"{bq_table['project_id']}.{bq_table['dataset_id']}." f"{bq_table['table_id']}"
    )
    process_folder = context["ti"].xcom_pull(
        task_ids="initial_load_from_input_bucket.create_process_folder",
        key="process_folder",
    )
    output_folder = f'{os.environ.get("DPU_PROCESS_BUCKET")}/{process_folder}/workflow-io/update_doc_registry'
    return cloud_run_utils.get_doc_registry_update_job_override(
        input_bq_table, output_folder
    )