Refactor HostFxr a bit to allow setting properties

Author: filmor

clr_loader/__init__.py

@@ -1,61 +1,10 @@
-from os.path import basename
-from typing import Optional, Any
-from .ffi import ffi
+from typing import Dict, Optional
 
+from .wrappers import Runtime
 
 __all__ = ["get_mono", "get_netfx", "get_coreclr"]
 
 
-RuntimeImpl = Any
-
-
-class ClrFunction:
-    def __init__(
-        self, runtime: RuntimeImpl, assembly: str, typename: str, func_name: str
-    ):
-        self._assembly = assembly
-        self._class = typename
-        self._name = func_name
-
-        self._callable = runtime.get_callable(assembly, typename, func_name)
-
-    def __call__(self, buffer: bytes) -> int:
-        buf_arr = ffi.from_buffer("char[]", buffer)
-        return self._callable(ffi.cast("void*", buf_arr), len(buf_arr))
-
-    def __repr__(self) -> str:
-        return f"<ClrFunction {self._class}.{self._name} in {basename(self._assembly)}>"
-
-
-class Assembly:
-    def __init__(self, runtime: RuntimeImpl, path: str):
-        self._runtime = runtime
-        self._path = path
-
-    def get_function(self, name: str, func: Optional[str] = None) -> ClrFunction:
-        if func is None:
-            name, func = name.rsplit(".", 1)
-
-        return ClrFunction(self._runtime, self._path, name, func)
-
-    def __getitem__(self, name: str) -> ClrFunction:
-        return self.get_function(name)
-
-    def __repr__(self) -> str:
-        return f"<Assembly {self._path} in {self._runtime}>"
-
-
-class Runtime:
-    def __init__(self, impl: RuntimeImpl):
-        self._impl = impl
-
-    def get_assembly(self, path: str) -> Assembly:
-        return Assembly(self._impl, path)
-
-    def __getitem__(self, path: str) -> Assembly:
-        return self.get_assembly(path)
-
-
 def get_mono(
     domain: Optional[str] = None,
     config_file: Optional[str] = None,
@@ -68,10 +17,17 @@ def get_mono(
     return Runtime(impl)
 
 
-def get_coreclr(runtime_config: str, dotnet_root: Optional[str] = None) -> Runtime:
-    from .hostfxr import HostFxr
+def get_coreclr(
+    runtime_config: str, dotnet_root: Optional[str] = None,
+    properties: Optional[Dict[str, str]] = None
+) -> Runtime:
+    from .hostfxr import DotnetCoreRuntime
+
+    impl = DotnetCoreRuntime(runtime_config=runtime_config, dotnet_root=dotnet_root)
+    if properties:
+        for key, value in properties:
+            impl[key] = value
 
-    impl = HostFxr(runtime_config=runtime_config, dotnet_root=dotnet_root)
     return Runtime(impl)
 
 

clr_loader/hostfxr.py

@@ -1,63 +1,69 @@
 import os
-import shutil
 import sys
+from typing import Optional
 
 from .ffi import ffi, load_hostfxr
-from .util import check_result
+from .util import check_result, find_dotnet_root
 
-__all__ = ["HostFxr"]
+__all__ = ["DotnetCoreRuntime"]
 
 
-class HostFxr:
-    # TODO: Allow generating runtime_config
-    def __init__(self, runtime_config, dotnet_root=None):
-        self._handle = None
+class DotnetCoreRuntime:
+    def __init__(self, runtime_config: str, dotnet_root: Optional[str] = None):
+        self._dotnet_root = dotnet_root or find_dotnet_root()
+        self._dll = load_hostfxr(self._dotnet_root)
+        self._is_finalized = False
+        self._handle = _get_handle(self._dll, self._dotnet_root, runtime_config)
+        self._load_func = _get_load_func(self._dll, self._handle)
 
-        if not dotnet_root:
-            dotnet_root = os.environ.get("DOTNET_ROOT", None)
+    @property
+    def dotnet_root(self):
+        return self._dotnet_root
 
-        if not dotnet_root and sys.platform == "win32":
-            # On Windows, the host library is stored separately from dotnet.exe for x86
-            if sys.maxsize > 2 ** 32:
-                possible_root = os.path.join(os.environ.get("ProgramFiles"), "dotnet")
-            else:
-                possible_root = os.path.join(
-                    os.environ.get("ProgramFiles(x86)"), "dotnet"
-                )
+    @property
+    def is_finalized(self):
+        return self._is_finalized
 
-            if os.path.isdir(possible_root):
-                dotnet_root = possible_root
+    def __getitem__(self, key: str) -> str:
+        buf = ffi.new("char_t**")
+        res = self._dll.hostfxr_get_runtime_property_value(
+            self._handle, encode(key), buf
+        )
+        if res != 0:
+            raise KeyError(key)
 
-        if not dotnet_root:
-            dotnet_path = shutil.which("dotnet")
-            if not dotnet_path:
-                raise RuntimeError("Can not determine dotnet root")
+        return decode(buf[0])
 
-            try:
-                # Pypy does not provide os.readlink right now
-                if hasattr(os, "readlink"):
-                    dotnet_tmp_path = os.readlink(dotnet_path)
-                else:
-                    dotnet_tmp_path = dotnet_path
+    def __setitem__(self, key: str, value: str) -> None:
+        if self.is_finalized:
+            raise RuntimeError("Already finalized")
 
-                if os.path.isabs(dotnet_tmp_path):
-                    dotnet_path = dotnet_tmp_path
-                else:
-                    dotnet_path = os.path.abspath(
-                        os.path.join(os.path.dirname(dotnet_path), dotnet_tmp_path)
-                    )
-            except OSError:
-                pass
+        res = self._dll.hostfxr_set_runtime_property_value(
+            self._handle, encode(key), encode(value)
+        )
+        check_result(res)
 
-            dotnet_root = os.path.dirname(dotnet_path)
+    def __iter__(self):
+        max_size = 100
+        size_ptr = ffi.new("size_t*")
+        size_ptr[0] = max_size
 
-        self._dll = load_hostfxr(dotnet_root)
-        self._dotnet_root = dotnet_root
+        keys_ptr = ffi.new("char_t*[]", max_size)
+        values_ptr = ffi.new("char_t*[]", max_size)
 
-        self._handle = _get_handle(self._dll, self._dotnet_root, runtime_config)
-        self._load_func = _get_load_func(self._dll, self._handle)
+        res = self._fxr._dll.hostfxr_get_runtime_properties(
+            self._fxr._handle, size_ptr, keys_ptr, values_ptr
+        )
+        check_result(res)
+
+        for i in range(size_ptr[0]):
+            yield (decode(keys_ptr[i]), decode(values_ptr[i]))
 
     def get_callable(self, assembly_path, typename, function):
+        # TODO: Maybe use coreclr_get_delegate as well, supported with newer API
+        # versions of hostfxr
+        self._is_finalized = True
+
         # Append assembly name to typename
         assembly_name, _ = os.path.splitext(os.path.basename(assembly_path))
         typename = f"{typename}, {assembly_name}"
@@ -82,47 +88,6 @@ def shutdown(self):
     def __del__(self):
         self.shutdown()
 
-    @property
-    def props(self):
-        return HostFxrProps(self)
-
-
-class HostFxrProps:
-    def __init__(self, fxr):
-        self._fxr = fxr
-
-    def __getitem__(self, key):
-        buf = ffi.new("char_t**")
-        res = self._fxr._dll.hostfxr_get_runtime_property_value(
-            self._fxr._handle, encode(key), buf
-        )
-        if res != 0:
-            raise KeyError(key)
-
-        return decode(buf[0])
-
-    def __setitem__(self, key, value):
-        res = self._fxr._dll.hostfxr_set_runtime_property_value(
-            self._fxr._handle, encode(key), encode(value)
-        )
-        check_result(res)
-
-    def __iter__(self):
-        max_size = 100
-        size_ptr = ffi.new("size_t*")
-        size_ptr[0] = max_size
-
-        keys_ptr = ffi.new("char_t*[]", max_size)
-        values_ptr = ffi.new("char_t*[]", max_size)
-
-        res = self._fxr._dll.hostfxr_get_runtime_properties(
-            self._fxr._handle, size_ptr, keys_ptr, values_ptr
-        )
-        check_result(res)
-
-        for i in range(size_ptr[0]):
-            yield (decode(keys_ptr[i]), decode(values_ptr[i]))
-
 
 def _get_handle(dll, dotnet_root, runtime_config):
     params = ffi.new("hostfxr_initialize_parameters*")

clr_loader/util/__init__.py

@@ -1,7 +1,10 @@
 from .clr_error import ClrError
 from .coreclr_errors import get_coreclr_error
+from .find import find_dotnet_root
 from .hostfxr_errors import get_hostfxr_error
 
+__all__ = ["check_result", "find_dotnet_root"]
+
 
 def check_result(err_code: int) -> None:
     """Check the error code of a .NET hosting API function and raise a

clr_loader/util/find.py

@@ -0,0 +1,44 @@
+import os
+import os.path
+import shutil
+import sys
+
+
+def find_dotnet_root() -> str:
+    dotnet_root = os.environ.get("DOTNET_ROOT", None)
+    if dotnet_root is not None:
+        return dotnet_root
+
+    if sys.platform == "win32":
+        # On Windows, the host library is stored separately from dotnet.exe for x86
+        if sys.maxsize > 2 ** 32:
+            prog_files = os.environ.get("ProgramFiles")
+        else:
+            prog_files = os.environ.get("ProgramFiles(x86)")
+
+        dotnet_root = os.path.join(prog_files, "dotnet")
+        if os.path.isdir(dotnet_root):
+            return dotnet_root
+
+    # Try to discover dotnet from PATH otherwise
+    dotnet_path = shutil.which("dotnet")
+    if not dotnet_path:
+        raise RuntimeError("Can not determine dotnet root")
+
+    try:
+        # Pypy does not provide os.readlink right now
+        if hasattr(os, "readlink"):
+            dotnet_tmp_path = os.readlink(dotnet_path)
+        else:
+            dotnet_tmp_path = dotnet_path
+
+        if os.path.isabs(dotnet_tmp_path):
+            dotnet_path = dotnet_tmp_path
+        else:
+            dotnet_path = os.path.abspath(
+                os.path.join(os.path.dirname(dotnet_path), dotnet_tmp_path)
+            )
+    except OSError:
+        pass
+
+    return os.path.dirname(dotnet_path)

clr_loader/wrappers.py

@@ -0,0 +1,52 @@
+from typing import Any, Optional
+from .ffi import ffi
+from os.path import basename
+
+RuntimeImpl = Any
+
+
+class ClrFunction:
+    def __init__(
+        self, runtime: RuntimeImpl, assembly: str, typename: str, func_name: str
+    ):
+        self._assembly = assembly
+        self._class = typename
+        self._name = func_name
+
+        self._callable = runtime.get_callable(assembly, typename, func_name)
+
+    def __call__(self, buffer: bytes) -> int:
+        buf_arr = ffi.from_buffer("char[]", buffer)
+        return self._callable(ffi.cast("void*", buf_arr), len(buf_arr))
+
+    def __repr__(self) -> str:
+        return f"<ClrFunction {self._class}.{self._name} in {basename(self._assembly)}>"
+
+
+class Assembly:
+    def __init__(self, runtime: RuntimeImpl, path: str):
+        self._runtime = runtime
+        self._path = path
+
+    def get_function(self, name: str, func: Optional[str] = None) -> ClrFunction:
+        if func is None:
+            name, func = name.rsplit(".", 1)
+
+        return ClrFunction(self._runtime, self._path, name, func)
+
+    def __getitem__(self, name: str) -> ClrFunction:
+        return self.get_function(name)
+
+    def __repr__(self) -> str:
+        return f"<Assembly {self._path} in {self._runtime}>"
+
+
+class Runtime:
+    def __init__(self, impl: RuntimeImpl):
+        self._impl = impl
+
+    def get_assembly(self, path: str) -> Assembly:
+        return Assembly(self._impl, path)
+
+    def __getitem__(self, path: str) -> Assembly:
+        return self.get_assembly(path)