Merge pull request #5 from MAK-Relic-Tool/fs.opener-support

fs.opener support

Author: ModernMAK

mypy.ini

@@ -23,6 +23,9 @@ strict_concatenate = True
 [mypy-serialization_tools.*]
 ignore_missing_imports = True
 
+[mypy-pkg_resources.*]
+ignore_missing_imports = True
+
 [mypy-tests.*]
 ignore_missing_imports = True
 ignore_errors = True

setup.cfg

@@ -27,5 +27,10 @@ install_requires =
     relic-tool-core
     fs
 
+[options.entry_points]
+fs.opener =
+    sga = relic.sga.core.filesystem:EssenceFSOpener
+
+
 [options.packages.find]
 where = src

src/relic/sga/core/__init__.py

@@ -3,4 +3,4 @@
 """
 from relic.sga.core.definitions import Version, MagicWord, StorageType, VerificationType
 
-__version__ = "1.0.0"
+__version__ = "1.1.0"

src/relic/sga/core/definitions.py

@@ -7,12 +7,19 @@
 from enum import Enum
 from typing import ClassVar, BinaryIO
 
+from relic.core.errors import MismatchError
 from serialization_tools.magic import MagicWordIO
 from serialization_tools.structx import Struct
 
 MagicWord = MagicWordIO(Struct("< 8s"), "_ARCHIVE".encode("ascii"))
 
 
+def _validate_magic_word(self: MagicWordIO, stream: BinaryIO, advance: bool) -> None:
+    magic = self.read_magic_word(stream, advance)
+    if magic != self.word:
+        raise MismatchError("MagicWord", magic, self.word)
+
+
 @dataclass
 class Version:
     """A Version object.

src/relic/sga/core/filesystem.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import abc
+import os
+from os.path import expanduser
 from typing import (
     Optional,
     Dict,
@@ -10,53 +13,128 @@
     Mapping,
     cast,
     Protocol,
+    TypeVar,
+    Generic,
+    runtime_checkable,
 )
 
+import fs.opener.errors
+import pkg_resources
 from fs import ResourceType, errors
 from fs.base import FS
 from fs.info import Info
 from fs.memoryfs import MemoryFS, _DirEntry, _MemoryFile
 from fs.multifs import MultiFS
+from fs.opener import Opener, registry as fs_registry
+from fs.opener.parse import ParseResult
 from fs.path import split
 
-from relic.sga.core.definitions import Version, MagicWord
+from relic.sga.core.definitions import Version, MagicWord, _validate_magic_word
 from relic.sga.core.errors import VersionNotSupportedError
 
 ESSENCE_NAMESPACE = "essence"
 
+TKey = TypeVar("TKey")
+TValue = TypeVar("TValue")
 
+
+class EntrypointRegistry(Generic[TKey, TValue]):
+    def __init__(self, entry_point_path: str, autoload: bool = False):
+        self._entry_point_path = entry_point_path
+        self._mapping: Dict[TKey, TValue] = {}
+        self._autoload = autoload
+
+    def register(self, key: TKey, value: TValue) -> None:
+        self._mapping[key] = value
+
+    @abc.abstractmethod
+    def auto_register(self, value: TValue) -> None:
+        raise NotImplementedError
+
+    def get(self, key: TKey, default: Optional[TValue] = None) -> Optional[TValue]:
+        if key in self._mapping:
+            return self._mapping[key]
+
+        if self._autoload:
+            try:
+                entry_point = next(
+                    pkg_resources.iter_entry_points(
+                        self._entry_point_path, self._key2entry_point_path(key)
+                    )
+                )
+            except StopIteration:
+                entry_point = None
+            if entry_point is None:
+                return default
+            self._auto_register_entrypoint(entry_point)
+            if key not in self._mapping:
+                raise NotImplementedError  # TODO specify autoload failed to load in a usable value
+            return self._mapping[key]
+        return default
+
+    @abc.abstractmethod
+    def _key2entry_point_path(self, key: TKey) -> str:
+        raise NotImplementedError
+
+    def _auto_register_entrypoint(self, entry_point: Any) -> None:
+        try:
+            entry_point_result = entry_point.load()
+        except:  # Wrap in exception
+            raise
+        return self._register_entrypoint(entry_point_result)
+
+    @abc.abstractmethod
+    def _register_entrypoint(self, entry_point_result: Any) -> None:
+        raise NotImplementedError
+
+
+@runtime_checkable
 class EssenceFSHandler(Protocol):
