metadata serialization

condax/condax/exceptions.py

@@ -16,3 +16,10 @@ def __init__(self, location: Path, msg: str = ""):
             102,
             f"{location} exists, is not empty, and is not a conda environment. {msg}",
         )
+
+
+class BadMetadataError(CondaxError):
+    def __init__(self, metadata_path: Path, msg: str):
+        super().__init__(
+            103, f"Error loading condax metadata at {metadata_path}: {msg}"
+        )

condax/condax/links.py

@@ -64,8 +64,8 @@ def create_link(env: Path, exe: Path, location: Path, is_forcing: bool = False)
 
     script_path = location / _get_wrapper_name(exe.name)
     if script_path.exists() and not is_forcing:
-        answer = input(f"{exe.name} already exists. Overwrite? (y/N) ").strip().lower()
-        if answer not in ("y", "yes"):
+        answer = input(f"{exe.name} already exists in {location}. Overwrite? (y/N) ")
+        if answer.strip().lower() not in ("y", "yes"):
             logger.warning(f"Skipped creating entrypoint: {exe.name}")
             return False
 

condax/condax/metadata.py

@@ -1,9 +1,11 @@
-from dataclasses import dataclass
+from abc import ABC, abstractmethod
 import json
 from pathlib import Path
-from typing import Iterable, List, Optional
+from typing import Any, Dict, Iterable, Optional, Type, TypeVar
 
 from condax.conda import env_info
+from condax.condax.exceptions import BadMetadataError
+from condax.utils import FullPath
 
 
 def create_metadata(env: Path, package: str, executables: Iterable[Path]):
@@ -16,55 +18,119 @@ def create_metadata(env: Path, package: str, executables: Iterable[Path]):
     meta.save()
 
 
-class _PackageBase:
-    def __init__(self, name: str, apps: List[str], include_apps: bool):
+S = TypeVar("S", bound="Serializable")
+
+
+class Serializable(ABC):
+    @classmethod
+    @abstractmethod
+    def deserialize(cls: Type[S], serialized: Dict[str, Any]) -> S:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def serialize(self) -> Dict[str, Any]:
+        raise NotImplementedError()
+
+
+class _PackageBase(Serializable):
+    def __init__(self, name: str, apps: Iterable[str], include_apps: bool):
         self.name = name
-        self.apps = apps
+        self.apps = set(apps)
         self.include_apps = include_apps
 
     def __lt__(self, other):
         return self.name < other.name
 
+    def serialize(self) -> Dict[str, Any]:
+        return {
+            "name": self.name,
+            "apps": list(self.apps),
+            "include_apps": self.include_apps,
+        }
+
+    @classmethod
+    def deserialize(cls, serialized: Dict[str, Any]):
+        assert isinstance(serialized, dict)
+        assert isinstance(serialized["name"], str)
+        assert isinstance(serialized["apps"], list)
+        assert all(isinstance(app, str) for app in serialized["apps"])
+        assert isinstance(serialized["include_apps"], bool)
+        serialized.update(apps=set(serialized["apps"]))
+        return cls(**serialized)
+
 
-@dataclass
 class MainPackage(_PackageBase):
-    name: str
-    prefix: Path
-    apps: List[str]
-    include_apps: bool = True
+    def __init__(
+        self, name: str, prefix: Path, apps: Iterable[str], include_apps: bool = True
+    ):
+        super().__init__(name, apps, include_apps)
+        self.prefix = prefix
+
+    def serialize(self) -> Dict[str, Any]:
+        return {
+            **super().serialize(),
+            "prefix": str(self.prefix),
+        }
+
+    @classmethod
+    def deserialize(cls, serialized: Dict[str, Any]):
+        assert isinstance(serialized["prefix"], str)
+        serialized.update(prefix=FullPath(serialized["prefix"]))
+        return super().deserialize(serialized)
 
 
 class InjectedPackage(_PackageBase):
     pass
 
 
-class CondaxMetaData:
+class CondaxMetaData(Serializable):
     """
     Handle metadata information written in `condax_metadata.json`
     placed in each environment.
     """
 
     metadata_file = "condax_metadata.json"
 
-    def __init__(self, main: MainPackage, injected: Iterable[InjectedPackage] = ()):
-        self.main_package = main
-        self.injected_packages = tuple(sorted(injected))
+    def __init__(
+        self,
+        main_package: MainPackage,
+        injected_packages: Iterable[InjectedPackage] = (),
+    ):
+        self.main_package = main_package
+        self.injected_packages = {pkg.name: pkg for pkg in injected_packages}
 
     def inject(self, package: InjectedPackage):
-        self.injected_packages = tuple(sorted(set(self.injected_packages) | {package}))
+        self.injected_packages[package.name] = package
 
     def uninject(self, name: str):
-        self.injected_packages = tuple(
-            p for p in self.injected_packages if p.name != name
+        self.injected_packages.pop(name, None)
+
+    def serialize(self) -> Dict[str, Any]:
+        return {
+            "main_package": self.main_package.serialize(),
+            "injected_packages": [
+                pkg.serialize() for pkg in self.injected_packages.values()
+            ],
+        }
+
+    @classmethod
+    def deserialize(cls, serialized: Dict[str, Any]):
+        assert isinstance(serialized, dict)
+        assert isinstance(serialized["main_package"], dict)
+        assert isinstance(serialized["injected_packages"], list)
+        serialized.update(
+            main_package=MainPackage.deserialize(serialized["main_package"]),
+            injected_packages=[
+                InjectedPackage.deserialize(pkg)
+                for pkg in serialized["injected_packages"]
+            ],
         )
-
-    def to_json(self) -> str:
-        return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)
+        return cls(**serialized)
 
     def save(self) -> None:
-        p = self.main_package.prefix / self.metadata_file
-        with open(p, "w") as fo:
-            fo.write(self.to_json())
+        metadata_path = self.main_package.prefix / self.metadata_file
+        with metadata_path.open("w") as f:
+            json.dump(self.serialize(), f, indent=4)
 
 
 def load(prefix: Path) -> Optional[CondaxMetaData]:
@@ -74,12 +140,10 @@ def load(prefix: Path) -> Optional[CondaxMetaData]:
 
     with open(p) as f:
         d = json.load(f)
-        if not d:
-            raise ValueError(f"Failed to read the metadata from {p}")
-    return _from_dict(d)
-
 
-def _from_dict(d: dict) -> CondaxMetaData:
-    main = MainPackage(**d["main_package"])
-    injected = [InjectedPackage(**p) for p in d["injected_packages"]]
-    return CondaxMetaData(main, injected)
+    try:
+        return CondaxMetaData.deserialize(d)
+    except AssertionError as e:
+        raise BadMetadataError(p, f"A value is of the wrong type. {e}") from e
+    except KeyError as e:
+        raise BadMetadataError(p, f"Key {e} is missing.") from e

condax/paths.py

@@ -1,5 +1,4 @@
 import logging
-import sys
 from pathlib import Path
 from typing import Union