Add CondaWorkspace spec for conda.toml

docs/source/api.rst

@@ -44,18 +44,19 @@ User Classes
     proj.conda_package.CondaRecipe
     proj.conda_package.RattlerRecipe
     proj.conda_project.CondaProject
+    proj.conda_workspace.CondaWorkspace
     proj.datapackage.DVCRepo
     proj.datapackage.DataPackage
     proj.dataworkflows.Dbt
     proj.dataworkflows.Quarto
     proj.dataworkflows.Prefect
-     proj.dataworkflows.Dagster
-     proj.dataworkflows.Kedro
-     proj.dataworkflows.Metaflow
-     proj.dataworkflows.MLFlow
-     proj.dataworkflows.Airflow
-     proj.dataworkflows.Snakemake
-     proj.dataworkflows.Nox
+    proj.dataworkflows.Dagster
+    proj.dataworkflows.Kedro
+    proj.dataworkflows.Metaflow
+    proj.dataworkflows.MLFlow
+    proj.dataworkflows.Airflow
+    proj.dataworkflows.Snakemake
+    proj.dataworkflows.Nox
     proj.documentation.Docusaurus
     proj.documentation.MDBook
     proj.documentation.MkDocs
@@ -124,6 +125,7 @@ User Classes
 .. autoclass:: projspec.proj.conda_package.CondaRecipe
 .. autoclass:: projspec.proj.conda_package.RattlerRecipe
 .. autoclass:: projspec.proj.conda_project.CondaProject
+.. autoclass:: projspec.proj.conda_workspace.CondaWorkspace
 .. autoclass:: projspec.proj.datapackage.DVCRepo
 .. autoclass:: projspec.proj.datapackage.DataPackage
 .. autoclass:: projspec.proj.dataworkflows.Dbt

src/projspec/proj/__init__.py

@@ -15,6 +15,7 @@
 )
 from projspec.proj.conda_package import CondaRecipe, RattlerRecipe
 from projspec.proj.conda_project import CondaProject
+from projspec.proj.conda_workspace import CondaWorkspace
 from projspec.proj.data_dir import Data
 from projspec.proj.datapackage import DataPackage, DVCRepo
 from projspec.proj.dataworkflows import (
@@ -78,6 +79,7 @@
     # Conda
     "CondaRecipe",
     "CondaProject",
+    "CondaWorkspace",
     # Data
     "Data",
     "DataPackage",

src/projspec/proj/conda_workspace.py

@@ -0,0 +1,145 @@
+import os
+
+import toml
+
+from projspec.proj import ParseFailed, ProjectSpec
+from projspec.proj.pixi import envs_from_lock, extract_tasks
+from projspec.utils import AttrDict, PickleableTomlDecoder
+
+
+class CondaWorkspace(ProjectSpec):
+    """A workspace managed by conda-workspaces (``conda.toml``).
+
+    conda-workspaces brings multi-environment workspace management and
+    task execution to the conda CLI as a plugin (``conda workspace`` /
+    ``conda task``).  The manifest is a conda-native sibling of
+    pixi.toml; the lockfile (``conda.lock``) is rattler-lock v6 with
+    a ``version: 1`` byte identifying it as conda-workspaces-owned.
+    """
+
+    icon = "🧰"
+    spec_doc = (
+        "https://conda-incubator.github.io/conda-workspaces/reference/conda-toml-spec/"
+    )
+
+    def _load_meta(self) -> dict:
+        """Merge metadata from ``pyproject.toml`` (``[tool.conda]``) and ``conda.toml``.
+
+        ``conda.toml`` wins on overlap, mirroring conda-workspaces' own
+        precedence rule.
+        """
+        meta = self.proj.pyproject.get("tool", {}).get("conda", {})
+        if "conda.toml" in self.proj.basenames:
+            try:
+                with self.proj.get_file("conda.toml", text=True) as f:
+                    meta = {
+                        **meta,
+                        **toml.loads(f.read(), decoder=PickleableTomlDecoder()),
+                    }
+            except (OSError, ValueError, UnicodeDecodeError, FileNotFoundError):
+                pass
+        return meta
+
+    def match(self) -> bool:
+        # A bare conda.toml without [workspace] is permitted by the spec
+        # as a tasks-only manifest; it does not constitute a workspace
+        # on its own, so do not match it here.
+        return (
+            bool(self.proj.pyproject.get("tool", {}).get("conda", {}))
+            or "conda.toml" in self.proj.basenames
+        )
+
+    def parse(self) -> None:
+        from projspec.artifact.python_env import CondaEnv, LockFile
+        from projspec.content.environment import Environment, Precision, Stack
+
+        meta = self._load_meta()
+        if not meta.get("workspace"):
+            raise ParseFailed
+
+        arts = AttrDict()
+        conts = AttrDict()
+        procs = AttrDict()
+        commands = AttrDict()
+
+        run_cmd = ("conda", "task", "run")
+        env_flag = "-e"
+
+        extract_tasks(
+            meta, procs, commands, self.proj, run_cmd=run_cmd, env_flag=env_flag
+        )
+        if "environments" in meta and "feature" in meta:
+            for env_name, details in meta["environments"].items():
+                feat: dict = {}
+                feats = set(
+                    details if isinstance(details, list) else details["features"]
+                )
+                for fname in feats:
+                    feat.update(meta["feature"].get(fname, {}))
+                if isinstance(details, list) or not details.get("no-default-feature"):
+                    feat.update(meta)
+                extract_tasks(
+                    feat,
+                    procs,
+                    commands,
+                    self.proj,
+                    env=env_name,
+                    run_cmd=run_cmd,
+                    env_flag=env_flag,
+                )
+
+        if procs:
+            arts["process"] = procs
+        if commands:
+            conts["commands"] = commands
+
+        # `envs-dir` is in the conda.toml spec but conda-workspaces' own
+        # parser does not yet read it from TOML (uses the model default).
+        # Reading it here keeps projspec aligned with the published spec.
+        envs_dir = meta.get("workspace", {}).get("envs-dir", ".conda/envs")
+        if "conda.lock" in self.proj.basenames:
+            conts["environments"] = AttrDict()
+            arts["conda_env"] = AttrDict()
+            with self.proj.fs.open(self.proj.basenames["conda.lock"], "rb") as f:
+                lock_envs = envs_from_lock(f)
+            for env_name, details in lock_envs.items():
+                arts["conda_env"][env_name] = CondaEnv(
+                    proj=self.proj,
+                    fn=f"{self.proj.url}/{envs_dir}/{env_name}",
+                    cmd=["conda", "workspace", "install", "-e", env_name],
+                )
+                conts["environments"][env_name] = Environment(
+                    proj=self.proj,
+                    packages=details["packages"],
+                    stack=Stack.CONDA,
+                    precision=Precision.LOCK,
+                    channels=details["channels"],
+                )
+
+        arts["lock_file"] = LockFile(
+            proj=self.proj,
+            fn=f"{self.proj.url}/conda.lock",
+            cmd=["conda", "workspace", "lock"],
+        )
+
+        self._artifacts = arts
+        self._contents = conts
+
+    @staticmethod
+    def _create(path: str) -> None:
+        name = os.path.basename(path)
+        with open(f"{path}/conda.toml", "wt") as f:
+            f.write(
+                f"""[workspace]
+name = "{name}"
+channels = ["conda-forge"]
+platforms = ["osx-arm64", "linux-64", "win-64"]
+version = "0.1.0"
+
+[tasks]
+hello = "echo 'hello world'"
+
+[dependencies]
+python = ">=3.10"
+"""
+            )

src/projspec/proj/pixi.py

@@ -182,30 +182,40 @@ def _create(path: str) -> None:
             )
 
 
