def get()

in sync/worktree.py [0:0]

• 33 lines of code
• 11 McCabe index (conditional complexity)

    def get(self) -> Repo:
        """Return the worktree.

        On first access, the worktree is reset to the current HEAD. Subsequent
        access doesn't perform the same check, so it's possible to retain state
        within a specific process."""
        # TODO: We can get the worktree to only checkout the paths we actually
        # need.
        # To do this we have to
        # * Enable sparse checkouts by setting core.sparseCheckouts
        # * Add the worktree with --no-checkout
        # * Add the list of paths to check out under $REPO/worktrees/info/sparse-checkout
        # * Go to the worktree and check it out
        if self._worktree is None:
            all_worktrees = {item.name: item for item in worktrees(self.pygit2_repo)}
            count = len(all_worktrees)
            max_count = get_max_worktree_count(self.repo)
            if max_count and count >= max_count:
                cleanup_repo(self.repo, self.pygit2_repo, max_count - 1)

            path_exists = os.path.exists(self.path)

            if self.worktree_name in all_worktrees and not path_exists:
                prune_worktrees(self.pygit2_repo)
                del all_worktrees[self.worktree_name]

            if self.worktree_name not in all_worktrees:
                if path_exists:
                    logger.warning("Found existing content in worktree path %s, removing" %
                                   self.path)
                    shutil.rmtree(self.path)
                logger.info(f"Creating worktree {self.worktree_name} at {self.path}")
                if not os.path.exists(os.path.dirname(self.path)):
                    os.makedirs(os.path.dirname(self.path))
                worktree = self.pygit2_repo.add_worktree(self.worktree_name,
                                                         os.path.abspath(self.path),
                                                         self.pygit2_repo.lookup_reference(
                                                             "refs/heads/%s" % self.process_name))
                wrapper = wrapper_get(self.repo)
                assert wrapper is not None
                wrapper.after_worktree_create(self.path)
            else:
                worktree = self.pygit2_repo.lookup_worktree(self.worktree_name)

            assert os.path.exists(self.path)
            assert worktree.path == self.path
            self._worktree = git.Repo(self.path)

        # TODO: In general the worktree should be on the right branch, but it would
        # be good to check. In the specific case of landing, we move the wpt worktree
        # around various commits, so it isn't necessarily on the correct branch
        assert self._worktree is not None
        return self._worktree