Refactor for respect single-responsability in transform module

Author: IvanildoBarauna

etl/models/transform/ResponseSplit.py

@@ -1,61 +1,16 @@
-import os
-
-import pandas as pd
-import concurrent.futures
-from tqdm import tqdm
-
-from etl.common.utils.common import (
-    DefaultTimestampStr,
-    DefaultOutputFolder,
-    DefaultUTCDatetime,
-)
 from etl.common.utils.logs import loggingInfo
 from etl.config.logFile import logFileName
 
 WORK_DIR = logFileName(file=__file__)
 
 
 class transformation:
-    def __init__(self, json_response: dict, params) -> None:
-        self.output_path = DefaultOutputFolder()
-        self.insert_timestamp = DefaultTimestampStr()
-        self.extracted_files = []
+    def __init__(self, json_response: dict, params, fila: object):
         self.json_response = json_response
-        self.totalParams = len(json_response)
         self.validParams = params
+        self.fila = fila
 
-        ## Parallel Processing data
-        with concurrent.futures.ThreadPoolExecutor(os.cpu_count()) as executor:
-            list(
-                tqdm(
-                    executor.map(self.__process_param__, enumerate(self.validParams)),
-                    total=self.totalParams,
-                    desc="Processing files",
-                )
-            )
-
-        loggingInfo(
-            f"{self.totalParams} files extracted in: {self.output_path}", WORK_DIR
-        )
-
-    def __process_param__(self, args):
-
-        index, param = args
-        dic = self.json_response[param.replace("-", "")]
-
-        # Convert 'dic' to a Pandas DataFrame
-        df = pd.DataFrame([dic])
-
-        # Add new columns to the DataFrame
-        df["symbol"] = param
-
-        # Add two columns with the current date and time
-        df["extracted_at"] = DefaultUTCDatetime()
-
-        # Write the DataFrame to a Parquet file
-        df.to_parquet(f"{self.output_path}{param}-{self.insert_timestamp}.parquet")
-
-        # Append list with the file path
-        self.extracted_files.append(f"{param}-{self.insert_timestamp}.parquet")
-
-        return None
+    def publish(self):
+        for param in self.validParams:
+            dic = self.json_response[param.replace("-", "")]
+            self.fila.put(dic)  # type: ignore

etl/models/transform/test_transform.py

@@ -1,5 +1,8 @@
 import pytest
 from etl.models.transform.ResponseSplit import transformation
+import queue
+
+fila = queue.Queue()
 
 
 @pytest.fixture
@@ -41,10 +44,11 @@ def valid_params() -> list:
 
 def test_transformation_process_param(json_response, valid_params):
     # Create an instance of the transformation class
-    transform = transformation(json_response, valid_params)
+    transform = transformation(json_response, valid_params, fila)
+    transform.publish()
 
-    assert len(transform.extracted_files) == 2
+    assert fila.qsize() == 2
 
-    for extracted_file in transform.extracted_files:
-        assert extracted_file.startswith("BRL-")
-        assert extracted_file.endswith(".parquet")
+    # for extracted_file in transform.extracted_files:
+    #     assert extracted_file.startswith("BRL-")
+    #     assert extracted_file.endswith(".parquet")