-def extract_feature(
+def extract_tasks(
     meta: dict,
     procs: AttrDict,
     commands: AttrDict,
-    pixi: Pixi,
+    proj,
     env: str | None = None,
+    run_cmd: tuple[str, ...] = ("pixi", "run"),
+    env_flag: str = "--environment",
 ):
-    """Consolidate metadata from 'features' to create commands and processes"""
+    """Consolidate metadata from 'features' to create commands and processes.
+
+    Shared by ``Pixi`` and ``CondaWorkspaces``: both formats share the
+    same ``[tasks]`` / ``[target.<platform>.tasks]`` shape, only the
+    runner CLI strings differ.  ``run_cmd`` is the prefix for the
+    runnable form (``pixi run`` vs ``conda task run``); ``env_flag`` is
+    the flag used to target an environment (``--environment`` vs
+    ``-e``).
+    """
     from projspec.artifact.process import Process
     from projspec.content.executable import Command
 
     for name, task in meta.get("tasks", {}).items():
         if env:
             name = f"{name}.{env}"
-        cmd = ["pixi", "run", name]
+        cmd = [*run_cmd, name]
         if env:
-            cmd.extend(["--environment", env])
-        art = Process(proj=pixi.proj, cmd=cmd)
+            cmd.extend([env_flag, env])
+        art = Process(proj=proj, cmd=cmd)
         procs[name] = art
         # tasks without a command are aliases
         cmd = task.get("cmd", "") if isinstance(task, dict) else task
-        # NB: these may have dependencies on other tasks and envs, but pixi
-        # manages those.
-        commands[name] = Command(proj=pixi.proj, cmd=cmd)
+        # NB: these may have dependencies on other tasks and envs, but the
+        # runner manages those.
+        commands[name] = Command(proj=proj, cmd=cmd)
     for platform, v in meta.get("target", {}).items():
         for name, task in v.get("tasks", {}).items():
             if env:
@@ -215,20 +225,36 @@ def extract_feature(
                 if isinstance(task, dict)
                 else task
             )
-            commands[name] = Command(proj=pixi.proj, cmd=cmd)
+            commands[name] = Command(proj=proj, cmd=cmd)
             if platform == this_platform():
                 # only commands on the current platform can be executed
-                cmd = ["pixi", "run", name]
+                cmd = [*run_cmd, name]
                 if env:
-                    cmd.extend(["--environment", env])
-                art = Process(proj=pixi.proj, cmd=cmd)
+                    cmd.extend([env_flag, env])
+                art = Process(proj=proj, cmd=cmd)
                 procs[name] = art
                 commands[name].artifacts.add(art)
 
 
+def extract_feature(
+    meta: dict,
+    procs: AttrDict,
+    commands: AttrDict,
+    pixi: "Pixi",
+    env: str | None = None,
+):
+    """Backwards-compatible wrapper around :func:`extract_tasks` for ``Pixi``."""
+    extract_tasks(meta, procs, commands, pixi.proj, env=env)
+
+
 def envs_from_lock(infile) -> dict:
-    """Extract the environments info from a pixi (yaml) lock file"""
-    # Developed for pixi format format 6
+    """Extract the environments info from a rattler-lock-shaped YAML lock file.
+
+    Developed for pixi lockfile format 6 (``pixi.lock``).  Also accepts
+    conda-workspaces' ``conda.lock``, which is the same schema with an
+    on-disk ``version: 1`` byte (see
+    https://conda-incubator.github.io/conda-workspaces/reference/conda-toml-spec/#lockfile-relationship).
+    """
     import yaml
 
     data = yaml.safe_load(infile)