Determine user requested specs

dof/_src/checkpoint.py

@@ -8,13 +8,17 @@
 from dof._src.models import package, environment
 from dof._src.utils import hash_string
 from dof._src.data.local import LocalData
+from dof._src.conda_meta.conda_meta import CondaMeta
 
 
 class Checkpoint():
     @classmethod
     def from_prefix(cls, prefix: str, uuid: str, tags: List[str] = []):
         packages = []
         channels = set()
+        meta = CondaMeta(prefix=prefix)
+        user_requested_specs_map = meta.get_requested_specs_map()
+
         for prefix_record in PrefixData(prefix, pip_interop_enabled=True).iter_records_sorted():
             if prefix_record.subdir == "pypi":
                 packages.append(
@@ -36,9 +40,9 @@ def from_prefix(cls, prefix: str, uuid: str, tags: List[str] = []):
                         conda_channel=prefix_record.channel.url(),
                         # TODO
                         arch="",
-                        # not sure here
-                        platform="linux-64",
-                        url=prefix_record.url
+                        platform=prefix_record.channel.platform,
+                        url=prefix_record.url,
+                        user_requested_spec=user_requested_specs_map.get(prefix_record.name, None)
                     )
                 )
 

dof/_src/conda_meta/conda.py

@@ -0,0 +1,44 @@
+import os
+from conda.core import envs_manager
+from conda.history import History
+
+class CondaCondaMeta:
+    @classmethod
+    def detect(cls, prefix):
+        """Detect if the given prefix is a conda based conda meta.
+        If it is, it will return an instance of CondaCondaMeta
+        """
+        known_prefixes = envs_manager.list_all_known_prefixes()
+        if prefix in known_prefixes:
+            return cls(prefix)
+        return None
+
+    def __init__(self, prefix):
+        self.prefix = prefix
+        history_file = f"{prefix}/conda-meta/history"
+        if not os.path.exists(history_file):
+            raise Exception(f"history file for prefix '{prefix}' does not exist")
+        self.history = History(prefix)
+
+    def get_requested_specs(self) -> list[str]:
+        """Return a list of all the MatchSpecs a user requested to be installed
+
+        Returns
+        -------
+        specs: list[str]
+            A list of all the MatchSpecs a user requested to be installed
+        """
+        requested_specs = self.history.get_requested_specs_map()
+        return [spec.spec for spec in requested_specs.values()]
+
+    def get_requested_specs_map(self) -> dict[str, str]:
+        """Return a dict of all the package name to MatchSpecs user requested
+          specs to be installed.
+
+        Returns
+        -------
+        specs: dict[str, str]
+            A list of all the package names to MatchSpecs a user requested to be installed
+        """
+        requested_specs = self.history.get_requested_specs_map()
+        return {k: v.spec for k,v in requested_specs.items()}

dof/_src/conda_meta/conda_meta.py

@@ -0,0 +1,69 @@
+# NOTE: 
+# There is a case for refactoring this into a pluggable or hook
+# based setup. For the purpose of exploring this approach we 
+# won't set that up here.
+
+import os
+
+from dof._src.conda_meta.conda import CondaCondaMeta
+from dof._src.conda_meta.pixi import PixiCondaMeta
+
+
+class CondaMeta():
+    def __init__(self, prefix):
+        """CondaMeta provides a way of interacting with the
+        conda-meta directory of an environment. Tools like conda
+        and pixi use conda-meta to keep important metadata about
+        the environment and it's history.
+
+        Parameters
+        ----------
+        prefix: str
+            The path to the environment
+        """
+        self.prefix = prefix
+
+        if not os.path.exists(prefix):
+            raise Exception(f"prefix {prefix} does not exist")
+
+        if not os.path.exists(f"{prefix}/conda-meta"):
+            raise Exception(f"invalid environment at {prefix}, conda-meta dir does not exist")
+
+        # detect which conda-meta flavour is used by the environment
+        for impl in [CondaCondaMeta, PixiCondaMeta]:
+            self.conda_meta = impl.detect(prefix)
+            if self.conda_meta is not None:
+                break
+
+        # if none is detected raise an exception
+        if self.conda_meta is None:
+            raise Exception("Could not detect conda or pixi based conda meta")
+
+    def get_requested_specs(self) -> list[str]:
+        """Return a list of all the MatchSpecs a user requested to be installed.
+
+        A user_requested_spec is one that the user explicitly asked to be 
+        installed. These are different from dependency_specs which are specs
+        that are installed because they are dependencies of the
+        requested_specs.
+
+        For example, when a user runs `conda install flask`, the user requested
+        spec is flask. And all the other installed packages are dependency_specs
+
+        Returns
+        -------
+        specs: list[str]
+            A list of all the MatchSpecs a user requested to be installed
+        """
+        return self.conda_meta.get_requested_specs()
+
+    def get_requested_specs_map(self) -> dict[str, str]:
+        """Return a dict of all the package name to MatchSpecs user requested
+          specs to be installed.
+
+        Returns
+        -------
+        specs: dict[str, str]
+            A list of all the package names to MatchSpecs a user requested to be installed
+        """
+        return self.conda_meta.get_requested_specs_map()

dof/_src/conda_meta/pixi.py

@@ -0,0 +1,39 @@
+import os
+
+class PixiCondaMeta:
+    @classmethod
+    def detect(cls, prefix):
+        """Detect if the given prefix is a pixi based conda meta.
+        If it is, it will return an instance of PixiCondaMeta
+        """
+        conda_meta_path = f"{prefix}/conda-meta"
+        # if the {prefix}/conda-meta/pixi path exists, then this is
+        # a pixi based conda meta environment
+        if os.path.exists(f"{conda_meta_path}/pixi"):
+            return cls(prefix)
+        return None
+
+    def __init__(self, prefix):
+        self.prefix = prefix
+
+    # TODO
+    def get_requested_specs(self) -> list[str]:
+        """Return a list of all the specs a user requested to be installed.
+        Returns
+        -------
+        specs: list[str]
+            A list of all the specs a user requested to be installed.
+        """
+        return []
+
+    # TODO
+    def get_requested_specs_map(self) -> dict[str, str]:
+        """Return a dict of all the package name to MatchSpecs user requested
+          specs to be installed.
+
+        Returns
+        -------
+        specs: dict[str, str]
+            A list of all the package names to MatchSpecs a user requested to be installed
+        """
+        return {}

dof/_src/data/local.py

@@ -1,3 +1,4 @@
+# TODO: rename this to `data_dir` and move up one module - this doesn't need it's whole own module
 from pathlib import Path
 from typing import List
 import os

dof/_src/lock.py

@@ -1,3 +1,4 @@
+# TODO: delete this whole thing
 import asyncio
 import yaml
 

dof/_src/models/environment.py

@@ -5,6 +5,7 @@
 from dof._src.models import package
 
 
+# TODO: delete this
 class CondaEnvironmentSpec(BaseModel):
     """Input conda environment.yaml spec"""
     name: Optional[str]

dof/_src/models/package.py

@@ -14,6 +14,10 @@ class CondaPackage(BaseModel):
     arch: str
     platform: str
     url: str
+    # the string representation of the matchspec that the user
+    # used to request the package. If this was not a package
+    # the user explicitly added, this will be none.
+    user_requested_spec: Optional[str] = None
 
     def to_repodata_record(self):
         """Converts a url package into a rattler compatible repodata record."""
@@ -28,7 +32,6 @@ def to_repodata_record(self):
             channel=self.conda_channel,
             url=self.url
         )
-
 
     def __str__(self):
         return f"conda: {self.name} - {self.version}"
@@ -59,6 +62,7 @@ def to_repodata_record(self):
         pass
 
 
+# TODO: probably remove?
 class UrlCondaPackage(BaseModel):
     url: str
 

dof/cli/root.py

@@ -7,6 +7,7 @@
 from dof._src.checkpoint import Checkpoint
 from dof._src.park.park import Park
 from dof.cli.checkpoint import checkpoint_command
+from dof._src.conda_meta.conda_meta import CondaMeta
 
 
 app = typer.Typer(
@@ -24,6 +25,7 @@
 )
 
 
+# TODO: Delete
 @app.command()
 def lock(
     env_file: str = typer.Option(
@@ -45,6 +47,40 @@ def lock(
             yaml.dump(solved_env.model_dump(), env_file)
 
 
+@app.command()
+def user_specs(
+    rev: str = typer.Option(
+        None,
+        help="uuid of the revision to inspect for user_specs"
+    ),
+    prefix: str = typer.Option(
+        None,
+        help="prefix to save"
+    ),
+):
+    """Demo command: output the list of user requested specs for a revision"""
+    if prefix is None:
+        prefix = os.environ.get("CONDA_PREFIX")
+    else:
+        prefix = os.path.abspath(prefix)
+
+    if rev is None:
+        meta = CondaMeta(prefix=prefix)
+        specs = meta.get_requested_specs()
+        print("the user requested specs in this environment are:")
+        # sort alphabetically for readability
+        for spec in sorted(specs):
+            print(f"  {spec}")
+    else:
+        chck = Checkpoint.from_uuid(prefix=prefix, uuid=rev)
+        pkgs = chck.list_packages()
+        print(f"the user requested specs rev {rev}:")
+        # sort alphabetically for readability
+        for spec in sorted(pkgs, key=lambda p: p.name):
+            if spec.user_requested_spec is not None:
+                print(f"  {spec.user_requested_spec}")
+
+
 @app.command()
 def push(
     target: Annotated[str, typer.Option(