feat(remote-parent): Implement doorstop

We added an implementation that pull a remote parent
for a doorstop document if not present.

It also validates if you try to import a prefix twice with different tags.

- This Implementation should be capable of doing recursive
pulling as well (not tested yet)

doorstop/core/builder.py

@@ -2,8 +2,15 @@
 
 """Functions to build a tree and access documents and items."""
 
+
 import os
+from traceback import format_exception
+
 from typing import List, Optional
+from collections.abc import Mapping, MutableMapping
+from pathlib import Path
+from dulwich import porcelain
+from dulwich.repo import Repo
 
 from doorstop import common
 from doorstop.common import DoorstopError
@@ -28,17 +35,23 @@ def build(cwd=None, root=None, request_next_number=None) -> Tree:
     :return: new :class:`~doorstop.core.tree.Tree`
 
     """
-    documents: List[Document] = []
+
+    documents: list[Document] = []
 
     # Find the root of the working copy
     cwd = cwd or os.getcwd()
     root = root or vcs.find_root(cwd)
 
+    # Remote parents inclusion
+    remote_documents: dict[str,str] = {} # Prefix, tag
+    remote_base_path: Path | None = Path(root, ".doorstop-remote") or None #TODO: make this configurable
+
+
     # Find all documents in the working copy
     log.info("looking for documents in {}...".format(root))
     skip_file_name = ".doorstop.skip-all"
     if not os.path.isfile(os.path.join(root, skip_file_name)):
-        _document_from_path(root, root, documents)
+        _document_from_path(root, root, documents, remote_documents, remote_base_path)
     exclude_dirnames = {".git", ".venv", "venv"}
     if not os.path.isfile(os.path.join(root, skip_file_name)):
         for dirpath, dirnames, _ in os.walk(root, topdown=True):
@@ -50,7 +63,7 @@ def build(cwd=None, root=None, request_next_number=None) -> Tree:
                 if os.path.isfile(os.path.join(path, skip_file_name)):
                     continue
                 whilelist_dirnames.append(dirname)
-                _document_from_path(path, root, documents)
+                _document_from_path(path, root, documents, remote_documents, remote_base_path)
             dirnames[:] = whilelist_dirnames
 
     # Build the tree
@@ -66,7 +79,7 @@ def build(cwd=None, root=None, request_next_number=None) -> Tree:
     return tree
 
 
-def _document_from_path(path, root, documents):
+def _document_from_path(path, root, documents:list[Document], remote_documents:dict[str,str], remote_base_path: Path):
     """Attempt to create and append a document from the specified path.
 
     :param path: path to a potential document
@@ -80,12 +93,27 @@ def _document_from_path(path, root, documents):
     except DoorstopError:
         pass  # no document in directory
     else:
+        if document.remote_parent_url and document.remote_parent_tag:
+            log.debug("""Document has remote parent
+                      Remote information:
+                        Remote Url: {}
+                        Remote Tag: {}""".format(
+                            document.remote_parent_url, 
+                            document.remote_parent_tag))
         if document.skip:
             log.debug("skipped document: {}".format(document))
         else:
             log.info("found document: {}".format(document))
+            if document.remote_parent_url and document.remote_parent_tag:
+                _download_remote_parent(remote_base_path,
+                                        document.remote_parent_url,
+                                        document.remote_parent_tag,
+                                        documents,
+                                        remote_documents)
             documents.append(document)
 
+            # log.debug("Remote documents in {}".format(remote_base_path / document.prefix))
+
 
 def find_document(prefix):
     """Find a document without an explicitly building a tree."""
@@ -120,3 +148,111 @@ def _clear_tree():
     """Force the shared tree to be rebuilt."""
     global _tree
     _tree = None
