refactor(pyamber): drop PyFilesystem2 for stdlib tempfile

amber/LICENSE-binary-python

@@ -247,15 +247,13 @@ Python packages:
   - annotated-doc==0.0.4
   - annotated-types==0.7.0
   - anyio==4.14.2
-  - appdirs==1.4.4
   - asn1crypto==1.5.1
   - attrs==26.1.0
   - betterproto==2.0.0b7
   - cachetools==6.2.6
   - charset-normalizer==3.4.9
   - filelock==3.29.7
   - fonttools==4.63.0
-  - fs==2.4.16
   - greenlet==3.5.3
   - h11==0.16.0
   - h2==4.3.0

amber/pyproject.toml

@@ -42,4 +42,4 @@ testpaths = ["src/test/python"]
 addopts = "--import-mode=importlib"
 markers = [
     "integration: end-to-end test routed to the amber-integration CI job",
-]
+]

amber/requirements.txt

@@ -15,8 +15,6 @@
 # specific language governing permissions and limitations
 # under the License.
 
-# Not imported directly: `fs` needs pkg_resources at import time (#4199).
-setuptools==80.10.2
 numpy==2.1.0
 pandas==2.2.3
 loguru==0.7.3
@@ -25,7 +23,6 @@ betterproto==2.0.0b7
 pampy==0.3.0
 overrides==7.7.0
 typing_extensions==4.14.1
-fs==2.4.16
 bidict==0.22.0
 cached_property==2.0.1
 psutil==7.2.2

amber/src/main/python/core/architecture/managers/executor_manager.py

@@ -15,13 +15,13 @@
 # specific language governing permissions and limitations
 # under the License.
 
-import fs
 import importlib
 import inspect
 import itertools
+import shutil
 import sys
+import tempfile
 from cached_property import cached_property
-from fs.base import FS
 from loguru import logger
 from pathlib import Path
 from typing import Tuple, Optional
@@ -48,28 +48,27 @@ def __init__(self):
         self.operator_module_name: Optional[str] = None
 
     @cached_property
-    def fs(self) -> FS:
+    def tmp_dir(self) -> Path:
         """
-        Creates a tmp fs for storing source code, which will be removed when the
-        workflow is completed.
+        Creates a tmp directory for storing source code, which will be removed
+        when the workflow is completed.
         :return:
         """
         # TODO:
         #       For various reasons when the workflow is not completed successfully,
-        #  the tmp fs could not be closed properly. This means it may leave files
-        #  in the /var/tmp folder after a partially started or failed execution.
-        #       A full-life-cycle management of tmp fs is required to consider all
-        #  possible errors happened during execution. However, the full-life-cycle
-        #  management could be hard due to errors from JAVA side which causes force
-        #  kill on the Python process.
+        #  the tmp directory could not be removed properly. This means it may leave
+        #  files in the OS temp folder after a partially started or failed execution.
+        #       A full-life-cycle management of the tmp directory is required to
+        #  consider all possible errors happened during execution. However, the
+        #  full-life-cycle management could be hard due to errors from JAVA side
+        #  which causes force kill on the Python process.
         #       As each python file is usually tiny in size, and the OS can
-        #  periodically clean up /var/tmp anyway, the full-life-cycle management is
-        #  not a priority to be fixed.
-        temp_fs = fs.open_fs("temp://")
-        root = Path(temp_fs.getsyspath("/"))
+        #  periodically clean up its temp folder anyway, the full-life-cycle
+        #  management is not a priority to be fixed.
+        root = Path(tempfile.mkdtemp(prefix="texera-udf-"))
         logger.debug(f"Opening a tmp directory at {root}.")
         sys.path.append(str(root))
-        return temp_fs
+        return root
 
     def gen_module_file_name(self) -> Tuple[str, str]:
         """
@@ -92,16 +91,19 @@ def load_executor_definition(self, code: str) -> type(Operator):
         """
         module_name, file_name = self.gen_module_file_name()
 
-        with self.fs.open(file_name, "w") as file:
+        # importlib decodes the source file as UTF-8 (PEP 3120), so the write
+        # side must be pinned to UTF-8 too — the locale default (e.g. cp1252
+        # on Windows) would break non-ASCII UDF code. newline="\n" keeps the
+        # written bytes identical across platforms.
+        with (self.tmp_dir / file_name).open(
+            "w", encoding="utf-8", newline="\n"
+        ) as file:
             file.write(code)
-        logger.debug(
-            "A tmp py file is written to "
-            f"{Path(self.fs.getsyspath('/')).joinpath(file_name)}."
-        )
+        logger.debug(f"A tmp py file is written to {self.tmp_dir / file_name}.")
 
         # gen_module_file_name guarantees module_name is unique across
         # the process, so import_module will always cleanly load source
-        # from the tmp fs we just wrote — no re-import / reload dance.
+        # from the tmp file we just wrote — no re-import / reload dance.
         executor_module = importlib.import_module(module_name)
         self.operator_module_name = module_name
 
