def _get_all_translated_files()

in src/translation/dags/batch_sql_translation.py [0:0]

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

def _get_all_translated_files(ti) -> None:
    translation_config = ti.xcom_pull(
        key="config", task_ids="create_translation_workflow"
    )

    target_path = translation_config["migrationTask"]["translationConfigDetails"][
        "gcsTargetPath"
    ]
    target_bucket, target_folder = gcs_util.parse_bucket_and_blob_from_path(target_path)

    translated_files = set()
    for blob in storage_client.list_blobs(target_bucket, prefix=target_folder):
        file_full_path = Path(blob.name)
        filename = file_full_path.name
        if (
            blob.name[-1] != "/"
            and filename != REPORT_FILENAME
            and filename != MAP_FILENAME
            and SKIP_VALIDATION_DIR_NAME not in file_full_path.parts
        ):
            translated_files.add(blob.name)

    ti.xcom_push(key="files", value=list(translated_files))