From d594085dff626d3e067557c95bbfe937e1267744 Mon Sep 17 00:00:00 2001 From: Abraham Murciano Date: Sun, 21 Aug 2022 00:24:39 +0300 Subject: [PATCH] metadata serialization --- condax/condax/exceptions.py | 7 ++ condax/condax/links.py | 4 +- condax/condax/metadata.py | 126 +++++++++++++++++++++++++++--------- condax/paths.py | 1 - 4 files changed, 104 insertions(+), 34 deletions(-) diff --git a/condax/condax/exceptions.py b/condax/condax/exceptions.py index 6f7d341..b15fa86 100644 --- a/condax/condax/exceptions.py +++ b/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}" + ) diff --git a/condax/condax/links.py b/condax/condax/links.py index 3aad779..9d81649 100644 --- a/condax/condax/links.py +++ b/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 diff --git a/condax/condax/metadata.py b/condax/condax/metadata.py index b41701f..a6470f0 100644 --- a/condax/condax/metadata.py +++ b/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,29 +18,72 @@ 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. @@ -46,25 +91,46 @@ class CondaxMetaData: 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 diff --git a/condax/paths.py b/condax/paths.py index 47f15a1..3ee1051 100644 --- a/condax/paths.py +++ b/condax/paths.py @@ -1,5 +1,4 @@ import logging -import sys from pathlib import Path from typing import Union