@@ -113,25 +115,25 @@ def load_executor_definition(self, code: str) -> type(Operator):
 
     def close(self) -> None:
         """
-        Close the tmp fs and release all resources created within it.
+        Remove the tmp directory and release all resources created within it.
         This also evicts the loaded operator module from ``sys.modules``
-        and removes the tmp fs path from ``sys.path`` so a single call
-        fully reverses every global side-effect performed by ``fs`` and
+        and removes the tmp directory path from ``sys.path`` so a single call
+        fully reverses every global side-effect performed by ``tmp_dir`` and
         ``load_executor_definition``.
         :return:
         """
-        if "fs" not in self.__dict__:
-            # fs was never materialized; nothing to clean up.
+        if "tmp_dir" not in self.__dict__:
+            # tmp_dir was never materialized; nothing to clean up.
             return
-        root = self.fs.getsyspath("/")
-        self.fs.close()
+        root = self.tmp_dir
+        shutil.rmtree(root, ignore_errors=True)
         try:
-            sys.path.remove(str(Path(root)))
+            sys.path.remove(str(root))
         except ValueError:
             pass
         if self.operator_module_name is not None:
             sys.modules.pop(self.operator_module_name, None)
-        logger.debug(f"Tmp directory {root} is closed and cleared.")
+        logger.debug(f"Tmp directory {root} is removed and cleared.")
 
     @staticmethod
     def is_concrete_operator(cls: type) -> bool:

amber/src/test/python/core/architecture/managers/test_executor_manager.py

@@ -39,6 +39,17 @@ def produce(self) -> Iterator[Union[TupleLike, TableLike, None]]:
         yield Tuple({"test": "data"})
 """
 
+NON_ASCII_OPERATOR_CODE = """
+from pytexera import *
+
+class NonAsciiOperator(UDFOperatorV2):
+    # コメント: user code may contain any Unicode text.
+    GREETING = "café"
+
+    def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]:
+        yield tuple_
+"""
+
 
 class TestExecutorManager:
     """Test suite for ExecutorManager, focusing on R UDF plugin support."""
@@ -255,26 +266,31 @@ def test_source_operator_mismatch_raises_error(self, executor_manager):
             )
         assert "SourceOperator API" in str(exc_info.value)
 
+    def test_non_ascii_udf_source_round_trip(self, executor_manager):
+        # importlib decodes the written source file as UTF-8 (PEP 3120), so
+        # the write side must be pinned to UTF-8 as well; the locale default
+        # (e.g. cp1252 on Windows) would break either the write or the import.
+        executor_manager.initialize_executor(
+            code=NON_ASCII_OPERATOR_CODE, is_source=False, language="python"
+        )
+        assert executor_manager.executor.GREETING == "café"
+
     def test_close_when_sys_path_entry_already_removed(self):
-        # Exercise the except ValueError branch: if the tmp fs path has
-        # already been pulled out of sys.path by something else, close()
+        # Exercise the except ValueError branch: if the tmp directory path
+        # has already been pulled out of sys.path by something else, close()
         # should swallow the error and finish cleanly.
-        from pathlib import Path
-
         manager = ExecutorManager()
-        root = Path(manager.fs.getsyspath("/"))
+        root = manager.tmp_dir
         sys.path.remove(str(root))
         manager.close()
         assert str(root) not in sys.path
 
-    def test_close_when_fs_materialized_but_no_executor_loaded(self):
-        # Exercise the branch where self.fs was touched (so the early
+    def test_close_when_tmp_dir_materialized_but_no_executor_loaded(self):
+        # Exercise the branch where self.tmp_dir was touched (so the early
         # return is skipped) but operator_module_name is still None,
         # meaning the sys.modules.pop branch must NOT execute.
-        from pathlib import Path
-
         manager = ExecutorManager()
-        root = Path(manager.fs.getsyspath("/"))
+        root = manager.tmp_dir
         assert manager.operator_module_name is None
         manager.close()
         assert str(root) not in sys.path

amber/src/test/python/core/runnables/test_main_loop.py

@@ -2613,8 +2613,8 @@ def test_two_main_loops_load_distinct_operator_classes(self):
         Regression test for executor-module contamination (#4705): executor
         modules were once named ``udf-v<per-instance-counter>``, so every
         loop's first executor was ``udf-v1`` in the process-wide
-        ``sys.modules``. A loop whose worker never completes never closes its
-        temp fs, so its ``udf-v1.py`` lingered on ``sys.path`` and the next
+        ``sys.modules``. A loop whose worker never completes never removes its
+        temp directory, so its ``udf-v1.py`` lingered on ``sys.path`` and the next
         loop re-resolved ``udf-v1`` to that older file, silently running the
         wrong operator. This test uses NO monkeypatch of
         ``gen_module_file_name`` -- module names must be process-globally
@@ -2633,9 +2633,9 @@ def test_two_main_loops_load_distinct_operator_classes(self):
         second = Context("worker-second", InternalQueue())
         try:
             # The first loop loads EchoOperator and is intentionally left "unfinished"
-            # until after the second loop is initialized: its temp fs is not closed yet,
-            # so its udf module and sys.path entry linger exactly as a crashed /
-            # never-completed worker's would.
+            # until after the second loop is initialized: its temp directory is not
+            # removed yet, so its udf module and sys.path entry linger exactly as a
+            # crashed / never-completed worker's would.
             first.executor_manager.initialize_executor(
                 echo_code, is_source=False, language="python"
             )