def save_translations()

in jobs/broken-site-report-ml/broken_site_report_ml/main.py [0:0]

• 35 lines of code
• 6 McCabe index (conditional complexity)

def save_translations(client, bq_dataset_id, results):
    res = []
    for uuid, result in results.items():
        if not result["status"]:
            bq_result = {
                "report_uuid": uuid,
                "translated_text": result["translated_text"],
                "language_code": result["language_code"],
            }
            res.append(bq_result)

    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
        schema=[
            bigquery.SchemaField("report_uuid", "STRING", mode="REQUIRED"),
            bigquery.SchemaField("translated_text", "STRING"),
            bigquery.SchemaField("language_code", "STRING"),
        ],
        write_disposition="WRITE_APPEND",
    )

    translations_table = f"{bq_dataset_id}.translations"

    job = client.load_table_from_json(
        res,
        translations_table,
        job_config=job_config,
    )

    logging.info("Writing to `translations` table")

    try:
        job.result()
    except Exception as e:
        print(f"ERROR: {e}")
        if job.errors:
            for error in job.errors:
                logging.error(error)

    table = client.get_table(translations_table)
    logging.info(f"Loaded {len(res)} rows into {table}")