Merge remote-tracking branch 'origin/development' into DAT-328

.github/workflows/main.yml

@@ -272,7 +272,7 @@ jobs:
       - name: Run SonarCloud Scan
         uses: SonarSource/sonarcloud-github-action@master
         env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}  # Needed to get PR information, if any
+#          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}  # Needed to get PR information, if any
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
 
   e2e_test:
@@ -479,4 +479,5 @@ jobs:
         env:
           EE_TOKEN: ${{ secrets.EE_TOKEN }}
         shell: bash
+
 

datamimic_ce/exporters/csv_exporter.py

@@ -36,7 +36,6 @@ def __init__(
         self._task_id = setup_context.task_id
 
         # Retrieve encoding and delimiter from setup_context or use defaults
-        kwargs["encoding"] = encoding or setup_context.default_encoding or "utf-8"
         self.delimiter = delimiter or setup_context.default_separator or ","
         self.quotechar = quotechar or '"'
         self.quoting = quoting or csv.QUOTE_MINIMAL
@@ -56,7 +55,7 @@ def __init__(
             **kwargs,
         )
         logger.info(
-            f"CSVEEExporter initialized with chunk size {chunk_size}, fieldnames '{fieldnames}', "
+            f"CSVExporter initialized with chunk size {chunk_size}, fieldnames '{fieldnames}', "
             f"encoding '{self._encoding}', delimiter '{self.delimiter}'"
         )
 
@@ -102,24 +101,3 @@ def _reset_state(self):
         """Resets the exporter state for reuse."""
         super()._reset_state()
         logger.debug("CSVEEExporter state has been reset.")
-
-    # def __init__(self, setup_context: SetupContext, product_name: str):
-    #     super().__init__("csv", setup_context, product_name)
-    #     self._separator = setup_context.default_separator
-    #     self._line_separator = setup_context.default_line_separator or os.linesep
-
-    # def consume(self, product: Tuple[str, List[Dict]]) -> None:
-    #     name = product[0]
-    #     data = product[1]
-    #
-    #     file_name = f"{name}.csv"
-    #     with open(self._exported_data_dir / file_name, "w") as file:
-    #         if data:
-    #             keys = data[0].keys()
-    #             file.write(f"{self._separator}".join(keys))
-    #             file.write(self._line_separator)
-    #             for row in data:
-    #                 file.write(f"{self._separator}".join([str(row[key]) for key in keys]))
-    #                 file.write(self._line_separator)
-    #
-    #     logger.debug(f"Exported data to {self._exported_data_dir / file_name}")

datamimic_ce/exporters/exporter_util.py

@@ -71,6 +71,8 @@ def create_exporter_list(
 
         :param setup_context:
         :param stmt:
+        :param targets:
+        :param mp_idx:
         :return:
         """
         consumers_with_operation = []

datamimic_ce/exporters/json_exporter.py

@@ -34,6 +34,7 @@ def __init__(
         mp_id: int | None,
         use_ndjson: bool = False,
         chunk_size: int = None,
+        encoding: str = None,
     ):
         self.use_ndjson = use_ndjson
         self._task_id = setup_context.task_id

datamimic_ce/exporters/txt_exporter.py

@@ -59,7 +59,6 @@ def __init__(
         )
 
         # Pass encoding via kwargs to the base class
-        kwargs["encoding"] = encoding or setup_context.default_encoding or "utf-8"
 
         super().__init__(
             "txt",

datamimic_ce/exporters/unified_buffered_exporter.py

@@ -65,11 +65,7 @@ def __init__(
         self.chunk_size = chunk_size  # Max entities per chunk
 
         # Prepare temporary buffer directory
-        #     / f"exporter_result_{self._task_id}_pid_{self._pid}_exporter_{self._exporter_type}_product_{self.product_name}"
-        self._buffer_tmp_dir = (
-            self._descriptor_dir
-            / f"exporter_result_{self._task_id}_exporter_{self._exporter_type}_product_{self.product_name}"
-        )
+        self._buffer_tmp_dir = self._get_buffer_tmp_dir()
         self._buffer_tmp_dir.mkdir(parents=True, exist_ok=True)
 
         # Initialize state variables
@@ -79,14 +75,19 @@ def __init__(
         # Handle any additional parameters from kwargs
         for key, value in kwargs.items():
             setattr(self, key, value)
-        # self._buffer_file = self._get_buffer_file()  # Initialize first buffer file
+
+    @property
+    def encoding(self) -> str:
+        return self._encoding
+
+    def _get_buffer_tmp_dir(self) -> Path:
+        return self._descriptor_dir / f"exporter_result_{self._task_id}_exporter_{self._exporter_type}_product_{self.product_name}"
 
     def _get_buffer_dir_by_pid(self) -> Path:
         """
         Returns the buffer directory for the current PID.
         :return:
         """
-        # / f"exporter_result_{self._task_id}_pid_{self._pid}_exporter_{self._exporter_type}_product_{self.product_name}"
         return (
             self._descriptor_dir
             / f"exporter_result_{self._task_id}_exporter_{self._exporter_type}_product_{self.product_name}"
@@ -168,7 +169,6 @@ def _rotate_chunk(self) -> None:
         self.chunk_index += 1
         self.current_counter = 0  # Reset the current count for new chunk
         self._save_state()
-        # self._buffer_file = self._get_buffer_file()  # Create new buffer file for the next chunk
         self._is_first_write = True  # Reset for the new chunk
 
     def store_data(self, data: list[dict]) -> None:
@@ -282,40 +282,6 @@ def consume(self, product: tuple):
         logger.debug(f"Storing data for '{self.product_name}' with {len(data)} records")
         self.store_data(data)
 
-    # def upload_to_storage(self, bucket: str, name: str) -> None:
-    #     """Uploads all buffered chunk files to the object storage in the specified format."""
-    #     buffer_dir = self._get_buffer_dir_by_pid()
-    #
-    #     # logger.debug(f"Found {len(buffer_dirs)} buffer directories for export.")
-    #
-    #     # Define file suffix
-    #     suffix = self._define_suffix()
-    #
-    #     buffer_files = sorted(
-    #         buffer_dir.glob(f"*.{suffix}"),
-    #         key=lambda x: int(re.search(r"_chunk_(\d+)", x.name).group(1)),
-    #     )
-    #     logger.debug(f"Found {len(buffer_files)} buffer files in directory {buffer_dir}.")
-    #
-    #     for buffer_file in buffer_files:
-    #         # Corresponding metadata file
-    #         metadata = self._load_metadata(buffer_file.with_suffix(".meta"))
-    #
-    #         total_count, uri = self._craft_uri(metadata, suffix)
-    #
-    #         with buffer_file.open("r", encoding=self._encoding) as f:
-    #             data_content = f.read()
-    #             data_buffer = BytesIO(data_content.encode(self._encoding))
-    #             content_type = self._get_content_type()
-    #             # Perform the upload
-    #             self._storage.write(bucket or self._default_bucket, uri, data_buffer, content_type)
-    #             logger.debug(f"Uploaded {uri} with {total_count} records")
-    #
-    #     logger.info(f"Exported buffer files successfully.")
-    #
-    #     # Cleanup buffer directories
-    #     self._reset_state()
-
     @abstractmethod
     def _define_suffix(self) -> str:
         """Defines the file suffix based on the format."""
@@ -342,44 +308,28 @@ def _craft_uri(self, metadata, suffix):
         )
 
         # Determine URI based on chunk size and multiprocessing
-        if self._mp:
-            if chunk_size is None:
-                uri = f"{product_name}_pid_{pid}.{suffix}"
-            elif chunk_size == 1:
-                uri = f"{product_name}_{chunk_start}_pid_{pid}.{suffix}"
-            else:
-                uri = f"{product_name}_{chunk_start}_{chunk_end}_pid_{pid}.{suffix}"
+        if chunk_size is None:
+            uri = f"{product_name}{self._pid_placeholder}.{suffix}"
+        elif chunk_size == 1:
+            uri = f"{product_name}_{chunk_start}{self._pid_placeholder}.{suffix}"
         else:
-            if chunk_size is None:
-                uri = f"{product_name}.{suffix}"
-            elif chunk_size == 1:
-                uri = f"{product_name}_{chunk_start}.{suffix}"
-            else:
-                uri = f"{product_name}_{chunk_start}_{chunk_end}.{suffix}"
+            uri = f"{product_name}_{chunk_start}_{chunk_end}{self._pid_placeholder}.{suffix}"
 
         return total_count, f"{self._task_id}/{uri}"
 
     def cleanup(self) -> None:
         """Cleans up the temporary buffer directories."""
         # Identify all buffer directories for this task
 
-        # buffer_dir = self._get_buffer_dir_by_pid()
-        #
-        # if buffer_dir.exists():
-        #     # Delete all files in the buffer directory
-        #     for file in buffer_dir.iterdir():
-        #         if file.is_file() and file.suffix == ".meta":
-        #             file.unlink()
-
         buffer_metadata_file = self._get_buffer_file().with_suffix(".meta")
         buffer_metadata_file.unlink()
         self._get_state_meta_file().unlink()
-        # pass
-        # # Remove the buffer directory
-        # try:
-        #     buffer_dir.rmdir()
-        # except OSError as e:
-        #     logger.error(f"Failed to remove buffer directory {buffer_dir}: {e}")
+                    # pass
+            # # Remove the buffer directory
+            # try:
+            #     buffer_dir.rmdir()
+            # except OSError as e:
+            #     logger.error(f"Failed to remove buffer directory {buffer_dir}: {e}")
         logger.debug("Buffer directory cleanup complete.")
 
     def _reset_state(self):

datamimic_ce/exporters/xml_exporter.py

@@ -52,12 +52,13 @@ def __init__(
         self.item_element = item_element
 
         # Pass encoding via kwargs to the base class
-        kwargs["encoding"] = encoding or setup_context.default_encoding or "utf-8"
+        # kwargs["encoding"] = encoding or setup_context.default_encoding or "utf-8"
 
         super().__init__(
             exporter_type="xml",
             setup_context=setup_context,
             product_name=product_name,
+            encoding=encoding,
             chunk_size=chunk_size,
             mp_idx=mp_id,
             **kwargs,

datamimic_ce/tasks/generate_task.py

@@ -959,6 +959,10 @@ def execute(self, context: SetupContext) -> dict[str, list] | None:
                     del result
                     # gc.collect()
 
+            # Clean temp directory on outermost gen_stmt
+            for temp_dir in context.descriptor_dir.glob(f"temp_result_{context.task_id}*"):
+                shutil.rmtree(temp_dir)
+
         # Just return product generated in single process if gen_stmt is inner one
         else:
             # Do not apply process by page for inner gen_stmt

tests_ce/functional_tests/csv_separatorr/test_separator.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestDataIteration:

tests_ce/functional_tests/data_source_cyclic/test_data_source_cyclic_functional.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestDataSourceCyclic:

tests_ce/functional_tests/test_casting_type/test_data_type_functional.py

@@ -9,7 +9,7 @@
 from pathlib import Path
 from types import NoneType
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestDataType:

tests_ce/functional_tests/test_condition/test_conditon.py

@@ -11,7 +11,7 @@
 
 import pytest
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestCondition(TestCase):

tests_ce/functional_tests/test_condition_element/test_conditon_element.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestConditionElement:

tests_ce/functional_tests/test_datetime/test_datetime.py

@@ -11,7 +11,7 @@
 
 import pytest
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestDateTime:

tests_ce/functional_tests/test_functional_array/test_array_functional.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestArrayFunctional:

tests_ce/functional_tests/test_functional_page_process/test_functional_page_process.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestFunctionalPageProcess:

tests_ce/functional_tests/test_functional_variable/test_variable_functional.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestVariableFunctional:

tests_ce/functional_tests/test_functional_weighted_data_source/test_functional_weighted_data_source.py

@@ -9,7 +9,7 @@
 import math
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestWeightedDataSourceFunctional:

tests_ce/functional_tests/test_generate_distribution/test_generate_distribution.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestGenerateDistribution:

tests_ce/functional_tests/test_generator_functional/test_generator_functional.py

@@ -11,7 +11,7 @@
 
 from datamimic_ce.generators.generator_util import GeneratorUtil
 from datamimic_ce.utils.file_util import FileUtil
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 def count_digits_after_decimal(number):

tests_ce/functional_tests/test_include/test_include.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestInclude:

tests_ce/functional_tests/test_memstore/test_memstore.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestMemStore:

tests_ce/functional_tests/test_mongodb/test_mongodb_functional.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestMongoDbFunction:

tests_ce/functional_tests/test_nested_generate/test_nested_generate.py

@@ -8,7 +8,7 @@
 
 from pathlib import Path
 
-from tests_ce.data_mimic_test import DataMimicTest
+from datamimic_ce.data_mimic_test import DataMimicTest
 
 
 class TestNestedGenerate: