From f6fbce36e18506347d4661e64b8efee1d01d669e Mon Sep 17 00:00:00 2001 From: Giambattista Bloisi Date: Wed, 27 Mar 2024 12:57:02 +0100 Subject: [PATCH] initial stage --- airflow/dags/import_EOSC_catalog.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/airflow/dags/import_EOSC_catalog.py b/airflow/dags/import_EOSC_catalog.py index d7a9148..d3882da 100644 --- a/airflow/dags/import_EOSC_catalog.py +++ b/airflow/dags/import_EOSC_catalog.py @@ -88,12 +88,11 @@ def import_EOSC_catalog(): num_batches = len(pieces)//kwargs["params"]["BATCH_LOADERS_NUM"] if num_batches > 0: - return list(split_list(pieces, num_batches)) - return list(split_list(pieces, len(pieces))) + return [*split_list(pieces, num_batches)] + return [*split_list(pieces, len(pieces))] @task - def bulk_load(**kwargs): - files = kwargs["files"] + def bulk_load(files: list[(str, str)], **kwargs): conn = BaseHook.get_connection(kwargs["params"]["OPENSEARCH_CONN_ID"]) client = OpenSearch( hosts=[{'host': conn.host, 'port': conn.port}],