Merge branch 'mr/switch_qualifiers_to_dict' into 'master'

Start to switch to qualifier dicts internally

See merge request it/e3-core!26

src/e3/anod/__init__.py

@@ -1 +1,51 @@
 from __future__ import annotations
+
+# The following two functions are used during transition of qualifiers
+# from str representation to dict by default
+
+
+def qualifier_dict_to_str(qual: dict) -> str:
+    """Serialize a dict qualifier into a str.
+
+    :param qual: dict qualifier
+    :return: a deterministic str that represent the dict qualifier
+    """
+    tmp = []
+    for k, v in qual.items():
+        if isinstance(v, str):
+            if v:
+                tmp.append(f"{k}={v}")
+            else:
+                # Empty string is used also for tag value. Should be removed
+                # once switch to dict is complete
+                tmp.append(f"{k}")
+        elif isinstance(v, bool):
+            if v:
+                tmp.append(k)
+        elif v:
+            tmp.append(";".join(sorted(v)))
+
+    return ",".join(sorted(tmp))
+
+
+def qualifier_str_to_dict(qual: str | None) -> dict:
+    """Parse a str qualifier into a dict.
+
+    :param qual: a string representing a qualifier
+    :return: a dict qualifier
+    """
+    if not qual:
+        return {}
+
+    result: dict[str, str | bool] = {}
+
+    for key, sep, value in (item.partition("=") for item in qual.split(",")):
+        if sep == "=":
+            if value:
+                result[key] = value
+            else:
+                result[key] = ""
+        else:
+            # Replace by bool once switch to dict is complete
+            result[key] = ""
+    return result

src/e3/anod/context.py

@@ -2,6 +2,7 @@
 
 from typing import TYPE_CHECKING
 import e3.log
