Adding options to log addition additional metadata into the lockfile

conda_lock/conda_lock.py

@@ -65,10 +65,13 @@
 from conda_lock.lookup import set_lookup_location
 from conda_lock.src_parser import (
     Dependency,
+    GitMeta,
+    InputMeta,
     LockedDependency,
     Lockfile,
     LockMeta,
     LockSpecification,
+    TimeMeta,
     UpdateSpecification,
     aggregate_lock_specs,
 )
@@ -283,6 +286,11 @@ def make_lock_files(
     filter_categories: bool = True,
     extras: Optional[AbstractSet[str]] = None,
     check_input_hash: bool = False,
+    add_inputs_metadata: bool = False,
+    add_git_metadata: bool = False,
+    add_time_metadata: bool = False,
+    metadata_jsons: Optional[List[pathlib.Path]] = None,
+    metadata_yamls: Optional[List[pathlib.Path]] = None,
 ) -> None:
     """
     Generate a lock file from the src files provided
@@ -316,6 +324,16 @@ def make_lock_files(
         Filter out unused categories prior to solving
     check_input_hash :
         Do not re-solve for each target platform for which specifications are unchanged
+    add_inputs_metadata:
+        If true add extra metadata to lockfile comments about the input files provided.
+    add_git_metadata:
+        If true add extra metadata to lockfile comments about the git-repo.
+    add_time_metadata:
+        If true add extra metadata to lockfile comments about the time of lockfile creation.
+    metadata_jsons:
+        JSON file(s) containing structured metadata to add to metadata section of the lockfile.
+    metadata_yamls:
+        YAML file(s) containing structured metadata to add to metadata section of the lockfile.
     """
 
     # initialize virtual package fake
@@ -392,6 +410,11 @@ def make_lock_files(
                 platforms=platforms_to_lock,
                 lockfile_path=lockfile_path,
                 update_spec=update_spec,
+                add_git_metadata=add_git_metadata,
+                add_time_metadata=add_time_metadata,
+                src_files=src_files if add_inputs_metadata else None,
+                metadata_jsons=metadata_jsons,
+                metadata_yamls=metadata_yamls,
             )
 
             if "lock" in kinds:
