Add openeo_driver.config.load.exec_py_file

related to Open-EO/openeo-geopyspark-driver#936

Author: soxofaan

CHANGELOG.md

@@ -21,6 +21,10 @@ and start a new "In Progress" section above it.
 
 ## In progress
 
+# 0.118.0
+
+- Add `openeo_driver.config.load.exec_py_file` (related to [Open-EO/openeo-geopyspark-driver#936)](https://github.com/Open-EO/openeo-geopyspark-driver/issues/936))
+
 # 0.116.0
 - Propagate alternate `href`s of job result assets ([Open-EO/openeo-geopyspark-driver#883](https://github.com/Open-EO/openeo-geopyspark-driver/issues/883))
 - Ensure that a top level UDF can return a DriverVectorCube. Previously it only returned a JSONResult ([#323](https://github.com/Open-EO/openeo-python-driver/issues/323))

openeo_driver/_version.py

@@ -1 +1 @@
-__version__ = "0.117.0a2"
+__version__ = "0.118.0a1"

openeo_driver/config/load.py

@@ -20,6 +20,20 @@
 _log = logging.getLogger(__name__)
 
 
+def exec_py_file(path: Union[str, Path]) -> dict:
+    """Compile and execute given python file, and get the globals from doing that."""
+    # Based on flask's Config.from_pyfile
+    path = Path(path)
+    with path.open(mode="r", encoding="utf-8") as f:
+        src = f.read()
+
+    ast = compile(src, path, "exec")
+    globals = {"__file__": str(path)}
+    exec(ast, globals)
+
+    return globals
+
+
 def load_from_py_file(
     path: Union[str, Path],
     variable: str = "config",
@@ -29,11 +43,7 @@ def load_from_py_file(
     path = Path(path)
     _log.debug(f"Loading configuration from Python file {path!r} (variable {variable!r})")
 
-    # Based on flask's Config.from_pyfile
-    with path.open(mode="rb") as f:
-        code = compile(f.read(), path, "exec")
-    globals = {"__file__": str(path)}
-    exec(code, globals)
+    globals = exec_py_file(path)
 
     try:
         config = globals[variable]

tests/test_config.py

@@ -3,10 +3,11 @@
 import re
 import textwrap
 from pathlib import Path
-from typing import List, Optional
+from typing import List, Optional, Callable
 
 import attrs.exceptions
 import pytest
+import dirty_equals
 
 import openeo_driver.config.load
 from openeo_driver.config import (
@@ -18,7 +19,7 @@
 )
 from openeo_driver.config.config import check_config_definition
 from openeo_driver.config.env import default_from_env_as_list
-from openeo_driver.config.load import load_from_py_file
+from openeo_driver.config.load import load_from_py_file, exec_py_file
 import openeo_driver.config.env
 from .conftest import enhanced_logging
 
@@ -31,6 +32,28 @@ def test_config_immutable():
     assert conf.id == "dontchangeme!"
 
 
+@pytest.mark.parametrize("path_type", [str, Path])
+def test_exec_py_file(tmp_path, path_type):
+    path = tmp_path / "custom.py"
+
+    content = f"""
+        def pi():
+            return 3.14159265359
+
+        x = pi()
+    """
+    path.write_text(textwrap.dedent(content))
+
+    globals = exec_py_file(path_type(path))
+    assert globals == dirty_equals.IsPartialDict(
+        {
+            "__file__": str(path),
+            "pi": dirty_equals.IsInstance(Callable),
+            "x": 3.14159265359,
+        }
+    )
+
+
 @pytest.mark.parametrize("path_type", [str, Path])
 def test_load_from_py_file_default(tmp_path, path_type):
     path = tmp_path / "myconfig.py"