Pseudo-Keys Part II: history metadata

onyo/lib/commands.py

@@ -296,7 +296,7 @@ def _edit_asset(inventory: Inventory,
     # Store original pseudo-keys of `asset`, in order to re-assign
     # them when loading edited file from disc. This is relevant, when
     # `operation` uses them (`Inventory.add_asset/modify_asset`).
-    reserved_keys = {k: v for k, v in asset.items() if k in list(PSEUDO_KEYS.keys()) + ['template']}
+    reserved_keys = {k: v for k, v in asset.items() if k in list(PSEUDO_KEYS.keys()) + ['template'] and v is not None}
     disallowed_keys = RESERVED_KEYS + list(PSEUDO_KEYS.keys())
     disallowed_keys.remove('onyo.is.directory')
     disallowed_keys.remove('onyo.path.parent')
@@ -332,7 +332,11 @@ def _edit_asset(inventory: Inventory,
             if any(k in disallowed_keys for k in tmp_asset.keys()):
                 raise ValueError(f"Can't set any of the keys ({', '.join(disallowed_keys)}).")
             asset = Item()
-            asset.data = tmp_asset.data  # keep exact objects for ruamel-based roundtrip
+            # keep exact objects for ruamel-based roundtrip:
+            asset.data = tmp_asset.data
+            # ^ This kills pseudo-keys, though. Re-add those, that aren't specified:
+            asset.update({k: v for k, v in PSEUDO_KEYS.items() if k not in tmp_asset})
+
             # When reading from file, we don't get reserved keys back, since they are not
             # part of the file content. We do need the object from reading the file to be
             # the basis, though, to get comment roundtrip from ruamel.

onyo/lib/git.py

@@ -9,7 +9,10 @@
 from onyo.lib.ui import ui
 
 if TYPE_CHECKING:
-    from typing import Iterable
+    from typing import (
+        Generator,
+        Iterable,
+    )
 
 log: logging.Logger = logging.getLogger('onyo.git')
 
@@ -401,3 +404,88 @@
 
     # TODO: git check-ignore --no-index --stdin (or via command call)  ->  lazy, check GitRepo.files once. (Same invalidation)
     #       -> OnyoRepo would use it to maintain a ignore list from a (top-level .onyoignore)? .onyo/ignore ? Both?
+
+    def _parse_log_output(self, lines: list[str]) -> dict:
+        """Generate a dict from the output of ``git log`` for one commit.
+
+        Internal helper that converts a list of git-log-output lines of
+        a single commit to a dictionary.
+        """
+        import datetime
+        import re
+        regex_author = re.compile(r"Author:\s+(?P<name>\b.*\b)\s+<(?P<email>[^\s]+)>$")
+        regex_committer = re.compile(r"Commit:\s+(?P<name>\b.*\b)\s+<(?P<email>[^\s]+)>$")
+        commit = dict()
+        for line in lines:
+            if line.startswith('commit '):
+                commit['hexsha'] = line.split()[1]
+                continue
+            elif line.startswith('Author:'):
+                try:
+                    commit['author'] = re.match(regex_author, line).groupdict()  # pyre-ignore [16] AttributeError is caught
+                except AttributeError as e:
+                    if str(e).endswith("'groupdict'"):
+                        raise RuntimeError(f"Unable to parse author-line:\n{line}") from e
+                    raise
+                continue
+            elif line.startswith('AuthorDate:'):
+                continue
+            elif line.startswith('Commit:'):
+                try:
+                    commit['committer'] = re.match(regex_committer, line).groupdict()
+                except AttributeError as e:
+                    if str(e).endswith("'groupdict'"):
+                        raise RuntimeError(f"Unable to parse committer-line:\n{line}") from e
+                    raise
+                continue
+            elif line.startswith('CommitDate:'):
+                commit['time'] = datetime.datetime.fromisoformat(line.split(maxsplit=1)[1])
+                continue
+            else:
+                # This line is part of the message.
+                if 'message' not in commit:
+                    commit['message'] = [line]
+                else:
+                    commit['message'].append(line)
+
+        return commit
+
+    def history(self, path: Path | None = None, n: int | None = None) -> Generator[dict, None, None]:
+        """Yields commit dicts representing the history of ``path``.
+
+        History according to git log (git log --follow if a path is given).
+
+        Parameters
+        ----------
+        path:
+          What file to follow. If `None`, get the history of HEAD instead.
+        n:
+          Limit history going back ``n`` commits.
+          ``None`` for no limit (default).
+        """
+        # TODO: Something like this may allow us to efficiently deal with multiline strings in git-log's output,
+        #       rather than splitting lines and iterating over them when assembling output that belongs to a single
+        #       commit, parsing the git data and then the operations record (also removes leading spaces for the commit
+        #       message):
+        #       --pretty='format:commit: %H%nAuthor: %an (%ae)%nCommitter: %cn (%ce)%nCommitDate: %cI%nMessage:%n%B'
+        limit = [f'-n{n}'] if n is not None else []
+        pathspec = ['--follow', '--', str(path)] if path else []
+        cmd = ['log', '--date=iso-strict', '--pretty=fuller'] + limit + pathspec
+        output = self._git(cmd)
+
+        # yield output on a per-commit-basis
+        commit_output = []
+        for line in output.splitlines():
+            if line.startswith('commit '):
+                # This is the first line of a new commit.
+                # 1. store previous commit output
+                if commit_output:
+                    # we just finished the previous commit.
+                    yield self._parse_log_output(commit_output)
+                # 2. start new commit output
+                commit_output = [line]
+            else:
+                # add to current commit output
+                commit_output.append(line)
+        if commit_output:
+            yield self._parse_log_output(commit_output)

onyo/lib/inventory.py

@@ -725,3 +725,6 @@ def raise_empty_keys(self, asset: dict | UserDict) -> None:
             # Note, that DotNotationWrapper.keys() delivers strings (and has to).
             # Hence, `None` as a key would show up here as 'None'.
             raise ValueError("Keys are not allowed to be empty or None-values.")
+
+    def get_history(self, path: Path | None = None, n: int | None = None) -> Generator[UserDict, None, None]:
+        yield from self.repo.get_history(path, n)

onyo/lib/items.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from copy import deepcopy
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -51,7 +52,10 @@ def __init__(self,
         self.update(onyo.lib.pseudokeys.PSEUDO_KEYS)
         self._path = None
 
-        if isinstance(item, Path):
+        if isinstance(item, Item):
+            self._path = item._path
+            self.data = deepcopy(item.data)
+        elif isinstance(item, Path):
             assert item.is_absolute()  # currently no support for relative. This is how all existing code should work ATM.
             self.update_from_path(item)
         elif item is not None:
@@ -95,15 +99,51 @@ def update_from_path(self, path: Path):
             loader = self.repo.get_asset_content if self.repo else get_asset_content
             self.update(loader(path))
 
-    # TODO: git metadata:
-    # def fill_created(self, what: str | None):
-    #     # self.repo.git.xxx   (self.get('onyo.path.file'))
-    #     # fill in all created keys
-    #     # switch `what` -> return whatever part was requesting this
-    #     pass#raise NotImplementedError
-    #
-    # def fill_modified(self, what: str | None):
-    #     pass#raise NotImplementedError
+    def fill_created(self, what: str | None):
+        """Initializer for the 'onyo.was.modified' pseudo-keys.
+
+        Fills in the entire sub-dict and returns the value specified by ``what``.
+
+        Note/TODO:
+        ----------
+        This is currently based on ``git log --follow <path>``. Looking back, the first appearance
+        of a 'new_assets'/'new_directories' operation should be it, assuming the history
+        was created by onyo commands.
+        However, if the history was created using the python interface, that assumption wouldn't hold
+        and we'd have to trace back moves and renames in order to know what path or file name we are looking
+        to match against these operations.
+        """
+        if self.repo and self['onyo.path.absolute']:
+            for commit in self.repo.get_history(self['onyo.path.file']):  # pyre-ignore[16]
+                if 'operations' in commit:
+                    if (self['onyo.is.asset'] and commit['operations']['new_assets']) or \
+                            (self['onyo.is.directory'] and commit['operations']['new_directories']):
+                        self['onyo.was.created'] = commit.data
+                        return commit[what] if what else None
+            return None
+
+    def fill_modified(self, what: str | None):
+        """Initializer for the 'onyo.was.modified' pseudo-keys.
+
+        Fills in the entire sub-dict and returns the value specified by ``what``.
+
+        Note/TODO:
+        ----------
+        See ``fill_created``.
+        """
+        if self.repo and self['onyo.path.absolute']:
+            for commit in self.repo.get_history(self['onyo.path.file']):  # pyre-ignore[16]
+                if 'operations' in commit:
+                    if (self['onyo.is.asset'] and
+                        (commit['operations']['modify_assets'] or
+                         commit['operations']['new_assets'])) or \
+                       (self['onyo.is.directory'] and
+                        (commit['operations']['new_directories'] or
+                         commit['operations']['move_directories'] or
+                         commit['operations']['rename_directories'])):
+                        self['onyo.was.modified'] = commit.data
+                        return commit[what] if what else None
+        return None
 
     def get_path_absolute(self):
         """Initializer for the 'onyo.path.absolute' pseudo-key."""
@@ -125,7 +165,14 @@ def get_path_parent(self):
 
     def get_path_file(self):
         """Initializer for the 'onyo.path.file' pseudo-key."""
-        return self._path.relative_to(self.repo.git.root) if self.repo and self['onyo.is.asset'] else None
+        if self.repo and self['onyo.path.absolute']:
+            if self['onyo.is.asset'] and not self['onyo.is.directory']:
+                return self['onyo.path.relative']
+            if self['onyo.is.asset'] and self['onyo.is.directory']:
+                return self['onyo.path.relative'] / onyo.lib.onyo.OnyoRepo.ASSET_DIR_FILE_NAME
+            if self['onyo.is.directory'] and not self['onyo.is.asset']:
+                return self['onyo.path.relative'] / onyo.lib.onyo.OnyoRepo.ANCHOR_FILE_NAME
+        return None
 
     def is_asset(self) -> bool | None:
         """Initializer for the 'onyo.is.asset' pseudo-key."""
@@ -147,7 +194,10 @@ def is_template(self) -> bool | None:
 
     def is_empty(self) -> bool | None:
         """Initializer for the 'onyo.is.empty' pseudo-key."""
-        return None  # TODO: Unclear what exactly this needs to consider. Probably child items? Or just assets?
+        if self['onyo.is.directory'] and self.repo and self._path:
+            # TODO: This likely can be faster when redoing/enhancing caching of repo paths.
+            return not any(p.parent == self._path for p in self.repo.get_asset_paths())
+        return None
 
 # TODO/Notes for next PR(s):
 # - Bug/Missing feature: pseudo-keys that are supposed to be settable by commands, are not yet

onyo/lib/onyo.py

@@ -17,7 +17,11 @@
 from .utils import get_asset_content, write_asset_file
 
 if TYPE_CHECKING:
-    from typing import Iterable, List
+    from typing import (
+        Generator,
+        Iterable,
+        List,
+    )
     from collections import UserDict
 
 log: logging.Logger = logging.getLogger('onyo.onyo')
@@ -738,3 +742,24 @@ def commit(self, paths: Iterable[Path] | Path, message: str):
         """
         self.git.commit(paths=paths, message=message)
         self.clear_cache()
+
+    def get_history(self, path: Path | None = None, n: int | None = None) -> Generator[UserDict, None, None]:
+        # TODO: This isn't quite right yet. operations records are defined in Inventory.
+        #       But Inventory shouldn't talk to GitRepo directly. So, either pass the record
+        #       from Inventory to OnyoRepo and turn it into a commit-message part only,
+        #       or have sort of a proxy in OnyoRepo.
+        #       -> May be: get_history(Item) in Inventory and get_history(path) in OnyoRepo.
+        from onyo.lib.parser import parse_operations_record
+        from onyo.lib.utils import DotNotationWrapper
+        for commit in self.git.history(path, n):
+            record = []
+            start = False
+            for line in commit['message']:
+                if line.strip() == "--- Inventory Operations ---":
+                    start = True
+                if start:
+                    record.append(line)
+            if record:
+
+                commit['operations'] = parse_operations_record(record)
+            yield DotNotationWrapper(commit)

onyo/lib/parser.py

@@ -0,0 +1,47 @@
+from pathlib import Path
+
+from onyo.lib.inventory import OPERATIONS_MAPPING
+
+
+def parse_operations_record(record: list[str]) -> dict:
+
+    if not record[0].strip() == "--- Inventory Operations ---":
+        raise RuntimeError("Invalid operations record.")
+    parsed_record = {k: [] for k in OPERATIONS_MAPPING.keys()}
+    collecting_key = None
+    for line in record[1:]:
+        line = line.strip()
+        if not line:
+            continue
+        match line:
+            case "New assets:":
+                collecting_key = "new_assets"
+            case "New directories:":
+                collecting_key = "new_directories"
+            case "Removed assets:":
+                collecting_key = "remove_assets"
+            case "Removed directories:":
+                collecting_key = "remove_directories"
+            case "Moved assets:":
+                collecting_key = "move_assets"
+            case "Moved directories:":
+                collecting_key = "move_directories"
+            case "Renamed directories:":
+                collecting_key = "rename_directories"
+            case "Renamed assets:":
+                collecting_key = "rename_assets"
+            case "Modified assets:":
+                collecting_key = "modify_assets"
+            case _:
+                if not collecting_key or line and not line.startswith("- "):
+                    raise RuntimeError("Invalid operations record.")
+                cleaned_line = line[2:].split(" -> ")
+                if len(cleaned_line) > 2:
+                    raise RuntimeError(f"Invalid operations record:{line}")
+                if len(cleaned_line) == 1:
+                    parsed_record[collecting_key].append(Path(cleaned_line[0]))
+                else:
+                    parsed_record[collecting_key].append(
+                        (Path(cleaned_line[0]), Path(cleaned_line[1]))
+                    )
+    return parsed_record