Initial work splitting porcelain (#2032)

dulwich/porcelain/lfs.py

@@ -0,0 +1,717 @@
+# lfs.py -- LFS porcelain
+# Copyright (C) 2013 Jelmer Vernooij <jelmer@jelmer.uk>
+#
+# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
+# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
+# General Public License as published by the Free Software Foundation; version 2.0
+# or (at your option) any later version. You can redistribute it and/or
+# modify it under the terms of either of these two licenses.
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# You should have received a copy of the licenses; if not, see
+# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
+# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
+# License, Version 2.0.
+#
+
+"""Porcelain functions for Git LFS support."""
+
+import fnmatch
+import logging
+import os
+import stat
+from collections.abc import Sequence
+from typing import Any
+
+from dulwich.index import (
+    ConflictedIndexEntry,
+    index_entry_from_stat,
+)
+from dulwich.objects import Blob, Commit, Tree
+from dulwich.refs import HEADREF, Ref
+from dulwich.repo import Repo
+
+
+def lfs_track(
+    repo: str | os.PathLike[str] | Repo = ".",
+    patterns: Sequence[str] | None = None,
+) -> list[str]:
+    """Track file patterns with Git LFS.
+
+    Args:
+      repo: Path to repository
+      patterns: List of file patterns to track (e.g., ["*.bin", "*.pdf"])
+                If None, returns current tracked patterns
+
+    Returns:
+      List of tracked patterns
+    """
+    from ..attrs import GitAttributes
+    from . import add, open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        gitattributes_path = os.path.join(r.path, ".gitattributes")
+
+        # Load existing GitAttributes
+        if os.path.exists(gitattributes_path):
+            gitattributes = GitAttributes.from_file(gitattributes_path)
+        else:
+            gitattributes = GitAttributes()
+
+        if patterns is None:
+            # Return current LFS tracked patterns
+            tracked = []
+            for pattern_obj, attrs in gitattributes:
+                if attrs.get(b"filter") == b"lfs":
+                    tracked.append(pattern_obj.pattern.decode())
+            return tracked
+
+        # Add new patterns
+        for pattern in patterns:
+            # Ensure pattern is bytes
+            pattern_bytes = pattern.encode() if isinstance(pattern, str) else pattern
+
+            # Set LFS attributes for the pattern
+            gitattributes.set_attribute(pattern_bytes, b"filter", b"lfs")
+            gitattributes.set_attribute(pattern_bytes, b"diff", b"lfs")
+            gitattributes.set_attribute(pattern_bytes, b"merge", b"lfs")
+            gitattributes.set_attribute(pattern_bytes, b"text", False)
+
+        # Write updated attributes
+        gitattributes.write_to_file(gitattributes_path)
+
+        # Stage the .gitattributes file
+        add(r, [".gitattributes"])
+
+        return lfs_track(r)  # Return updated list
+
+
+def lfs_untrack(
+    repo: str | os.PathLike[str] | Repo = ".",
+    patterns: Sequence[str] | None = None,
+) -> list[str]:
+    """Untrack file patterns from Git LFS.
+
+    Args:
+      repo: Path to repository
+      patterns: List of file patterns to untrack
+
+    Returns:
+      List of remaining tracked patterns
+    """
+    from ..attrs import GitAttributes
+    from . import add, open_repo_closing
+
+    if not patterns:
+        return lfs_track(repo)
+
+    with open_repo_closing(repo) as r:
+        gitattributes_path = os.path.join(r.path, ".gitattributes")
+
+        if not os.path.exists(gitattributes_path):
+            return []
+
+        # Load existing GitAttributes
+        gitattributes = GitAttributes.from_file(gitattributes_path)
+
+        # Remove specified patterns
+        for pattern in patterns:
+            pattern_bytes = pattern.encode() if isinstance(pattern, str) else pattern
+
+            # Check if pattern is tracked by LFS
+            for pattern_obj, attrs in list(gitattributes):
+                if (
+                    pattern_obj.pattern == pattern_bytes
+                    and attrs.get(b"filter") == b"lfs"
+                ):
+                    gitattributes.remove_pattern(pattern_bytes)
+                    break
+
+        # Write updated attributes
+        gitattributes.write_to_file(gitattributes_path)
+
+        # Stage the .gitattributes file
+        add(r, [".gitattributes"])
+
+        return lfs_track(r)  # Return updated list
+
+
+def lfs_init(repo: str | os.PathLike[str] | Repo = ".") -> None:
+    """Initialize Git LFS in a repository.
+
+    Args:
+      repo: Path to repository
+
+    Returns:
+      None
+    """
+    from ..lfs import LFSStore
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        # Create LFS store
+        LFSStore.from_repo(r, create=True)
+
+        # Set up Git config for LFS
+        config = r.get_config()
+        config.set((b"filter", b"lfs"), b"process", b"git-lfs filter-process")
+        config.set((b"filter", b"lfs"), b"required", b"true")
+        config.set((b"filter", b"lfs"), b"clean", b"git-lfs clean -- %f")
+        config.set((b"filter", b"lfs"), b"smudge", b"git-lfs smudge -- %f")
+        config.write_to_path()
+
+
+def lfs_clean(
+    repo: str | os.PathLike[str] | Repo = ".",
+    path: str | os.PathLike[str] | None = None,
+) -> bytes:
+    """Clean a file by converting it to an LFS pointer.
+
+    Args:
+      repo: Path to repository
+      path: Path to file to clean (relative to repo root)
+
+    Returns:
+      LFS pointer content as bytes
+    """
+    from ..lfs import LFSFilterDriver, LFSStore
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        if path is None:
+            raise ValueError("Path must be specified")
+
+        # Get LFS store
+        lfs_store = LFSStore.from_repo(r)
+        filter_driver = LFSFilterDriver(lfs_store, config=r.get_config())
+
+        # Read file content
+        full_path = os.path.join(r.path, path)
+        with open(full_path, "rb") as f:
+            content = f.read()
+
+        # Clean the content (convert to LFS pointer)
+        return filter_driver.clean(content)
+
+
+def lfs_smudge(
+    repo: str | os.PathLike[str] | Repo = ".",
+    pointer_content: bytes | None = None,
+) -> bytes:
+    """Smudge an LFS pointer by retrieving the actual content.
+
+    Args:
+      repo: Path to repository
+      pointer_content: LFS pointer content as bytes
+
+    Returns:
+      Actual file content as bytes
+    """
+    from ..lfs import LFSFilterDriver, LFSStore
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        if pointer_content is None:
+            raise ValueError("Pointer content must be specified")
+
+        # Get LFS store
+        lfs_store = LFSStore.from_repo(r)
+        filter_driver = LFSFilterDriver(lfs_store, config=r.get_config())
+
+        # Smudge the pointer (retrieve actual content)
+        return filter_driver.smudge(pointer_content)
+
+
+def lfs_ls_files(
+    repo: str | os.PathLike[str] | Repo = ".",
+    ref: str | bytes | None = None,
+) -> list[tuple[bytes, str, int]]:
+    """List files tracked by Git LFS.
+
+    Args:
+      repo: Path to repository
+      ref: Git ref to check (defaults to HEAD)
+
+    Returns:
+      List of (path, oid, size) tuples for LFS files
+    """
+    from ..lfs import LFSPointer
+    from ..object_store import iter_tree_contents
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        if ref is None:
+            ref = b"HEAD"
+        elif isinstance(ref, str):
+            ref = ref.encode()
+
+        # Get the commit and tree
+        try:
+            commit = r[ref]
+            assert isinstance(commit, Commit)
+            tree = r[commit.tree]
+            assert isinstance(tree, Tree)
+        except KeyError:
+            return []
+
+        lfs_files = []
+
+        # Walk the tree
+        for path, mode, sha in iter_tree_contents(r.object_store, tree.id):
+            assert path is not None
+            assert mode is not None
+            assert sha is not None
+            if not stat.S_ISREG(mode):
+                continue
+
+            # Check if it's an LFS pointer
+            obj = r.object_store[sha]
+            if not isinstance(obj, Blob):
+                raise AssertionError(f"Expected Blob object, got {type(obj).__name__}")
+            pointer = LFSPointer.from_bytes(obj.data)
+            if pointer is not None:
+                lfs_files.append((path, pointer.oid, pointer.size))
+
+        return lfs_files
+
+
+def lfs_migrate(
+    repo: str | os.PathLike[str] | Repo = ".",
+    include: list[str] | None = None,
+    exclude: list[str] | None = None,
+    everything: bool = False,
+) -> int:
+    """Migrate files to Git LFS.
+
+    Args:
+      repo: Path to repository
+      include: Patterns of files to include
+      exclude: Patterns of files to exclude
+      everything: Migrate all files above a certain size
+
+    Returns:
+      Number of migrated files
+    """
+    from ..lfs import LFSFilterDriver, LFSStore
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        # Initialize LFS if needed
+        lfs_store = LFSStore.from_repo(r, create=True)
+        filter_driver = LFSFilterDriver(lfs_store, config=r.get_config())
+
+        # Get current index
+        index = r.open_index()
+
+        migrated = 0
+
+        # Determine files to migrate
+        files_to_migrate = []
+
+        if everything:
+            # Migrate all files above 100MB
+            for path, entry in index.items():
+                full_path = os.path.join(r.path, path.decode())
+                if os.path.exists(full_path):
+                    size = os.path.getsize(full_path)
+                    if size > 100 * 1024 * 1024:  # 100MB
+                        files_to_migrate.append(path.decode())
+        else:
+            # Use include/exclude patterns
+            for path, entry in index.items():
+                path_str = path.decode()
+
+                # Check include patterns
+                if include:
+                    matched = any(
+                        fnmatch.fnmatch(path_str, pattern) for pattern in include
+                    )
+                    if not matched:
+                        continue
+
+                # Check exclude patterns
+                if exclude:
+                    excluded = any(
+                        fnmatch.fnmatch(path_str, pattern) for pattern in exclude
+                    )
+                    if excluded:
+                        continue
+
+                files_to_migrate.append(path_str)
+
+        # Migrate files
+        for path_str in files_to_migrate:
+            full_path = os.path.join(r.path, path_str)
+            if not os.path.exists(full_path):
+                continue
+
+            # Read file content
+            with open(full_path, "rb") as f:
+                content = f.read()
+
+            # Convert to LFS pointer
+            pointer_content = filter_driver.clean(content)
+
+            # Write pointer back to file
+            with open(full_path, "wb") as f:
+                f.write(pointer_content)
+
+            # Create blob for pointer content and update index
+            blob = Blob()
+            blob.data = pointer_content
+            r.object_store.add_object(blob)
+
+            st = os.stat(full_path)
+            index_entry = index_entry_from_stat(st, blob.id, 0)
+            path_bytes = path_str.encode() if isinstance(path_str, str) else path_str
+            index[path_bytes] = index_entry
+
+            migrated += 1
+
+        # Write updated index
+        index.write()
+
+        # Track patterns if include was specified
+        if include:
+            lfs_track(r, include)
+
+        return migrated
+
+
+def lfs_pointer_check(
+    repo: str | os.PathLike[str] | Repo = ".",
+    paths: Sequence[str] | None = None,
+) -> dict[str, Any | None]:
+    """Check if files are valid LFS pointers.
+
+    Args:
+      repo: Path to repository
+      paths: List of file paths to check (if None, check all files)
+
+    Returns:
+      Dict mapping paths to LFSPointer objects (or None if not a pointer)
+    """
+    from ..lfs import LFSPointer
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        results = {}
+
+        if paths is None:
+            # Check all files in index
+            index = r.open_index()
+            paths = [path.decode() for path in index]
+
+        for path in paths:
+            full_path = os.path.join(r.path, path)
+            if os.path.exists(full_path):
+                try:
+                    with open(full_path, "rb") as f:
+                        content = f.read()
+                    pointer = LFSPointer.from_bytes(content)
+                    results[path] = pointer
+                except OSError:
+                    results[path] = None
+            else:
+                results[path] = None
+
+        return results
+
+
+def lfs_fetch(
+    repo: str | os.PathLike[str] | Repo = ".",
+    remote: str = "origin",
+    refs: list[str | bytes] | None = None,
+) -> int:
+    """Fetch LFS objects from remote.
+
+    Args:
+      repo: Path to repository
+      remote: Remote name (default: origin)
+      refs: Specific refs to fetch LFS objects for (default: all refs)
+
+    Returns:
+      Number of objects fetched
+    """
+    from ..lfs import LFSClient, LFSPointer, LFSStore
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        # Get LFS server URL from config
+        config = r.get_config()
+        lfs_url_bytes = config.get((b"lfs",), b"url")
+        if not lfs_url_bytes:
+            # Try remote URL
+            remote_url = config.get((b"remote", remote.encode()), b"url")
+            if remote_url:
+                # Append /info/lfs to remote URL
+                remote_url_str = remote_url.decode()
+                if remote_url_str.endswith(".git"):
+                    remote_url_str = remote_url_str[:-4]
+                lfs_url = f"{remote_url_str}/info/lfs"
+            else:
+                raise ValueError(f"No LFS URL configured for remote {remote}")
+        else:
+            lfs_url = lfs_url_bytes.decode()
+
+        # Get authentication
+        auth = None
+        # TODO: Support credential helpers and other auth methods
+
+        # Create LFS client and store
+        client = LFSClient(lfs_url, auth)
+        store = LFSStore.from_repo(r)
+
+        # Find all LFS pointers in the refs
+        pointers_to_fetch = []
+
+        if refs is None:
+            # Get all refs
+            refs = list(r.refs.keys())
+
+        for ref in refs:
+            if isinstance(ref, str):
+                ref_key = Ref(ref.encode())
+            elif isinstance(ref, bytes):
+                ref_key = Ref(ref)
+            else:
+                ref_key = ref
+            try:
+                commit = r[r.refs[ref_key]]
+            except KeyError:
+                continue
+
+            # Walk the commit tree
+            assert isinstance(commit, Commit)
+            for path, mode, sha in r.object_store.iter_tree_contents(commit.tree):
+                assert sha is not None
+                try:
+                    obj = r.object_store[sha]
+                except KeyError:
+                    pass
+                else:
+                    if isinstance(obj, Blob):
+                        pointer = LFSPointer.from_bytes(obj.data)
+                        if pointer and pointer.is_valid_oid():
+                            # Check if we already have it
+                            try:
+                                with store.open_object(pointer.oid):
+                                    pass  # Object exists, no need to fetch
+                            except KeyError:
+                                pointers_to_fetch.append((pointer.oid, pointer.size))
+
+        # Fetch missing objects
+        fetched = 0
+        for oid, size in pointers_to_fetch:
+            content = client.download(oid, size)
+            store.write_object([content])
+            fetched += 1
+
+        return fetched
+
+
+def lfs_pull(repo: str | os.PathLike[str] | Repo = ".", remote: str = "origin") -> int:
+    """Pull LFS objects for current checkout.
+
+    Args:
+      repo: Path to repository
+      remote: Remote name (default: origin)
+
+    Returns:
+      Number of objects fetched
+    """
+    from ..lfs import LFSPointer, LFSStore
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        # First do a fetch for HEAD
+        fetched = lfs_fetch(repo, remote, [b"HEAD"])
+
+        # Then checkout LFS files in working directory
+        store = LFSStore.from_repo(r)
+        index = r.open_index()
+
+        for path, entry in index.items():
+            full_path = os.path.join(r.path, path.decode())
+            if os.path.exists(full_path):
+                with open(full_path, "rb") as f:
+                    content = f.read()
+
+                pointer = LFSPointer.from_bytes(content)
+                if pointer and pointer.is_valid_oid():
+                    try:
+                        # Replace pointer with actual content
+                        with store.open_object(pointer.oid) as lfs_file:
+                            lfs_content = lfs_file.read()
+                        with open(full_path, "wb") as f:
+                            f.write(lfs_content)
+                    except KeyError:
+                        # Object not available
+                        pass
+
+        return fetched
+
+
+def lfs_push(
+    repo: str | os.PathLike[str] | Repo = ".",
+    remote: str = "origin",
+    refs: list[str | bytes] | None = None,
+) -> int:
+    """Push LFS objects to remote.
+
+    Args:
+      repo: Path to repository
+      remote: Remote name (default: origin)
+      refs: Specific refs to push LFS objects for (default: current branch)
+
+    Returns:
+      Number of objects pushed
+    """
+    from ..lfs import LFSClient, LFSPointer, LFSStore
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        # Get LFS server URL from config
+        config = r.get_config()
+        lfs_url_bytes = config.get((b"lfs",), b"url")
+        if not lfs_url_bytes:
+            # Try remote URL
+            remote_url = config.get((b"remote", remote.encode()), b"url")
+            if remote_url:
+                # Append /info/lfs to remote URL
+                remote_url_str = remote_url.decode()
+                if remote_url_str.endswith(".git"):
+                    remote_url_str = remote_url_str[:-4]
+                lfs_url = f"{remote_url_str}/info/lfs"
+            else:
+                raise ValueError(f"No LFS URL configured for remote {remote}")
+        else:
+            lfs_url = lfs_url_bytes.decode()
+
+        # Get authentication
+        auth = None
+        # TODO: Support credential helpers and other auth methods
+
+        # Create LFS client and store
+        client = LFSClient(lfs_url, auth)
+        store = LFSStore.from_repo(r)
+
+        # Find all LFS objects to push
+        if refs is None:
+            # Push current branch
+            head_ref = r.refs.read_ref(HEADREF)
+            refs = [head_ref] if head_ref else []
+
+        objects_to_push = set()
+
+        for ref in refs:
+            if isinstance(ref, str):
+                ref_bytes = ref.encode()
+            else:
+                ref_bytes = ref
+            try:
+                if ref_bytes.startswith(b"refs/"):
+                    commit = r[r.refs[Ref(ref_bytes)]]
+                else:
+                    commit = r[ref_bytes]
+            except KeyError:
+                continue
+
+            # Walk the commit tree
+            assert isinstance(commit, Commit)
+            for path, mode, sha in r.object_store.iter_tree_contents(commit.tree):
+                assert sha is not None
+                try:
+                    obj = r.object_store[sha]
+                except KeyError:
+                    pass
+                else:
+                    if isinstance(obj, Blob):
+                        pointer = LFSPointer.from_bytes(obj.data)
+                        if pointer and pointer.is_valid_oid():
+                            objects_to_push.add((pointer.oid, pointer.size))
+
+        # Push objects
+        pushed = 0
+        for oid, size in objects_to_push:
+            try:
+                with store.open_object(oid) as f:
+                    content = f.read()
+            except KeyError:
+                # Object not in local store
+                logging.warn("LFS object %s not found locally", oid)
+            else:
+                client.upload(oid, size, content)
+                pushed += 1
+
+        return pushed
+
+
+def lfs_status(repo: str | os.PathLike[str] | Repo = ".") -> dict[str, list[str]]:
+    """Show status of LFS files.
+
+    Args:
+      repo: Path to repository
+
+    Returns:
+      Dict with status information
+    """
+    from ..lfs import LFSPointer, LFSStore
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        store = LFSStore.from_repo(r)
+        index = r.open_index()
+
+        status: dict[str, list[str]] = {
+            "tracked": [],
+            "not_staged": [],
+            "not_committed": [],
+            "not_pushed": [],
+            "missing": [],
+        }
+
+        # Check working directory files
+        for path, entry in index.items():
+            path_str = path.decode()
+            full_path = os.path.join(r.path, path_str)
+
+            if os.path.exists(full_path):
+                with open(full_path, "rb") as f:
+                    content = f.read()
+
+                pointer = LFSPointer.from_bytes(content)
+                if pointer and pointer.is_valid_oid():
+                    status["tracked"].append(path_str)
+
+                    # Check if object exists locally
+                    try:
+                        with store.open_object(pointer.oid):
+                            pass  # Object exists locally
+                    except KeyError:
+                        status["missing"].append(path_str)
+
+                    # Check if file has been modified
+                    if isinstance(entry, ConflictedIndexEntry):
+                        continue  # Skip conflicted entries
+                    try:
+                        staged_obj = r.object_store[entry.sha]
+                    except KeyError:
+                        pass
+                    else:
+                        if not isinstance(staged_obj, Blob):
+                            raise AssertionError(
+                                f"Expected Blob object, got {type(staged_obj).__name__}"
+                            )
+                        staged_pointer = LFSPointer.from_bytes(staged_obj.data)
+                        if staged_pointer and staged_pointer.oid != pointer.oid:
+                            status["not_staged"].append(path_str)
+
+        # TODO: Check for not committed and not pushed files
+
+        return status

dulwich/porcelain/notes.py

@@ -0,0 +1,185 @@
+# notes.py -- Porcelain-like interface for Git notes
+# Copyright (C) 2013 Jelmer Vernooij <jelmer@jelmer.uk>
+#
+# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
+# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
+# General Public License as published by the Free Software Foundation; version 2.0
+# or (at your option) any later version. You can redistribute it and/or
+# modify it under the terms of either of these two licenses.
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# You should have received a copy of the licenses; if not, see
+# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
+# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
+# License, Version 2.0.
+#
+
+"""Porcelain-like interface for Git notes."""
+
+from typing import TYPE_CHECKING
+
+from dulwich.objects import ObjectID
+from dulwich.objectspec import parse_object
+
+from ..refs import LOCAL_NOTES_PREFIX
+
+if TYPE_CHECKING:
+    from . import RepoPath
+
+
+def _make_notes_ref(name: bytes) -> bytes:
+    """Make a notes ref name."""
+    if name.startswith(b"refs/notes/"):
+        return name
+    return LOCAL_NOTES_PREFIX + name
+
+
+def notes_add(
+    repo: "RepoPath",
+    object_sha: bytes,
+    note: bytes,
+    ref: bytes = b"commits",
+    author: bytes | None = None,
+    committer: bytes | None = None,
+    message: bytes | None = None,
+) -> bytes:
+    """Add or update a note for an object.
+
+    Args:
+      repo: Path to repository
+      object_sha: SHA of the object to annotate
+      note: Note content
+      ref: Notes ref to use (defaults to "commits" for refs/notes/commits)
+      author: Author identity (defaults to committer)
+      committer: Committer identity (defaults to config)
+      message: Commit message for the notes update
+
+    Returns:
+      SHA of the new notes commit
+    """
+    from . import DEFAULT_ENCODING, open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        # Parse the object to get its SHA
+        obj = parse_object(r, object_sha)
+        object_sha = obj.id
+
+        if isinstance(note, str):
+            note = note.encode(DEFAULT_ENCODING)
+        if isinstance(ref, str):
+            ref = ref.encode(DEFAULT_ENCODING)
+
+        notes_ref = _make_notes_ref(ref)
+        config = r.get_config_stack()
+
+        return r.notes.set_note(
+            object_sha,
+            note,
+            notes_ref,
+            author=author,
+            committer=committer,
+            message=message,
+            config=config,
+        )
+
+
+def notes_remove(
+    repo: "RepoPath",
+    object_sha: bytes,
+    ref: bytes = b"commits",
+    author: bytes | None = None,
+    committer: bytes | None = None,
+    message: bytes | None = None,
+) -> bytes | None:
+    """Remove a note for an object.
+
+    Args:
+      repo: Path to repository
+      object_sha: SHA of the object to remove notes from
+      ref: Notes ref to use (defaults to "commits" for refs/notes/commits)
+      author: Author identity (defaults to committer)
+      committer: Committer identity (defaults to config)
+      message: Commit message for the notes removal
+
+    Returns:
+      SHA of the new notes commit, or None if no note existed
+    """
+    from . import DEFAULT_ENCODING, open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        # Parse the object to get its SHA
+        obj = parse_object(r, object_sha)
+        object_sha = obj.id
+
+        if isinstance(ref, str):
+            ref = ref.encode(DEFAULT_ENCODING)
+
+        notes_ref = _make_notes_ref(ref)
+        config = r.get_config_stack()
+
+        return r.notes.remove_note(
+            object_sha,
+            notes_ref,
+            author=author,
+            committer=committer,
+            message=message,
+            config=config,
+        )
+
+
+def notes_show(
+    repo: "RepoPath", object_sha: bytes, ref: bytes = b"commits"
+) -> bytes | None:
+    """Show the note for an object.
+
+    Args:
+      repo: Path to repository
+      object_sha: SHA of the object
+      ref: Notes ref to use (defaults to "commits" for refs/notes/commits)
+
+    Returns:
+      Note content as bytes, or None if no note exists
+    """
+    from . import DEFAULT_ENCODING, open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        # Parse the object to get its SHA
+        obj = parse_object(r, object_sha)
+        object_sha = obj.id
+
+        if isinstance(ref, str):
+            ref = ref.encode(DEFAULT_ENCODING)
+
+        notes_ref = _make_notes_ref(ref)
+        config = r.get_config_stack()
+
+        return r.notes.get_note(object_sha, notes_ref, config=config)
+
+
+def notes_list(
+    repo: "RepoPath", ref: bytes = b"commits"
+) -> list[tuple[ObjectID, bytes]]:
+    """List all notes in a notes ref.
+
+    Args:
+      repo: Path to repository
+      ref: Notes ref to use (defaults to "commits" for refs/notes/commits)
+
+    Returns:
+      List of tuples of (object_sha, note_content)
+    """
+    from . import DEFAULT_ENCODING, open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        if isinstance(ref, str):
+            ref = ref.encode(DEFAULT_ENCODING)
+
+        notes_ref = _make_notes_ref(ref)
+        config = r.get_config_stack()
+
+        return r.notes.list_notes(notes_ref, config=config)

dulwich/porcelain/submodule.py

@@ -0,0 +1,259 @@
+# submodule.py -- Submodule porcelain
+# Copyright (C) 2013 Jelmer Vernooij <jelmer@jelmer.uk>
+#
+# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
+# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
+# General Public License as published by the Free Software Foundation; version 2.0
+# or (at your option) any later version. You can redistribute it and/or
+# modify it under the terms of either of these two licenses.
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# You should have received a copy of the licenses; if not, see
+# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
+# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
+# License, Version 2.0.
+#
+
+"""Porcelain functions for working with submodules."""
+
+import os
+from collections.abc import Iterator, Sequence
+from typing import TYPE_CHECKING, BinaryIO
+
+from ..config import ConfigFile, read_submodules
+from ..objects import Commit
+from ..repo import Repo
+
+if TYPE_CHECKING:
+    from . import RepoPath
+
+
+def submodule_add(
+    repo: str | os.PathLike[str] | Repo,
+    url: str,
+    path: str | os.PathLike[str] | None = None,
+    name: str | None = None,
+) -> None:
+    """Add a new submodule.
+
+    Args:
+      repo: Path to repository
+      url: URL of repository to add as submodule
+      path: Path where submodule should live
+      name: Name for the submodule
+    """
+    from . import Error, _canonical_part, open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        if path is None:
+            path = os.path.relpath(_canonical_part(url), r.path)
+        if name is None:
+            name = os.fsdecode(path) if path is not None else None
+
+        if name is None:
+            raise Error("Submodule name must be specified or derivable from path")
+
+        # TODO(jelmer): Move this logic to dulwich.submodule
+        gitmodules_path = os.path.join(r.path, ".gitmodules")
+        try:
+            config = ConfigFile.from_path(gitmodules_path)
+        except FileNotFoundError:
+            config = ConfigFile()
+            config.path = gitmodules_path
+        config.set(("submodule", name), "url", url)
+        config.set(("submodule", name), "path", os.fsdecode(path))
+        config.write_to_path()
+
+
+def submodule_init(repo: str | os.PathLike[str] | Repo) -> None:
+    """Initialize submodules.
+
+    Args:
+      repo: Path to repository
+    """
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        config = r.get_config()
+        gitmodules_path = os.path.join(r.path, ".gitmodules")
+        for path, url, name in read_submodules(gitmodules_path):
+            config.set((b"submodule", name), b"active", True)
+            config.set((b"submodule", name), b"url", url)
+        config.write_to_path()
+
+
+def submodule_list(repo: "RepoPath") -> Iterator[tuple[str, str]]:
+    """List submodules.
+
+    Args:
+      repo: Path to repository
+    """
+    from ..submodule import iter_cached_submodules
+    from . import DEFAULT_ENCODING, open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        head_commit = r[r.head()]
+        assert isinstance(head_commit, Commit)
+        for path, sha in iter_cached_submodules(r.object_store, head_commit.tree):
+            yield path.decode(DEFAULT_ENCODING), sha.decode(DEFAULT_ENCODING)
+
+
+def submodule_update(
+    repo: str | os.PathLike[str] | Repo,
+    paths: Sequence[str | bytes | os.PathLike[str]] | None = None,
+    init: bool = False,
+    force: bool = False,
+    recursive: bool = False,
+    errstream: BinaryIO | None = None,
+) -> None:
+    """Update submodules.
+
+    Args:
+      repo: Path to repository
+      paths: Optional list of specific submodule paths to update. If None, updates all.
+      init: If True, initialize submodules first
+      force: Force update even if local changes exist
+      recursive: If True, recursively update nested submodules
+      errstream: Error stream for error messages
+    """
+    from ..client import get_transport_and_path
+    from ..index import build_index_from_tree
+    from ..refs import HEADREF
+    from ..submodule import iter_cached_submodules
+    from . import (
+        DEFAULT_ENCODING,
+        clone,
+        open_repo_closing,
+        reset,
+    )
+
+    with open_repo_closing(repo) as r:
+        if init:
+            submodule_init(r)
+
+        config = r.get_config()
+        gitmodules_path = os.path.join(r.path, ".gitmodules")
+
+        # Get list of submodules to update
+        submodules_to_update = []
+        head_commit = r[r.head()]
+        assert isinstance(head_commit, Commit)
+        for path, sha in iter_cached_submodules(r.object_store, head_commit.tree):
+            path_str = (
+                path.decode(DEFAULT_ENCODING) if isinstance(path, bytes) else path
+            )
+            if paths is None or path_str in paths:
+                submodules_to_update.append((path, sha))
+
+        # Read submodule configuration
+        for path, target_sha in submodules_to_update:
+            path_str = (
+                path.decode(DEFAULT_ENCODING) if isinstance(path, bytes) else path
+            )
+
+            # Find the submodule name from ..gitmodules
+            submodule_name: bytes | None = None
+            for sm_path, sm_url, sm_name in read_submodules(gitmodules_path):
+                if sm_path == path:
+                    submodule_name = sm_name
+                    break
+
+            if not submodule_name:
+                continue
+
+            # Get the URL from config
+            section = (
+                b"submodule",
+                submodule_name
+                if isinstance(submodule_name, bytes)
+                else submodule_name.encode(),
+            )
+            try:
+                url_value = config.get(section, b"url")
+                if isinstance(url_value, bytes):
+                    url = url_value.decode(DEFAULT_ENCODING)
+                else:
+                    url = url_value
+            except KeyError:
+                # URL not in config, skip this submodule
+                continue
+
+            # Get or create the submodule repository paths
+            submodule_path = os.path.join(r.path, path_str)
+            submodule_git_dir = os.path.join(r.controldir(), "modules", path_str)
+
+            # Clone or fetch the submodule
+            if not os.path.exists(submodule_git_dir):
+                # Clone the submodule as bare repository
+                os.makedirs(os.path.dirname(submodule_git_dir), exist_ok=True)
+
+                # Clone to the git directory
+                sub_repo = clone(url, submodule_git_dir, bare=True, checkout=False)
+                sub_repo.close()
+
+                # Create the submodule directory if it doesn't exist
+                if not os.path.exists(submodule_path):
+                    os.makedirs(submodule_path)
+
+                # Create .git file in the submodule directory
+                relative_git_dir = os.path.relpath(submodule_git_dir, submodule_path)
+                git_file_path = os.path.join(submodule_path, ".git")
+                with open(git_file_path, "w") as f:
+                    f.write(f"gitdir: {relative_git_dir}\n")
+
+                # Set up working directory configuration
+                with open_repo_closing(submodule_git_dir) as sub_repo:
+                    sub_config = sub_repo.get_config()
+                    sub_config.set(
+                        (b"core",),
+                        b"worktree",
+                        os.path.abspath(submodule_path).encode(),
+                    )
+                    sub_config.write_to_path()
+
+                    # Checkout the target commit
+                    sub_repo.refs[HEADREF] = target_sha
+
+                    # Build the index and checkout files
+                    tree = sub_repo[target_sha]
+                    if hasattr(tree, "tree"):  # If it's a commit, get the tree
+                        tree_id = tree.tree
+                    else:
+                        tree_id = target_sha
+
+                    build_index_from_tree(
+                        submodule_path,
+                        sub_repo.index_path(),
+                        sub_repo.object_store,
+                        tree_id,
+                    )
+            else:
+                # Fetch and checkout in existing submodule
+                with open_repo_closing(submodule_git_dir) as sub_repo:
+                    # Fetch from remote
+                    client, path_segments = get_transport_and_path(url)
+                    client.fetch(path_segments.encode(), sub_repo)
+
+                    # Update to the target commit
+                    sub_repo.refs[HEADREF] = target_sha
+
+                    # Reset the working directory
+                    reset(sub_repo, "hard", target_sha)
+
+            # Recursively update nested submodules if requested
+            if recursive:
+                submodule_gitmodules = os.path.join(submodule_path, ".gitmodules")
+                if os.path.exists(submodule_gitmodules):
+                    submodule_update(
+                        submodule_path,
+                        paths=None,
+                        init=True,  # Always initialize nested submodules
+                        force=force,
+                        recursive=True,
+                        errstream=errstream,
+                    )

dulwich/porcelain/tag.py

@@ -0,0 +1,216 @@
+# tag.py -- Porcelain-like tag functions for Dulwich
+# Copyright (C) 2013 Jelmer Vernooij <jelmer@jelmer.uk>
+#
+# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
+# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
+# General Public License as published by the Free Software Foundation; version 2.0
+# or (at your option) any later version. You can redistribute it and/or
+# modify it under the terms of either of these two licenses.
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# You should have received a copy of the licenses; if not, see
+# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
+# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
+# License, Version 2.0.
+#
+
+"""Porcelain-like tag functions for Dulwich."""
+
+import sys
+import time
+from typing import TYPE_CHECKING, TextIO
+
+from dulwich.objects import Tag, parse_timezone
+
+from ..objectspec import (
+    parse_object,
+)
+from ..refs import (
+    Ref,
+    local_tag_name,
+)
+from ..repo import get_user_identity
+
+if TYPE_CHECKING:
+    from . import RepoPath
+
+
+def _make_tag_ref(name: str | bytes) -> Ref:
+    from . import DEFAULT_ENCODING
+
+    if isinstance(name, str):
+        name = name.encode(DEFAULT_ENCODING)
+    return local_tag_name(name)
+
+
+def verify_tag(
+    repo: "RepoPath",
+    tagname: str | bytes,
+    keyids: list[str] | None = None,
+) -> None:
+    """Verify GPG signature on a tag.
+
+    Args:
+      repo: Path to repository
+      tagname: Name of tag to verify
+      keyids: Optional list of trusted key IDs. If provided, the tag
+        must be signed by one of these keys. If not provided, just verifies
+        that the tag has a valid signature.
+
+    Raises:
+      gpg.errors.BadSignatures: if GPG signature verification fails
+      gpg.errors.MissingSignatures: if tag was not signed by a key
+        specified in keyids
+    """
+    from . import Error, open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        if isinstance(tagname, str):
+            tagname = tagname.encode()
+        tag_ref = _make_tag_ref(tagname)
+        tag_id = r.refs[tag_ref]
+        tag_obj = r[tag_id]
+        if not isinstance(tag_obj, Tag):
+            raise Error(f"{tagname!r} does not point to a tag object")
+        tag_obj.verify(keyids)
+
+
+def tag_create(
+    repo: "RepoPath",
+    tag: str | bytes,
+    author: str | bytes | None = None,
+    message: str | bytes | None = None,
+    annotated: bool = False,
+    objectish: str | bytes = "HEAD",
+    tag_time: int | None = None,
+    tag_timezone: int | None = None,
+    sign: bool | None = None,
+    encoding: str | None = None,
+) -> None:
+    """Creates a tag in git via dulwich calls.
+
+    Args:
+      repo: Path to repository
+      tag: tag string
+      author: tag author (optional, if annotated is set)
+      message: tag message (optional)
+      annotated: whether to create an annotated tag
+      objectish: object the tag should point at, defaults to HEAD
+      tag_time: Optional time for annotated tag
+      tag_timezone: Optional timezone for annotated tag
+      sign: GPG Sign the tag (bool, defaults to False,
+        pass True to use default GPG key,
+        pass a str containing Key ID to use a specific GPG key)
+      encoding: Encoding to use for tag messages
+    """
+    from . import (
+        DEFAULT_ENCODING,
+        get_user_timezones,
+        open_repo_closing,
+    )
+
+    if encoding is None:
+        encoding = DEFAULT_ENCODING
+    with open_repo_closing(repo) as r:
+        object = parse_object(r, objectish)
+
+        if isinstance(tag, str):
+            tag = tag.encode(encoding)
+
+        if annotated:
+            # Create the tag object
+            tag_obj = Tag()
+            if author is None:
+                author = get_user_identity(r.get_config_stack())
+            elif isinstance(author, str):
+                author = author.encode(encoding)
+            else:
+                assert isinstance(author, bytes)
+            tag_obj.tagger = author
+            if isinstance(message, str):
+                message = message.encode(encoding)
+            elif isinstance(message, bytes):
+                pass
+            else:
+                message = b""
+            tag_obj.message = message + "\n".encode(encoding)
+            tag_obj.name = tag
+            tag_obj.object = (type(object), object.id)
+            if tag_time is None:
+                tag_time = int(time.time())
+            tag_obj.tag_time = tag_time
+            if tag_timezone is None:
+                tag_timezone = get_user_timezones()[1]
+            elif isinstance(tag_timezone, str):
+                tag_timezone = parse_timezone(tag_timezone.encode())
+            tag_obj.tag_timezone = tag_timezone
+
+            # Check if we should sign the tag
+            config = r.get_config_stack()
+
+            if sign is None:
+                # Check tag.gpgSign configuration when sign is not explicitly set
+                try:
+                    should_sign = config.get_boolean(
+                        (b"tag",), b"gpgsign", default=False
+                    )
+                except KeyError:
+                    should_sign = False  # Default to not signing if no config
+            else:
+                should_sign = sign
+
+            # Get the signing key from config if signing is enabled
+            keyid = None
+            if should_sign:
+                try:
+                    keyid_bytes = config.get((b"user",), b"signingkey")
+                    keyid = keyid_bytes.decode() if keyid_bytes else None
+                except KeyError:
+                    keyid = None
+                tag_obj.sign(keyid)
+
+            r.object_store.add_object(tag_obj)
+            tag_id = tag_obj.id
+        else:
+            tag_id = object.id
+
+        r.refs[_make_tag_ref(tag)] = tag_id
+
+
+def tag_list(repo: "RepoPath", outstream: TextIO = sys.stdout) -> list[Ref]:
+    """List all tags.
+
+    Args:
+      repo: Path to repository
+      outstream: Stream to write tags to
+    """
+    from . import open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        tags: list[Ref] = sorted(r.refs.as_dict(Ref(b"refs/tags")))
+        return tags
+
+
+def tag_delete(repo: "RepoPath", name: str | bytes) -> None:
+    """Remove a tag.
+
+    Args:
+      repo: Path to repository
+      name: Name of tag to remove
+    """
+    from . import Error, open_repo_closing
+
+    with open_repo_closing(repo) as r:
+        if isinstance(name, bytes):
+            names = [name]
+        elif isinstance(name, list):
+            names = name
+        else:
+            raise Error(f"Unexpected tag name type {name!r}")
+        for name in names:
+            del r.refs[_make_tag_ref(name)]

tests/__init__.py

@@ -171,12 +171,6 @@ def self_test_suite() -> unittest.TestSuite:
         "object_store",
         "pack",
         "patch",
-        "porcelain",
-        "porcelain_cherry_pick",
-        "porcelain_filters",
-        "porcelain_lfs",
-        "porcelain_merge",
-        "porcelain_notes",
         "protocol",
         "rebase",
         "reflog",
@@ -197,6 +191,16 @@ def self_test_suite() -> unittest.TestSuite:
         "worktree",
     ]
     module_names = ["tests.test_" + name for name in names]
+    porcelain_names = [
+        "cherry_pick",
+        "filters",
+        "lfs",
+        "merge",
+        "notes",
+    ]
+    module_names += ["tests.porcelain"] + [
+        "tests.porcelain.test_" + name for name in porcelain_names
+    ]
     loader = unittest.TestLoader()
     return loader.loadTestsFromNames(module_names)
 

tests/compat/test_porcelain.py

@@ -27,7 +27,7 @@ from unittest import skipIf
 from dulwich import porcelain
 from dulwich.tests.utils import build_commit_graph
 
-from ..test_porcelain import PorcelainGpgTestCase
+from ..porcelain import PorcelainGpgTestCase
 from .utils import CompatTestCase, run_git_or_fail
 
 try:

tests/test_porcelain.py → tests/porcelain/__init__.py

@@ -53,7 +53,7 @@ from dulwich.server import DictBackend
 from dulwich.tests.utils import build_commit_graph, make_commit, make_object
 from dulwich.web import make_server, make_wsgi_chain
 
-from . import TestCase
+from .. import TestCase
 
 try:
     import gpg

tests/test_porcelain_cherry_pick.py → tests/porcelain/test_cherry_pick.py

@@ -27,7 +27,7 @@ import tempfile
 
 from dulwich import porcelain
 
-from . import DependencyMissing, TestCase
+from .. import DependencyMissing, TestCase
 
 
 class PorcelainCherryPickTests(TestCase):

tests/test_porcelain_filters.py → tests/porcelain/test_filters.py

@@ -29,8 +29,8 @@ from io import BytesIO
 from dulwich import porcelain
 from dulwich.repo import Repo
 
-from . import TestCase
-from .compat.utils import rmtree_ro
+from .. import TestCase
+from ..compat.utils import rmtree_ro
 
 
 class PorcelainFilterTests(TestCase):

tests/test_porcelain_merge.py → tests/porcelain/test_merge.py

@@ -29,7 +29,7 @@ import unittest
 from dulwich import porcelain
 from dulwich.repo import Repo
 
-from . import DependencyMissing, TestCase
+from .. import DependencyMissing, TestCase
 
 
 class PorcelainMergeTests(TestCase):