Refactoring to improve API and allow configless .NET Core loading (#31)

- Drop `Runtime` wrapper in favour of an abstract base class
- Use `pathlib` throughout and allow it as parameter
- Expose a `shutdown` method (that is a noop in most cases, but at least
  it exists)
- Add functions to find all installed .NET Core runtimes
- If no runtime configuration is given, generate a simple one in a
  temporary directory

Author: filmor

clr_loader/__init__.py

@@ -1,56 +1,106 @@
+from pathlib import Path
+from tempfile import TemporaryDirectory
 from typing import Dict, Optional, Sequence
 
-from .wrappers import Runtime
-from .util.find import find_libmono, find_dotnet_root
+from .types import Assembly, Runtime, RuntimeInfo
+from .util import StrOrPath
+from .util.find import find_dotnet_root, find_libmono, find_runtimes
+from .util.runtime_spec import DotnetCoreRuntimeSpec
 
-__all__ = ["get_mono", "get_netfx", "get_coreclr"]
+__all__ = [
+    "get_mono",
+    "get_netfx",
+    "get_coreclr",
+    "find_dotnet_root",
+    "find_libmono",
+    "find_runtimes",
+    "Runtime",
+    "Assembly",
+    "RuntimeInfo",
+]
 
 
 def get_mono(
+    *,
     domain: Optional[str] = None,
-    config_file: Optional[str] = None,
-    global_config_file: Optional[str] = None,
-    libmono: Optional[str] = None,
+    config_file: Optional[StrOrPath] = None,
+    global_config_file: Optional[StrOrPath] = None,
+    libmono: Optional[StrOrPath] = None,
     sgen: bool = True,
     debug: bool = False,
     jit_options: Optional[Sequence[str]] = None,
 ) -> Runtime:
     from .mono import Mono
 
+    libmono = _maybe_path(libmono)
     if libmono is None:
         libmono = find_libmono(sgen)
 
     impl = Mono(
         domain=domain,
         debug=debug,
         jit_options=jit_options,
-        config_file=config_file,
-        global_config_file=global_config_file,
+        config_file=_maybe_path(config_file),
+        global_config_file=_maybe_path(global_config_file),
         libmono=libmono,
     )
-    return Runtime(impl)
+    return impl
 
 
 def get_coreclr(
-    runtime_config: str,
-    dotnet_root: Optional[str] = None,
+    *,
+    runtime_config: Optional[StrOrPath] = None,
+    dotnet_root: Optional[StrOrPath] = None,
     properties: Optional[Dict[str, str]] = None,
+    runtime_spec: Optional[DotnetCoreRuntimeSpec] = None,
 ) -> Runtime:
     from .hostfxr import DotnetCoreRuntime
 
+    dotnet_root = _maybe_path(dotnet_root)
     if dotnet_root is None:
         dotnet_root = find_dotnet_root()
 
+    temp_dir = None
+    runtime_config = _maybe_path(runtime_config)
+    if runtime_config is None:
+        if runtime_spec is None:
+            candidates = [
+                rt for rt in find_runtimes() if rt.name == "Microsoft.NETCore.App"
+            ]
+            candidates.sort(key=lambda spec: spec.version, reverse=True)
+            if not candidates:
+                raise RuntimeError("Failed to find a suitable runtime")
+
+            runtime_spec = candidates[0]
+
+        temp_dir = TemporaryDirectory()
+        runtime_config = Path(temp_dir.name) / "runtimeconfig.json"
+
+        with open(runtime_config, "w") as f:
+            runtime_spec.write_config(f)
+
     impl = DotnetCoreRuntime(runtime_config=runtime_config, dotnet_root=dotnet_root)
     if properties:
         for key, value in properties.items():
             impl[key] = value
 
-    return Runtime(impl)
+    if temp_dir:
+        temp_dir.cleanup()
 
+    return impl
 
-def get_netfx(name: Optional[str] = None, config_file: Optional[str] = None) -> Runtime:
+
+def get_netfx(
+    *, name: Optional[str] = None, config_file: Optional[StrOrPath] = None
+) -> Runtime:
     from .netfx import NetFx
 
-    impl = NetFx(name=name, config_file=config_file)
-    return Runtime(impl)
+    impl = NetFx(name=name, config_file=_maybe_path(config_file))
+    return impl
+
+
+def _maybe_path(p: Optional[StrOrPath]) -> Optional[Path]:
+    if p is None:
+        return None
+    else:
+        return Path(p)

clr_loader/ffi/__init__.py

@@ -1,6 +1,5 @@
-import glob
-import os
 import sys
+from pathlib import Path
 from typing import Optional
 
 import cffi  # type: ignore
@@ -9,46 +8,51 @@
 
 __all__ = ["ffi", "load_hostfxr", "load_mono", "load_netfx"]
 
-ffi = cffi.FFI()
+ffi = cffi.FFI()  # type: ignore
 
 for cdef in hostfxr.cdef + mono.cdef + netfx.cdef:
     ffi.cdef(cdef)
 
 
-def load_hostfxr(dotnet_root: str):
+def load_hostfxr(dotnet_root: Path):
     hostfxr_name = _get_dll_name("hostfxr")
-    hostfxr_path = os.path.join(dotnet_root, "host", "fxr", "?.*", hostfxr_name)
 
-    for hostfxr_path in reversed(sorted(glob.glob(hostfxr_path))):
+    # This will fail as soon as .NET hits version 10, but hopefully by then
+    # we'll have a more robust way of finding the libhostfxr
+    hostfxr_path = dotnet_root / "host" / "fxr"
+    hostfxr_paths = hostfxr_path.glob(f"?.*/{hostfxr_name}")
+
+    for hostfxr_path in reversed(sorted(hostfxr_paths)):
         try:
-            return ffi.dlopen(hostfxr_path)
+            return ffi.dlopen(str(hostfxr_path))
         except Exception:
             pass
 
     raise RuntimeError(f"Could not find a suitable hostfxr library in {dotnet_root}")
 
 
-def load_mono(path: Optional[str] = None):
+def load_mono(path: Optional[Path] = None):
     # Preload C++ standard library, Mono needs that and doesn't properly link against it
-    if sys.platform.startswith("linux"):
+    if sys.platform == "linux":
         ffi.dlopen("stdc++", ffi.RTLD_GLOBAL)
 
-    return ffi.dlopen(path, ffi.RTLD_GLOBAL)
+    path_str = str(path) if path else None
+    return ffi.dlopen(path_str, ffi.RTLD_GLOBAL)
 
 
 def load_netfx():
     if sys.platform != "win32":
         raise RuntimeError(".NET Framework is only supported on Windows")
 
-    dirname = os.path.join(os.path.dirname(__file__), "dlls")
+    dirname = Path(__file__).parent / "dlls"
     if sys.maxsize > 2**32:
         arch = "amd64"
     else:
         arch = "x86"
 
-    path = os.path.join(dirname, arch, "ClrLoader.dll")
+    path = dirname / arch / "ClrLoader.dll"
 
-    return ffi.dlopen(path)
+    return ffi.dlopen(str(path))
 
 
 def _get_dll_name(name: str) -> str:

clr_loader/ffi/hostfxr.py

@@ -2,6 +2,7 @@
 
 import sys
 
+
 cdef = []
 
 if sys.platform == "win32":

clr_loader/hostfxr.py

@@ -1,29 +1,48 @@
-import os
 import sys
+from pathlib import Path
+from typing import Generator, Tuple
 
 from .ffi import ffi, load_hostfxr
-from .util import check_result, find_dotnet_root
+from .types import Runtime, RuntimeInfo, StrOrPath
+from .util import check_result
 
 __all__ = ["DotnetCoreRuntime"]
 
+_IS_SHUTDOWN = False
 
-class DotnetCoreRuntime:
-    def __init__(self, runtime_config: str, dotnet_root: str):
-        self._dotnet_root = dotnet_root or find_dotnet_root()
+
+class DotnetCoreRuntime(Runtime):
+    def __init__(self, runtime_config: Path, dotnet_root: Path, **params: str):
+        if _IS_SHUTDOWN:
+            raise RuntimeError("Runtime can not be reinitialized")
+
+        self._dotnet_root = Path(dotnet_root)
         self._dll = load_hostfxr(self._dotnet_root)
-        self._is_finalized = False
+        self._is_initialized = False
         self._handle = _get_handle(self._dll, self._dotnet_root, runtime_config)
         self._load_func = _get_load_func(self._dll, self._handle)
 
+        for key, value in params.items():
+            self[key] = value
+
+        # TODO: Get version
+        self._version = "<undefined>"
+
     @property
-    def dotnet_root(self) -> str:
+    def dotnet_root(self) -> Path:
         return self._dotnet_root
 
     @property
-    def is_finalized(self) -> bool:
-        return self._is_finalized
+    def is_initialized(self) -> bool:
+        return self._is_initialized
+
+    @property
+    def is_shutdown(self) -> bool:
+        return _IS_SHUTDOWN
 
     def __getitem__(self, key: str) -> str:
+        if self.is_shutdown:
+            raise RuntimeError("Runtime is shut down")
         buf = ffi.new("char_t**")
         res = self._dll.hostfxr_get_runtime_property_value(
             self._handle, encode(key), buf
@@ -34,15 +53,17 @@ def __getitem__(self, key: str) -> str:
         return decode(buf[0])
 
     def __setitem__(self, key: str, value: str) -> None:
-        if self.is_finalized:
-            raise RuntimeError("Already finalized")
+        if self.is_initialized:
+            raise RuntimeError("Already initialized")
 
         res = self._dll.hostfxr_set_runtime_property_value(
             self._handle, encode(key), encode(value)
         )
         check_result(res)
 
-    def __iter__(self):
+    def __iter__(self) -> Generator[Tuple[str, str], None, None]:
+        if self.is_shutdown:
+            raise RuntimeError("Runtime is shut down")
         max_size = 100
         size_ptr = ffi.new("size_t*")
         size_ptr[0] = max_size
@@ -51,25 +72,26 @@ def __iter__(self):
         values_ptr = ffi.new("char_t*[]", max_size)
 
         res = self._dll.hostfxr_get_runtime_properties(
-            self._dll._handle, size_ptr, keys_ptr, values_ptr
+            self._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: str, typename: str, function: str):
+    def get_callable(self, assembly_path: StrOrPath, typename: str, function: str):
         # TODO: Maybe use coreclr_get_delegate as well, supported with newer API
         # versions of hostfxr
-        self._is_finalized = True
+        self._is_initialized = True
 
         # Append assembly name to typename
-        assembly_name, _ = os.path.splitext(os.path.basename(assembly_path))
+        assembly_path = Path(assembly_path)
+        assembly_name = assembly_path.stem
         typename = f"{typename}, {assembly_name}"
 
         delegate_ptr = ffi.new("void**")
         res = self._load_func(
-            encode(assembly_path),
+            encode(str(assembly_path)),
             encode(typename),
             encode(function),
             ffi.NULL,
@@ -79,27 +101,39 @@ def get_callable(self, assembly_path: str, typename: str, function: str):
         check_result(res)
         return ffi.cast("component_entry_point_fn", delegate_ptr[0])
 
+    def _check_initialized(self) -> None:
+        if self._handle is None:
+            raise RuntimeError("Runtime is shut down")
+        elif not self._is_initialized:
+            raise RuntimeError("Runtime is not initialized")
+
     def shutdown(self) -> None:
         if self._handle is not None:
             self._dll.hostfxr_close(self._handle)
             self._handle = None
 
-    def __del__(self):
-        self.shutdown()
+    def info(self):
+        return RuntimeInfo(
+            kind="CoreCLR",
+            version=self._version,
+            initialized=self._handle is not None,
+            shutdown=self._handle is None,
+            properties=dict(self) if not _IS_SHUTDOWN else {},
+        )
 
 
-def _get_handle(dll, dotnet_root: str, runtime_config: str):
+def _get_handle(dll, dotnet_root: StrOrPath, runtime_config: StrOrPath):
     params = ffi.new("hostfxr_initialize_parameters*")
     params.size = ffi.sizeof("hostfxr_initialize_parameters")
     # params.host_path = ffi.new("char_t[]", encode(sys.executable))
     params.host_path = ffi.NULL
-    dotnet_root_p = ffi.new("char_t[]", encode(dotnet_root))
+    dotnet_root_p = ffi.new("char_t[]", encode(str(Path(dotnet_root))))
     params.dotnet_root = dotnet_root_p
 
     handle_ptr = ffi.new("hostfxr_handle*")
 
     res = dll.hostfxr_initialize_for_runtime_config(
-        encode(runtime_config), params, handle_ptr
+        encode(str(Path(runtime_config))), params, handle_ptr
     )
     check_result(res)
 
@@ -119,16 +153,16 @@ def _get_load_func(dll, handle):
 
 if sys.platform == "win32":
 
-    def encode(string):
+    def encode(string: str):
         return string
 
-    def decode(char_ptr):
+    def decode(char_ptr) -> str:
         return ffi.string(char_ptr)
 
 else:
 
-    def encode(string):
+    def encode(string: str):
         return string.encode("utf8")
 
-    def decode(char_ptr):
+    def decode(char_ptr) -> str:
         return ffi.string(char_ptr).decode("utf8")