wip

Author: ebonnal

pythoned/__init__.py

@@ -1,34 +1,59 @@
 import importlib
 import os
 import re
+import tempfile
 from types import ModuleType
-from typing import Dict, Iterator
+from typing import Callable, Dict, Iterator, List, TextIO, Union, cast
 
 
 _TYPE_ERROR_MSG = "The provided expression must be an str (editing) or a bool (filtering), but got {}."
 
+def _construct_edit_func(expression: str, modules: List[str]) -> str:
+    return f"""
+{"\n".join("import " + module for module in modules)}
+from typing import Union
 
-def edit(lines: Iterator[str], expression) -> Iterator[str]:
-    modules: Dict[str, ModuleType] = {}
+def process_line(_: str) -> Union[str, bool]:
+    return {expression}
+"""
+
+def _get_process_line_func(
+    expression: str,
+    modules: List[str],
+    process_line_file: TextIO,
+) -> Callable[[str], Union[str, bool]]:
+    process_line_file.write(_construct_edit_func(expression, modules).encode())
+    process_line_module = importlib.import_module(process_line_file.name)
+    return cast(
+        Callable[[str], Union[str, bool]],
+        process_line_module.process_line,
+    )
+
+def edit(
+    lines: Iterator[str],
+    expression,
+    process_line_file: TextIO,
+) -> Iterator[str]:
+    modules: List[str] = []
+    process_line = _get_process_line_func(expression, modules, process_line_file)
     for line in lines:
         linesep = ""
         if line.endswith(os.linesep):
             linesep, line = os.linesep, line[: -len(os.linesep)]
-        globals = {"_": line, **modules}
         try:
-            value = eval(expression, globals)
+            value = process_line(line)
         except NameError as name_error:
             match = re.match(r"name '([A-Za-z]+)'.*", str(name_error))
             if match:
                 module = match.group(1)
             else:
                 raise name_error
             try:
-                modules[module] = importlib.import_module(module)
-                globals = {"_": line, **modules}
+                modules.append(module)
+                process_line = _get_process_line_func(expression, modules, process_line_file)
             except:
                 raise name_error
-            value = eval(expression, globals)
+            value = process_line(line)
         if isinstance(value, str):
             yield value + linesep
         elif isinstance(value, bool):

tests/test.py

@@ -1,3 +1,4 @@
+import tempfile
 from typing import Iterator
 import unittest
 
@@ -10,28 +11,29 @@ def lines() -> Iterator[str]:
 
 class TestStream(unittest.TestCase):
     def test_edit(self) -> None:
-        self.assertEqual(
-            list(edit(lines(), "_[-1]")),
-            ["0\n", "r\n", "r"],
-            msg="str expression must edit the lines",
-        )
-        self.assertEqual(
-            list(edit(lines(), 're.sub(r"\d", "X", _)')),
-            ["fXX\n", "bar\n", "fXXbar"],
-            msg="re should be supported out-of-the-box",
-        )
-        self.assertEqual(
-            list(edit(lines(), '"0" in _')),
-            ["f00\n", "f00bar"],
-            msg="bool expression must filter the lines",
-        )
-        self.assertEqual(
-            list(edit(lines(), "len(_) == 3")),
-            ["f00\n", "bar\n"],
-            msg="_ must exclude linesep",
-        )
-        self.assertEqual(
-            list(edit(lines(), "str(int(math.pow(10, len(_))))")),
-            ["1000\n", "1000\n", "1000000"],
-            msg="modules should be auto-imported",
-        )
+        with tempfile.NamedTemporaryFile(delete=True) as process_line_file:
+            self.assertEqual(
+                list(edit(lines(), "_[-1]", process_line_file)),
+                ["0\n", "r\n", "r"],
+                msg="str expression must edit the lines",
+            )
+            self.assertEqual(
+                list(edit(lines(), 're.sub(r"\d", "X", _)', process_line_file)),
+                ["fXX\n", "bar\n", "fXXbar"],
+                msg="re should be supported out-of-the-box",
+            )
+            self.assertEqual(
+                list(edit(lines(), '"0" in _', process_line_file)),
+                ["f00\n", "f00bar"],
+                msg="bool expression must filter the lines",
+            )
+            self.assertEqual(
+                list(edit(lines(), "len(_) == 3", process_line_file)),
+                ["f00\n", "bar\n"],
+                msg="_ must exclude linesep",
+            )
+            self.assertEqual(
+                list(edit(lines(), "str(int(math.pow(10, len(_))))", process_line_file)),
+                ["1000\n", "1000\n", "1000000"],
+                msg="modules should be auto-imported",
+            )