+    version: Version
+
     def read(self, stream: BinaryIO) -> EssenceFS:
         raise NotImplementedError
 
     def write(self, stream: BinaryIO, essence_fs: EssenceFS) -> int:
         raise NotImplementedError
 
 
-class EssenceFSFactory:
-    def __init__(self) -> None:
-        self.handler_map: Dict[Version, EssenceFSHandler] = {}
+class EssenceFSFactory(EntrypointRegistry[Version, EssenceFSHandler]):
+    def _key2entry_point_path(self, key: Version) -> str:
+        return f"v{key.major}.{key.minor}"
 
-    def register_handler(self, version: Version, handler: EssenceFSHandler) -> None:
-        if version is None:
-            raise ValueError
-        if handler is None:
-            raise ValueError
-        #     self.default_handler = handler
-        # else:
-        self.handler_map[version] = handler
+    def _register_entrypoint(self, entry_point_result: Any) -> None:
+        if isinstance(entry_point_result, EssenceFSHandler):
+            self.auto_register(entry_point_result)
+        elif isinstance(entry_point_result, (list, tuple, Collection)):
+            version, handler = entry_point_result
+            if not isinstance(handler, EssenceFSHandler):
+                handler = handler()
+            self.register(version, handler)
+        else:
+            # Callable; register nested result
+            self._register_entrypoint(entry_point_result())
+
+    def auto_register(self, value: EssenceFSHandler) -> None:
+        self.register(value.version, value)
+
+    def __init__(self, autoload: bool = True) -> None:
+        super().__init__("relic.sga.handler", autoload)
 
     @staticmethod
     def _read_magic_and_version(sga_stream: BinaryIO) -> Version:
-        sga_stream.seek(0)
-        MagicWord.read_magic_word(sga_stream)
-        return Version.unpack(sga_stream)
+        # sga_stream.seek(0)
+        jump_back = sga_stream.tell()
+        _validate_magic_word(MagicWord, sga_stream, advance=True)
+        version = Version.unpack(sga_stream)
+        sga_stream.seek(jump_back)
+        return version
 
     def _get_handler(self, version: Version) -> EssenceFSHandler:
-        handler = self.handler_map.get(version)
+        handler = self.get(version)
         if handler is None:
             # This may raise a 'false positive' if a Null handler is registered