+
+def _create_remote_dir(remote_base_path:Path):
+    """Create the folder that will store pulled documents"""
+    match remote_base_path.exists():
+        case True: log.debug("Skipping creation folder already exists {} in {}.".format(remote_base_path.parent, remote_base_path.parts[-1]))
+        case False: log.info("Creating folder {} in {}".format(remote_base_path.parts[-1], remote_base_path.parent))
+    try:
+        remote_base_path.mkdir(exist_ok=True)
+    except Exception as e:
+        tb_str = ''.join(format_exception(None, e, e.__traceback__))
+        raise DoorstopError("""Something unexpected has happened when trying to create folder {} in {}\n.
+                            Details: {}""".format(remote_base_path.parent, remote_base_path.parts[-1], tb_str))
+
+def _remote_documents_check(existing_remote_docs: Mapping[str,str], new_remote_docs: Mapping[str,str]) -> dict[str,bool]: 
+    results: dict[str,bool] = {}
+
+    for key in new_remote_docs.keys():
+        if key in existing_remote_docs: 
+            results[key] = existing_remote_docs[key] == new_remote_docs[key] 
+
+    if list(results.values()).count(False) >=1 :
+        raise DoorstopError(f"""
+Different version for a prefix detected please check the remote document definitions.
+Affected prefix(es) { [ x for x in results.keys() if results[x] == False ] }
+                             """)
+    return results
+
+def _validate_remote_documents(new_remote_doc: Path, 
+                               remote_document_tag: str,
+                               remote_documents: dict[str,str],
+                               doorstop_documents: list[Document]
+                               ):
+
+        root = new_remote_doc.parent.parent
+
+        def find_remote_docs(remote_path:Path):
+            """
+            Recreating part of the build here to discover documents inside the downloaded repo.
+            """
+
+            remote_docs: list[Document] = []
+            skip_file_name = ".doorstop.skip-all"
+            exclude_dirnames = {".git", ".venv", "venv"}
+            for dirpath, dirnames, _ in os.walk(remote_path, topdown=True):
+                for dirname in dirnames:
+                    if dirname in exclude_dirnames:
+                        continue
+                    path = os.path.join(dirpath, dirname)
+                    if os.path.isfile(os.path.join(path, skip_file_name)):
+                        continue
+                    _document_from_path(path, str(root), remote_docs, remote_documents, new_remote_doc)
+            return remote_docs
+
+        new_remote_docs = find_remote_docs(new_remote_doc)
+        entries = { str(d.prefix): remote_document_tag for d in new_remote_docs}
+        document_check = _remote_documents_check(remote_documents, entries)
+        remote_documents.update(entries) 
+        data=[ e for e in new_remote_docs if str(e.prefix) not in document_check.keys()]
+        doorstop_documents.extend(data)
+
+
+
+def _git_pull(git_url:str, git_tag:str, target_folder:Path):
+
+      def exact_want(refs, depth=None):
+                tag = b'refs/heads/' + git_tag.encode() if git_tag in ['main', 'master'] else b'refs/tags/' + git_tag.encode()
+                if tag in refs:
+                    return tag
+
+                raise DoorstopError(
+                    "ref {} not found in remote {}".format(git_tag, git_url)
+                )
+
+      path = str(target_folder.absolute())
+      with open(os.devnull, "wb") as f:
+        if not (target_folder / ".git").exists():
+            Repo.init(path, mkdir=False)
+        porcelain.pull(path, git_url, refspecs=exact_want(porcelain.fetch(path, git_url)), outstream=f,force=True)
+
+
+
+
+def _download_remote_parent(remote_base_path:Path, 
+                            remote_parent_url:str,
+                            remote_parent_tag: str, 
+                            doorstop_documents: list[Document], 
+                            remote_documents: dict[str,str]):
+
+    _create_remote_dir(remote_base_path)     
+
+    folder_name = remote_parent_url.split("/")[-1].split(".")[0] 
+
+    if not folder_name:
+        raise DoorstopError(
+                "Couldn't guess a name to storage remote parent, check the config for {}"
+                .format(remote_parent_url))
+
+    try:
+        target = remote_base_path / folder_name
+        _create_remote_dir(target)
+        _git_pull(remote_parent_url, remote_parent_tag, target) 
+        _validate_remote_documents(target , remote_parent_tag,remote_documents,doorstop_documents)
+    except Exception as e:
+        tb_str = ''.join(format_exception(None, e, e.__traceback__))
+        raise DoorstopError("""Unexpected error when downloading remote parent from {}
+ERROR Details: 
+    {} """.format(remote_parent_url, tb_str))
+

doorstop/core/document.py

@@ -77,6 +77,8 @@ def __init__(self, path, root=os.getcwd(), **kwargs):
         self._items: List[Item] = []
         self._itered = False
         self.children: List[Document] = []
+        self.remote_parent_url: str|None = None 
+        self.remote_parent_tag: str|None = None 
 
         if not self._data["itemformat"]:
             self._data["itemformat"] = Item.DEFAULT_ITEMFORMAT
@@ -236,6 +238,8 @@ def load(self, reload=False):
                 raise DoorstopError(msg)
 
         self.extensions = data.get("extensions", {})
+        self.remote_parent_url = data.get("remote_parent_url", None)
+        self.remote_parent_tag = data.get("remote_parent_tag", None)
 
         # Set meta attributes
         self._loaded = True

pyproject.toml

@@ -55,6 +55,7 @@ plantuml-markdown = "^3.4.2"
 six = "*" # fixes https://github.com/dougn/python-plantuml/issues/11
 openpyxl = ">=3.1.2"
 setuptools = { version = ">=70", python = ">=3.12" }
+dulwich = "^0.22.1"
 
 [tool.poetry.dev-dependencies]
 

reqs/.doorstop.yml

@@ -1,6 +1,9 @@
+remote_parent_url: git@gitlab.codethink.co.uk:sif/process/trustable.git
+remote_parent_tag: main
 settings:
   digits: 3
   prefix: REQ
+  parent: T
   sep: ''
 attributes:
   defaults: