diff --git a/dof/_src/checkpoint.py b/dof/_src/checkpoint.py index 484ff9a..1c0e836 100644 --- a/dof/_src/checkpoint.py +++ b/dof/_src/checkpoint.py @@ -8,6 +8,7 @@ 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(): @@ -15,6 +16,9 @@ class Checkpoint(): 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) ) ) diff --git a/dof/_src/conda_meta/__init__.py b/dof/_src/conda_meta/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dof/_src/conda_meta/conda.py b/dof/_src/conda_meta/conda.py new file mode 100644 index 0000000..8c3cae4 --- /dev/null +++ b/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()} diff --git a/dof/_src/conda_meta/conda_meta.py b/dof/_src/conda_meta/conda_meta.py new file mode 100644 index 0000000..3993cbe --- /dev/null +++ b/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() diff --git a/dof/_src/conda_meta/pixi.py b/dof/_src/conda_meta/pixi.py new file mode 100644 index 0000000..78aba81 --- /dev/null +++ b/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 {} diff --git a/dof/_src/data/local.py b/dof/_src/data/local.py index f573580..9cd89e9 100644 --- a/dof/_src/data/local.py +++ b/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 diff --git a/dof/_src/lock.py b/dof/_src/lock.py index 372da91..966b651 100644 --- a/dof/_src/lock.py +++ b/dof/_src/lock.py @@ -1,3 +1,4 @@ +# TODO: delete this whole thing import asyncio import yaml diff --git a/dof/_src/models/environment.py b/dof/_src/models/environment.py index 330bec0..cbf2182 100644 --- a/dof/_src/models/environment.py +++ b/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] diff --git a/dof/_src/models/package.py b/dof/_src/models/package.py index 97db516..2617adc 100644 --- a/dof/_src/models/package.py +++ b/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 diff --git a/dof/cli/root.py b/dof/cli/root.py index b41f30f..dd3e355 100644 --- a/dof/cli/root.py +++ b/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(