-            raise VersionNotSupportedError(version, list(self.handler_map.keys()))
+            raise VersionNotSupportedError(version, list(self._mapping.keys()))
         return handler
 
     def _get_handler_from_stream(
@@ -87,6 +165,55 @@ def write(
         return handler.write(sga_stream, sga_fs)
 
 
+registry = EssenceFSFactory(True)
+
+
+# @fs_registry.install
+# Can't use decorator; it breaks subclassing for entrypoints
+class EssenceFSOpener(Opener):
+    def __init__(self, factory: Optional[EssenceFSFactory] = None):
+        if factory is None:
+            factory = registry
+        self.factory = factory
+
+    protocols = ["sga"]
+
+    def open_fs(
+        self,
+        fs_url: str,
+        parse_result: ParseResult,
+        writeable: bool,
+        create: bool,
+        cwd: str,
+    ) -> FS:
+        # All EssenceFS should be writable; so we can ignore that
+
+        # Resolve Path
+        if fs_url == "sga://":
+            if create:
+                return EssenceFS()
+            else:
+                raise fs.opener.errors.OpenerError(
+                    "No path was given and opener not marked for 'create'!"
+                )
+
+        _path = os.path.abspath(os.path.join(cwd, expanduser(parse_result.resource)))
+        path = os.path.normpath(_path)
+
+        # Create will always create a new EssenceFS if needed
+        try:
+            with open(path, "rb") as sga_file:
+                return self.factory.read(sga_file)
+        except FileNotFoundError as e:
+            if create:
+                return EssenceFS()
+            else:
+                raise
+
+
+fs_registry.install(EssenceFSOpener)
+
+
 class _EssenceFile(_MemoryFile):
     ...  # I plan on allowing lazy file loading from the archive; I'll likely need to implement this to do that
 
@@ -111,14 +238,30 @@ def to_info(self, namespaces=None):
 
 
 class _EssenceDriveFS(MemoryFS):
-    def __init__(self) -> None:
+    def __init__(self, alias: str) -> None:
         super().__init__()
+        self.alias = alias
 
     def _make_dir_entry(
         self, resource_type: ResourceType, name: str
     ) -> _EssenceDirEntry:
         return _EssenceDirEntry(resource_type, name)
 
+    def validatepath(self, path: str) -> str:
+        if ":" in path:
+            parts = path.split(":", 1)
+            if parts[0][0] == "/":
+                parts[0] = parts[0][1:]
+            if parts[0] != self.alias:
+                raise fs.errors.InvalidPath(
+                    path,
+                    f"Alias `{parts[0]}` does not math the Drive's Alias `{self.alias}`",
+                )
+            fixed_path = parts[1]
+        else:
+            fixed_path = path
+        return super().validatepath(fixed_path)
+
     def setinfo(self, path: str, info: Mapping[str, Mapping[str, object]]) -> None:
         _path = self.validatepath(path)
         with self._lock:
@@ -171,7 +314,7 @@ def getessence(self, path: str) -> Info:
         return self.getinfo(path, [ESSENCE_NAMESPACE])
 
     def create_drive(self, name: str) -> _EssenceDriveFS:
-        drive = _EssenceDriveFS()
+        drive = _EssenceDriveFS(name)
         self.add_fs(name, drive)
         return drive
 
@@ -227,4 +370,6 @@ def _delegate(self, path):
     "_EssenceDirEntry",
     "_EssenceDriveFS",
     "EssenceFS",
+    "registry",
+    "EssenceFSOpener",
 ]

src/relic/sga/core/serialization.py

@@ -23,7 +23,12 @@
 from serialization_tools.size import KiB, MiB
 from serialization_tools.structx import Struct
 
-from relic.sga.core.definitions import StorageType, Version, MagicWord
+from relic.sga.core.definitions import (
+    StorageType,
+    Version,
+    MagicWord,
+    _validate_magic_word,
+)
 from relic.sga.core.errors import (
     MD5MismatchError,
     VersionMismatchError,
@@ -747,7 +752,7 @@ def __init__(
 
     def read(self, stream: BinaryIO) -> EssenceFS:
         # Magic & Version; skippable so that we can check for a valid file and read the version elsewhere
-        MagicWord.assert_magic_word(stream, advance=True)
+        _validate_magic_word(MagicWord, stream, advance=True)
         stream_version = Version.unpack(stream)
         if stream_version != self.version:
             raise VersionMismatchError(stream_version, self.version)
@@ -776,7 +781,10 @@ def read(self, stream: BinaryIO) -> EssenceFS:
         )
         essence_fs = EssenceFS()
         assembler.assemble(essence_fs)
-        essence_info: Dict[str, object] = {"name": name}
+        essence_info: Dict[str, object] = {
+            "name": name,
+            "version": {"major": stream_version.major, "minor": stream_version.minor},
+        }
         if metadata is not None:
             essence_info.update(metadata)
 

tests/test_filesystem.py

@@ -1,5 +1,6 @@
 import unittest
 
+import fs
 from fs.test import FSTestCases
 
 from relic.sga.core.filesystem import EssenceFS, _EssenceDriveFS
@@ -8,10 +9,19 @@
 class TestEssenceFS(FSTestCases, unittest.TestCase):
     def make_fs(self):
         essence_fs = EssenceFS()
-        essence_fs.add_fs("data", _EssenceDriveFS(), True)
+        # EssenceFS shouldn't be writeable by default;
+        #   being an emulator for Window's hard drives.
+        #       With no 'drive' installed, there's nothing to write to!
+        essence_fs.add_fs("data", _EssenceDriveFS("data"), True)
         return essence_fs
 
 
 class TestEssenceDriveFS(FSTestCases, unittest.TestCase):
     def make_fs(self):
-        return _EssenceDriveFS()
+        return _EssenceDriveFS("")
+
+
+class TestOpener:
+    def test_open_fs(self):
+        with fs.open_fs("sga://", create=True) as sga:
+            pass

tests/test_regressions.py

@@ -39,6 +39,8 @@ def test_import_module(submodule: str):
     "_EssenceDirEntry",
     "_EssenceDriveFS",
     "EssenceFS",
+    "registry",
+    "EssenceFSOpener",
 ]
 protocols__all__ = ["T", "StreamSerializer"]
 serialization__all__ = [