@@ -716,13 +739,58 @@ def _solve_for_arch(
     return list(conda_deps.values()) + list(pip_deps.values())
 
 
+def convert_structured_metadata_yaml(in_path: pathlib.Path) -> Dict[str, Any]:
+    with in_path.open("r") as infile:
+        metadata = yaml.safe_load(infile)
+    return metadata
+
+
+def convert_structured_metadata_json(in_path: pathlib.Path) -> Dict[str, Any]:
+    import json
+
+    with in_path.open("r") as infile:
+        metadata = json.load(infile)
+    return metadata
+
+
+def update_metadata(to_change: Dict[str, Any], change_source: Dict[str, Any]) -> None:
+    for key in change_source:
+        if key in to_change:
+            logger.warning(
+                f"Custom metadata field {key} provided twice, overwriting value "
+                + f"{to_change[key]} with {change_source[key]}"
+            )
+    to_change.update(change_source)
+
+
+def get_custom_metadata(
+    metadata_jsons: Optional[List[pathlib.Path]] = None,
+    metadata_yamls: Optional[List[pathlib.Path]] = None,
+) -> Dict[str, str]:
+    custom_metadata_dict: Dict[str, Any] = {}
+    if metadata_jsons is not None:
+        for json_path in metadata_jsons:
+            new_metadata = convert_structured_metadata_json(json_path)
+            update_metadata(custom_metadata_dict, new_metadata)
+    if metadata_yamls is not None:
+        for yaml_path in metadata_yamls:
+            new_metadata = convert_structured_metadata_yaml(yaml_path)
+            update_metadata(custom_metadata_dict, new_metadata)
+    return custom_metadata_dict
+
+
 def create_lockfile_from_spec(
     *,
     conda: PathLike,
     spec: LockSpecification,
     platforms: List[str] = [],
     lockfile_path: pathlib.Path,
     update_spec: Optional[UpdateSpecification] = None,
+    add_git_metadata: bool = False,
+    add_time_metadata: bool = False,
+    metadata_jsons: Optional[List[pathlib.Path]] = None,
+    metadata_yamls: Optional[List[pathlib.Path]] = None,
+    src_files: Optional[List[pathlib.Path]] = None,
 ) -> Lockfile:
     """
     Solve or update specification
@@ -745,13 +813,32 @@ def create_lockfile_from_spec(
         for dep in deps:
             locked[(dep.manager, dep.name, dep.platform)] = dep
 
+    git_metadata = GitMeta.create() if add_git_metadata else None
+    time_metadata = TimeMeta.create() if add_time_metadata else None
+    inputs_metadata: Optional[Dict[str, InputMeta]] = (
+        {str(src_file): InputMeta.create(src_file=src_file) for src_file in src_files}
+        if src_files is not None
+        else None
+    )
+    custom_metadata: Optional[Dict[str, str]] = (
+        get_custom_metadata(
+            metadata_jsons=metadata_jsons, metadata_yamls=metadata_yamls
+        )
+        if metadata_jsons is not None or metadata_yamls is not None
+        else None
+    )
+
     return Lockfile(
         package=[locked[k] for k in locked],
         metadata=LockMeta(
             content_hash=spec.content_hash(),
             channels=[c for c in spec.channels],
             platforms=spec.platforms,
             sources=[str(source.resolve()) for source in spec.sources],
+            git_metadata=git_metadata,
+            time_metadata=time_metadata,
+            inputs_metadata=inputs_metadata,
+            custom_metadata=custom_metadata,
         ),
     )
 
@@ -916,6 +1003,11 @@ def run_lock(
     virtual_package_spec: Optional[pathlib.Path] = None,
     update: Optional[List[str]] = None,
     filter_categories: bool = False,
+    add_inputs_metadata: bool = False,
+    add_git_metadata: bool = False,
+    add_time_metadata: bool = False,
+    metadata_jsons: Optional[List[pathlib.Path]] = None,
+    metadata_yamls: Optional[List[pathlib.Path]] = None,
 ) -> None:
     if environment_files == DEFAULT_FILES:
         if lockfile_path.exists():
@@ -960,6 +1052,11 @@ def run_lock(
         extras=extras,
         check_input_hash=check_input_hash,
         filter_categories=filter_categories,
+        add_inputs_metadata=add_inputs_metadata,
+        add_git_metadata=add_git_metadata,
+        add_time_metadata=add_time_metadata,
+        metadata_jsons=metadata_jsons,
+        metadata_yamls=metadata_yamls,
     )
 
 
@@ -1085,6 +1182,35 @@ def main() -> None:
     multiple=True,
     help="Packages to update to their latest versions. If empty, update all.",
 )
+@click.option(
+    "--add-inputs-metadata",
+    is_flag=True,
+    help="If true add extra metadata to lockfile comments about the input files provided.",
+)
+@click.option(
+    "--add-git-metadata",
+    is_flag=True,
+    help="If true add extra metadata to lockfile comments about the git-repo.",
+)
+@click.option(
+    "--add-time-metadata",
+    is_flag=True,
+    help="If true add extra metadata to lockfile comments about the time of lockfile creation.",
+)
+@click.option(
+    "--metadata-jsons",
+    default=None,
+    multiple=True,
+    type=click.Path(),
+    help="JSON file(s) containing structured metadata to add to metadata section of the lockfile.",
+)
+@click.option(
+    "--metadata-yamls",
+    default=None,
+    multiple=True,
+    type=click.Path(),
+    help="YAML file(s) containing structured metadata to add to metadata section of the lockfile.",
+)
 @click.option(
     "--pypi_to_conda_lookup_file",
     type=str,
@@ -1112,6 +1238,11 @@ def lock(
     virtual_package_spec: Optional[PathLike],
     pypi_to_conda_lookup_file: Optional[str],
     update: Optional[List[str]] = None,
+    add_inputs_metadata: bool = False,
+    add_git_metadata: bool = False,
+    add_time_metadata: bool = False,
+    metadata_jsons: Optional[List[pathlib.Path]] = None,
+    metadata_yamls: Optional[List[pathlib.Path]] = None,
 ) -> None:
     """Generate fully reproducible lock files for conda environments.
 
@@ -1127,6 +1258,10 @@ def lock(
     """
     logging.basicConfig(level=log_level)
 
+    if metadata_jsons is not None:
+        metadata_jsons = [pathlib.Path(path) for path in metadata_jsons]
+    if metadata_yamls is not None:
+        metadata_yamls = [pathlib.Path(path) for path in metadata_yamls]
     # Set Pypi <--> Conda lookup file location
     if pypi_to_conda_lookup_file:
         set_lookup_location(pypi_to_conda_lookup_file)
@@ -1175,6 +1310,11 @@ def lock(
         virtual_package_spec=virtual_package_spec,
         update=update,
         filter_categories=filter_categories,
+        add_inputs_metadata=add_inputs_metadata,
+        add_git_metadata=add_git_metadata,
+        add_time_metadata=add_time_metadata,
+        metadata_jsons=metadata_jsons,
+        metadata_yamls=metadata_yamls,
     )
     if strip_auth:
         with tempfile.TemporaryDirectory() as tempdir:

conda_lock/src_parser/__init__.py

@@ -1,5 +1,6 @@
 import hashlib
 import json
+import logging
 import pathlib
 import typing
 
@@ -16,6 +17,9 @@
 from conda_lock.virtual_package import FakeRepoData
 
 
+logger = logging.getLogger(__name__)
+
+
 class StrictModel(BaseModel):
     class Config:
         extra = "forbid"
@@ -101,6 +105,66 @@ def validate_hash(cls, v: HashModel, values: Dict[str, typing.Any]) -> HashModel
         return v
 
 
+class TimeMeta(StrictModel):
+    """Stores information about when the lockfile was generated."""
+
+    created_at: str = Field(..., description="Time stamp of lock-file creation time")
+
+    @classmethod
+    def create(cls) -> "TimeMeta":
+        import time
+
+        return cls(created_at=datetime.datetime.utcnow().isoformat() + 'Z')
+
+
+class GitMeta(StrictModel):
+    """
+    Stores information about the git repo the lockfile is being generated in (if applicable) and
+    the git user generating the file.
+    """
+
+    git_user_name: str = Field(..., description="Git user.name field of global config")
+    git_user_email: str = Field(
+        ..., description="Git user.email field of global config"
+    )
+    git_sha: Optional[str] = Field(
+        default=None, description="sha256 hash of the most recent git commit"
+    )
+
+    @classmethod
+    def create(cls) -> "GitMeta":
+        import git
+
+        git_sha: Optional[str]
+        try:
+            repo = git.Repo(search_parent_directories=True)
+            git_sha = f"{repo.head.object.hexsha}{'-dirty' if repo.is_dirty() else ''}"
+        except git.exc.InvalidGitRepositoryError:
+            git_sha = None
+        return cls(
+            git_user_name=git.Git()().config("user.name"),
+            git_user_email=git.Git()().config("user.email"),
+            git_sha=git_sha,
+        )
+
+
+class InputMeta(StrictModel):
+    """Stores information about an input provided to generate the lockfile."""
+
+    md5: str = Field(..., description="md5 checksum for an input file")
+
+    @classmethod
+    def create(cls, src_file: pathlib.Path) -> "InputMeta":
+        return cls(md5=f"{cls.get_input_md5(src_file=src_file)}")
+
+    @staticmethod
+    def get_input_md5(src_file: pathlib.Path) -> str:
+        hasher = hashlib.md5()
+        with src_file.open("r") as infile:
+            hasher.update(infile.read().encode("utf-8"))
+        return hasher.hexdigest()
+
+
 class LockMeta(StrictModel):
     content_hash: Dict[str, str] = Field(
         ..., description="Hash of dependencies for each target platform"
@@ -113,6 +177,23 @@ class LockMeta(StrictModel):
         ...,
         description="paths to source files, relative to the parent directory of the lockfile",
     )
+    time_metadata: Optional[TimeMeta] = Field(
+        None, description="Metadata dealing with the time lockfile was created"
+    )
+    git_metadata: Optional[GitMeta] = Field(
+        None,
+        description=(
+            "Metadata dealing with the git repo the lockfile was created in and the user that created it"
+        ),
+    )
+    inputs_metadata: Optional[Dict[str, InputMeta]] = Field(
+        None,
+        description="Metadata dealing with the input files used to create the lockfile",
+    )
+    custom_metadata: Optional[Dict[str, str]] = Field(
+        None,
+        description="Custom metadata provided by the user to be added to the lockfile",
+    )
 
     def __or__(self, other: "LockMeta") -> "LockMeta":
         """merge other into self"""
@@ -126,6 +207,10 @@ def __or__(self, other: "LockMeta") -> "LockMeta":
             channels=self.channels,
             platforms=sorted(set(self.platforms).union(other.platforms)),
             sources=ordered_union([self.sources, other.sources]),
+            time_metadata=other.time_metadata,
+            git_metadata=other.git_metadata,
+            inputs_metadata=other.inputs_metadata,
+            custom_metadata=other.custom_metadata,
         )
 
     @validator("channels", pre=True, always=True)

conda_lock/src_parser/lockfile.py

@@ -77,7 +77,9 @@ def write_section(text: str) -> None:
         yaml.dump(
             {
                 "version": Lockfile.version,
-                **json.loads(content.json(by_alias=True, exclude_unset=True)),
+                **json.loads(
+                    content.json(by_alias=True, exclude_unset=True, exclude_none=True)
+                ),
             },
             f,
         )

requirements.txt

@@ -9,3 +9,4 @@ pydantic
 poetry
 ruamel.yaml
 typing-extensions
+gitpython