+from e3.anod import qualifier_str_to_dict, qualifier_dict_to_str
 from e3.anod.action import (
     Build,
     BuildOrDownload,
@@ -32,7 +33,7 @@
 from e3.error import E3Error
 
 if TYPE_CHECKING:
-    from typing import cast, NoReturn, Optional, Tuple
+    from typing import cast, NoReturn, Optional, Tuple, Iterable
     from collections.abc import Callable
     from e3.anod.action import Action
     from e3.anod.package import SourceBuilder
@@ -130,7 +131,7 @@ def load(
         self,
         name: str,
         env: BaseEnv | None,
-        qualifier: str | None,
+        qualifier: str | dict[str, str | bool | Iterable[str]] | None,
         kind: PRIMITIVE,
         sandbox: SandBox | None = None,
         source_name: str | None = None,
@@ -150,7 +151,11 @@ def load(
             env = self.default_env
 
         # Key used for the spec instance cache
-        key = (name, env.build, env.host, env.target, qualifier, kind, source_name)
+        if isinstance(qualifier, str) or qualifier is None:
+            qualifier_key = qualifier
+        else:
+            qualifier_key = qualifier_dict_to_str(qualifier)
+        key = (name, env.build, env.host, env.target, qualifier_key, kind, source_name)
 
         if key not in self.cache:
             # Spec is not in cache so create a new instance
@@ -283,6 +288,14 @@ def add_plan_action(
         elif TYPE_CHECKING:
             primitive = cast(PRIMITIVE, primitive)
 
+        qual_dict: dict[str, str | bool | Iterable[str]]
+        if isinstance(plan_action_env.qualifier, str):
+            qual_dict = qualifier_str_to_dict(plan_action_env.qualifier)
+        elif plan_action_env.qualifier is None:
+            qual_dict = {}
+        else:
+            qual_dict = plan_action_env.qualifier
+
         return self.add_anod_action(
             name=plan_action_env.module,
             env=(
@@ -291,7 +304,7 @@ def add_plan_action(
                 else BaseEnv.from_env(plan_action_env)
             ),
             primitive=primitive,
-            qualifier=plan_action_env.qualifier,
+            qualifier=qual_dict,
             source_packages=plan_action_env.source_packages,
             upload=plan_action_env.push_to_store,
             plan_line=plan_action_env.plan_line,
@@ -304,7 +317,7 @@ def add_anod_action(
         name: str,
         env: BaseEnv,
         primitive: PRIMITIVE,
-        qualifier: str | None = None,
+        qualifier: dict[str, str | bool | Iterable[str]] | None = None,
         source_packages: list[str] | None = None,
         upload: bool = True,
         plan_line: str | None = None,
@@ -330,6 +343,9 @@ def add_anod_action(
         :param sandbox: the SandBox object that will be used to run commands
         :return: the root added action
         """
+        if qualifier is None:
+            qualifier = {}
+
         # First create the subtree for the spec
         result = self.add_spec(
             name,
@@ -389,7 +405,7 @@ def add_spec(
         name: str,
         env: BaseEnv,
         primitive: PRIMITIVE,
-        qualifier: str | None = None,
+        qualifier: str | dict[str, str | bool | Iterable[str]] | None = None,
         source_packages: list[str] | None = None,
         expand_build: bool = True,
         source_name: str | None = None,
@@ -680,7 +696,7 @@ def add_dep(spec_instance: Anod, dep: Dependency, dep_instance: Anod) -> None:
                     name=e.name,
                     env=e.env(spec, self.default_env),
                     primitive=e.kind if e.kind != "download" else "install",
-                    qualifier=e.qualifier,
+                    qualifier=e.parsed_qualifier,
                     plan_args=None,
                     plan_line=plan_line,
                     sandbox=sandbox,

src/e3/anod/deps.py

@@ -4,13 +4,18 @@
 from typing import TYPE_CHECKING
 
 import e3.anod.error
+import e3.log
+from e3.anod import qualifier_str_to_dict
 from e3.env import BaseEnv
 
+
 if TYPE_CHECKING:
     from typing import Any, Hashable, Literal
     from e3.anod.spec import Anod, DEPENDENCY_PRIMITIVE
     from e3.mypy import assert_never
 
+logger = e3.log.getLogger("e3.anod.deps")
+
 
 class BuildVar:
     """Declare a dependency between an Anod spec and a variable."""
@@ -83,25 +88,20 @@ def __init__(
         self.build = build
         self.local_name = local_name if local_name is not None else name
 
-        self.qualifier: str | None
-        if isinstance(qualifier, dict):
-            for q in set(qualifier.keys()):
-                v = qualifier[q]
-                if isinstance(v, bool):
-                    # It's a tag qualifier
-                    if v:
-                        qualifier[q] = ""
-                    else:
-                        qualifier.pop(q)
-
-                # Compute a sorted representation for list, set and frozenset
-                elif not isinstance(v, str):
-                    qualifier[q] = ";".join(sorted(v))
-            self.qualifier = ",".join(
-                [f"{key}{'=' if val else ''}{val}" for key, val in qualifier.items()]
-            )
+        self.parsed_qualifier: dict
+
+        if isinstance(qualifier, str):
+            self.parsed_qualifier = qualifier_str_to_dict(qualifier)
+
+        elif isinstance(qualifier, dict):
+            # Ensure we get a copy as in that case qualifier is mutable
+            self.parsed_qualifier = dict(qualifier)
+
+        elif qualifier is None:
+            self.parsed_qualifier = {}
+
         else:
-            self.qualifier = qualifier
+            raise e3.anod.error.SpecError(f"invalid qualifier type: {qualifier}")
 
         if require not in (
             "build_tree",

src/e3/anod/driver.py

@@ -57,7 +57,7 @@ def activate(self, sandbox: SandBox, spec_repository: AnodSpecRepository) -> Non
             if isinstance(e, self.anod_instance.Dependency):
                 dep_class = spec_repository.load(e.name)
                 dep_instance = dep_class(
-                    qualifier=e.qualifier,
+                    qualifier=e.parsed_qualifier,
                     kind=e.kind,
                     env=e.env(self.anod_instance, BaseEnv.from_env()),
                 )

src/e3/anod/qualifiers_manager.py

@@ -9,6 +9,7 @@
 
 
 if TYPE_CHECKING:
+    from typing import Iterable
     from e3.anod.spec import Anod
 
 
@@ -78,8 +79,10 @@ def default(self) -> str | bool | frozenset[str] | None:
         return None  # all: no cover
 
     @abc.abstractmethod
-    def value(self, value: str) -> str | bool | frozenset[str]:
-        """Compute the value of qualifier given the use input."""
+    def value(
+        self, value: str | bool | Iterable[str] | None
+    ) -> str | bool | frozenset[str]:
+        """Compute the value of qualifier given the user input."""
 
     @abc.abstractmethod
     def repr(
@@ -147,10 +150,20 @@ def default(self) -> str | bool | frozenset[str] | None:
         """See QualifierDeclaration.default."""
         return self._default
 
-    def value(self, value: str) -> str | bool | frozenset[str]:
+    def value(
+        self, value: str | bool | Iterable[str] | None
+    ) -> str | bool | frozenset[str]:
         """See QualifierDeclaration.value."""
+        # Temporary until full switch to dict
+        if isinstance(value, bool):
+            value = ""
+
         if not isinstance(value, str):
-            raise AnodError("Key value qualifiers can only parse a string value.")
+            raise AnodError(
+                f"{self.origin}: Invalid value for qualifier {self.name}: "
+                f"requires a str value, got {type(value)}({value})"
+            )
+
         if self.choices is not None and value not in self.choices:
             choices_str = ", ".join((f"'{choice}'" for choice in self.choices))
             raise AnodError(
@@ -204,10 +217,22 @@ def default(self) -> str | bool | frozenset[str] | None:
         """See QualifierDeclaration.value."""
         return False
 
-    def value(self, value: str) -> str | bool | frozenset[str]:
+    def value(
+        self, value: str | bool | Iterable[str] | None
+    ) -> str | bool | frozenset[str]:
         """See QualifierDeclaration.value."""
         # As soon as a tag qualifier is passed, its value is True
-        return True
+        if isinstance(value, str):
+            return True
+        elif value is None:
+            return True
+        elif isinstance(value, bool):
+            return value
+        else:
+            raise AnodError(
+                f"{self.origin}: Invalid value for qualifier {self.name}: "
+                f"requires a str, bool or None value, got {type(value)}({value})"
+            )
 
     def repr(
         self, value: str | bool | frozenset[str], hash_pool: list[str] | None
@@ -282,15 +307,43 @@ def default(self) -> str | bool | frozenset[str] | None:
         """See QualifierDeclaration.default."""
         return self._default
 
-    def value(self, value: str) -> str | bool | frozenset[str]:
+    def value(
+        self, value: str | bool | Iterable[str] | None
+    ) -> str | bool | frozenset[str]:
         """See QualifierDeclaration.value."""
-        if not isinstance(value, str):
-            raise AnodError("Key set qualifiers can only parse a string value.")
+        if isinstance(value, bool):
+            raise AnodError(
+                f"{self.origin}: Invalid value for qualifier {self.name}: "
+                f"requires a str or Iterable[str], got bool"
+            )
+        elif value is None:
+            raise AnodError(
+                f"{self.origin}: Invalid value for qualifier {self.name}: "
+                f"requires a str or Iterable[str], got None"
+            )
+        elif isinstance(value, str):
+            # Make sure '' value is the empty set
+            value_set = (
+                frozenset(value.split(self.LIST_SEPARATOR)) if value else frozenset({})
+            )
+        else:
+            try:
+                result = all((isinstance(el, str) for el in value))
+                if not result:
+                    raise AnodError(
+                        f"{self.origin}: Invalid value for qualifier {self.name}: "
+                        f"one of the element in the Iterable is not a str: "
+                        f"got {type(value)}({value})"
+                    )
+            except TypeError as e:
 
-        # Make sure '' value is the empty set
-        value_set = (
-            frozenset(value.split(self.LIST_SEPARATOR)) if value else frozenset({})
-        )
+                raise AnodError(
+                    f"{self.origin}: Invalid value for qualifier {self.name}: "
+                    f"requires a str or Iterable[str], "
+                    f"got {type(value)}({value})"
+                ) from e
+
+            value_set = frozenset(value)
 
         # Check if the values are in choices
         if self.choices: