# porcelain.py -- Porcelain-like layer on top of Dulwich # Copyright (C) 2013 Jelmer Vernooij # # 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 # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Simple wrapper that provides porcelain-like functions on top of Dulwich. Currently implemented: * am * apply_patch * archive * add * bisect{_start,_bad,_good,_skip,_reset,_log,_replay} * branch{_create,_delete,_list} * check_ignore * checkout * checkout_branch * clone * cone mode{_init, _set, _add} * commit * commit_tree * daemon * describe * diff_tree * fetch * filter_branch * for_each_ref * grep * init * interpret_trailers * ls_files * ls_remote * ls_tree * mailsplit * merge * merge_tree * mv/move * prune * pull * push * rm * remote{_add} * receive_pack * replace{_create,_delete,_list} * request_pull * rerere{_status,_diff,_forget,_clear,_gc} * reset * revert * sparse_checkout * submodule_add * submodule_init * submodule_list * rev_list * tag{_create,_delete,_list} * upload_pack * update_server_info * var * write_commit_graph * status * shortlog * symbolic_ref * worktree{_add,_list,_remove,_prune,_lock,_unlock,_move} These functions are meant to behave similarly to the git subcommands. Differences in behaviour are considered bugs. Note: one of the consequences of this is that paths tend to be interpreted relative to the current working directory rather than relative to the repository root. Functions should generally accept both unicode strings and bytestrings """ __all__ = [ "INFINITE_DEPTH", "CheckoutError", "CountObjectsResult", "DivergedBranches", "Error", "NoneStream", "RemoteExists", "TimezoneFormatError", "TransportKwargs", "active_branch", "add", "am", "am_abort", "am_continue", "am_quit", "am_skip", "annotate", "archive", "bisect_bad", "bisect_good", "bisect_log", "bisect_replay", "bisect_reset", "bisect_skip", "bisect_start", "branch_create", "branch_delete", "branch_list", "branch_remotes_list", "branches_containing", "cat_file_content", "cat_file_size", "cat_file_type", "check_diverged", "check_ignore", "check_mailmap", "checkout", "cherry", "cherry_pick", "clean", "clone", "commit", "commit_decode", "commit_encode", "commit_tree", "cone_mode_add", "cone_mode_disable", "cone_mode_init", "cone_mode_list", "cone_mode_set", "count_objects", "daemon", "describe", "diff", "diff_tree", "fetch", "filter_branch", "filter_branches_by_pattern", "find_unique_abbrev", "for_each_ref", "format_patch", "fsck", "gc", "get_branch_merge", "get_branch_remote", "get_object_by_path", "get_remote_repo", "get_tree_changes", "get_untracked_paths", "get_user_timezones", "grep", "hash_object", "independent_commits", "init", "interpret_trailers", "is_ancestor", "is_interactive_rebase", "lfs_clean", "lfs_fetch", "lfs_init", "lfs_ls_files", "lfs_migrate", "lfs_pointer_check", "lfs_pull", "lfs_push", "lfs_smudge", "lfs_status", "lfs_track", "lfs_untrack", "log", "ls_files", "ls_remote", "ls_tree", "mailinfo", "mailsplit", "maintenance_register", "maintenance_run", "maintenance_unregister", "merge", "merge_base", "merge_tree", "merged_branches", "mktag", "mv", "no_merged_branches", "notes_add", "notes_list", "notes_remove", "notes_show", "open_repo", "open_repo_closing", "pack_objects", "pack_refs", "parse_timezone_format", "path_to_tree_path", "print_commit", "print_name_only", "print_name_status", "print_oneline", "print_stat", "print_tag", "prune", "pull", "push", "range_diff", "rebase", "receive_pack", "reflog", "reflog_delete", "reflog_expire", "remote_add", "remote_remove", "remove", "repack", "replace_create", "replace_delete", "replace_list", "request_pull", "rerere", "rerere_clear", "rerere_diff", "rerere_forget", "rerere_gc", "rerere_status", "reset", "reset_file", "restore", "rev_list", "rev_parse", "revert", "set_branch_tracking", "shortlog", "show", "show_blob", "show_branch", "show_commit", "show_index", "show_object", "show_ref", "show_tag", "show_tree", "sparse_checkout", "stash_drop", "stash_list", "stash_pop", "stash_push", "status", "stripspace", "submodule_add", "submodule_init", "submodule_list", "submodule_update", "subtree_add", "subtree_merge", "subtree_pull", "subtree_push", "subtree_split", "switch", "symbolic_ref", "tag_create", "tag_delete", "tag_list", "unpack_objects", "update_head", "update_ref", "update_server_info", "upload_pack", "var", "var_list", "verify_commit", "verify_tag", "web_daemon", "worktree_add", "worktree_list", "worktree_lock", "worktree_move", "worktree_prune", "worktree_remove", "worktree_repair", "worktree_unlock", "write_commit_graph", "write_tree", ] import datetime import fnmatch import logging import os import posixpath import re import stat import sys import time import warnings from collections import namedtuple from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence from collections.abc import Set as AbstractSet from contextlib import AbstractContextManager, closing, contextmanager, suppress from dataclasses import dataclass from io import BytesIO, RawIOBase from pathlib import Path from typing import ( IO, TYPE_CHECKING, Any, BinaryIO, TextIO, TypedDict, TypeVar, cast, overload, ) from typing import ( Protocol as TypingProtocol, ) if sys.version_info >= (3, 12): from typing import override else: from typing_extensions import override from .._typing import Buffer if TYPE_CHECKING: import urllib3 from ..diff import ColorizedDiffStream from ..filter_branch import CommitData from ..gc import GCStats from ..maintenance import MaintenanceResult from ..objects import RawObjectID from ..archive import tar_stream from ..bisect import BisectState from ..client import ( FetchPackResult, LsRemoteResult, SendPackResult, get_transport_and_path, ) from ..config import Config, StackedConfig, env_config from ..diff_tree import ( CHANGE_ADD, CHANGE_COPY, CHANGE_DELETE, CHANGE_MODIFY, CHANGE_RENAME, RENAME_CHANGE_TYPES, TreeChange, tree_changes, ) from ..errors import SendPackError from ..file import open_nofollow from ..graph import can_fast_forward from ..ignore import IgnoreFilterManager from ..index import ( ConflictedIndexEntry, Index, IndexEntry, _fs_to_tree_path, blob_from_path_and_stat, build_file_from_blob, get_path_element_validator, get_unstaged_changes, symlink, update_working_tree, ) from ..object_store import BaseObjectStore, tree_lookup_path from ..objects import ( Blob, Commit, ObjectID, ShaFile, Tag, Tree, TreeEntry, format_timezone, parse_timezone, pretty_format_tree_entry, ) from ..objectspec import ( parse_commit, parse_object, parse_ref, parse_reftuples, parse_tree, ) from ..pack import UnpackedObject, write_pack_from_container, write_pack_index from ..patch import ( MailinfoResult, get_summary, parse_unified_diff, write_commit_patch, write_object_diff, write_tree_diff, ) from ..patch import ( apply_patches as apply_file_patches, ) from ..protocol import ZERO_SHA, Protocol from ..refs import ( HEADREF, LOCAL_BRANCH_PREFIX, LOCAL_REMOTE_PREFIX, LOCAL_REPLACE_PREFIX, LOCAL_TAG_PREFIX, DictRefsContainer, Ref, SymrefLoop, _import_remote_refs, filter_ref_prefix, local_branch_name, local_replace_name, parse_remote_ref, shorten_ref_name, ) from ..repo import BaseRepo, Repo, _get_default_identity from ..server import ( FileSystemBackend, ReceivePackHandler, TCPGitServer, UploadPackHandler, ) from ..server import update_server_info as server_update_server_info from ..sparse_patterns import ( SparseCheckoutConflictError, apply_included_paths, determine_included_paths, ) from ..trailers import add_trailer_to_message, format_trailers, parse_trailers from .lfs import ( lfs_clean, lfs_fetch, lfs_init, lfs_ls_files, lfs_migrate, lfs_pointer_check, lfs_pull, lfs_push, lfs_smudge, lfs_status, lfs_track, lfs_untrack, ) from .notes import ( notes_add, notes_list, notes_remove, notes_show, ) from .submodule import ( submodule_add, submodule_init, submodule_list, submodule_update, ) from .subtree import ( subtree_add, subtree_merge, subtree_pull, subtree_push, subtree_split, ) from .tag import tag_create, tag_delete, tag_list, verify_tag from .worktree import ( worktree_add, worktree_list, worktree_lock, worktree_move, worktree_prune, worktree_remove, worktree_repair, worktree_unlock, ) logger = logging.getLogger(__name__) # Module level tuple definition for status output GitStatus = namedtuple("GitStatus", "staged unstaged untracked") # TypeVar for preserving BaseRepo subclass types T = TypeVar("T", bound="BaseRepo") # Type alias for common repository parameter pattern RepoPath = str | os.PathLike[str] | Repo # Depth value used to fetch complete history (unshallow) INFINITE_DEPTH = 0x7FFFFFFF class TransportKwargs(TypedDict, total=False): """Keyword arguments accepted by get_transport_and_path.""" operation: str | None thin_packs: bool report_activity: Callable[[int, str], None] | None quiet: bool include_tags: bool username: str | None password: str | None key_filename: str | None ssh_command: str | None pool_manager: "urllib3.PoolManager | None" @dataclass class CountObjectsResult: """Result of counting objects in a repository. Attributes: count: Number of loose objects size: Total size of loose objects in bytes in_pack: Number of objects in pack files packs: Number of pack files size_pack: Total size of pack files in bytes """ count: int size: int in_pack: int | None = None packs: int | None = None size_pack: int | None = None class NoneStream(RawIOBase): """Fallback if stdout or stderr are unavailable, does nothing.""" def read(self, size: int = -1) -> None: """Read from stream (returns None as this is a null stream).""" return None def readall(self) -> bytes: """Read all bytes (returns empty bytes). Returns: Empty bytes object """ return b"" if sys.version_info >= (3, 12): @override def readinto(self, b: Buffer) -> int | None: return 0 @override def write(self, b: Buffer) -> int | None: return len(cast(bytes, b)) if b else 0 else: @override def readinto(self, b: bytearray | memoryview) -> int | None: # type: ignore[override] return 0 @override def write(self, b: bytes | bytearray | memoryview) -> int | None: # type: ignore[override] return len(b) if b else 0 default_bytes_out_stream: BinaryIO = cast( BinaryIO, getattr(sys.stdout, "buffer", None) or NoneStream() ) default_bytes_err_stream: BinaryIO = cast( BinaryIO, getattr(sys.stderr, "buffer", None) or NoneStream() ) DEFAULT_ENCODING = "utf-8" class Error(Exception): """Porcelain-based error.""" def __init__(self, msg: str) -> None: """Initialize Error with message.""" super().__init__(msg) class RemoteExists(Error): """Raised when the remote already exists.""" class TimezoneFormatError(Error): """Raised when the timezone cannot be determined from a given string.""" class CheckoutError(Error): """Indicates that a checkout cannot be performed.""" def _checked_worktree_path(repo: "Repo", tree_path: bytes) -> bytes: """Resolve a working-tree path, refusing anything that escapes the repo. ``checkout``, ``restore`` and ``reset_file`` write files at caller-supplied paths. Those paths are normally trusted, but validating them as a matter of defense in depth keeps a stray ``../`` or ``.git`` component from writing outside the work tree or into the control directory, matching the bar git applies to its own path operands. In addition to the name-based validation, refuse any path whose leading directory components already exist in the work tree as a symlink. A malicious repository could otherwise leave e.g. ``sub`` as a symlink to ``.git/hooks`` and then have ``checkout(paths=["sub/post-checkout"])`` write attacker content through it. Args: repo: Repository the path is relative to. tree_path: Path in tree form (``/``-separated), as produced by ``_fs_to_tree_path``. Returns: The filesystem path under the repository root, as bytes. Raises: Error: If the path is absolute, carries a component the configured ``core.protectNTFS``/``core.protectHFS`` validator rejects, or resolves through a symlink already present in the work tree. """ from ..index import ( InvalidPathError, _has_dos_drive_prefix, get_path_element_validator, validate_path, verify_leading_dirs, ) # Tree paths use "/" as the separator, so a leading "/", "\\" or (on # Windows) drive-letter prefix would make os.path.join discard the # repository root. absolute = tree_path.startswith((b"/", b"\\")) or ( os.name == "nt" and _has_dos_drive_prefix(tree_path) ) if absolute: raise Error(f"refusing to write path outside repository: {tree_path!r}") validator = get_path_element_validator(repo.get_config_stack()) if not validate_path(tree_path, validator): raise Error(f"refusing to write unsafe path: {tree_path!r}") repo_path = os.fsencode(repo.path) try: verify_leading_dirs(tree_path, [], repo_path) except InvalidPathError: raise Error(f"refusing to write through symlink: {tree_path!r}") return os.path.join(repo_path, tree_path) def parse_timezone_format(tz_str: str) -> int: """Parse given string and attempt to return a timezone offset. Different formats are considered in the following order: - Git internal format: - RFC 2822: e.g. Mon, 20 Nov 1995 19:12:08 -0500 - ISO 8601: e.g. 1995-11-20T19:12:08-0500 Args: tz_str: datetime string Returns: Timezone offset as integer Raises: TimezoneFormatError: if timezone information cannot be extracted """ import re # Git internal format internal_format_pattern = re.compile("^[0-9]+ [+-][0-9]{,4}$") if re.match(internal_format_pattern, tz_str): try: tz_internal = parse_timezone(tz_str.split(" ")[1].encode(DEFAULT_ENCODING)) return tz_internal[0] except ValueError: pass # RFC 2822 import email.utils rfc_2822 = email.utils.parsedate_tz(tz_str) if rfc_2822 and rfc_2822[9] is not None: return rfc_2822[9] # ISO 8601 # Supported offsets: # sHHMM, sHH:MM, sHH iso_8601_pattern = re.compile( "[0-9] ?([+-])([0-9]{2})(?::(?=[0-9]{2}))?([0-9]{2})?$" ) match = re.search(iso_8601_pattern, tz_str) total_secs = 0 if match: sign, hours, minutes = match.groups() total_secs += int(hours) * 3600 if minutes: total_secs += int(minutes) * 60 total_secs = -total_secs if sign == "-" else total_secs return total_secs # YYYY.MM.DD, MM/DD/YYYY, DD.MM.YYYY contain no timezone information raise TimezoneFormatError(tz_str) def _ssh_command_from_env(env: Mapping[str, str] | None = None) -> str | None: """Return the ssh command requested via ``GIT_SSH_COMMAND`` / ``GIT_SSH``. ``GIT_SSH_COMMAND`` wins over ``GIT_SSH``, matching git's own precedence. Returns ``None`` when neither is set, so callers can fall through to the ``core.sshCommand`` config or the transport default. Env lookup lives here in porcelain rather than in the transport library so that :mod:`dulwich.client` stays process-environment-free. """ if env is None: env = os.environ env_ssh_command = env.get("GIT_SSH_COMMAND") if env_ssh_command: return env_ssh_command env_ssh = env.get("GIT_SSH") if env_ssh: return env_ssh return None def _protocol_version_from_env(env: Mapping[str, str] | None = None) -> int | None: """Parse the version from the ``GIT_PROTOCOL`` environment variable. Git uses a colon-separated ``key=value`` format for ``GIT_PROTOCOL`` (for example ``version=2`` or ``feature=extra:version=2``). Return the requested version as an ``int`` when present and parseable, otherwise ``None``. Env lookup lives here in porcelain rather than in the transport library so that :mod:`dulwich.client` stays process-environment-free. """ if env is None: env = os.environ value = env.get("GIT_PROTOCOL") if not value: return None for pair in value.split(":"): key, sep, raw_val = pair.partition("=") if not sep or key.strip() != "version": # TODO: extract and surface features (e.g. ``feature=extra``). # For now dulwich only consumes ``version``; other keys are # ignored rather than being forwarded to the transport. logger.warning("Ignoring unsupported GIT_PROTOCOL pair %r", pair) continue try: return int(raw_val.strip()) except ValueError: logger.warning("Ignoring unparsable GIT_PROTOCOL version %r", raw_val) return None logger.warning("GIT_PROTOCOL %r has no version= pair; ignoring", value) return None def _config_stack( repo: BaseRepo, env: Mapping[str, str] | None = None ) -> "StackedConfig": """Build a config stack for ``repo`` honouring the environment. ``GIT_CONFIG_COUNT``/``GIT_CONFIG_KEY_n``/``GIT_CONFIG_VALUE_n`` take precedence over every file-based backend, including the repository's own config. """ if env is None: env = os.environ config = repo.get_config_stack() override = env_config(env) if override is not None: config.backends.insert(0, override) return config def _get_user_identity( config: "StackedConfig", kind: str | None = None, env: Mapping[str, str] | None = None, ) -> bytes: """Determine the identity to use for new commits, honouring ``env``. Checks GIT_${KIND}_NAME/GIT_${KIND}_EMAIL, then user.name/user.email from the config, then the identity of the user running the process. Resolving this here rather than in :func:`dulwich.repo.get_user_identity` keeps the environment lookup in porcelain, where ``env`` can override it. """ if env is None: env = os.environ user: bytes | None = None email: bytes | None = None if kind: name_var = env.get("GIT_" + kind + "_NAME") if name_var is not None: user = name_var.encode("utf-8") email_var = env.get("GIT_" + kind + "_EMAIL") if email_var is not None: email = email_var.encode("utf-8") if user is None: with suppress(KeyError): user = config.get(("user",), "name") if email is None: with suppress(KeyError): email = config.get(("user",), "email") if user is None or email is None: default_user, default_email = _get_default_identity(env=env) if user is None: user = default_user.encode("utf-8") if email is None: email = default_email.encode("utf-8") if email.startswith(b"<") and email.endswith(b">"): email = email[1:-1] return user + b" <" + email + b">" def get_user_timezones(env: Mapping[str, str] | None = None) -> tuple[int, int]: """Retrieve local timezone as described in git documentation. https://raw.githubusercontent.com/git/git/v2.3.0/Documentation/date-formats.txt Args: env: Environment to read GIT_AUTHOR_DATE and GIT_COMMITTER_DATE from (defaults to os.environ) Returns: A tuple containing author timezone, committer timezone. """ if env is None: env = os.environ local_timezone = time.localtime().tm_gmtoff if env.get("GIT_AUTHOR_DATE"): author_timezone = parse_timezone_format(env["GIT_AUTHOR_DATE"]) else: author_timezone = local_timezone if env.get("GIT_COMMITTER_DATE"): commit_timezone = parse_timezone_format(env["GIT_COMMITTER_DATE"]) else: commit_timezone = local_timezone return author_timezone, commit_timezone def _parse_ceiling_dirs(env: Mapping[str, str]) -> list[str] | None: """Parse ``GIT_CEILING_DIRECTORIES`` into an absolute-path list. Entries are separated by ``os.pathsep``. Non-empty entries are resolved through :func:`os.path.realpath` by default, matching git's default behaviour of following symlinks in ceiling directories. An empty entry acts as a mode toggle: entries that follow it are kept as-is (only made absolute) so that symlinks on the walking path can be compared against a symlinked ceiling. """ raw = env.get("GIT_CEILING_DIRECTORIES") if not raw: return None resolve = True result: list[str] = [] for part in raw.split(os.pathsep): if not part: resolve = False continue result.append(os.path.realpath(part) if resolve else os.path.abspath(part)) return result def _parse_env_bool(env: Mapping[str, str], name: str) -> bool: """Parse a boolean Git environment variable. Follows git's rules: ``true``/``yes``/``on`` and any non-zero integer are true, ``false``/``no``/``off``, ``0`` and the empty string are false, all case-insensitively. Anything else is an error, as in git. """ raw = env.get(name) if raw is None: return False value = raw.strip().lower() if value in ("true", "yes", "on"): return True if value in ("", "false", "no", "off"): return False try: return int(value) != 0 except ValueError: raise ValueError( f"bad boolean environment value {raw!r} for {name!r}" ) from None def _repo_from_env( path_or_repo: str | bytes | os.PathLike[str] | None, env: Mapping[str, str] | None, ) -> Repo: """Resolve a Repo from a path or from the Git environment variables. An explicit ``path_or_repo`` always wins over environment variables so that callers who pass a path get what they asked for. Otherwise the Git environment variables ``GIT_DIR``, ``GIT_COMMON_DIR``, ``GIT_WORK_TREE``, ``GIT_OBJECT_DIRECTORY``, ``GIT_ALTERNATE_OBJECT_DIRECTORIES``, ``GIT_INDEX_FILE``, ``GIT_CEILING_DIRECTORIES`` and ``GIT_DISCOVERY_ACROSS_FILESYSTEM`` are consulted; if none of the locator vars are set, discovery walks up from the current directory, bounded by ``GIT_CEILING_DIRECTORIES`` and, unless ``GIT_DISCOVERY_ACROSS_FILESYSTEM`` is true, by filesystem boundaries. """ if path_or_repo is not None: return Repo(path_or_repo) if env is None: env = os.environ git_dir = env.get("GIT_DIR") git_common_dir = env.get("GIT_COMMON_DIR") git_work_tree = env.get("GIT_WORK_TREE") git_object_dir = env.get("GIT_OBJECT_DIRECTORY") git_alternates = env.get("GIT_ALTERNATE_OBJECT_DIRECTORIES") git_index_file = env.get("GIT_INDEX_FILE") ceilings = _parse_ceiling_dirs(env) across_fs = _parse_env_bool(env, "GIT_DISCOVERY_ACROSS_FILESYSTEM") alternates = ( [p for p in git_alternates.split(os.pathsep) if p] if git_alternates else None ) if ( git_dir is None and git_work_tree is None and git_common_dir is None and git_object_dir is None and alternates is None and git_index_file is None ): return Repo.discover(ceiling_dirs=ceilings, across_filesystem=across_fs) if git_dir is None: # An override without GIT_DIR: fall back to discovery from cwd for # the control dir, then layer the overrides on top. Preserve the # discovered worktree so we don't silently downgrade a non-bare # repo when only GIT_OBJECT_DIRECTORY (etc.) is set. with closing( Repo.discover(ceiling_dirs=ceilings, across_filesystem=across_fs) ) as found: git_dir = found.controldir() if git_work_tree is None and not found.bare: git_work_tree = found.path return Repo( controldir=git_dir, commondir=git_common_dir, worktree=git_work_tree, object_directory=git_object_dir, alternates=alternates, index_file=git_index_file, ) @overload def open_repo(path_or_repo: T) -> AbstractContextManager[T]: ... @overload def open_repo( path_or_repo: str | os.PathLike[str] | None = ..., env: Mapping[str, str] | None = ..., ) -> AbstractContextManager[Repo]: ... def open_repo( path_or_repo: str | os.PathLike[str] | T | None = None, env: Mapping[str, str] | None = None, ) -> AbstractContextManager[T | Repo]: """Open an argument that can be a repository or a path for a repository. When ``path_or_repo`` is ``None`` the ``GIT_DIR``, ``GIT_COMMON_DIR``, ``GIT_WORK_TREE``, ``GIT_OBJECT_DIRECTORY``, ``GIT_ALTERNATE_OBJECT_DIRECTORIES``, ``GIT_INDEX_FILE``, ``GIT_CEILING_DIRECTORIES`` and ``GIT_DISCOVERY_ACROSS_FILESYSTEM`` environment variables are consulted, falling back to discovery from the current directory. """ if isinstance(path_or_repo, BaseRepo): return _noop_context_manager(path_or_repo) return _repo_from_env(path_or_repo, env) @contextmanager def _noop_context_manager(obj: T) -> Iterator[T]: """Context manager that has the same api as closing but does nothing.""" yield obj def _get_reflog_message( default_message: bytes, explicit_message: bytes | None = None, env: Mapping[str, str] | None = None, ) -> bytes: """Get reflog message, checking GIT_REFLOG_ACTION environment variable. Args: default_message: Default message to use if no explicit message or env var explicit_message: Explicit message passed as argument (takes precedence) env: Environment to read GIT_REFLOG_ACTION from (defaults to os.environ) Returns: The reflog message with priority: 1. explicit_message if provided 2. GIT_REFLOG_ACTION environment variable if set 3. default_message otherwise """ if explicit_message is not None: return explicit_message if env is None: env = os.environ env_action = env.get("GIT_REFLOG_ACTION") if env_action is not None: return env_action.encode("utf-8") return default_message @overload def open_repo_closing(path_or_repo: T) -> AbstractContextManager[T]: ... @overload def open_repo_closing( path_or_repo: str | bytes | os.PathLike[str] | None = ..., env: Mapping[str, str] | None = ..., ) -> AbstractContextManager[Repo]: ... def open_repo_closing( path_or_repo: str | bytes | os.PathLike[str] | T | None = None, env: Mapping[str, str] | None = None, ) -> AbstractContextManager[T | Repo]: """Open an argument that can be a repository or a path for a repository. Returns a context manager that will close the repo on exit if the argument is a path (or the repo was resolved from environment variables), else does nothing if the argument is an already-open repo. When ``path_or_repo`` is ``None`` the ``GIT_DIR``, ``GIT_COMMON_DIR``, ``GIT_WORK_TREE``, ``GIT_OBJECT_DIRECTORY``, ``GIT_ALTERNATE_OBJECT_DIRECTORIES``, ``GIT_INDEX_FILE``, ``GIT_CEILING_DIRECTORIES`` and ``GIT_DISCOVERY_ACROSS_FILESYSTEM`` environment variables are consulted, falling back to discovery from the current directory. """ if isinstance(path_or_repo, BaseRepo): return _noop_context_manager(path_or_repo) return closing(_repo_from_env(path_or_repo, env)) def path_to_tree_path( repopath: str | bytes | os.PathLike[str], path: str | bytes | os.PathLike[str], tree_encoding: str = DEFAULT_ENCODING, ) -> bytes: """Convert a path to a path usable in an index, e.g. bytes and relative to the repository root. Args: repopath: Repository path, absolute or relative to the cwd path: A path, absolute or relative to the cwd tree_encoding: Encoding to use for tree paths Returns: A path formatted for use in e.g. an index """ # Resolve might returns a relative path on Windows # https://bugs.python.org/issue38671 if sys.platform == "win32": path = os.path.abspath(path) # Convert bytes paths to str for Path if isinstance(path, bytes): path = os.fsdecode(path) path = Path(path) resolved_path = path.resolve() # Resolve and abspath seems to behave differently regarding symlinks, # as we are doing abspath on the file path, we need to do the same on # the repo path or they might not match if sys.platform == "win32": repopath = os.path.abspath(repopath) # Convert bytes paths to str for Path if isinstance(repopath, bytes): repopath = os.fsdecode(repopath) repopath = Path(repopath).resolve() try: relpath = resolved_path.relative_to(repopath) except ValueError: # If path is a symlink that points to a file outside the repo, we # want the relpath for the link itself, not the resolved target if path.is_symlink(): parent = path.parent.resolve() relpath = (parent / path.name).relative_to(repopath) else: raise if sys.platform == "win32": return str(relpath).replace(os.path.sep, "/").encode(tree_encoding) else: return bytes(relpath) def tree_path_to_fs_path( tree_path: bytes, tree_encoding: str = DEFAULT_ENCODING, ) -> bytes: """Convert a git tree path to a filesystem path (relative). Args: tree_path: Path from git tree (bytes with "/" separators, UTF-8 encoded) tree_encoding: Encoding used for tree paths (default: utf-8) Returns: Filesystem path as bytes (with os.sep, filesystem encoding) """ # Decode from tree encoding path_str = tree_path.decode(tree_encoding) # Replace / with OS separator if needed if os.sep != "/": path_str = path_str.replace("/", os.sep) # Encode for filesystem return os.fsencode(path_str) class DivergedBranches(Error): """Branches have diverged and fast-forward is not possible.""" def __init__(self, current_sha: bytes, new_sha: bytes) -> None: """Initialize DivergedBranches error with current and new SHA values.""" self.current_sha = current_sha self.new_sha = new_sha def check_diverged(repo: BaseRepo, current_sha: ObjectID, new_sha: ObjectID) -> None: """Check if updating to a sha can be done with fast forwarding. Args: repo: Repository object current_sha: Current head sha new_sha: New head sha """ try: can = can_fast_forward(repo, current_sha, new_sha) except KeyError: can = False if not can: raise DivergedBranches(current_sha, new_sha) def archive( repo: str | BaseRepo | None = None, committish: str | bytes | Commit | Tag | None = None, outstream: BinaryIO | RawIOBase = default_bytes_out_stream, errstream: BinaryIO | RawIOBase = default_bytes_err_stream, remote: str | bytes | None = None, ssh_command: str | None = None, env: Mapping[str, str] | None = None, ) -> None: """Create an archive. Args: repo: Path of repository for which to generate an archive. Ignored when ``remote`` is set. committish: Commit SHA1 or ref to use outstream: Output stream (defaults to stdout) errstream: Error stream (defaults to stderr) remote: Location of a remote repository to retrieve the archive from, rather than generating it from ``repo``. ssh_command: SSH command to use. Defaults to the command configured in ``env``. env: Environment to read Git variables from (defaults to os.environ) """ if committish is None: committish = "HEAD" if remote is not None: if ssh_command is None: ssh_command = _ssh_command_from_env(env) remote_str = remote.decode() if isinstance(remote, bytes) else remote client, path = get_transport_and_path(remote_str, ssh_command=ssh_command) committish_bytes: bytes if isinstance(committish, Commit | Tag): committish_bytes = committish.id elif isinstance(committish, str): committish_bytes = committish.encode(DEFAULT_ENCODING) else: committish_bytes = committish def write_data(data: bytes) -> None: outstream.write(data) def write_error(data: bytes) -> None: errstream.write(data) client.archive( path.encode(DEFAULT_ENCODING) if isinstance(path, str) else path, committish_bytes, write_data, write_error=write_error, ) return with open_repo_closing(repo) as repo_obj: c = parse_commit(repo_obj, committish) tree = repo_obj.object_store[c.tree] assert isinstance(tree, Tree) for chunk in tar_stream(repo_obj.object_store, tree, c.commit_time): outstream.write(chunk) def update_server_info(repo: RepoPath | None = None) -> None: """Update server info files for a repository. Args: repo: path to the repository """ with open_repo_closing(repo) as r: server_update_server_info(r) def write_commit_graph(repo: RepoPath | None = None, reachable: bool = True) -> None: """Write a commit graph file for a repository. Args: repo: path to the repository or a Repo object reachable: if True, include all commits reachable from refs. if False, only include direct ref targets. """ with open_repo_closing(repo) as r: # Get all refs refs = list(r.refs.as_dict().values()) if refs: r.object_store.write_commit_graph(refs, reachable=reachable) def symbolic_ref(repo: RepoPath, ref_name: str | bytes, force: bool = False) -> None: """Set git symbolic ref into HEAD. Args: repo: path to the repository ref_name: short name of the new ref force: force settings without checking if it exists in refs/heads """ with open_repo_closing(repo) as repo_obj: ref_path = _make_branch_ref(ref_name) if not force and ref_path not in repo_obj.refs.keys(): ref_name_str = ( ref_name.decode("utf-8", "replace") if isinstance(ref_name, bytes) else ref_name ) raise Error(f"fatal: ref `{ref_name_str}` is not a ref") repo_obj.refs.set_symbolic_ref(HEADREF, ref_path) def pack_refs(repo: RepoPath, all: bool = False) -> None: """Pack loose references into packed-refs file.""" with open_repo_closing(repo) as repo_obj: repo_obj.refs.pack_refs(all=all) def _get_variables( repo: RepoPath | None = None, env: Mapping[str, str] | None = None ) -> dict[str, str]: """Internal function to get all Git logical variables. Args: repo: Path to the repository env: Environment to read variables from (defaults to os.environ) Returns: A dictionary of all logical variables with values """ if env is None: env = os.environ with open_repo_closing(repo) as repo_obj: config = _config_stack(repo_obj, env=env) # Define callbacks for each logical variable def get_author_ident() -> str | None: """Get GIT_AUTHOR_IDENT.""" try: author_identity = _get_user_identity(config, kind="AUTHOR", env=env) author_tz, _ = get_user_timezones(env=env) timestamp = int(time.time()) return f"{author_identity.decode('utf-8', 'replace')} {timestamp} {author_tz:+05d}" except Exception: return None def get_committer_ident() -> str | None: """Get GIT_COMMITTER_IDENT.""" try: committer_identity = _get_user_identity( config, kind="COMMITTER", env=env ) _, committer_tz = get_user_timezones(env=env) timestamp = int(time.time()) return f"{committer_identity.decode('utf-8', 'replace')} {timestamp} {committer_tz:+05d}" except Exception: return None def get_editor() -> str | None: """Get GIT_EDITOR.""" editor = env.get("GIT_EDITOR") if editor is None: try: editor_bytes = config.get(("core",), "editor") editor = editor_bytes.decode("utf-8", "replace") except KeyError: editor = env.get("VISUAL") or env.get("EDITOR") return editor def get_sequence_editor() -> str | None: """Get GIT_SEQUENCE_EDITOR.""" sequence_editor = env.get("GIT_SEQUENCE_EDITOR") if sequence_editor is None: try: seq_editor_bytes = config.get(("sequence",), "editor") sequence_editor = seq_editor_bytes.decode("utf-8", "replace") except KeyError: # Falls back to GIT_EDITOR if not set sequence_editor = get_editor() return sequence_editor def get_pager() -> str | None: """Get GIT_PAGER.""" pager = env.get("GIT_PAGER") if pager is None: try: pager_bytes = config.get(("core",), "pager") pager = pager_bytes.decode("utf-8", "replace") except KeyError: pager = env.get("PAGER") return pager def get_default_branch() -> str: """Get GIT_DEFAULT_BRANCH.""" try: default_branch_bytes = config.get(("init",), "defaultBranch") return default_branch_bytes.decode("utf-8", "replace") except KeyError: # Git's default is "master" return "master" # Dictionary mapping variable names to their getter callbacks variable_callbacks: dict[str, Callable[[], str | None]] = { "GIT_AUTHOR_IDENT": get_author_ident, "GIT_COMMITTER_IDENT": get_committer_ident, "GIT_EDITOR": get_editor, "GIT_SEQUENCE_EDITOR": get_sequence_editor, "GIT_PAGER": get_pager, "GIT_DEFAULT_BRANCH": get_default_branch, } # Build the variables dictionary by calling callbacks variables: dict[str, str] = {} for var_name, callback in variable_callbacks.items(): value = callback() if value is not None: variables[var_name] = value return variables def var_list( repo: RepoPath | None = None, env: Mapping[str, str] | None = None ) -> dict[str, str]: """List all Git logical variables. Args: repo: Path to the repository env: Environment to read variables from (defaults to os.environ) Returns: A dictionary of all logical variables with their values """ return _get_variables(repo, env=env) def var( repo: RepoPath | None = None, variable: str = "GIT_AUTHOR_IDENT", env: Mapping[str, str] | None = None, ) -> str: """Get the value of a specific Git logical variable. Args: repo: Path to the repository variable: The variable to query (e.g., 'GIT_AUTHOR_IDENT') env: Environment to read variables from (defaults to os.environ) Returns: The value of the requested variable as a string Raises: KeyError: If the requested variable has no value """ variables = _get_variables(repo, env=env) if variable in variables: return variables[variable] else: raise KeyError(f"Variable {variable} has no value") def commit( repo: RepoPath | None = None, message: str | bytes | Callable[[Any, Commit], bytes] | None = None, author: bytes | None = None, author_timestamp: float | None = None, author_timezone: int | None = None, committer: bytes | None = None, commit_timestamp: float | None = None, commit_timezone: int | None = None, encoding: bytes | None = None, no_verify: bool = False, signoff: bool | None = None, all: bool = False, amend: bool = False, sign: bool | None = None, env: Mapping[str, str] | None = None, ) -> bytes: """Create a new commit. Args: repo: Path to repository message: Optional commit message (string/bytes or callable that takes (repo, commit) and returns bytes) author: Optional author name and email author_timestamp: Author timestamp (defaults to commit timestamp) author_timezone: Author timestamp timezone committer: Optional committer name and email commit_timestamp: Commit timestamp (defaults to now) commit_timezone: Commit timestamp timezone encoding: Encoding to use for commit message no_verify: Skip pre-commit and commit-msg hooks signoff: Add Signed-off-by line to commit message. If None, uses format.signoff config. all: Automatically stage all tracked files that have been modified amend: Replace the tip of the current branch by creating a new commit sign: GPG sign the commit. If None, uses commit.gpgsign config. If True, signs with default GPG key. If False, does not sign. env: Environment to read Git variables from (defaults to os.environ) Returns: SHA1 of the new commit """ encoding_str = encoding.decode("ascii") if encoding else DEFAULT_ENCODING if isinstance(message, str): message = message.encode(encoding_str) if isinstance(author, str): author = author.encode(encoding_str) if isinstance(committer, str): committer = committer.encode(encoding_str) local_timezone = get_user_timezones(env=env) if author_timezone is None: author_timezone = local_timezone[0] if commit_timezone is None: commit_timezone = local_timezone[1] with open_repo_closing(repo) as r: commit_config = _config_stack(r, env=env) # Handle amend logic merge_heads = None if amend: try: head_commit = r[r.head()] assert isinstance(head_commit, Commit) except KeyError: raise ValueError("Cannot amend: no existing commit found") # If message not provided, use the message from the current HEAD if message is None: message = head_commit.message # If author not provided, use the author from the current HEAD if author is None: author = head_commit.author if author_timezone is None: author_timezone = head_commit.author_timezone # Use the parent(s) of the current HEAD as our parent(s) merge_heads = list(head_commit.parents) # If -a flag is used, stage all modified tracked files if all: index = r.open_index(config=commit_config) normalizer = r.get_blob_normalizer(config=commit_config) # Pass the normalizer's checkin_normalize method directly if normalizer is not None: filter_callback = normalizer.checkin_normalize else: filter_callback = None trust_ctime = commit_config.get_boolean(b"core", b"trustctime", True) unstaged_changes = list( get_unstaged_changes( index, r.path, filter_callback, trust_ctime=trust_ctime ) ) if unstaged_changes: # Convert bytes paths to strings for add function modified_files: list[str | bytes | os.PathLike[str]] = [] for path in unstaged_changes: if isinstance(path, bytes): modified_files.append(path.decode()) else: modified_files.append(path) add(r, paths=modified_files) # Resolve identities here so the worktree does not fall back to # get_user_identity(), which reads os.environ directly. if author is None: author = _get_user_identity(commit_config, kind="AUTHOR", env=env) if committer is None: committer = _get_user_identity(commit_config, kind="COMMITTER", env=env) # For amend, create dangling commit to avoid adding current HEAD as parent if amend: commit_sha = r.get_worktree().commit( message=message, author=author, author_timestamp=author_timestamp, author_timezone=author_timezone, committer=committer, commit_timestamp=commit_timestamp, commit_timezone=commit_timezone, encoding=encoding, no_verify=no_verify, sign=sign, signoff=signoff, merge_heads=merge_heads, ref=None, config=commit_config, ) # Update HEAD to point to the new commit with reflog message try: old_head = r.refs[HEADREF] except KeyError: old_head = None # Get the actual commit message from the created commit commit_obj = r[commit_sha] assert isinstance(commit_obj, Commit) commit_message = commit_obj.message default_message = b"commit (amend): " + commit_message # Truncate message if too long for reflog if len(default_message) > 100: default_message = default_message[:97] + b"..." reflog_message = _get_reflog_message(default_message, env=env) # Pass committer explicitly: Repo._write_reflog would otherwise # resolve it via get_user_identity(), which reads os.environ. r.refs.set_if_equals( HEADREF, old_head, commit_sha, committer=committer, message=reflog_message, ) return commit_sha else: # TODO(jelmer): WorkTree.commit() hardcodes the "commit: " # reflog message, so GIT_REFLOG_ACTION is not honoured here (nor was # it before `env` was added). Threading it through needs a # WorkTree.commit() signature change. return r.get_worktree().commit( message=message, author=author, author_timestamp=author_timestamp, author_timezone=author_timezone, committer=committer, commit_timestamp=commit_timestamp, commit_timezone=commit_timezone, encoding=encoding, no_verify=no_verify, sign=sign, signoff=signoff, merge_heads=merge_heads, config=commit_config, ) def commit_tree( repo: RepoPath, tree: ObjectID, message: str | bytes | None = None, author: bytes | None = None, committer: bytes | None = None, ) -> ObjectID: """Create a new commit object. Args: repo: Path to repository tree: An existing tree object message: Commit message author: Optional author name and email committer: Optional committer name and email """ with open_repo_closing(repo) as r: if isinstance(message, str): message = message.encode(DEFAULT_ENCODING) return r.get_worktree().commit( message=message, tree=tree, committer=committer, author=author, config=r.get_config_stack(), ) def interpret_trailers( message: str | bytes, *, trailers: list[tuple[str, str]] | None = None, trim_empty: bool = False, only_trailers: bool = False, only_input: bool = False, unfold: bool = False, parse: bool = False, where: str = "end", if_exists: str = "addIfDifferentNeighbor", if_missing: str = "add", separators: str = ":", ) -> bytes: r"""Parse and manipulate trailers in a commit message. This function implements the functionality of ``git interpret-trailers``, allowing parsing and manipulation of structured metadata (trailers) in commit messages. Trailers are key-value pairs at the end of commit messages, formatted like: Signed-off-by: Alice Reviewed-by: Bob Args: message: The commit message (string or bytes) trailers: List of (key, value) tuples to add as new trailers trim_empty: Remove trailers with empty values only_trailers: Output only the trailers, not the message body only_input: Don't add new trailers, only parse existing ones unfold: Join multiline trailer values into a single line parse: Shorthand for --only-trailers --only-input --unfold where: Where to add new trailers ('end', 'start', 'after', 'before') if_exists: How to handle duplicate keys - 'add': Always add - 'replace': Replace all existing - 'addIfDifferent': Add only if value differs from all existing - 'addIfDifferentNeighbor': Add only if value differs from neighbors - 'doNothing': Don't add if key exists if_missing: What to do if key doesn't exist ('add' or 'doNothing') separators: Valid separator characters (default ':') Returns: The processed message as bytes Examples: >>> msg = b"Subject\\n\\nBody text\\n" >>> interpret_trailers(msg, trailers=[("Signed-off-by", "Alice ")]) b'Subject\\n\\nBody text\\n\\nSigned-off-by: Alice \\n' >>> msg = b"Subject\\n\\nSigned-off-by: Alice\\n" >>> interpret_trailers(msg, only_trailers=True) b'Signed-off-by: Alice\\n' """ # Handle --parse shorthand if parse: only_trailers = True only_input = True unfold = True # Convert message to bytes if isinstance(message, str): message_bytes = message.encode("utf-8") else: message_bytes = message # Parse existing trailers _message_body, parsed_trailers = parse_trailers(message_bytes, separators) # Apply unfold if requested if unfold: for trailer in parsed_trailers: # Replace newlines and multiple spaces with single space trailer.value = " ".join(trailer.value.split()) # Apply trim_empty if requested if trim_empty: parsed_trailers = [t for t in parsed_trailers if t.value.strip()] # Add new trailers if requested and not only_input if not only_input and trailers: for key, value in trailers: message_bytes = add_trailer_to_message( message_bytes, key, value, separators[0], # Use first separator as default where=where, if_exists=if_exists, if_missing=if_missing, ) # Re-parse to get updated trailers for output if only_trailers: _message_body, parsed_trailers = parse_trailers(message_bytes, separators) # Return based on only_trailers flag if only_trailers: return format_trailers(parsed_trailers) else: return message_bytes def stripspace( text: str | bytes, *, strip_comments: bool = False, comment_char: str = "#", comment_lines: bool = False, ) -> bytes: r"""Strip unnecessary whitespace from text. This function implements the functionality of ``git stripspace``, commonly used to clean up commit messages and other text content. Args: text: The text to process (string or bytes) strip_comments: If True, remove lines that begin with comment_char comment_char: The comment character to use (default: "#") comment_lines: If True, prepend comment_char to each line Returns: The processed text as bytes The function performs the following operations: 1. If comment_lines is True, prepend comment_char + space to each line 2. Strip trailing whitespace from each line 3. If strip_comments is True, remove lines starting with comment_char 4. Collapse multiple consecutive blank lines into a single blank line 5. Remove leading blank lines 6. Remove trailing blank lines 7. Ensure the text ends with a newline (unless empty) Examples: >>> stripspace(b" hello \\n\\n\\nworld \\n\\n") b'hello\\n\\nworld\\n' >>> stripspace(b"# comment\\ntext\\n", strip_comments=True) b'text\\n' >>> stripspace(b"line\\n", comment_lines=True) b'# line\\n' """ from ..stripspace import stripspace as _stripspace # Convert text to bytes if isinstance(text, str): text_bytes = text.encode("utf-8") else: text_bytes = text # Convert comment_char to bytes comment_char_bytes = ( comment_char.encode("utf-8") if isinstance(comment_char, str) else comment_char ) return _stripspace( text_bytes, strip_comments=strip_comments, comment_char=comment_char_bytes, comment_lines=comment_lines, ) def init( path: str | os.PathLike[str] = ".", *, bare: bool = False, symlinks: bool | None = None, object_format: str | None = None, ) -> Repo: """Create a new git repository. Args: path: Path to repository. bare: Whether to create a bare repository. symlinks: Whether to create actual symlinks (defaults to autodetect) object_format: Object format to use ("sha1" or "sha256", defaults to "sha1") Returns: A Repo instance """ if not os.path.exists(path): os.mkdir(path) if bare: return Repo.init_bare(path, object_format=object_format) else: return Repo.init(path, symlinks=symlinks, object_format=object_format) def _filter_transport_kwargs(**kwargs: object) -> TransportKwargs: """Filter kwargs to only include parameters accepted by get_transport_and_path. Args: **kwargs: Arbitrary keyword arguments Returns: Dictionary containing only the kwargs that get_transport_and_path accepts """ valid_params = { "operation", "thin_packs", "report_activity", "quiet", "include_tags", "username", "password", "key_filename", "ssh_command", "pool_manager", } return cast(TransportKwargs, {k: v for k, v in kwargs.items() if k in valid_params}) def clone( source: str | bytes | Repo, target: str | os.PathLike[str] | None = None, bare: bool = False, checkout: bool | None = None, errstream: BinaryIO | RawIOBase = default_bytes_err_stream, outstream: BinaryIO | None = None, origin: str | None = "origin", depth: int | None = None, branch: str | bytes | None = None, config: Config | None = None, filter_spec: str | None = None, protocol_version: int | None = None, recurse_submodules: bool = False, ssh_command: str | None = None, env: Mapping[str, str] | None = None, **kwargs: str | bytes | Sequence[str | bytes], ) -> Repo: """Clone a local or remote git repository. Args: source: Path or URL for source repository target: Path to target repository (optional) bare: Whether or not to create a bare repository checkout: Whether or not to check-out HEAD after cloning errstream: Optional stream to write progress to outstream: Optional stream to write progress to (deprecated) origin: Name of remote from the repository used to clone depth: Depth to fetch at branch: Optional branch or tag to be used as HEAD in the new repository instead of the cloned repository's HEAD. config: Configuration to use filter_spec: A git-rev-list-style object filter spec, as an ASCII string. Only used if the server supports the Git protocol-v2 'filter' feature, and ignored otherwise. protocol_version: desired Git protocol version. Defaults to the version requested in ``env``, and otherwise to the highest mutually supported protocol version. recurse_submodules: Whether to initialize and clone submodules ssh_command: Optional custom SSH command. Defaults to the command configured in ``env``. env: Environment to read Git variables from (defaults to os.environ) **kwargs: Additional keyword arguments including refspecs to fetch. Can be a bytestring, a string, or a list of bytestring/string. Returns: The new repository """ if outstream is not None: import warnings warnings.warn( "outstream= has been deprecated in favour of errstream=.", DeprecationWarning, stacklevel=3, ) # TODO(jelmer): Capture logging output and stream to errstream if config is None: config = StackedConfig.default() env_override = env_config(os.environ if env is None else env) if env_override is not None: config.backends.insert(0, env_override) if protocol_version is None: protocol_version = _protocol_version_from_env(env) if checkout is None: checkout = not bare if checkout and bare: raise Error("checkout and bare are incompatible") if target is None: if isinstance(source, Repo): raise ValueError("target must be specified when cloning from a Repo object") elif isinstance(source, bytes): target = source.split(b"/")[-1].decode() else: target = source.split("/")[-1] if isinstance(branch, str): branch = branch.encode(DEFAULT_ENCODING) mkdir = not os.path.exists(target) if isinstance(source, Repo): # For direct repo cloning, use LocalGitClient from ..client import GitClient, LocalGitClient client: GitClient = LocalGitClient(config=config) path = source.path else: source_str = source.decode() if isinstance(source, bytes) else source transport_kwargs = _filter_transport_kwargs(**kwargs) if ssh_command is None: ssh_command = _ssh_command_from_env(env) if ssh_command is not None: transport_kwargs["ssh_command"] = ssh_command (client, path) = get_transport_and_path( source_str, config=config, **transport_kwargs ) filter_spec_bytes: bytes | None = None if filter_spec: filter_spec_bytes = filter_spec.encode("ascii") repo = client.clone( path, str(target), # Convert PathLike to str mkdir=mkdir, bare=bare, origin=origin, checkout=checkout, branch=branch.decode() if branch else None, # Convert bytes to str progress=lambda data: (errstream.write(data), None)[1], depth=depth, filter_spec=filter_spec_bytes, protocol_version=protocol_version, ) # Initialize and update submodules if requested if recurse_submodules and not bare: try: submodule_init(repo) submodule_update(repo, init=True, recursive=True) except FileNotFoundError as e: # .gitmodules file doesn't exist - no submodules to process logger.debug("No .gitmodules file found: %s", e) except KeyError as e: # Submodule configuration missing logger.warning("Submodule configuration error: %s", e) if errstream: errstream.write( f"Warning: Submodule configuration error: {e}\n".encode() ) return repo def add( repo: str | os.PathLike[str] | Repo | None = None, paths: Sequence[str | bytes | os.PathLike[str]] | str | bytes | os.PathLike[str] | None = None, ) -> tuple[list[str], set[str]]: """Add files to the staging area. Args: repo: Repository for the files paths: Paths to add. If None, stages all untracked and modified files from the current working directory (mimicking 'git add .' behavior). Returns: Tuple with set of added files and ignored files If the repository contains ignored directories, the returned set will contain the path to an ignored directory (with trailing slash). Individual files within ignored directories will not be returned. Note: When paths=None, this function adds all untracked and modified files from the entire repository, mimicking 'git add -A' behavior. """ ignored = set() with open_repo_closing(repo) as r: repo_path = Path(r.path).resolve() ignore_manager = IgnoreFilterManager.from_repo(r, config=r.get_config_stack()) # Get unstaged changes once for the entire operation index = r.open_index(config=r.get_config_stack()) normalizer = r.get_blob_normalizer(config=r.get_config_stack()) if normalizer is not None: filter_callback = normalizer.checkin_normalize else: filter_callback = None # Read config once for filesystem compatibility options config = r.get_config_stack() preload_index = config.get_boolean(b"core", b"preloadIndex", False) trust_ctime = config.get_boolean(b"core", b"trustctime", True) precompose_unicode = config.get_boolean(b"core", b"precomposeunicode", False) all_unstaged_paths = list( get_unstaged_changes( index, r.path, filter_callback, preload_index, trust_ctime ) ) if paths is None: # When no paths specified, add all untracked and modified files from repo root paths = [str(repo_path)] relpaths = [] if isinstance(paths, str | bytes | os.PathLike): paths = [paths] for p in paths: # Handle bytes paths by decoding them if isinstance(p, bytes): p = p.decode("utf-8") path = Path(p) if not path.is_absolute(): # Make relative paths relative to the repo directory path = repo_path / path # Don't resolve symlinks completely - only resolve the parent directory # to avoid issues when symlinks point outside the repository if path.is_symlink(): # For symlinks, resolve only the parent directory parent_resolved = path.parent.resolve() resolved_path = parent_resolved / path.name else: # For regular files/dirs, resolve normally resolved_path = path.resolve() try: relpath = str(resolved_path.relative_to(repo_path)).replace(os.sep, "/") except ValueError as e: # Path is not within the repository p_str = p.decode() if isinstance(p, bytes) else str(p) raise ValueError( f"Path {p_str} is not within repository {repo_path}" ) from e # Handle directories by scanning their contents if resolved_path.is_dir(): # Check if the directory itself is ignored dir_relpath = posixpath.join(relpath, "") if relpath != "." else "" if dir_relpath and ignore_manager.is_ignored(dir_relpath): ignored.add(dir_relpath) continue # When adding a directory, add all untracked files within it current_untracked = list( get_untracked_paths( str(resolved_path), str(repo_path), index, precompose_unicode=precompose_unicode, repo=r, ) ) for untracked_path in current_untracked: # If we're scanning a subdirectory, adjust the path if relpath != ".": untracked_path = posixpath.join(relpath, untracked_path) if not ignore_manager.is_ignored(untracked_path): relpaths.append(untracked_path) else: ignored.add(untracked_path) # Also add unstaged (modified) files within this directory for unstaged_path in all_unstaged_paths: if isinstance(unstaged_path, bytes): unstaged_path_str = unstaged_path.decode("utf-8") else: unstaged_path_str = unstaged_path # Check if this unstaged file is within the directory we're processing unstaged_full_path = repo_path / unstaged_path_str try: unstaged_full_path.relative_to(resolved_path) # File is within this directory, add it if not ignore_manager.is_ignored(unstaged_path_str): relpaths.append(unstaged_path_str) else: ignored.add(unstaged_path_str) except ValueError: # File is not within this directory, skip it continue continue # FIXME: Support patterns if ignore_manager.is_ignored(relpath): ignored.add(relpath) continue relpaths.append(relpath) r.get_worktree().stage(relpaths, config=r.get_config_stack()) return (relpaths, ignored) def _is_subdir( subdir: str | os.PathLike[str], parentdir: str | os.PathLike[str] ) -> bool: """Check whether subdir is parentdir or a subdir of parentdir. If parentdir or subdir is a relative path, it will be disamgibuated relative to the pwd. """ parentdir_abs = os.path.realpath(parentdir) + os.path.sep subdir_abs = os.path.realpath(subdir) + os.path.sep return subdir_abs.startswith(parentdir_abs) # TODO: option to remove ignored files also, in line with `git clean -fdx` def clean( repo: str | os.PathLike[str] | Repo | None = None, target_dir: str | os.PathLike[str] | None = None, ) -> None: """Remove any untracked files from the target directory recursively. Equivalent to running ``git clean -fd`` in target_dir. Args: repo: Repository where the files may be tracked target_dir: Directory to clean - current directory if None """ if target_dir is None: target_dir = os.getcwd() with open_repo_closing(repo) as r: if not _is_subdir(target_dir, r.path): raise Error("target_dir must be in the repo's working dir") config = r.get_config_stack() config.get_boolean((b"clean",), b"requireForce", True) # TODO(jelmer): if require_force is set, then make sure that -f, -i or # -n is specified. index = r.open_index(config=r.get_config_stack()) ignore_manager = IgnoreFilterManager.from_repo(r, config=r.get_config_stack()) paths_in_wd = _walk_working_dir_paths(target_dir, r.path) # Reverse file visit order, so that files and subdirectories are # removed before containing directory for ap, is_dir in reversed(list(paths_in_wd)): # target_dir and r.path are both str, so ap must be str assert isinstance(ap, str) if is_dir: # All subdirectories and files have been removed if untracked, # so dir contains no tracked files iff it is empty. is_empty = len(os.listdir(ap)) == 0 if is_empty: os.rmdir(ap) else: ip = path_to_tree_path(r.path, ap) is_tracked = ip in index rp = os.path.relpath(ap, r.path) is_ignored = ignore_manager.is_ignored(rp) if not is_tracked and not is_ignored: os.remove(ap) def remove( repo: str | os.PathLike[str] | Repo | None = None, paths: Sequence[str | bytes | os.PathLike[str]] = [], cached: bool = False, ) -> None: """Remove files from the staging area. Args: repo: Repository for the files paths: Paths to remove. Can be absolute or relative to the repository root. cached: Only remove from index, not from working directory """ with open_repo_closing(repo) as r: index = r.open_index(config=r.get_config_stack()) blob_normalizer = r.get_blob_normalizer(config=r.get_config_stack()) for p in paths: # If path is absolute, use it as-is. Otherwise, treat it as relative to repo p_str = os.fsdecode(p) if isinstance(p, bytes) else str(p) if os.path.isabs(p_str): full_path = p_str else: # Treat relative paths as relative to the repository root full_path = os.path.join(r.path, p_str) tree_path = index.canonical_path(path_to_tree_path(r.path, full_path)) # Convert to bytes for file operations full_path_bytes = os.fsencode(full_path) try: entry = index[tree_path] if isinstance(entry, ConflictedIndexEntry): raise Error(f"{p_str} has conflicts in the index") index_sha = entry.sha except KeyError as exc: raise Error(f"{p_str} did not match any files") from exc if not cached: try: st = os.lstat(full_path_bytes) except OSError: pass else: try: blob = blob_from_path_and_stat(full_path_bytes, st) # Apply checkin normalization to compare apples to apples if blob_normalizer is not None: blob = blob_normalizer.checkin_normalize(blob, tree_path) except OSError: pass else: try: head_commit = r[r.head()] assert isinstance(head_commit, Commit) committed_sha = tree_lookup_path( r.__getitem__, head_commit.tree, tree_path )[1] except KeyError: committed_sha = None if blob.id != index_sha and index_sha != committed_sha: raise Error( "file has staged content differing " f"from both the file and head: {p_str}" ) if index_sha != committed_sha: raise Error(f"file has staged changes: {p_str}") os.remove(full_path_bytes) del index[tree_path] index.write() rm = remove def mv( repo: str | os.PathLike[str] | Repo, source: str | bytes | os.PathLike[str], destination: str | bytes | os.PathLike[str], force: bool = False, ) -> None: """Move or rename a file, directory, or symlink. Args: repo: Path to the repository source: Path to move from destination: Path to move to force: Force move even if destination exists Raises: Error: If source doesn't exist, is not tracked, or destination already exists (without force) """ with open_repo_closing(repo) as r: index = r.open_index(config=r.get_config_stack()) # Handle paths - convert to string if necessary if isinstance(source, bytes): source = source.decode(sys.getfilesystemencoding()) elif hasattr(source, "__fspath__"): source = os.fspath(source) else: source = str(source) if isinstance(destination, bytes): destination = destination.decode(sys.getfilesystemencoding()) elif hasattr(destination, "__fspath__"): destination = os.fspath(destination) else: destination = str(destination) # Get full paths if os.path.isabs(source): source_full_path = source else: # Treat relative paths as relative to the repository root source_full_path = os.path.join(r.path, source) if os.path.isabs(destination): destination_full_path = destination else: # Treat relative paths as relative to the repository root destination_full_path = os.path.join(r.path, destination) # Check if destination is a directory if os.path.isdir(destination_full_path): # Move source into destination directory basename = os.path.basename(source_full_path) destination_full_path = os.path.join(destination_full_path, basename) # Convert to tree paths for index source_tree_path = path_to_tree_path(r.path, source_full_path) destination_tree_path = path_to_tree_path(r.path, destination_full_path) # Check if source exists in index if source_tree_path not in index: raise Error(f"source '{source}' is not under version control") # Check if source exists in filesystem if not os.path.exists(source_full_path): raise Error(f"source '{source}' does not exist") # Check if destination already exists if os.path.exists(destination_full_path) and not force: raise Error(f"destination '{destination}' already exists (use -f to force)") # Check if destination is already in index if destination_tree_path in index and not force: raise Error( f"destination '{destination}' already exists in index (use -f to force)" ) # Get the index entry for the source source_entry = index[source_tree_path] # Convert to bytes for file operations source_full_path_bytes = os.fsencode(source_full_path) destination_full_path_bytes = os.fsencode(destination_full_path) # Create parent directory for destination if needed dest_dir = os.path.dirname(destination_full_path_bytes) if dest_dir and not os.path.exists(dest_dir): os.makedirs(dest_dir) # Move the file in the filesystem if os.path.exists(destination_full_path_bytes) and force: os.remove(destination_full_path_bytes) os.rename(source_full_path_bytes, destination_full_path_bytes) # Update the index del index[source_tree_path] index[destination_tree_path] = source_entry index.write() move = mv def commit_decode( commit: Commit, contents: bytes, default_encoding: str = DEFAULT_ENCODING ) -> str: """Decode commit contents using the commit's encoding or default.""" if commit.encoding: encoding = commit.encoding.decode("ascii") else: encoding = default_encoding return contents.decode(encoding, "replace") def commit_encode( commit: Commit, contents: str, default_encoding: str = DEFAULT_ENCODING ) -> bytes: """Encode commit contents using the commit's encoding or default.""" if commit.encoding: encoding = commit.encoding.decode("ascii") else: encoding = default_encoding return contents.encode(encoding) class _TextStream(TypingProtocol): """Minimal write-only text stream protocol used for human-readable output.""" def write(self, data: str, /) -> int: ... def print_commit( commit: Commit, decode: Callable[[bytes], str], outstream: _TextStream = sys.stdout, abbrev_commit: bool = False, ) -> None: """Write a human-readable commit log entry. Args: commit: A `Commit` object decode: Function to decode commit data outstream: A stream file to write to abbrev_commit: If True, abbreviate commit hashes """ outstream.write("-" * 50 + "\n") commit_id = commit.id.decode("ascii") if abbrev_commit: commit_id = commit_id[:7] outstream.write("commit: " + commit_id + "\n") if len(commit.parents) > 1: outstream.write( "merge: " + "...".join([c.decode("ascii") for c in commit.parents[1:]]) + "\n" ) outstream.write("Author: " + decode(commit.author) + "\n") if commit.author != commit.committer: outstream.write("Committer: " + decode(commit.committer) + "\n") time_tuple = time.gmtime(commit.author_time + commit.author_timezone) time_str = time.strftime("%a %b %d %Y %H:%M:%S", time_tuple) timezone_str = format_timezone(commit.author_timezone).decode("ascii") outstream.write("Date: " + time_str + " " + timezone_str + "\n") if commit.message: outstream.write("\n") outstream.write(decode(commit.message) + "\n") outstream.write("\n") def print_tag( tag: Tag, decode: Callable[[bytes], str], outstream: TextIO = sys.stdout ) -> None: """Write a human-readable tag. Args: tag: A `Tag` object decode: Function for decoding bytes to unicode string outstream: A stream to write to """ outstream.write("Tagger: " + decode(tag.tagger) + "\n") time_tuple = time.gmtime(tag.tag_time + tag.tag_timezone) time_str = time.strftime("%a %b %d %Y %H:%M:%S", time_tuple) timezone_str = format_timezone(tag.tag_timezone).decode("ascii") outstream.write("Date: " + time_str + " " + timezone_str + "\n") outstream.write("\n") outstream.write(decode(tag.message)) outstream.write("\n") def show_blob( repo: RepoPath, blob: Blob, decode: Callable[[bytes], str], outstream: TextIO = sys.stdout, ) -> None: """Write a blob to a stream. Args: repo: A `Repo` object blob: A `Blob` object decode: Function for decoding bytes to unicode string outstream: A stream file to write to """ outstream.write(decode(blob.data)) def show_commit( repo: RepoPath, commit: Commit, decode: Callable[[bytes], str], outstream: "_TextStream | ColorizedDiffStream" = sys.stdout, ) -> None: """Show a commit to a stream. Args: repo: A `Repo` object commit: A `Commit` object decode: Function for decoding bytes to unicode string outstream: Stream to write to """ from ..diff import ColorizedDiffStream # Adapter that lets print_commit (which writes str) feed a # ColorizedDiffStream (which writes bytes). class _StreamWrapper: def __init__(self, stream: "ColorizedDiffStream") -> None: self.stream = stream def write(self, data: str) -> int: self.stream.write(data.encode("utf-8")) return len(data) with open_repo_closing(repo) as r: # Use wrapper for ColorizedDiffStream, direct stream for others if isinstance(outstream, ColorizedDiffStream): wrapped_stream = _StreamWrapper(outstream) print_commit(commit, decode=decode, outstream=wrapped_stream) # Write diff directly to the ColorizedDiffStream as bytes write_tree_diff( outstream, r.object_store, commit.parents[0] if commit.parents else None, commit.tree, ) else: print_commit(commit, decode=decode, outstream=outstream) if commit.parents: parent_commit = r[commit.parents[0]] assert isinstance(parent_commit, Commit) base_tree = parent_commit.tree else: base_tree = None # Traditional path: buffer diff and write as decoded text diffstream = BytesIO() write_tree_diff(diffstream, r.object_store, base_tree, commit.tree) diffstream.seek(0) outstream.write(commit_decode(commit, diffstream.getvalue())) def show_tree( repo: RepoPath, tree: Tree, decode: Callable[[bytes], str], outstream: TextIO = sys.stdout, ) -> None: """Print a tree to a stream. Args: repo: A `Repo` object tree: A `Tree` object decode: Function for decoding bytes to unicode string outstream: Stream to write to """ for n in tree: outstream.write(decode(n) + "\n") def show_tag( repo: RepoPath, tag: Tag, decode: Callable[[bytes], str], outstream: TextIO = sys.stdout, ) -> None: """Print a tag to a stream. Args: repo: A `Repo` object tag: A `Tag` object decode: Function for decoding bytes to unicode string outstream: Stream to write to """ with open_repo_closing(repo) as r: print_tag(tag, decode, outstream) obj = r[tag.object[1]] assert isinstance(obj, Tree | Blob | Commit | Tag) show_object(repo, obj, decode, outstream) def show_object( repo: RepoPath, obj: Tree | Blob | Commit | Tag, decode: Callable[[bytes], str], outstream: TextIO, ) -> None: """Show details of a git object.""" handlers: dict[bytes, Callable[[RepoPath, Any, Any, TextIO], None]] = { b"tree": show_tree, b"blob": show_blob, b"commit": show_commit, b"tag": show_tag, } handler = handlers.get(obj.type_name) if handler is None: raise ValueError(f"Unknown object type: {obj.type_name.decode()}") handler(repo, obj, decode, outstream) def print_name_status(changes: Iterator[TreeChange]) -> Iterator[str]: """Print a simple status summary, listing changed files.""" for change in changes: if not change: continue change_item: TreeChange if isinstance(change, list): change_item = cast(TreeChange, change[0]) else: change_item = change if change_item.type == CHANGE_ADD: assert change_item.new is not None path1 = change_item.new.path assert path1 is not None path2: bytes = b"" kind = "A" elif change_item.type == CHANGE_DELETE: assert change_item.old is not None path1 = change_item.old.path assert path1 is not None path2 = b"" kind = "D" elif change_item.type == CHANGE_MODIFY: assert change_item.new is not None path1 = change_item.new.path assert path1 is not None path2 = b"" kind = "M" elif change_item.type in RENAME_CHANGE_TYPES: assert change_item.old is not None and change_item.new is not None path1 = change_item.old.path assert path1 is not None path2_opt = change_item.new.path assert path2_opt is not None path2 = path2_opt if change_item.type == CHANGE_RENAME: kind = "R" elif change_item.type == CHANGE_COPY: kind = "C" else: kind = "?" else: raise ValueError(f"Unknown change type: {change_item.type}") path1_str = ( path1.decode("utf-8", errors="replace") if isinstance(path1, bytes) else path1 ) path2_str = ( path2.decode("utf-8", errors="replace") if isinstance(path2, bytes) else path2 ) yield f"{kind:<8}{path1_str:<20}{path2_str:<20}" def print_name_only(changes: Iterator[TreeChange]) -> Iterator[str]: """Print only the names of changed files. Args: changes: Iterator of TreeChange objects Yields: Formatted name-only strings for each change """ for change in changes: if not change: continue change_item: TreeChange if isinstance(change, list): change_item = cast(TreeChange, change[0]) else: change_item = change if change_item.type == CHANGE_DELETE: assert change_item.old is not None path = change_item.old.path else: assert change_item.new is not None path = change_item.new.path assert path is not None path_str = ( path.decode("utf-8", errors="replace") if isinstance(path, bytes) else path ) yield path_str def print_oneline( commit: Commit, decode: Callable[[bytes], str], outstream: TextIO = sys.stdout, abbrev_commit: bool = True, ) -> None: """Write a single-line commit log entry. Args: commit: A `Commit` object decode: Function to decode commit data outstream: A stream file to write to abbrev_commit: If True, abbreviate commit hashes """ commit_id = commit.id.decode("ascii") if abbrev_commit: commit_id = commit_id[:7] message = decode(commit.message).split("\n", 1)[0] if commit.message else "" outstream.write(f"{commit_id} {message}\n") def print_stat( store: "BaseObjectStore", commit: Commit, outstream: TextIO = sys.stdout, ) -> None: """Write a diffstat summary for a commit. Args: store: ObjectStore for looking up objects commit: A `Commit` object outstream: A stream file to write to """ if commit.parents: parent = store[commit.parents[0]] assert isinstance(parent, Commit) base_tree = parent.tree else: base_tree = None diffstream = BytesIO() write_tree_diff(diffstream, store, base_tree, commit.tree) diffstream.seek(0) diff_lines = diffstream.getvalue().split(b"\n") from ..diffstat import diffstat stat_output = diffstat(diff_lines) outstream.write(stat_output.decode("utf-8", errors="replace") + "\n") def log( repo: RepoPath | None = None, paths: Sequence[str | bytes] | None = None, outstream: TextIO = sys.stdout, max_entries: int | None = None, reverse: bool = False, name_status: bool = False, name_only: bool = False, author: str | None = None, committer: str | None = None, grep: str | None = None, since: str | int | None = None, until: str | int | None = None, no_merges: bool = False, merges: bool = False, oneline: bool = False, abbrev_commit: bool = False, stat: bool = False, patch: bool = False, follow: bool = False, ) -> None: """Write commit logs. Args: repo: Path to repository paths: Optional set of specific paths to print entries for outstream: Stream to write log output to reverse: Reverse order in which entries are printed name_status: Print name/status for each changed file name_only: Print only names of changed files max_entries: Optional maximum number of entries to display author: Filter commits by author pattern committer: Filter commits by committer pattern grep: Filter commits by message pattern since: Show commits after this date (timestamp or date string) until: Show commits before this date (timestamp or date string) no_merges: Exclude merge commits merges: Only show merge commits oneline: Show each commit on a single line abbrev_commit: Abbreviate commit hashes stat: Show diffstat for each commit patch: Show patch (diff) for each commit follow: Follow file renames """ import re since_ts: int | None = None until_ts: int | None = None if since is not None: if isinstance(since, int): since_ts = since else: from ..approxidate import parse_approxidate since_ts = parse_approxidate(since) if until is not None: if isinstance(until, int): until_ts = until else: from ..approxidate import parse_approxidate until_ts = parse_approxidate(until) author_re = re.compile(author.encode(), re.IGNORECASE) if author else None committer_re = re.compile(committer.encode(), re.IGNORECASE) if committer else None grep_re = re.compile(grep.encode(), re.IGNORECASE) if grep else None with open_repo_closing(repo) as r: try: include = [r.head()] except KeyError: include = [] # Convert paths to bytes if needed paths_bytes = None if paths: paths_bytes = [p.encode() if isinstance(p, str) else p for p in paths] walker = r.get_walker( include=include, max_entries=None, # We filter ourselves to handle author/grep/merges paths=paths_bytes, reverse=reverse, since=since_ts, until=until_ts, follow=follow, ) count = 0 for entry in walker: commit = entry.commit # Filter by merge status if no_merges and len(commit.parents) > 1: continue if merges and len(commit.parents) <= 1: continue # Filter by author if author_re and not author_re.search(commit.author): continue # Filter by committer if committer_re and not committer_re.search(commit.committer): continue # Filter by commit message if grep_re and (not commit.message or not grep_re.search(commit.message)): continue # Check max_entries after filtering if max_entries is not None and count >= max_entries: break count += 1 def decode_wrapper(x: bytes) -> str: return commit_decode(entry.commit, x) if oneline: print_oneline(commit, decode_wrapper, outstream, abbrev_commit=True) else: print_commit( commit, decode_wrapper, outstream, abbrev_commit=abbrev_commit, ) if name_status: outstream.writelines( [ line + "\n" for line in print_name_status( cast(Iterator[TreeChange], entry.changes()) ) ] ) if name_only: outstream.writelines( [ line + "\n" for line in print_name_only( cast(Iterator[TreeChange], entry.changes()) ) ] ) if stat: print_stat(r.object_store, commit, outstream) if patch: if commit.parents: parent = r[commit.parents[0]] assert isinstance(parent, Commit) base_tree = parent.tree else: base_tree = None diffstream = BytesIO() write_tree_diff(diffstream, r.object_store, base_tree, commit.tree) diffstream.seek(0) outstream.write(commit_decode(commit, diffstream.getvalue())) # TODO(jelmer): better default for encoding? def show( repo: RepoPath | None = None, objects: Sequence[str | bytes] | None = None, outstream: TextIO = sys.stdout, default_encoding: str = DEFAULT_ENCODING, ) -> None: """Print the changes in a commit. Args: repo: Path to repository objects: Objects to show (defaults to [HEAD]) outstream: Stream to write to default_encoding: Default encoding to use if none is set in the commit """ if objects is None: objects = ["HEAD"] if isinstance(objects, str | bytes): objects = [objects] def _make_decode(obj: ShaFile) -> Callable[[bytes], str]: if isinstance(obj, Commit): return lambda x: commit_decode(obj, x, default_encoding) return lambda x: x.decode(default_encoding) with open_repo_closing(repo) as r: for objectish in objects: o = parse_object(r, objectish, config=r.get_config_stack()) assert isinstance(o, Tree | Blob | Commit | Tag) show_object(r, o, _make_decode(o), outstream) def diff_tree( repo: RepoPath, old_tree: str | bytes | Tree, new_tree: str | bytes | Tree, outstream: BinaryIO = default_bytes_out_stream, ) -> None: """Compares the content and mode of blobs found via two tree objects. Args: repo: Path to repository old_tree: Id of old tree new_tree: Id of new tree outstream: Stream to write to """ with open_repo_closing(repo) as r: if isinstance(old_tree, Tree): old_tree_id: ObjectID | None = old_tree.id elif isinstance(old_tree, str): old_tree_id = ObjectID(old_tree.encode()) else: old_tree_id = ObjectID(old_tree) if isinstance(new_tree, Tree): new_tree_id: ObjectID | None = new_tree.id elif isinstance(new_tree, str): new_tree_id = ObjectID(new_tree.encode()) else: new_tree_id = ObjectID(new_tree) write_tree_diff(outstream, r.object_store, old_tree_id, new_tree_id) def diff( repo: RepoPath | None = None, commit: str | bytes | Commit | None = None, commit2: str | bytes | Commit | None = None, staged: bool = False, paths: Sequence[str | bytes] | None = None, outstream: BinaryIO = default_bytes_out_stream, diff_algorithm: str | None = None, ) -> None: """Show diff. Args: repo: Path to repository commit: First commit to compare. If staged is True, compare index to this commit. If staged is False, compare working tree to this commit. If None, defaults to HEAD for staged and index for unstaged. commit2: Second commit to compare against first commit. If provided, show diff between commit and commit2 (ignoring staged flag). staged: If True, show staged changes (index vs commit). If False, show unstaged changes (working tree vs commit/index). Ignored if commit2 is provided. paths: Optional list of paths to limit diff outstream: Stream to write to diff_algorithm: Algorithm to use for diffing ("myers" or "patience"), defaults to the underlying function's default if None """ from .. import diff as diff_module with open_repo_closing(repo) as r: # Normalize paths to bytes byte_paths: list[bytes] | None = None if paths is not None and paths: # Check if paths is not empty byte_paths = [] for p in paths: if isinstance(p, str): byte_paths.append(p.encode("utf-8")) else: byte_paths.append(p) elif paths == []: # Convert empty list to None byte_paths = None else: byte_paths = None # Resolve commit refs to SHAs if provided if commit is not None: if isinstance(commit, Commit): # Already a Commit object commit_sha = commit.id commit_obj = commit else: # parse_commit handles both refs and SHAs, and always returns a Commit object commit_obj = parse_commit(r, commit) commit_sha = commit_obj.id else: commit_sha = None commit_obj = None if commit2 is not None: # Compare two commits if isinstance(commit2, Commit): commit2_obj = commit2 else: commit2_obj = parse_commit(r, commit2) # Get trees from commits old_tree = commit_obj.tree if commit_obj else None new_tree = commit2_obj.tree # Use tree_changes to get the changes and apply path filtering changes = r.object_store.tree_changes(old_tree, new_tree) for (oldpath, newpath), (oldmode, newmode), (oldsha, newsha) in changes: # Skip if paths are specified and this change doesn't match if byte_paths: path_to_check = newpath or oldpath assert path_to_check is not None if not any( path_to_check == p or path_to_check.startswith(p + b"/") for p in byte_paths ): continue write_object_diff( outstream, r.object_store, (oldpath, oldmode, oldsha), (newpath, newmode, newsha), diff_algorithm=diff_algorithm, ) elif staged: # Show staged changes (index vs commit) diff_module.diff_index_to_tree( r, outstream, commit_sha, byte_paths, diff_algorithm=diff_algorithm, config=r.get_config_stack(), ) elif commit is not None: # Compare working tree to a specific commit assert ( commit_sha is not None ) # mypy: commit_sha is set when commit is not None diff_module.diff_working_tree_to_tree( r, outstream, commit_sha, byte_paths, diff_algorithm=diff_algorithm, config=r.get_config_stack(), ) else: # Compare working tree to index diff_module.diff_working_tree_to_index( r, outstream, byte_paths, diff_algorithm=diff_algorithm, config=r.get_config_stack(), ) def rev_list( repo: RepoPath, commits: Sequence[str | bytes], outstream: BinaryIO = default_bytes_out_stream, ) -> None: """Lists commit objects in reverse chronological order. Args: repo: Path to repository commits: Commits over which to iterate outstream: Stream to write to """ with open_repo_closing(repo) as r: for entry in r.get_walker( include=[r[c if isinstance(c, bytes) else c.encode()].id for c in commits] ): outstream.write(entry.commit.id + b"\n") def _canonical_part(url: str) -> str: name = url.rsplit("/", 1)[-1] if name.endswith(".git"): name = name[:-4] return name def verify_commit( repo: RepoPath, committish: str | bytes = "HEAD", keyids: list[str] | None = None, ) -> None: """Verify signature on a commit. Args: repo: Path to repository committish: Commit to verify (defaults to HEAD) keyids: Optional list of trusted key IDs. If provided, the commit must be signed by one of these keys. If not provided, just verifies that the commit has a valid signature. Raises: gpg.errors.BadSignatures: if GPG signature verification fails gpg.errors.MissingSignatures: if commit was not signed by a key specified in keyids """ from dulwich.signature import ( UntrustedSignature, get_signature_vendor_for_signature, ) with open_repo_closing(repo) as r: commit = parse_commit(r, committish) payload, signature, _sig_type = commit.extract_signature() if signature is None: if keyids: raise UntrustedSignature( "commit is not signed by any of the trusted keys: " "no signature present", trusted_keys=list(keyids), ) return vendor = get_signature_vendor_for_signature( signature, config=r.get_config_stack(), keyids=keyids ) vendor.verify(payload, signature) def replace_list(repo: RepoPath) -> list[tuple[ObjectID, ObjectID]]: """List all replacement refs. Args: repo: Path to repository Returns: List of tuples of (object_sha, replacement_sha) where object_sha is the object being replaced and replacement_sha is what it's replaced with """ with open_repo_closing(repo) as r: replacements: list[tuple[ObjectID, ObjectID]] = [] for ref in r.refs.keys(): if ref.startswith(LOCAL_REPLACE_PREFIX): object_sha = ObjectID(ref[len(LOCAL_REPLACE_PREFIX) :]) replacement_sha = r.refs[ref] replacements.append((object_sha, replacement_sha)) return replacements def replace_delete(repo: RepoPath, object_sha: ObjectID | str) -> None: """Delete a replacement ref. Args: repo: Path to repository object_sha: SHA of the object whose replacement should be removed """ with open_repo_closing(repo) as r: # Convert to ObjectID if string if isinstance(object_sha, str): object_sha_id = ObjectID(object_sha.encode("ascii")) else: object_sha_id = object_sha replace_ref = _make_replace_ref(object_sha_id) if replace_ref not in r.refs: raise KeyError( f"No replacement ref found for {object_sha_id.decode('ascii')}" ) del r.refs[replace_ref] def replace_create( repo: RepoPath, object_sha: str | ObjectID, replacement_sha: str | ObjectID, ) -> None: """Create a replacement ref to replace one object with another. Args: repo: Path to repository object_sha: SHA of the object to replace replacement_sha: SHA of the replacement object """ with open_repo_closing(repo) as r: # Convert to ObjectID if string if isinstance(object_sha, str): object_sha_id = ObjectID(object_sha.encode("ascii")) else: object_sha_id = object_sha if isinstance(replacement_sha, str): replacement_sha_id = ObjectID(replacement_sha.encode("ascii")) else: replacement_sha_id = replacement_sha # Create the replacement ref replace_ref = _make_replace_ref(object_sha_id) r.refs[replace_ref] = replacement_sha_id def reset( repo: str | os.PathLike[str] | Repo, mode: str, treeish: str | bytes | Commit | Tree | Tag = "HEAD", env: Mapping[str, str] | None = None, ) -> None: """Reset current HEAD to the specified state. Args: repo: Path to repository mode: Mode ("hard", "soft", "mixed") treeish: Treeish to reset to env: Environment to read GIT_REFLOG_ACTION from (defaults to os.environ) """ with open_repo_closing(repo) as r: # Parse the target tree tree = parse_tree(r, treeish) # Only parse as commit if treeish is not a Tree object if isinstance(treeish, Tree): # For Tree objects, we can't determine the commit, skip updating HEAD target_commit = None else: target_commit = parse_commit(r, treeish) # Update HEAD to point to the target commit if target_commit is not None: # Get the current HEAD value for set_if_equals try: old_head = r.refs[HEADREF] except KeyError: old_head = None # Create reflog message treeish_str = ( treeish.decode("utf-8") if isinstance(treeish, bytes) else str(treeish) if not isinstance(treeish, Commit | Tree | Tag) else target_commit.id.hex() ) default_message = f"reset: moving to {treeish_str}".encode() reflog_message = _get_reflog_message(default_message, env=env) # Pass committer explicitly: Repo._write_reflog would otherwise # resolve it via get_user_identity(), which reads os.environ. r.refs.set_if_equals( HEADREF, old_head, target_commit.id, committer=_get_user_identity( _config_stack(r, env=env), kind="COMMITTER", env=env ), message=reflog_message, ) if mode == "soft": # Soft reset: only update HEAD, leave index and working tree unchanged return elif mode == "mixed": # Mixed reset: update HEAD and index, but leave working tree unchanged from ..object_store import iter_tree_contents # Open the index index = r.open_index(config=r.get_config_stack()) # Clear the current index index.clear() # Populate index from the target tree for entry in iter_tree_contents(r.object_store, tree.id): # Create an IndexEntry from the tree entry # Use zeros for filesystem-specific fields since we're not touching the working tree assert ( entry.mode is not None and entry.sha is not None and entry.path is not None ) index_entry = IndexEntry( ctime=(0, 0), mtime=(0, 0), dev=0, ino=0, mode=entry.mode, uid=0, gid=0, size=0, # Size will be 0 since we're not reading from disk sha=entry.sha, flags=0, ) index[entry.path] = index_entry # Write the updated index index.write() elif mode == "hard": # Hard reset: update HEAD, index, and working tree # For reset --hard, use current index tree as old tree to get proper deletions index = r.open_index(config=r.get_config_stack()) if len(index) > 0: index_tree_id = index.commit(r.object_store) else: # Empty index index_tree_id = None # Get configuration for working tree updates honor_filemode, validate_path_element, symlink_fn = ( _get_worktree_update_config(r) ) blob_normalizer = r.get_blob_normalizer(config=r.get_config_stack()) changes = tree_changes( r.object_store, index_tree_id, tree.id, want_unchanged=True ) update_working_tree( r, index_tree_id, tree.id, change_iterator=changes, honor_filemode=honor_filemode, validate_path_element=validate_path_element, symlink_fn=symlink_fn, blob_normalizer=blob_normalizer, allow_overwrite_modified=True, # Allow overwriting modified files, config=r.get_config_stack(), ) else: raise Error(f"Invalid reset mode: {mode}") def get_remote_repo( repo: Repo, remote_location: str | bytes | None = None ) -> tuple[str | None, str]: """Get the remote repository information. Args: repo: Local repository object remote_location: Optional remote name or URL; defaults to branch remote Returns: Tuple of (remote_name, remote_url) where remote_name may be None if remote_location is a URL rather than a configured remote """ config = repo.get_config() if remote_location is None: remote_location = get_branch_remote(repo) if isinstance(remote_location, str): encoded_location = remote_location.encode() else: encoded_location = remote_location section = (b"remote", encoded_location) remote_name: str | None = None if config.has_section(section): remote_name = encoded_location.decode() encoded_location = config.get(section, "url") else: remote_name = None return (remote_name, encoded_location.decode()) def _find_reachable_tags( r: BaseRepo, pushed_shas: set[ObjectID], already_included: set[Ref], remote_refs: dict[Ref, ObjectID], ) -> Iterator[tuple[Ref, ObjectID]]: """Yield annotated tags whose targets are reachable from pushed commits. Args: r: Local repository pushed_shas: SHAs being pushed (non-deletion) already_included: Refs already being pushed remote_refs: Current remote refs """ reachable: set[ObjectID] = set() for sha in pushed_shas: try: obj = r[sha] except KeyError: continue if isinstance(obj, Commit): for entry in r.get_walker([sha]): reachable.add(entry.commit.id) for ref in r.refs.keys(): if not ref.startswith(LOCAL_TAG_PREFIX): continue if ref in already_included or ref in remote_refs: continue tag_sha = r.refs[ref] try: tag_obj = r[tag_sha] except KeyError: continue if isinstance(tag_obj, Tag) and tag_obj.object[1] in reachable: yield (Ref(ref), tag_sha) def _select_push_refs( r: BaseRepo, remote_refs: dict[Ref, ObjectID], refspecs: str | bytes | Sequence[str | bytes] | None, *, force: bool = False, mirror_mode: bool = False, all: bool = False, tags: bool = False, delete: bool = False, ) -> list[tuple[Ref | None, Ref | None, bool]]: """Select which refs to push based on mode flags or user refspecs. Returns a list of (local_ref, remote_ref, force) tuples. local_ref is None for deletions. Args: r: Local repository remote_refs: Current remote refs dict refspecs: User-provided refspecs (may be None) force: Force overwriting refs mirror_mode: Push all refs all: Push all branches tags: Push all tags delete: Delete the specified remote refs """ result: list[tuple[Ref | None, Ref | None, bool]] = [] if mirror_mode: for ref in r.refs.keys(): result.append((Ref(ref), Ref(ref), True)) elif all: for ref in r.refs.keys(): if ref.startswith(LOCAL_BRANCH_PREFIX): result.append((Ref(ref), Ref(ref), force)) elif delete: assert refspecs is not None if isinstance(refspecs, str | bytes): refspecs = [refspecs] remote_container = DictRefsContainer(remote_refs) # type: ignore[arg-type] for spec in refspecs: try: resolved = parse_ref(remote_container, spec) except KeyError: resolved = Ref(spec.encode() if isinstance(spec, str) else spec) result.append((None, resolved, force)) elif tags and refspecs is None: for ref in r.refs.keys(): if ref.startswith(LOCAL_TAG_PREFIX): result.append((Ref(ref), Ref(ref), force)) else: # Parse user-provided refspecs (or default to active branch) if refspecs is None: active_ref = r.refs.follow(HEADREF)[0][1] if not active_ref.startswith(LOCAL_BRANCH_PREFIX): raise ValueError(active_ref) refspecs = [active_ref[len(LOCAL_BRANCH_PREFIX) :]] elif isinstance(refspecs, str | bytes): refspecs = [refspecs] refspecs_bytes = [ spec.encode() if isinstance(spec, str) else spec for spec in refspecs ] remote_container = DictRefsContainer(remote_refs) # type: ignore[arg-type] result.extend( parse_reftuples(r.refs, remote_container, refspecs_bytes, force=force) ) # --tags combined with refspecs: also include all local tags if tags and refspecs is not None and not all and not mirror_mode: already = {rh for _, rh, _ in result} for ref in r.refs.keys(): if ref.startswith(LOCAL_TAG_PREFIX) and Ref(ref) not in already: result.append((Ref(ref), Ref(ref), force)) return result def push( repo: RepoPath, remote_location: str | bytes | None = None, refspecs: str | bytes | Sequence[str | bytes] | None = None, outstream: BinaryIO = default_bytes_out_stream, errstream: BinaryIO | RawIOBase = default_bytes_err_stream, force: bool = False, push_options: list[str] | None = None, atomic: bool = False, all: bool = False, tags: bool = False, delete: bool = False, dry_run: bool = False, prune: bool = False, set_upstream: bool = False, follow_tags: bool = False, mirror: bool = False, env: Mapping[str, str] | None = None, **kwargs: object, ) -> SendPackResult: """Remote push with dulwich via dulwich.client. Args: repo: Path to repository remote_location: Location of the remote refspecs: Refs to push to remote outstream: A stream file to write output errstream: A stream file to write errors force: Force overwriting refs push_options: Optional list of push options to send to the server (e.g. for AGit flow: ["topic=my-branch", "title=My PR"]) atomic: If True, request atomic push (all refs update or none do) all: If True, push all branches tags: If True, push all tags delete: If True, delete the specified remote refs dry_run: If True, do everything except actually send the updates prune: If True, remove remote refs that don't exist locally set_upstream: If True, set upstream tracking info for pushed branches follow_tags: If True, push annotated tags reachable from pushed commits mirror: If True, mirror all refs (implies force, push all refs and delete remote refs not present locally) env: Environment to read Git variables from (defaults to os.environ) **kwargs: Additional keyword arguments for the client """ if delete and not refspecs: raise Error("--delete requires ref arguments") # Open the repo with open_repo_closing(repo) as r: (remote_name, remote_location) = get_remote_repo(r, remote_location) # Check if mirror mode is enabled (via flag or config) mirror_mode = mirror if not mirror_mode and remote_name: try: mirror_mode_val = r.get_config_stack().get_boolean( (b"remote", remote_name.encode()), b"mirror" ) if mirror_mode_val is not None: mirror_mode = mirror_mode_val except KeyError: pass if mirror_mode: force = True # Get the client and path transport_kwargs = _filter_transport_kwargs(**kwargs) if transport_kwargs.get("ssh_command") is None: transport_kwargs["ssh_command"] = _ssh_command_from_env(env) client, path = get_transport_and_path( remote_location, config=r.get_config_stack(), **transport_kwargs, ) selected_refs: list[tuple[Ref | None, Ref | None, bool]] = [] remote_changed_refs: dict[Ref, ObjectID | None] = {} def update_refs(refs: dict[Ref, ObjectID]) -> dict[Ref, ObjectID]: new_refs: dict[Ref, ObjectID] = {} selected_refs.extend( _select_push_refs( r, refs, refspecs, force=force, mirror_mode=mirror_mode, all=all, tags=tags, delete=delete, ) ) # Prune remote branches not present locally if prune or mirror_mode: local_refs = set(r.refs.keys()) for remote_ref in refs: should_prune = (mirror_mode and remote_ref not in local_refs) or ( prune and not mirror_mode and remote_ref.startswith(LOCAL_BRANCH_PREFIX) and Ref(remote_ref[len(LOCAL_BRANCH_PREFIX) :]) not in set(r.refs.keys(base=Ref(LOCAL_BRANCH_PREFIX))) ) if should_prune: new_refs[remote_ref] = ZERO_SHA remote_changed_refs[remote_ref] = None # Apply selected ref mappings for lh, rh, force_ref in selected_refs: if lh is None: assert rh is not None new_refs[rh] = ZERO_SHA remote_changed_refs[rh] = None else: try: localsha = r.refs[lh] except KeyError as exc: raise Error( f"No valid ref {lh.decode() if isinstance(lh, bytes) else lh} in local repository" ) from exc assert rh is not None if not force_ref and rh in refs: check_diverged(r, refs[rh], localsha) new_refs[rh] = localsha remote_changed_refs[rh] = localsha # --follow-tags: push annotated tags reachable from pushed commits if follow_tags: pushed_shas = {sha for sha in new_refs.values() if sha != ZERO_SHA} for ref, sha in _find_reachable_tags( r, pushed_shas, set(new_refs), refs ): new_refs[ref] = sha remote_changed_refs[ref] = sha return new_refs err_encoding = getattr(errstream, "encoding", None) or DEFAULT_ENCODING remote_location = client.get_url(path) if dry_run: # Fetch remote refs without pushing, then compute what would change result = client.send_pack( path.encode(), lambda refs: refs, generate_pack_data=lambda have, want, **kw: (0, iter([])), progress=lambda data: (errstream.write(data), None)[1], ) update_refs({k: v for k, v in (result.refs or {}).items() if v is not None}) errstream.write( b"Push to " + remote_location.encode(err_encoding) + b" (dry run).\n" ) for rh, sha in remote_changed_refs.items(): action = "delete" if sha is None else "update" errstream.write( f"Would {action} {rh.decode('utf-8', 'replace')}\n".encode() ) return result try: def generate_pack_data_wrapper( have: AbstractSet[ObjectID], want: AbstractSet[ObjectID], *, ofs_delta: bool = False, progress: Callable[..., None] | None = None, ) -> tuple[int, Iterator[UnpackedObject]]: return r.generate_pack_data( set(have), set(want), progress=progress, ofs_delta=ofs_delta ) push_options_bytes: list[bytes] | None = None if push_options is not None: push_options_bytes = [ opt.encode() if isinstance(opt, str) else opt for opt in push_options ] result = client.send_pack( path.encode(), update_refs, generate_pack_data=generate_pack_data_wrapper, progress=lambda data: (errstream.write(data), None)[1], push_options=push_options_bytes, atomic=atomic, ) except SendPackError as exc: raise Error( "Push to " + remote_location + " failed -> " + exc.args[0].decode(), ) from exc else: errstream.write( b"Push to " + remote_location.encode(err_encoding) + b" successful.\n" ) for ref, error in (result.ref_status or {}).items(): if error is not None: errstream.write( f"Push of ref {ref.decode('utf-8', 'replace')} failed: {error}\n".encode( err_encoding ) ) else: errstream.write( f"Ref {ref.decode('utf-8', 'replace')} updated\n".encode() ) if remote_name is not None: _import_remote_refs(r.refs, remote_name, remote_changed_refs) # --set-upstream: configure tracking for pushed branches if set_upstream and remote_name is not None: config = r.get_config() for local_ref, remote_ref, _force_ref in selected_refs: if ( local_ref is not None and remote_ref is not None and local_ref.startswith(LOCAL_BRANCH_PREFIX) ): branch_name = local_ref[len(LOCAL_BRANCH_PREFIX) :] config.set( (b"branch", branch_name), b"remote", remote_name.encode(), ) config.set( (b"branch", branch_name), b"merge", remote_ref, ) config.write_to_path() return result # Trigger auto GC if needed from ..gc import maybe_auto_gc with open_repo_closing(repo) as r: maybe_auto_gc(r) def pull( repo: RepoPath, remote_location: str | bytes | None = None, refspecs: str | bytes | Sequence[str | bytes] | None = None, outstream: BinaryIO = default_bytes_out_stream, errstream: BinaryIO | RawIOBase = default_bytes_err_stream, fast_forward: bool = True, ff_only: bool = False, force: bool = False, filter_spec: str | None = None, protocol_version: int | None = None, env: Mapping[str, str] | None = None, **kwargs: object, ) -> None: """Pull from remote via dulwich.client. Args: repo: Path to repository remote_location: Location of the remote refspecs: refspecs to fetch. Can be a bytestring, a string, or a list of bytestring/string. outstream: A stream file to write to output errstream: A stream file to write to errors fast_forward: If True, raise an exception when fast-forward is not possible ff_only: If True, only allow fast-forward merges. Raises DivergedBranches when branches have diverged rather than performing a merge. force: If True, allow overwriting local changes in the working tree. If False, pull will abort if it would overwrite uncommitted changes. filter_spec: A git-rev-list-style object filter spec, as an ASCII string. Only used if the server supports the Git protocol-v2 'filter' feature, and ignored otherwise. protocol_version: desired Git protocol version. Defaults to the version requested in ``env``, and otherwise to the highest mutually supported protocol version. env: Environment to read Git variables from (defaults to os.environ) **kwargs: Additional keyword arguments for the client """ if protocol_version is None: protocol_version = _protocol_version_from_env(env) # Open the repo with open_repo_closing(repo) as r: (remote_name, remote_location) = get_remote_repo(r, remote_location) selected_refs = [] if refspecs is None: refspecs_normalized: bytes | list[bytes] = [b"HEAD"] elif isinstance(refspecs, str): refspecs_normalized = refspecs.encode() elif isinstance(refspecs, bytes): refspecs_normalized = refspecs else: refspecs_normalized = [] for spec in refspecs: if isinstance(spec, str): refspecs_normalized.append(spec.encode()) else: refspecs_normalized.append(spec) def determine_wants( remote_refs: dict[Ref, ObjectID], depth: int | None = None ) -> list[ObjectID]: remote_refs_container = DictRefsContainer(remote_refs) # type: ignore[arg-type] selected_refs.extend( parse_reftuples( remote_refs_container, r.refs, refspecs_normalized, force=force ) ) return [ remote_refs[lh] for (lh, rh, force_ref) in selected_refs if lh is not None and lh in remote_refs and remote_refs[lh] not in r.object_store ] transport_kwargs = _filter_transport_kwargs(**kwargs) if transport_kwargs.get("ssh_command") is None: transport_kwargs["ssh_command"] = _ssh_command_from_env(env) client, path = get_transport_and_path( remote_location, config=r.get_config_stack(), **transport_kwargs, ) if filter_spec: filter_spec_bytes: bytes | None = filter_spec.encode("ascii") else: filter_spec_bytes = None def progress(data: bytes) -> None: errstream.write(data) fetch_result = client.fetch( path.encode(), r, progress=progress, determine_wants=determine_wants, # type: ignore[arg-type] # Function matches protocol but mypy can't verify filter_spec=filter_spec_bytes, protocol_version=protocol_version, ) # Store the old HEAD tree before making changes try: old_head = r.refs[HEADREF] old_commit = r[old_head] assert isinstance(old_commit, Commit) old_tree_id = old_commit.tree except KeyError: old_tree_id = None merged = False for lh, rh, force_ref in selected_refs: if not force_ref and rh is not None and rh in r.refs: try: assert lh is not None followed_ref = r.refs.follow(rh)[1] assert followed_ref is not None lh_ref = fetch_result.refs[lh] assert lh_ref is not None check_diverged(r, followed_ref, lh_ref) except DivergedBranches as exc: if ff_only or fast_forward: raise else: # Perform merge assert lh is not None merge_ref = fetch_result.refs[lh] assert merge_ref is not None _merge_result, conflicts = _do_merge(r, merge_ref) if conflicts: raise Error( f"Merge conflicts occurred: {conflicts}" ) from exc merged = True # Skip updating ref since merge already updated HEAD continue if rh is not None and lh is not None: lh_value = fetch_result.refs[lh] if lh_value is not None: r.refs[Ref(rh)] = lh_value # Only update HEAD if we didn't perform a merge if selected_refs and not merged: lh, rh, _ = selected_refs[0] if lh is not None: ref_value = fetch_result.refs[lh] if ref_value is not None: r[b"HEAD"] = ref_value # Update working tree to match the new HEAD # Skip if merge was performed as merge already updates the working tree if not merged and old_tree_id is not None: head_commit = r[b"HEAD"] assert isinstance(head_commit, Commit) new_tree_id = head_commit.tree blob_normalizer = r.get_blob_normalizer(config=r.get_config_stack()) changes = tree_changes(r.object_store, old_tree_id, new_tree_id) update_working_tree( r, old_tree_id, new_tree_id, change_iterator=changes, blob_normalizer=blob_normalizer, allow_overwrite_modified=force, config=r.get_config_stack(), ) if remote_name is not None: _import_remote_refs(r.refs, remote_name, fetch_result.refs) # Trigger auto GC if needed from ..gc import maybe_auto_gc with open_repo_closing(repo) as r: maybe_auto_gc(r) def status( repo: str | os.PathLike[str] | Repo | None = None, ignored: bool = False, untracked_files: str = "normal", ) -> GitStatus: """Returns staged, unstaged, and untracked changes relative to the HEAD. Args: repo: Path to repository or repository object ignored: Whether to include ignored files in untracked untracked_files: How to handle untracked files, defaults to "all": "no": do not return untracked files "normal": return untracked directories, not their contents "all": include all files in untracked directories Using untracked_files="no" can be faster than "all" when the worktree contains many untracked files/directories. Using untracked_files="normal" provides a good balance, only showing directories that are entirely untracked without listing all their contents. Returns: GitStatus tuple, staged - dict with lists of staged paths (filesystem paths as bytes) unstaged - list of unstaged paths (filesystem paths as bytes) untracked - list of untracked paths (filesystem paths as bytes) """ with open_repo_closing(repo) as r: # Open the index once and reuse it for both staged and unstaged checks index = r.open_index(config=r.get_config_stack()) # 1. Get status of staged tracked_changes = get_tree_changes(r, index) # 2. Get status of unstaged normalizer = r.get_blob_normalizer(config=r.get_config_stack()) # Pass the normalizer's checkin_normalize method directly if normalizer is not None: filter_callback = normalizer.checkin_normalize else: filter_callback = None # Read config once for filesystem compatibility options config = r.get_config_stack() preload_index = config.get_boolean(b"core", b"preloadIndex", False) trust_ctime = config.get_boolean(b"core", b"trustctime", True) try: max_stat = int(config.get(b"core", b"maxStat")) except KeyError: max_stat = None precompose_unicode = config.get_boolean(b"core", b"precomposeunicode", False) unstaged_changes_tree = list( get_unstaged_changes( index, r.path, filter_callback, preload_index, trust_ctime, max_stat, ) ) untracked_paths = get_untracked_paths( r.path, r.path, index, exclude_ignored=not ignored, untracked_files=untracked_files, precompose_unicode=precompose_unicode, repo=r, ) # Convert all paths to filesystem encoding # Convert staged changes (dict with lists of tree paths) staged_fs = {} for change_type, paths in tracked_changes.items(): staged_fs[change_type] = [ tree_path_to_fs_path(p) if isinstance(p, bytes) else os.fsencode(p) for p in paths ] # Convert unstaged changes (list of tree paths) unstaged_fs = [tree_path_to_fs_path(p) for p in unstaged_changes_tree] # Convert untracked changes (list of strings) untracked_fs = [os.fsencode(p) for p in untracked_paths] return GitStatus(staged_fs, unstaged_fs, untracked_fs) def shortlog( repo: str | os.PathLike[str] | Repo, summary_only: bool = False, sort_by_commits: bool = False, ) -> list[dict[str, str]]: """Summarize commits by author, like git shortlog. Args: repo: Path to repository or Repo object. summary_only: If True, only show counts per author. sort_by_commits: If True, sort authors by number of commits. Returns: A list where each item is a dict containing: - "author": the author's name as a string - "messages": all commit messages concatenated into a single string """ with open_repo_closing(repo) as r: walker = r.get_walker() authors: dict[str, list[str]] = {} for entry in walker: commit = entry.commit author = commit.author.decode(commit.encoding or "utf-8") message = commit.message.decode(commit.encoding or "utf-8").strip() authors.setdefault(author, []).append(message) # Convert messages to single string per author items: list[dict[str, str]] = [ {"author": author, "messages": "\n".join(msgs)} for author, msgs in authors.items() ] if sort_by_commits: # Sort by number of commits (lines in messages) items.sort(key=lambda x: len(x["messages"].splitlines()), reverse=True) return items def _precompose_unicode_path(path: str) -> str: """Normalize a filesystem path to NFC (precomposed) Unicode form. On macOS, HFS+/APFS filesystems return filenames in NFD (decomposed) form. This function normalizes them to NFC so they match the paths stored in the git index. """ import unicodedata return unicodedata.normalize("NFC", path) def _walk_working_dir_paths( frompath: str | bytes | os.PathLike[str], basepath: str | bytes | os.PathLike[str], prune_dirnames: Callable[[str, list[str]], list[str]] | None = None, precompose_unicode: bool = False, ) -> Iterator[tuple[str | bytes, bool]]: """Get path, is_dir for files in working dir from frompath. Args: frompath: Path to begin walk basepath: Path to compare to prune_dirnames: Optional callback to prune dirnames during os.walk dirnames will be set to result of prune_dirnames(dirpath, dirnames) precompose_unicode: If True, normalize paths to NFC Unicode form """ # Convert paths to strings for os.walk compatibility for dirpath, dirnames, filenames in os.walk(frompath): # type: ignore[type-var] # Skip .git and below. if ".git" in dirnames: dirnames.remove(".git") if dirpath != basepath: continue if ".git" in filenames: filenames.remove(".git") if dirpath != basepath: continue if precompose_unicode and isinstance(dirpath, str): dirpath = _precompose_unicode_path(dirpath) dirnames[:] = [ _precompose_unicode_path(d) for d in dirnames if isinstance(d, str) ] filenames = [ _precompose_unicode_path(f) for f in filenames if isinstance(f, str) ] if dirpath != frompath: yield dirpath, True # type: ignore[misc] for filename in filenames: filepath = os.path.join(dirpath, filename) # type: ignore[call-overload] yield filepath, False if prune_dirnames: dirnames[:] = prune_dirnames(dirpath, dirnames) # type: ignore[arg-type] def get_untracked_paths( frompath: str | bytes | os.PathLike[str], basepath: str | bytes | os.PathLike[str], index: Index, exclude_ignored: bool = False, untracked_files: str = "all", precompose_unicode: bool = False, repo: Repo | None = None, ) -> Iterator[str]: """Get untracked paths. Args: frompath: Path to walk basepath: Path to compare to index: Index to check against exclude_ignored: Whether to exclude ignored paths untracked_files: How to handle untracked files: - "no": return an empty list - "all": return all files in untracked directories - "normal": return untracked directories without listing their contents precompose_unicode: If True, normalize filesystem paths to NFC Unicode form. This is needed on macOS where the filesystem returns NFD paths. repo: Repository to read ignore patterns from. Required when the working tree is not the repository root, as it is with ``core.worktree``; defaults to opening a repository at ``basepath``. Note: ignored directories will never be walked for performance reasons. If exclude_ignored is False, only the path to an ignored directory will be yielded, no files inside the directory will be returned """ if untracked_files not in ("no", "all", "normal"): raise ValueError("untracked_files must be one of (no, all, normal)") if untracked_files == "no": return # Normalize paths to str frompath_str = os.fsdecode(os.fspath(frompath)) basepath_str = os.fsdecode(os.fspath(basepath)) with open_repo_closing(repo if repo is not None else basepath_str) as r: ignore_manager = IgnoreFilterManager.from_repo(r, config=r.get_config_stack()) ignored_dirs = [] # List to store untracked directories found during traversal untracked_dir_list = [] def directory_has_non_ignored_files(dir_path: str, base_rel_path: str) -> bool: """Recursively check if directory contains any non-ignored files.""" try: for entry in os.listdir(dir_path): entry_path = os.path.join(dir_path, entry) rel_entry = os.path.join(base_rel_path, entry) if os.path.isfile(entry_path): if ignore_manager.is_ignored(rel_entry) is not True: return True elif os.path.isdir(entry_path): if directory_has_non_ignored_files(entry_path, rel_entry): return True return False except OSError: # If we can't read the directory, assume it has non-ignored files return True def prune_dirnames(dirpath: str, dirnames: list[str]) -> list[str]: for i in range(len(dirnames) - 1, -1, -1): path = os.path.join(dirpath, dirnames[i]) ip = os.path.join(os.path.relpath(path, basepath_str), "") # Check if directory is ignored if ignore_manager.is_ignored(ip) is True: if not exclude_ignored: ignored_dirs.append( os.path.join(os.path.relpath(path, frompath_str), "") ) del dirnames[i] continue # For "normal" mode, check if the directory is entirely untracked if untracked_files == "normal": # Convert directory path to tree path for index lookup dir_tree_path = path_to_tree_path(basepath_str, path) # Check if any file in this directory is tracked dir_prefix = dir_tree_path + b"/" if dir_tree_path else b"" has_tracked_files = any(name.startswith(dir_prefix) for name in index) if not has_tracked_files: # This directory is entirely untracked rel_path_base = os.path.relpath(path, basepath_str) rel_path_from = os.path.join( os.path.relpath(path, frompath_str), "" ) # If excluding ignored, check if directory contains any non-ignored files if exclude_ignored: if not directory_has_non_ignored_files(path, rel_path_base): # Directory only contains ignored files, skip it del dirnames[i] continue # Check if it should be excluded due to ignore rules is_ignored = ignore_manager.is_ignored(rel_path_base) if not exclude_ignored or not is_ignored: untracked_dir_list.append(rel_path_from) del dirnames[i] return dirnames # For "all" mode, use the original behavior if untracked_files == "all": for ap, is_dir in _walk_working_dir_paths( frompath_str, basepath_str, prune_dirnames=prune_dirnames, precompose_unicode=precompose_unicode, ): # frompath_str and basepath_str are both str, so ap must be str assert isinstance(ap, str) if not is_dir: ip = path_to_tree_path(basepath_str, ap) if ip not in index: if not exclude_ignored or not ignore_manager.is_ignored( os.path.relpath(ap, basepath_str) ): yield os.path.relpath(ap, frompath_str) else: # "normal" mode # Walk directories, handling both files and directories for ap, is_dir in _walk_working_dir_paths( frompath_str, basepath_str, prune_dirnames=prune_dirnames, precompose_unicode=precompose_unicode, ): # frompath_str and basepath_str are both str, so ap must be str assert isinstance(ap, str) # This part won't be reached for pruned directories if is_dir: # Check if this directory is entirely untracked dir_tree_path = path_to_tree_path(basepath_str, ap) dir_prefix = dir_tree_path + b"/" if dir_tree_path else b"" has_tracked_files = any(name.startswith(dir_prefix) for name in index) if not has_tracked_files: if not exclude_ignored or not ignore_manager.is_ignored( os.path.relpath(ap, basepath_str) ): yield os.path.join(os.path.relpath(ap, frompath_str), "") else: # Check individual files in directories that contain tracked files ip = path_to_tree_path(basepath_str, ap) if ip not in index: if not exclude_ignored or not ignore_manager.is_ignored( os.path.relpath(ap, basepath_str) ): yield os.path.relpath(ap, frompath_str) # Yield any untracked directories found during pruning yield from untracked_dir_list yield from ignored_dirs def grep( repo: RepoPath, pattern: str | bytes, *, outstream: TextIO = sys.stdout, rev: str | bytes | None = None, pathspecs: Sequence[str | bytes] | None = None, ignore_case: bool = False, line_number: bool = False, max_depth: int | None = None, respect_ignores: bool = True, ) -> None: """Search for a pattern in tracked files. Args: repo: Path to repository or Repo object pattern: Regular expression pattern to search for outstream: Stream to write results to rev: Revision to search in (defaults to HEAD) pathspecs: Optional list of path patterns to limit search ignore_case: Whether to perform case-insensitive matching line_number: Whether to output line numbers max_depth: Maximum directory depth to search respect_ignores: Whether to respect .gitignore patterns """ from ..object_store import iter_tree_contents # Compile the pattern flags = re.IGNORECASE if ignore_case else 0 try: if isinstance(pattern, bytes): compiled_pattern = re.compile(pattern, flags) else: compiled_pattern = re.compile(pattern.encode("utf-8"), flags) except re.error as e: raise ValueError(f"Invalid regular expression: {e}") from e with open_repo_closing(repo) as r: # Get the tree to search if rev is None: try: commit = r[b"HEAD"] assert isinstance(commit, Commit) except KeyError as e: raise ValueError("No HEAD commit found") from e else: rev_bytes = rev if isinstance(rev, bytes) else rev.encode("utf-8") commit_obj = parse_commit(r, rev_bytes) if commit_obj is None: raise ValueError(f"Invalid revision: {rev}") commit = commit_obj tree = r[commit.tree] assert isinstance(tree, Tree) # Set up ignore filter if requested ignore_manager = None if respect_ignores: ignore_manager = IgnoreFilterManager.from_repo( r, config=r.get_config_stack() ) # Convert pathspecs to bytes pathspecs_bytes: list[bytes] | None = None if pathspecs: pathspecs_bytes = [ p if isinstance(p, bytes) else p.encode("utf-8") for p in pathspecs ] # Iterate through all files in the tree for entry in iter_tree_contents(r.object_store, tree.id): path, mode, sha = entry.path, entry.mode, entry.sha assert path is not None assert mode is not None assert sha is not None # Skip directories if stat.S_ISDIR(mode): continue # Check max depth if max_depth is not None: depth = path.count(b"/") if depth > max_depth: continue # Check pathspecs if pathspecs_bytes: matches_pathspec = False for pathspec in pathspecs_bytes: # Simple prefix matching (could be enhanced with full pathspec support) if path.startswith(pathspec) or fnmatch.fnmatch( path.decode("utf-8", errors="replace"), pathspec.decode("utf-8", errors="replace"), ): matches_pathspec = True break if not matches_pathspec: continue # Check ignore patterns if ignore_manager: path_str = path.decode("utf-8", errors="replace") if ignore_manager.is_ignored(path_str) is True: continue # Get the blob content blob = r[sha] assert isinstance(blob, Blob) # Search for pattern in the blob content = blob.data lines = content.split(b"\n") for line_num, line in enumerate(lines, 1): if compiled_pattern.search(line): path_str = path.decode("utf-8", errors="replace") line_str = line.decode("utf-8", errors="replace") if line_number: outstream.write(f"{path_str}:{line_num}:{line_str}\n") else: outstream.write(f"{path_str}:{line_str}\n") def get_tree_changes( repo: RepoPath, index: Index | None = None ) -> dict[str, list[str | bytes]]: """Return add/delete/modify changes to tree by comparing index to HEAD. Args: repo: repo path or object index: optional Index object to reuse (avoids re-opening the index) Returns: dict with lists for each type of change """ with open_repo_closing(repo) as r: if index is None: index = r.open_index(config=r.get_config_stack()) # Compares the Index to the HEAD & determines changes # Iterate through the changes and report add/delete/modify # TODO: call out to dulwich.diff_tree somehow. tracked_changes: dict[str, list[str | bytes]] = { "add": [], "delete": [], "modify": [], } try: head_commit = r[b"HEAD"] assert isinstance(head_commit, Commit) tree_id = head_commit.tree except KeyError: tree_id = None for change in index.changes_from_tree(r.object_store, tree_id): if not change[0][0]: assert change[0][1] is not None tracked_changes["add"].append(change[0][1]) elif not change[0][1]: assert change[0][0] is not None tracked_changes["delete"].append(change[0][0]) elif change[0][0] == change[0][1]: assert change[0][0] is not None tracked_changes["modify"].append(change[0][0]) else: raise NotImplementedError("git mv ops not yet supported") return tracked_changes def daemon( path: str | os.PathLike[str] = ".", address: str | None = None, port: int | None = None, ) -> None: """Run a daemon serving Git requests over TCP/IP. Args: path: Path to the directory to serve. address: Optional address to listen on (defaults to ::) port: Optional port to listen on (defaults to TCP_GIT_PORT) """ # TODO(jelmer): Support git-daemon-export-ok and --export-all. backend = FileSystemBackend(os.fspath(path)) server = TCPGitServer(backend, address or "localhost", port or 9418) server.serve_forever() def web_daemon( path: str | os.PathLike[str] = ".", address: str | None = None, port: int | None = None, ) -> None: """Run a daemon serving Git requests over HTTP. Args: path: Path to the directory to serve address: Optional address to listen on (defaults to ::) port: Optional port to listen on (defaults to 80) """ from ..web import ( WSGIRequestHandlerLogger, WSGIServerLogger, make_server, make_wsgi_chain, ) backend = FileSystemBackend(os.fspath(path)) app = make_wsgi_chain(backend) server = make_server( address or "::", port or 80, app, handler_class=WSGIRequestHandlerLogger, server_class=WSGIServerLogger, ) server.serve_forever() def upload_pack( path: str | os.PathLike[str] = ".", inf: BinaryIO | None = None, outf: BinaryIO | None = None, ) -> int: """Upload a pack file after negotiating its contents using smart protocol. Args: path: Path to the repository inf: Input stream to communicate with client outf: Output stream to communicate with client """ if outf is None: outf = sys.stdout.buffer if inf is None: inf = sys.stdin.buffer assert outf is not None assert inf is not None path = os.path.expanduser(path) backend = FileSystemBackend(path) def send_fn(data: bytes) -> None: outf.write(data) outf.flush() proto = Protocol(inf.read, send_fn) handler = UploadPackHandler(backend, [path], proto) # FIXME: Catch exceptions and write a single-line summary to outf. handler.handle() return 0 def receive_pack( path: str | os.PathLike[str] = ".", inf: BinaryIO | None = None, outf: BinaryIO | None = None, ) -> int: """Receive a pack file after negotiating its contents using smart protocol. Args: path: Path to the repository inf: Input stream to communicate with client outf: Output stream to communicate with client """ if outf is None: outf = sys.stdout.buffer if inf is None: inf = sys.stdin.buffer assert outf is not None assert inf is not None path = os.path.expanduser(path) backend = FileSystemBackend(path) def send_fn(data: bytes) -> None: outf.write(data) outf.flush() proto = Protocol(inf.read, send_fn) handler = ReceivePackHandler(backend, [path], proto) # FIXME: Catch exceptions and write a single-line summary to outf. handler.handle() return 0 def _make_branch_ref(name: str | bytes) -> Ref: if isinstance(name, str): name = name.encode(DEFAULT_ENCODING) return local_branch_name(name) def _make_replace_ref(name: str | bytes | ObjectID) -> Ref: if isinstance(name, str): name = name.encode(DEFAULT_ENCODING) return local_replace_name(name) def branch_delete(repo: RepoPath, name: str | bytes | Sequence[str | bytes]) -> None: """Delete a branch. Args: repo: Path to the repository name: Name of the branch """ with open_repo_closing(repo) as r: names: Sequence[str | bytes] if isinstance(name, str | bytes): names = [name] else: names = name for branch_name in names: del r.refs[_make_branch_ref(branch_name)] def branch_create( repo: str | os.PathLike[str] | Repo, name: str | bytes, objectish: str | bytes | None = None, force: bool = False, env: Mapping[str, str] | None = None, ) -> None: """Create a branch. Args: repo: Path to the repository name: Name of the new branch objectish: Target object to point new branch at (defaults to HEAD) force: Force creation of branch, even if it already exists env: Environment to read GIT_REFLOG_ACTION from (defaults to os.environ) """ with open_repo_closing(repo) as r: if objectish is None: objectish = "HEAD" # Try to expand branch shorthand before parsing original_objectish = objectish objectish_bytes = ( objectish.encode(DEFAULT_ENCODING) if isinstance(objectish, str) else objectish ) if Ref(b"refs/remotes/" + objectish_bytes) in r.refs: objectish = b"refs/remotes/" + objectish_bytes elif local_branch_name(objectish_bytes) in r.refs: objectish = local_branch_name(objectish_bytes) object = parse_object(r, objectish, config=r.get_config_stack()) refname = _make_branch_ref(name) default_message = ( b"branch: Created from " + original_objectish.encode(DEFAULT_ENCODING) if isinstance(original_objectish, str) else b"branch: Created from " + original_objectish ) ref_message = _get_reflog_message(default_message, env=env) # Pass committer explicitly: Repo._write_reflog would otherwise resolve # it via get_user_identity(), which reads os.environ. ref_committer = _get_user_identity( _config_stack(r, env=env), kind="COMMITTER", env=env ) if force: r.refs.set_if_equals( refname, None, object.id, committer=ref_committer, message=ref_message ) else: if not r.refs.add_if_new( refname, object.id, committer=ref_committer, message=ref_message ): name_str = name.decode() if isinstance(name, bytes) else name raise Error(f"Branch with name {name_str} already exists.") # Check if we should set up tracking config = r.get_config_stack() try: auto_setup_merge = config.get((b"branch",), b"autoSetupMerge").decode() except KeyError: auto_setup_merge = "true" # Default value # Determine if the objectish refers to a remote-tracking branch objectish_ref = None if original_objectish != "HEAD": # Try to resolve objectish as a ref objectish_bytes = ( original_objectish.encode(DEFAULT_ENCODING) if isinstance(original_objectish, str) else original_objectish ) if Ref(objectish_bytes) in r.refs: objectish_ref = objectish_bytes elif Ref(b"refs/remotes/" + objectish_bytes) in r.refs: objectish_ref = b"refs/remotes/" + objectish_bytes elif local_branch_name(objectish_bytes) in r.refs: objectish_ref = local_branch_name(objectish_bytes) else: # HEAD might point to a remote-tracking branch head_ref = r.refs.follow(HEADREF)[0][1] if head_ref.startswith(b"refs/remotes/"): objectish_ref = head_ref # Set up tracking if appropriate if objectish_ref and ( (auto_setup_merge == "always") or ( auto_setup_merge == "true" and objectish_ref.startswith(b"refs/remotes/") ) ): # Extract remote name and branch from the ref if objectish_ref.startswith(b"refs/remotes/"): parts = objectish_ref[len(b"refs/remotes/") :].split(b"/", 1) if len(parts) == 2: remote_name = parts[0] remote_branch = local_branch_name(parts[1]) # Set up tracking repo_config = r.get_config() branch_name_bytes = ( name.encode(DEFAULT_ENCODING) if isinstance(name, str) else name ) repo_config.set( (b"branch", branch_name_bytes), b"remote", remote_name ) repo_config.set( (b"branch", branch_name_bytes), b"merge", remote_branch ) repo_config.write_to_path() def filter_branches_by_pattern(branches: Iterable[bytes], pattern: str) -> list[bytes]: """Filter branches by fnmatch pattern. Args: branches: Iterable of branch names as bytes pattern: Pattern to match against Returns: List of filtered branch names """ return [ branch for branch in branches if fnmatch.fnmatchcase(branch.decode(), pattern) ] def branch_list(repo: RepoPath) -> list[Ref]: """List all branches. Args: repo: Path to the repository Returns: List of branch names (without refs/heads/ prefix) """ with open_repo_closing(repo) as r: branches: list[Ref] = list(r.refs.keys(base=Ref(LOCAL_BRANCH_PREFIX))) # Check for branch.sort configuration config = r.get_config_stack() try: sort_key = config.get((b"branch",), b"sort").decode() except KeyError: # Default is refname (alphabetical) sort_key = "refname" # Parse sort key reverse = False if sort_key.startswith("-"): reverse = True sort_key = sort_key[1:] # Apply sorting if sort_key == "refname": # Simple alphabetical sort (default) branches.sort(reverse=reverse) elif sort_key in ("committerdate", "authordate"): # Sort by date def get_commit_date(branch_name: bytes) -> int: ref = local_branch_name(branch_name) sha = r.refs[ref] commit = r.object_store[sha] assert isinstance(commit, Commit) if sort_key == "committerdate": return cast(int, commit.commit_time) else: # authordate return cast(int, commit.author_time) # Sort branches by date # Note: Python's sort naturally orders smaller values first (ascending) # For dates, this means oldest first by default # Use a stable sort with branch name as secondary key for consistent ordering if reverse: # For reverse sort, we want newest dates first but alphabetical names second branches.sort(key=lambda b: (-get_commit_date(b), b)) else: branches.sort(key=lambda b: (get_commit_date(b), b)) else: # Unknown sort key, fall back to default branches.sort() return branches def branch_remotes_list(repo: RepoPath) -> list[bytes]: """List the short names of all remote branches. Args: repo: Path to the repository Returns: List of branch names (without refs/remotes/ prefix, and without remote name; e.g. 'main' from 'origin/main') """ with open_repo_closing(repo) as r: branches = [bytes(ref) for ref in r.refs.keys(base=Ref(LOCAL_REMOTE_PREFIX))] config = r.get_config_stack() try: sort_key = config.get((b"branch",), b"sort").decode() except KeyError: # Default is refname (alphabetical) sort_key = "refname" # Parse sort key reverse = False if sort_key.startswith("-"): reverse = True sort_key = sort_key[1:] # Apply sorting if sort_key == "refname": # Simple alphabetical sort (default) branches.sort(reverse=reverse) elif sort_key in ("committerdate", "authordate"): # Sort by date def get_commit_date(branch_name: bytes) -> int: ref = LOCAL_REMOTE_PREFIX + branch_name sha = r.refs[Ref(ref)] commit = r.object_store[sha] assert isinstance(commit, Commit) if sort_key == "committerdate": return cast(int, commit.commit_time) else: # authordate return cast(int, commit.author_time) # Sort branches by date # Note: Python's sort naturally orders smaller values first (ascending) # For dates, this means oldest first by default # Use a stable sort with branch name as secondary key for consistent ordering if reverse: # For reverse sort, we want newest dates first but alphabetical names second branches.sort(key=lambda b: (-get_commit_date(b), b)) else: branches.sort(key=lambda b: (get_commit_date(b), b)) else: # Unknown sort key raise ValueError(f"Unknown sort key: {sort_key}") return branches def _get_branch_merge_status(repo: RepoPath) -> Iterator[tuple[bytes, bool]]: """Get merge status for all branches relative to current HEAD. Args: repo: Path to the repository Yields: tuple of (``branch_name``, ``is_merged``) where: - ``branch_name``: Branch name without refs/heads/ prefix - ``is_merged``: True if branch is merged into HEAD, False otherwise """ with open_repo_closing(repo) as r: current_sha = r.refs[HEADREF] for branch_ref, branch_sha in r.refs.as_dict(base=Ref(b"refs/heads/")).items(): # Check if branch is an ancestor of HEAD (fully merged) is_merged = can_fast_forward(r, branch_sha, current_sha) yield branch_ref, is_merged def merged_branches(repo: RepoPath) -> Iterator[bytes]: """List branches that have been merged into the current branch. Args: repo: Path to the repository Yields: Branch names (without refs/heads/ prefix) that are merged into the current HEAD """ for branch_name, is_merged in _get_branch_merge_status(repo): if is_merged: yield branch_name def no_merged_branches(repo: RepoPath) -> Iterator[bytes]: """List branches that have been merged into the current branch. Args: repo: Path to the repository Yields: Branch names (without refs/heads/ prefix) that are merged into the current HEAD """ for branch_name, is_merged in _get_branch_merge_status(repo): if not is_merged: yield branch_name def branches_containing(repo: RepoPath, commit: str) -> Iterator[bytes]: """List branches that contain the specified commit. Args: repo: Path to the repository commit: Commit-ish string (SHA, branch name, tag, etc.) Yields: Branch names (without refs/heads/ prefix) that contain the commit Raises: ValueError: If the commit reference is malformed KeyError: If the commit reference does not exist """ with open_repo_closing(repo) as r: commit_obj = parse_commit(r, commit) commit_sha = commit_obj.id for branch_ref, branch_sha in r.refs.as_dict( base=Ref(LOCAL_BRANCH_PREFIX) ).items(): if can_fast_forward(r, commit_sha, branch_sha): yield branch_ref def active_branch(repo: RepoPath) -> bytes: """Return the active branch in the repository, if any. Args: repo: Repository to open Returns: branch name Raises: KeyError: if the repository does not have a working tree IndexError: if HEAD is floating """ with open_repo_closing(repo) as r: active_ref = r.refs.follow(HEADREF)[0][1] if not active_ref.startswith(LOCAL_BRANCH_PREFIX): raise ValueError(active_ref) return active_ref[len(LOCAL_BRANCH_PREFIX) :] def get_branch_remote(repo: str | os.PathLike[str] | Repo) -> bytes: """Return the active branch's remote name, if any. Args: repo: Repository to open Returns: remote name Raises: KeyError: if the repository does not have a working tree """ with open_repo_closing(repo) as r: branch_name = active_branch(r.path) config = r.get_config() try: remote_name = config.get((b"branch", branch_name), b"remote") except KeyError: remote_name = b"origin" return remote_name def get_branch_merge(repo: RepoPath, branch_name: bytes | None = None) -> bytes: """Return the branch's merge reference (upstream branch), if any. Args: repo: Repository to open branch_name: Name of the branch (defaults to active branch) Returns: merge reference name (e.g. b"refs/heads/main") Raises: KeyError: if the branch does not have a merge configuration """ with open_repo_closing(repo) as r: if branch_name is None: branch_name = active_branch(r.path) config = r.get_config() return config.get((b"branch", branch_name), b"merge") def set_branch_tracking( repo: str | os.PathLike[str] | Repo, branch_name: bytes, remote_name: bytes, remote_ref: bytes, ) -> None: """Set up branch tracking configuration. Args: repo: Repository to open branch_name: Name of the local branch remote_name: Name of the remote (e.g. b"origin") remote_ref: Remote reference to track (e.g. b"refs/heads/main") """ with open_repo_closing(repo) as r: config = r.get_config() config.set((b"branch", branch_name), b"remote", remote_name) config.set((b"branch", branch_name), b"merge", remote_ref) config.write_to_path() def fetch( repo: RepoPath, remote_location: str | bytes | None = None, outstream: TextIO = sys.stdout, errstream: BinaryIO | RawIOBase = default_bytes_err_stream, message: bytes | None = None, depth: int | None = None, prune: bool = False, prune_tags: bool = False, force: bool = False, operation: str | None = None, thin_packs: bool = True, report_activity: Callable[[int, str], None] | None = None, quiet: bool = False, include_tags: bool = False, username: str | None = None, password: str | None = None, key_filename: str | None = None, ssh_command: str | None = None, shallow_since: str | None = None, shallow_exclude: list[str] | None = None, unshallow: bool = False, env: Mapping[str, str] | None = None, ) -> FetchPackResult: """Fetch objects from a remote server. Args: repo: Path to the repository remote_location: String identifying a remote server outstream: Output stream (defaults to stdout) errstream: Error stream (defaults to stderr) message: Reflog message (defaults to b"fetch: from ") depth: Depth to fetch at prune: Prune remote removed refs prune_tags: Prune remote removed tags force: Force fetching even if it would overwrite local changes operation: Git operation for authentication (e.g., "fetch") thin_packs: Whether to use thin packs report_activity: Optional callback for reporting transport activity quiet: Whether to suppress progress output include_tags: Whether to include tags username: Username for authentication password: Password for authentication key_filename: SSH key filename ssh_command: SSH command to use. Defaults to the command configured in ``env``. shallow_since: Deepen or shorten the history to include commits after this date shallow_exclude: Deepen or shorten the history to exclude commits reachable from these refs unshallow: Convert a shallow repository to a complete one env: Environment to read Git variables from (defaults to os.environ) Returns: Dictionary with refs on the remote """ if ssh_command is None: ssh_command = _ssh_command_from_env(env) with open_repo_closing(repo) as r: (remote_name, remote_location) = get_remote_repo(r, remote_location) default_message = b"fetch: from " + remote_location.encode(DEFAULT_ENCODING) message = _get_reflog_message(default_message, message, env=env) # Handle unshallow option if unshallow: if depth is not None: raise ValueError("--unshallow and --depth are mutually exclusive") # Use a very large depth to fetch complete history (same as Git) depth = INFINITE_DEPTH client, path = get_transport_and_path( remote_location, config=r.get_config_stack(), operation=operation, thin_packs=thin_packs, report_activity=report_activity, quiet=quiet, include_tags=include_tags, username=username, password=password, key_filename=key_filename, ssh_command=ssh_command, ) def progress(data: bytes) -> None: errstream.write(data) fetch_result = client.fetch( path.encode(), r, progress=progress, depth=depth, shallow_since=shallow_since, shallow_exclude=shallow_exclude, ) if remote_name is not None: _import_remote_refs( r.refs, remote_name, fetch_result.refs, message, prune=prune, prune_tags=prune_tags, ) # Trigger auto GC if needed from ..gc import maybe_auto_gc with open_repo_closing(repo) as r: maybe_auto_gc(r) return fetch_result def for_each_ref( repo: Repo | str | None = None, pattern: str | bytes | None = None, ) -> list[tuple[bytes, bytes, bytes]]: """Iterate over all refs that match the (optional) pattern. Args: repo: Path to the repository pattern: Optional glob (7) patterns to filter the refs with Returns: List of bytes tuples with: (sha, object_type, ref_name) """ if isinstance(pattern, str): pattern = os.fsencode(pattern) with open_repo_closing(repo) as r: refs = r.get_refs() if pattern: matching_refs: dict[Ref, ObjectID] = {} pattern_parts = pattern.split(b"/") for ref, sha in refs.items(): matches = False # git for-each-ref uses glob (7) style patterns, but fnmatch # is greedy and also matches slashes, unlike glob.glob. # We have to check parts of the pattern individually. # See https://github.com/python/cpython/issues/72904 ref_parts = ref.split(b"/") if len(ref_parts) > len(pattern_parts): continue for pat, ref_part in zip(pattern_parts, ref_parts): matches = fnmatch.fnmatchcase(ref_part, pat) if not matches: break if matches: matching_refs[ref] = sha refs = matching_refs ret: list[tuple[bytes, bytes, bytes]] = [ (sha, r.get_object(sha).type_name, ref) for ref, sha in sorted( refs.items(), key=lambda ref_sha: ref_sha[0], ) if ref != b"HEAD" ] return ret def show_ref( repo: Repo | str | None = None, patterns: list[str | bytes] | None = None, head: bool = False, branches: bool = False, tags: bool = False, dereference: bool = False, verify: bool = False, ) -> list[tuple[bytes, bytes]]: """List references in a local repository. Args: repo: Path to the repository patterns: Optional list of patterns to filter refs (matched from the end) head: Show the HEAD reference branches: Limit to local branches (refs/heads/) tags: Limit to local tags (refs/tags/) dereference: Dereference tags into object IDs verify: Enable stricter reference checking (exact path match) Returns: List of tuples with (sha, ref_name) or (sha, ref_name^{}) for dereferenced tags """ # Convert string patterns to bytes byte_patterns: list[bytes] | None = None if patterns: byte_patterns = [os.fsencode(p) if isinstance(p, str) else p for p in patterns] with open_repo_closing(repo) as r: refs = r.get_refs() # Filter by branches/tags if specified if branches or tags: prefixes = [] if branches: prefixes.append(LOCAL_BRANCH_PREFIX) if tags: prefixes.append(LOCAL_TAG_PREFIX) filtered_refs = filter_ref_prefix(refs, prefixes) else: # By default, show tags, heads, and remote refs (but not HEAD) filtered_refs = filter_ref_prefix(refs, [b"refs/"]) # Add HEAD if requested if head and HEADREF in refs: filtered_refs[HEADREF] = refs[HEADREF] # Filter by patterns if specified if byte_patterns: matching_refs: dict[Ref, ObjectID] = {} for ref, sha in filtered_refs.items(): for pattern in byte_patterns: if verify: # Verify mode requires exact match if ref == pattern: matching_refs[ref] = sha break else: # Pattern matching from the end of the full name # Only complete parts are matched # E.g., "master" matches "refs/heads/master" but not "refs/heads/mymaster" pattern_parts = pattern.split(b"/") ref_parts = ref.split(b"/") # Try to match from the end if len(pattern_parts) <= len(ref_parts): # Check if the end of ref matches the pattern matches = True for i in range(len(pattern_parts)): if ( ref_parts[-(len(pattern_parts) - i)] != pattern_parts[i] ): matches = False break if matches: matching_refs[ref] = sha break filtered_refs = matching_refs # Sort by ref name sorted_refs = sorted(filtered_refs.items(), key=lambda x: x[0]) # Build result list result: list[tuple[bytes, bytes]] = [] for ref, sha in sorted_refs: result.append((sha, ref)) # Dereference tags if requested if dereference and ref.startswith(LOCAL_TAG_PREFIX): try: obj = r.get_object(sha) # Peel tag objects to get the underlying commit/object while obj.type_name == b"tag": assert isinstance(obj, Tag) _obj_class, sha = obj.object obj = r.get_object(sha) result.append((sha, ref + b"^{}")) except KeyError: # Object not found, skip dereferencing pass return result def show_branch( repo: Repo | str | None = None, branches: list[str | bytes] | None = None, all_branches: bool = False, remotes: bool = False, current: bool = False, topo_order: bool = False, more: int | None = None, list_branches: bool = False, independent_branches: bool = False, merge_base: bool = False, ) -> list[str]: """Display branches and their commits. Args: repo: Path to the repository branches: List of specific branches to show (default: all local branches) all_branches: Show both local and remote branches remotes: Show only remote branches current: Include current branch if not specified topo_order: Show in topological order instead of chronological more: Show N more commits beyond common ancestor (negative to show only headers) list_branches: Synonym for more=-1 (show only branch headers) independent_branches: Show only branches not reachable from others merge_base: Show merge bases instead of commit list Returns: List of output lines """ from ..graph import find_octopus_base, independent output_lines: list[str] = [] with open_repo_closing(repo) as r: refs = r.get_refs() # Determine which branches to show branch_refs: dict[Ref, ObjectID] = {} if branches: # Specific branches requested for branch in branches: branch_bytes = ( os.fsencode(branch) if isinstance(branch, str) else branch ) # Try as full ref name first branch_ref_check = Ref(branch_bytes) if branch_ref_check in refs: branch_refs[branch_ref_check] = refs[branch_ref_check] else: # Try as branch name branch_ref = local_branch_name(branch_bytes) if branch_ref in refs: branch_refs[branch_ref] = refs[branch_ref] # Try as remote branch else: remote_ref = Ref(LOCAL_REMOTE_PREFIX + branch_bytes) if remote_ref in refs: branch_refs[remote_ref] = refs[remote_ref] else: # Default behavior: show local branches if all_branches: # Show both local and remote branches branch_refs = filter_ref_prefix( refs, [LOCAL_BRANCH_PREFIX, LOCAL_REMOTE_PREFIX] ) elif remotes: # Show only remote branches branch_refs = filter_ref_prefix(refs, [LOCAL_REMOTE_PREFIX]) else: # Show only local branches branch_refs = filter_ref_prefix(refs, [LOCAL_BRANCH_PREFIX]) # Add current branch if requested and not already included if current: try: head_refs, _ = r.refs.follow(HEADREF) if head_refs: head_ref = head_refs[0] if head_ref not in branch_refs and head_ref in refs: branch_refs[head_ref] = refs[head_ref] except (KeyError, TypeError): # HEAD doesn't point to a branch or doesn't exist pass if not branch_refs: return output_lines # Sort branches for consistent output sorted_branches = sorted(branch_refs.items(), key=lambda x: x[0]) branch_sha_list: list[ObjectID] = [sha for _, sha in sorted_branches] # Handle --independent flag if independent_branches: independent_shas = independent(r, branch_sha_list) for ref_name, sha in sorted_branches: if sha in independent_shas: ref_str = os.fsdecode(shorten_ref_name(ref_name)) output_lines.append(ref_str) return output_lines # Handle --merge-base flag if merge_base: if len(branch_sha_list) < 2: # Need at least 2 branches for merge base return output_lines merge_bases = find_octopus_base(r, branch_sha_list) for sha in merge_bases: output_lines.append(sha.decode("ascii")) return output_lines # Get current branch for marking current_branch: bytes | None = None try: head_refs, _ = r.refs.follow(HEADREF) if head_refs: current_branch = head_refs[0] except (KeyError, TypeError): pass # Collect commit information for each branch branch_commits: list[tuple[bytes, str]] = [] # (sha, message) for ref_name, sha in sorted_branches: try: commit = r[sha] if isinstance(commit, Commit): message = commit.message.decode("utf-8", errors="replace").split( "\n" )[0] else: message = "" branch_commits.append((sha, message)) except KeyError: branch_commits.append((sha, "")) # Handle --list flag (show only branch headers) if list_branches or (more is not None and more < 0): # Just show the branch headers for i, (ref_name, sha) in enumerate(sorted_branches): is_current = ref_name == current_branch marker = "*" if is_current else "!" # Create spacing for alignment prefix = " " * i + marker + " " * (len(sorted_branches) - i - 1) ref_str = os.fsdecode(shorten_ref_name(ref_name)) _, message = branch_commits[i] output_lines.append(f"{prefix}[{ref_str}] {message}") return output_lines # Build commit history for visualization # Collect all commits reachable from any branch all_commits: dict[ bytes, tuple[int, list[ObjectID], str] ] = {} # sha -> (timestamp, parents, message) def collect_commits(sha: bytes, branch_idx: int, visited: set[bytes]) -> None: """Recursively collect commits.""" if sha in visited: return visited.add(sha) try: commit = r[sha] except KeyError: # Commit not found, stop traversal pass else: if not isinstance(commit, Commit): return timestamp = commit.commit_time parents = commit.parents if isinstance(commit, Commit) else [] message = ( commit.message.decode("utf-8", errors="replace").split("\n")[0] if isinstance(commit, Commit) else "" ) if sha not in all_commits: all_commits[sha] = (timestamp, parents, message) # Recurse to parents for parent in parents: collect_commits(parent, branch_idx, visited) # Collect commits from all branches for i, (_, sha) in enumerate(sorted_branches): collect_commits(sha, i, set()) # Find common ancestor common_ancestor_sha = None if len(branch_sha_list) >= 2: try: merge_bases = find_octopus_base(r, branch_sha_list) if merge_bases: common_ancestor_sha = merge_bases[0] except (KeyError, IndexError): pass # Sort commits (chronological by default, or topological if requested) if topo_order: # Topological sort is more complex, for now use chronological # TODO: Implement proper topological ordering sorted_commits = sorted(all_commits.items(), key=lambda x: -x[1][0]) else: # Reverse chronological order (newest first) sorted_commits = sorted(all_commits.items(), key=lambda x: -x[1][0]) # Determine how many commits to show if more is not None: # Find index of common ancestor if common_ancestor_sha and common_ancestor_sha in all_commits: ancestor_idx = next( ( i for i, (sha, _) in enumerate(sorted_commits) if sha == common_ancestor_sha ), None, ) if ancestor_idx is not None: # Show commits up to ancestor + more sorted_commits = sorted_commits[: ancestor_idx + 1 + more] # Determine which branches contain which commits branch_contains: list[set[bytes]] = [] for ref_name, sha in sorted_branches: reachable = set() def mark_reachable(commit_sha: bytes) -> None: if commit_sha in reachable: return reachable.add(commit_sha) if commit_sha in all_commits: _, parents, _ = all_commits[commit_sha] for parent in parents: mark_reachable(parent) mark_reachable(sha) branch_contains.append(reachable) # Output branch headers for i, (ref_name, sha) in enumerate(sorted_branches): is_current = ref_name == current_branch marker = "*" if is_current else "!" # Create spacing for alignment prefix = " " * i + marker + " " * (len(sorted_branches) - i - 1) ref_str = os.fsdecode(shorten_ref_name(ref_name)) _, message = branch_commits[i] output_lines.append(f"{prefix}[{ref_str}] {message}") # Output separator output_lines.append("-" * (len(sorted_branches) + 2)) # Output commits for commit_sha, (_, _, message) in sorted_commits: # Build marker string markers = [] for i, (ref_name, branch_sha) in enumerate(sorted_branches): if commit_sha == branch_sha: # This is the tip of the branch markers.append("*") elif commit_sha in branch_contains[i]: # This commit is in the branch markers.append("+") else: # This commit is not in the branch markers.append(" ") marker_str = "".join(markers) output_lines.append(f"{marker_str} [{message}]") # Limit output to 26 branches (git show-branch limitation) if len(sorted_branches) > 26: break return output_lines def ls_remote( remote: str | bytes, config: Config | None = None, operation: str | None = None, thin_packs: bool = True, report_activity: Callable[[int, str], None] | None = None, quiet: bool = False, include_tags: bool = False, username: str | None = None, password: str | None = None, key_filename: str | None = None, ssh_command: str | None = None, env: Mapping[str, str] | None = None, ) -> LsRemoteResult: """List the refs in a remote. Args: remote: Remote repository location config: Configuration to use operation: Operation type thin_packs: Whether to use thin packs report_activity: Function to report activity quiet: Whether to suppress output include_tags: Whether to include tags username: Username for authentication password: Password for authentication key_filename: SSH key filename ssh_command: SSH command to use. Defaults to the command configured in ``env``. env: Environment to read Git variables from (defaults to os.environ) Returns: LsRemoteResult object with refs and symrefs """ if config is None: config = StackedConfig.default() env_override = env_config(os.environ if env is None else env) if env_override is not None: config.backends.insert(0, env_override) if ssh_command is None: ssh_command = _ssh_command_from_env(env) remote_str = remote.decode() if isinstance(remote, bytes) else remote client, host_path = get_transport_and_path( remote_str, config=config, operation=operation, thin_packs=thin_packs, report_activity=report_activity, quiet=quiet, include_tags=include_tags, username=username, password=password, key_filename=key_filename, ssh_command=ssh_command, ) return client.get_refs( host_path.encode() if isinstance(host_path, str) else host_path ) def repack(repo: RepoPath, write_bitmaps: bool = False) -> None: """Repack loose files in a repository. Currently this only packs loose objects. Args: repo: Path to the repository write_bitmaps: Whether to write bitmap indexes for packs """ with open_repo_closing(repo) as r: r.object_store.pack_loose_objects() if write_bitmaps: # Update pack cache to pick up newly created packs r.object_store._update_pack_cache() r.object_store.generate_pack_bitmaps(r.refs.as_dict()) def pack_objects( repo: RepoPath, object_ids: Sequence[ObjectID], packf: BinaryIO, idxf: BinaryIO | None, delta_window_size: int | None = None, deltify: bool | None = None, reuse_deltas: bool = True, pack_index_version: int | None = None, ) -> None: """Pack objects into a file. Args: repo: Path to the repository object_ids: List of object ids to write packf: File-like object to write to idxf: File-like object to write to (can be None) delta_window_size: Sliding window size for searching for deltas; Set to None for default window size. deltify: Whether to deltify objects reuse_deltas: Allow reuse of existing deltas while deltifying pack_index_version: Pack index version to use (1, 2, or 3). If None, uses default version. """ with open_repo_closing(repo) as r: entries, data_sum = write_pack_from_container( packf.write, r.object_store, [(oid, None) for oid in object_ids], deltify=deltify, delta_window_size=delta_window_size, reuse_deltas=reuse_deltas, object_format=r.object_format, ) if idxf is not None: index_entries = sorted([(k, v[0], v[1]) for (k, v) in entries.items()]) write_pack_index(idxf, index_entries, data_sum, version=pack_index_version) def ls_tree( repo: RepoPath, treeish: str | bytes | Commit | Tree | Tag = b"HEAD", outstream: TextIO | BinaryIO = sys.stdout, recursive: bool = False, name_only: bool = False, ) -> None: """List contents of a tree. Args: repo: Path to the repository treeish: Tree id to list outstream: Output stream (defaults to stdout) recursive: Whether to recursively list files name_only: Only print item name """ def list_tree(store: BaseObjectStore, treeid: ObjectID, base: bytes) -> None: tree = store[treeid] assert isinstance(tree, Tree) for name, mode, sha in tree.iteritems(): assert name is not None assert mode is not None assert sha is not None if base: name = posixpath.join(base, name) if name_only: if isinstance(outstream, BinaryIO): outstream.write(name + b"\n") else: outstream.write(name.decode("utf-8", "replace") + "\n") else: formatted = pretty_format_tree_entry(name, mode, sha) if isinstance(outstream, BinaryIO): outstream.write(formatted.encode("utf-8")) else: outstream.write(formatted) if stat.S_ISDIR(mode) and recursive: list_tree(store, sha, name) with open_repo_closing(repo) as r: tree = parse_tree(r, treeish) list_tree(r.object_store, tree.id, b"") def remote_add( repo: RepoPath, name: bytes | str, url: bytes | str, ) -> None: """Add a remote. Args: repo: Path to the repository name: Remote name url: Remote URL """ if not isinstance(name, bytes): name = name.encode(DEFAULT_ENCODING) if not isinstance(url, bytes): url = url.encode(DEFAULT_ENCODING) with open_repo_closing(repo) as r: c = r.get_config() section = (b"remote", name) if c.has_section(section): raise RemoteExists(f"Remote {name.decode()} already exists") c.set(section, b"url", url) c.write_to_path() def remote_remove(repo: Repo, name: bytes | str) -> None: """Remove a remote. Args: repo: Path to the repository name: Remote name """ if not isinstance(name, bytes): name = name.encode(DEFAULT_ENCODING) with open_repo_closing(repo) as r: c = r.get_config() section = (b"remote", name) del c[section] c.write_to_path() def _quote_path(path: str) -> str: """Quote a path using C-style quoting similar to git's core.quotePath. Args: path: Path to quote Returns: Quoted path string """ # Check if path needs quoting (non-ASCII or special characters) needs_quoting = False for char in path: if ord(char) > 127 or char in '"\\': needs_quoting = True break if not needs_quoting: return path # Apply C-style quoting quoted = '"' for char in path: if ord(char) > 127: # Non-ASCII character, encode as octal escape utf8_bytes = char.encode("utf-8") for byte in utf8_bytes: quoted += f"\\{byte:03o}" elif char == '"': quoted += '\\"' elif char == "\\": quoted += "\\\\" else: quoted += char quoted += '"' return quoted def check_ignore( repo: RepoPath, paths: Sequence[str | bytes | os.PathLike[str]], no_index: bool = False, quote_path: bool = True, ) -> Iterator[str]: r"""Debug gitignore files. Args: repo: Path to the repository paths: List of paths to check for no_index: Don't check index quote_path: If True, quote non-ASCII characters in returned paths using C-style octal escapes (e.g. "тест.txt" becomes "\\321\\202\\320\\265\\321\\201\\321\\202.txt"). If False, return raw unicode paths. Returns: List of ignored files """ with open_repo_closing(repo) as r: index = r.open_index(config=r.get_config_stack()) ignore_manager = IgnoreFilterManager.from_repo(r, config=r.get_config_stack()) for original_path in paths: # Convert path to string for consistent handling original_path_fspath = os.fspath(original_path) # Normalize to str original_path_str = os.fsdecode(original_path_fspath) if not no_index and path_to_tree_path(r.path, original_path_str) in index: continue # Preserve whether the original path had a trailing slash had_trailing_slash = original_path_str.endswith(("/", os.path.sep)) if os.path.isabs(original_path_str): path = os.path.relpath(original_path_str, r.path) # Normalize Windows paths to use forward slashes if os.path.sep != "/": path = path.replace(os.path.sep, "/") else: path = original_path_str # Restore trailing slash if it was in the original if had_trailing_slash and not path.endswith("/"): path = path + "/" # For directories, check with trailing slash to get correct ignore behavior test_path = path path_without_slash = path.rstrip("/") is_directory = os.path.isdir(os.path.join(r.path, path_without_slash)) # If this is a directory path, ensure we test it correctly if is_directory and not path.endswith("/"): test_path = path + "/" if ignore_manager.is_ignored(test_path): # Return relative path (like git does) when absolute path was provided if os.path.isabs(original_path_str): output_path = path else: output_path = original_path_str yield _quote_path(output_path) if quote_path else output_path def _get_current_head_tree(repo: Repo) -> ObjectID | None: """Get the current HEAD tree ID. Args: repo: Repository object Returns: Tree ID of current HEAD, or None if no HEAD exists (empty repo) """ try: current_head = repo.refs[HEADREF] current_commit = repo[current_head] assert isinstance(current_commit, Commit), "Expected a Commit object" tree_id: ObjectID = current_commit.tree return tree_id except KeyError: # No HEAD yet (empty repo) return None def _check_uncommitted_changes( repo: Repo, target_tree_id: ObjectID, force: bool = False ) -> None: """Check for uncommitted changes that would conflict with a checkout/switch. Args: repo: Repository object target_tree_id: Tree ID to check conflicts against force: If True, skip the check Raises: CheckoutError: If there are conflicting local changes """ if force: return # Get current HEAD tree for comparison current_tree_id = _get_current_head_tree(repo) if current_tree_id is None: # No HEAD yet (empty repo) return status_report = status(repo) changes = [] # staged is a dict with 'add', 'delete', 'modify' keys if isinstance(status_report.staged, dict): changes.extend(status_report.staged.get("add", [])) changes.extend(status_report.staged.get("delete", [])) changes.extend(status_report.staged.get("modify", [])) # unstaged is a list changes.extend(status_report.unstaged) if changes: # Check if any changes would conflict with checkout target_tree_obj = repo[target_tree_id] assert isinstance(target_tree_obj, Tree), "Expected a Tree object" target_tree = target_tree_obj for change in changes: if isinstance(change, str): change = change.encode(DEFAULT_ENCODING) try: target_tree.lookup_path(repo.object_store.__getitem__, change) except KeyError: # File doesn't exist in target tree - change can be preserved pass else: # File exists in target tree - would overwrite local changes raise CheckoutError( f"Your local changes to '{change.decode()}' would be " "overwritten. Please commit or stash before switching." ) def _get_worktree_update_config( repo: Repo, ) -> tuple[ bool, Callable[[bytes], bool], Callable[[str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None], ]: """Get configuration for working tree updates. Args: repo: Repository object Returns: Tuple of (honor_filemode, validate_path_element, symlink_fn) """ config = repo.get_config() honor_filemode = config.get_boolean(b"core", b"filemode", os.name != "nt") # core.protectNTFS defaults to True on all platforms (matching Git's # PROTECT_NTFS_DEFAULT=1) and protectHFS defaults to True on macOS; both # apply together, so defer to the shared selector rather than picking one. validate_path_element = get_path_element_validator(config) if config.get_boolean(b"core", b"symlinks", True): def symlink_wrapper( source: str | bytes | os.PathLike[str], target: str | bytes | os.PathLike[str], ) -> None: symlink(source, target) # type: ignore[arg-type,unused-ignore] symlink_fn = symlink_wrapper else: def symlink_fallback( source: str | bytes | os.PathLike[str], target: str | bytes | os.PathLike[str], ) -> None: mode = "w" + ("b" if isinstance(source, bytes) else "") with open(target, mode) as f: f.write(source) symlink_fn = symlink_fallback return honor_filemode, validate_path_element, symlink_fn def _perform_tree_switch( repo: Repo, current_tree_id: ObjectID | None, target_tree_id: ObjectID, force: bool = False, ) -> None: """Perform the actual working tree switch. Args: repo: Repository object current_tree_id: Current tree ID (or None for empty repo) target_tree_id: Target tree ID to switch to force: If True, allow overwriting modified files """ honor_filemode, validate_path_element, symlink_fn = _get_worktree_update_config( repo ) config = repo.get_config_stack() # Get blob normalizer for line ending conversion blob_normalizer = repo.get_blob_normalizer(config=config) # Update working tree tree_change_iterator: Iterator[TreeChange] = tree_changes( repo.object_store, current_tree_id, target_tree_id ) update_working_tree( repo, current_tree_id, target_tree_id, change_iterator=tree_change_iterator, honor_filemode=honor_filemode, validate_path_element=validate_path_element, symlink_fn=symlink_fn, blob_normalizer=blob_normalizer, allow_overwrite_modified=force, config=config, ) def update_head( repo: RepoPath, target: str | bytes, detached: bool = False, new_branch: str | bytes | None = None, ) -> None: """Update HEAD to point at a new branch/commit. Note that this does not actually update the working tree. Args: repo: Path to the repository detached: Create a detached head target: Branch or committish to switch to new_branch: New branch to create """ with open_repo_closing(repo) as r: if new_branch is not None: to_set = _make_branch_ref(new_branch) else: to_set = HEADREF if detached: # TODO(jelmer): Provide some way so that the actual ref gets # updated rather than what it points to, so the delete isn't # necessary. del r.refs[to_set] r.refs[to_set] = parse_commit(r, target).id else: r.refs.set_symbolic_ref(to_set, parse_ref(r, target)) if new_branch is not None: r.refs.set_symbolic_ref(HEADREF, to_set) def checkout( repo: str | os.PathLike[str] | Repo, target: str | bytes | Commit | Tag | None = None, force: bool = False, new_branch: bytes | str | None = None, paths: list[bytes | str] | None = None, ) -> None: """Switch to a branch or commit, updating both HEAD and the working tree. This is similar to 'git checkout', allowing you to switch to a branch, tag, or specific commit. Unlike update_head, this function also updates the working tree to match the target. Args: repo: Path to repository or repository object target: Branch name, tag, or commit SHA to checkout. If None and paths is specified, restores files from HEAD force: Force checkout even if there are local changes new_branch: Create a new branch at target (like git checkout -b) paths: List of specific paths to checkout. If specified, only these paths are updated and HEAD is not changed Raises: CheckoutError: If checkout cannot be performed due to conflicts KeyError: If the target reference cannot be found """ with open_repo_closing(repo) as r: # Store the original target for later reference checks original_target = target worktree = r.get_worktree() # Handle path-specific checkout (like git checkout -- ) if paths is not None: # Convert paths to bytes byte_paths = [] for path in paths: if isinstance(path, str): byte_paths.append(path.encode(DEFAULT_ENCODING)) else: byte_paths.append(path) # If no target specified, use HEAD if target is None: try: target = r.refs[HEADREF] except KeyError: raise CheckoutError("No HEAD reference found") else: if isinstance(target, str): target = target.encode(DEFAULT_ENCODING) # Get the target commit and tree target_tree = parse_tree(r, target) # Get blob normalizer for line ending conversion blob_normalizer = r.get_blob_normalizer(config=r.get_config_stack()) # Restore specified paths from target tree for path in byte_paths: try: # Look up the path in the target tree mode, sha = target_tree.lookup_path( r.object_store.__getitem__, path ) obj = r[sha] assert isinstance(obj, Blob), "Expected a Blob object" except KeyError: # Path doesn't exist in target tree pass else: # Defense in depth: refuse a path that would escape the # work tree before creating directories or writing. file_path = _checked_worktree_path(r, path) # Create directories if needed os.makedirs(os.path.dirname(file_path), exist_ok=True) # Write the file content. Route through build_file_from_blob # rather than a raw os.open: it replaces a symlink left at the # target instead of following it, so a malicious repository # cannot use a symlink to write outside the work tree. if stat.S_ISREG(mode): # Apply checkout filters (smudge) if blob_normalizer: obj = blob_normalizer.checkout_normalize(obj, path) build_file_from_blob(obj, mode, file_path) elif stat.S_ISLNK(mode): build_file_from_blob(obj, mode, file_path) # Update the index worktree.stage(path, config=r.get_config_stack()) return # Normal checkout (switching branches/commits) if target is None: raise ValueError("Target must be specified for branch/commit checkout") if isinstance(target, str): target_bytes = target.encode(DEFAULT_ENCODING) elif isinstance(target, bytes): target_bytes = target else: # For Commit/Tag objects, we'll use their SHA target_bytes = target.id if isinstance(new_branch, str): new_branch = new_branch.encode(DEFAULT_ENCODING) # Parse the target to get the commit assert ( original_target is not None ) # Guaranteed by earlier check for normal checkout target_commit = parse_commit(r, original_target) target_tree_id = target_commit.tree # Get current HEAD tree for comparison current_tree_id = _get_current_head_tree(r) # Check for uncommitted changes if not forcing if current_tree_id is not None: _check_uncommitted_changes(r, target_tree_id, force) # Update working tree _perform_tree_switch(r, current_tree_id, target_tree_id, force) # Update HEAD if new_branch: # Create new branch and switch to it branch_create(r, new_branch, objectish=target_commit.id.decode("ascii")) update_head(r, new_branch) # Set up tracking if creating from a remote branch if isinstance(original_target, bytes) and target_bytes.startswith( LOCAL_REMOTE_PREFIX ): try: remote_name, branch_name = parse_remote_ref(target_bytes) # Set tracking to refs/heads/ on the remote set_branch_tracking( r, new_branch, remote_name, local_branch_name(branch_name) ) except ValueError: # Invalid remote ref format, skip tracking setup pass else: # Check if target is a branch name (with or without refs/heads/ prefix) branch_ref = None if ( isinstance(original_target, str | bytes) and target_bytes in r.refs.keys() ): if target_bytes.startswith(LOCAL_BRANCH_PREFIX): branch_ref = target_bytes else: # Try adding refs/heads/ prefix potential_branch = ( _make_branch_ref(target_bytes) if isinstance(original_target, str | bytes) else None ) if potential_branch in r.refs.keys(): branch_ref = potential_branch if branch_ref: # It's a branch - update HEAD symbolically update_head(r, branch_ref) else: # It's a tag, other ref, or commit SHA - detached HEAD update_head(r, target_commit.id.decode("ascii"), detached=True) def restore( repo: str | os.PathLike[str] | Repo, paths: list[bytes | str], source: str | bytes | Commit | Tag | None = None, staged: bool = False, worktree: bool = True, ) -> None: """Restore working tree files. This is similar to 'git restore', allowing you to restore specific files from a commit or the index without changing HEAD. Args: repo: Path to repository or repository object paths: List of specific paths to restore source: Branch name, tag, or commit SHA to restore from. If None, restores staged files from HEAD, or worktree files from index staged: Restore files in the index (--staged) worktree: Restore files in the working tree (default: True) Raises: CheckoutError: If restore cannot be performed ValueError: If neither staged nor worktree is specified KeyError: If the source reference cannot be found """ if not staged and not worktree: raise ValueError("At least one of staged or worktree must be True") with open_repo_closing(repo) as r: from ..index import _fs_to_tree_path, build_file_from_blob # Determine the source tree if source is None: if staged: # Restoring staged files from HEAD try: source = r.refs[HEADREF] except KeyError: raise CheckoutError("No HEAD reference found") elif worktree: # Restoring worktree files from index from ..index import ConflictedIndexEntry, IndexEntry index = r.open_index(config=r.get_config_stack()) for path in paths: if isinstance(path, str): tree_path = _fs_to_tree_path(path) else: tree_path = path try: index_entry = index[tree_path] if isinstance(index_entry, ConflictedIndexEntry): raise CheckoutError( f"Path '{path if isinstance(path, str) else path.decode(DEFAULT_ENCODING)}' has conflicts" ) blob = r[index_entry.sha] assert isinstance(blob, Blob), "Expected a Blob object" # Defense in depth: refuse a path that would escape # the work tree. full_path = _checked_worktree_path(r, tree_path) mode = index_entry.mode # Use build_file_from_blob to write the file build_file_from_blob(blob, mode, full_path) except KeyError: # Path doesn't exist in index raise CheckoutError( f"Path '{path if isinstance(path, str) else path.decode(DEFAULT_ENCODING)}' not in index" ) return # source is not None at this point assert source is not None # Get the source tree source_tree = parse_tree(r, treeish=source) # Restore specified paths from source tree for path in paths: if isinstance(path, str): tree_path = _fs_to_tree_path(path) else: tree_path = path try: # Look up the path in the source tree mode, sha = source_tree.lookup_path( r.object_store.__getitem__, tree_path ) blob = r[sha] assert isinstance(blob, Blob), "Expected a Blob object" except KeyError: # Path doesn't exist in source tree raise CheckoutError( f"Path '{path if isinstance(path, str) else path.decode(DEFAULT_ENCODING)}' not found in source" ) # Defense in depth: refuse a path that would escape the work tree. full_path = _checked_worktree_path(r, tree_path) if worktree: # Use build_file_from_blob to restore to working tree build_file_from_blob(blob, mode, full_path) if staged: # Update the index with the blob from source from ..index import IndexEntry index = r.open_index(config=r.get_config_stack()) # When only updating staged (not worktree), we want to reset the index # to the source, but invalidate the stat cache so Git knows to check # the worktree file. Use zeros for stat fields. if not worktree: # Invalidate stat cache by using zeros new_entry = IndexEntry( ctime=(0, 0), mtime=(0, 0), dev=0, ino=0, mode=mode, uid=0, gid=0, size=0, sha=sha, ) else: # If we also updated worktree, use actual stat from ..index import index_entry_from_stat st = os.lstat(full_path) new_entry = index_entry_from_stat(st, sha, mode) index[tree_path] = new_entry index.write() def switch( repo: str | os.PathLike[str] | Repo, target: str | bytes | Commit | Tag, create: str | bytes | None = None, force: bool = False, detach: bool = False, ) -> None: """Switch branches. This is similar to 'git switch', allowing you to switch to a different branch or commit, updating both HEAD and the working tree. Args: repo: Path to repository or repository object target: Branch name, tag, or commit SHA to switch to create: Create a new branch at target before switching (like git switch -c) force: Force switch even if there are local changes detach: Switch to a commit in detached HEAD state (like git switch --detach) Raises: CheckoutError: If switch cannot be performed due to conflicts KeyError: If the target reference cannot be found ValueError: If both create and detach are specified """ if create and detach: raise ValueError("Cannot use both create and detach options") with open_repo_closing(repo) as r: # Store the original target for later reference checks original_target = target if isinstance(target, str): target_bytes = target.encode(DEFAULT_ENCODING) elif isinstance(target, bytes): target_bytes = target else: # For Commit/Tag objects, we'll use their SHA target_bytes = target.id if isinstance(create, str): create = create.encode(DEFAULT_ENCODING) # Parse the target to get the commit target_commit = parse_commit(r, original_target) target_tree_id = target_commit.tree # Get current HEAD tree for comparison current_tree_id = _get_current_head_tree(r) # Check for uncommitted changes if not forcing if current_tree_id is not None: _check_uncommitted_changes(r, target_tree_id, force) # Update working tree _perform_tree_switch(r, current_tree_id, target_tree_id, force) # Update HEAD if create: # Create new branch and switch to it branch_create(r, create, objectish=target_commit.id.decode("ascii")) update_head(r, create) # Set up tracking if creating from a remote branch if isinstance(original_target, bytes) and target_bytes.startswith( LOCAL_REMOTE_PREFIX ): try: remote_name, branch_name = parse_remote_ref(target_bytes) # Set tracking to refs/heads/ on the remote set_branch_tracking( r, create, remote_name, local_branch_name(branch_name) ) except ValueError: # Invalid remote ref format, skip tracking setup pass elif detach: # Detached HEAD mode update_head(r, target_commit.id.decode("ascii"), detached=True) else: # Check if target is a branch name (with or without refs/heads/ prefix) branch_ref = None if ( isinstance(original_target, str | bytes) and target_bytes in r.refs.keys() ): if target_bytes.startswith(LOCAL_BRANCH_PREFIX): branch_ref = target_bytes else: # Try adding refs/heads/ prefix potential_branch = ( _make_branch_ref(target_bytes) if isinstance(original_target, str | bytes) else None ) if potential_branch in r.refs.keys(): branch_ref = potential_branch if branch_ref: # It's a branch - update HEAD symbolically update_head(r, branch_ref) else: # It's a tag, other ref, or commit SHA # In git switch, this would be an error unless --detach is used raise CheckoutError( f"'{target_bytes.decode(DEFAULT_ENCODING)}' is not a branch. " "Use detach=True to switch to a commit in detached HEAD state." ) def reset_file( repo: Repo, file_path: str, target: str | bytes | Commit | Tree | Tag = b"HEAD", symlink_fn: Callable[ [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None ] | None = None, ) -> None: """Reset the file to specific commit or branch. Args: repo: dulwich Repo object file_path: file to reset, relative to the repository path target: branch or commit or b'HEAD' to reset symlink_fn: Function to use for creating symlinks """ tree = parse_tree(repo, treeish=target) tree_path = _fs_to_tree_path(file_path) file_entry = tree.lookup_path(repo.object_store.__getitem__, tree_path) # Defense in depth: refuse a path that would escape the work tree. full_path = _checked_worktree_path(repo, tree_path) blob = repo.object_store[file_entry[1]] assert isinstance(blob, Blob) mode = file_entry[0] build_file_from_blob(blob, mode, full_path, symlink_fn=symlink_fn) def sparse_checkout( repo: str | os.PathLike[str] | Repo, patterns: list[str] | None = None, force: bool = False, cone: bool | None = None, ) -> None: """Perform a sparse checkout in the repository (either 'full' or 'cone mode'). Perform sparse checkout in either 'cone' (directory-based) mode or 'full pattern' (.gitignore) mode, depending on the ``cone`` parameter. If ``cone`` is ``None``, the mode is inferred from the repository's ``core.sparseCheckoutCone`` config setting. Steps: 1) If ``patterns`` is provided, write them to ``.git/info/sparse-checkout``. 2) Determine which paths in the index are included vs. excluded. - If ``cone=True``, use "cone-compatible" directory-based logic. - If ``cone=False``, use standard .gitignore-style matching. 3) Update the index's skip-worktree bits and add/remove files in the working tree accordingly. 4) If ``force=False``, refuse to remove files that have local modifications. Args: repo: Path to the repository or a Repo object. patterns: Optional list of sparse-checkout patterns to write. force: Whether to force removal of locally modified files (default False). cone: Boolean indicating cone mode (True/False). If None, read from config. Returns: None """ with open_repo_closing(repo) as repo_obj: # --- 0) Possibly infer 'cone' from config --- if cone is None: cone = repo_obj.get_worktree().infer_cone_mode() # --- 1) Read or write patterns --- if patterns is None: lines = repo_obj.get_worktree().get_sparse_checkout_patterns() if lines is None: raise Error("No sparse checkout patterns found.") else: lines = patterns repo_obj.get_worktree().set_sparse_checkout_patterns(patterns) # --- 2) Determine the set of included paths --- index = repo_obj.open_index(config=repo_obj.get_config_stack()) included_paths = determine_included_paths(index, lines, cone) # --- 3) Apply those results to the index & working tree --- try: apply_included_paths( repo_obj, included_paths, force=force, config=repo_obj.get_config_stack(), ) except SparseCheckoutConflictError as exc: raise CheckoutError(*exc.args) from exc def cone_mode_init(repo: str | os.PathLike[str] | Repo) -> None: """Initialize a repository to use sparse checkout in 'cone' mode. Sets ``core.sparseCheckout`` and ``core.sparseCheckoutCone`` in the config. Writes an initial ``.git/info/sparse-checkout`` file that includes only top-level files (and excludes all subdirectories), e.g. ``["/*", "!/*/"]``. Then performs a sparse checkout to update the working tree accordingly. If no directories are specified, then only top-level files are included: https://git-scm.com/docs/git-sparse-checkout#_internalscone_mode_handling Args: repo: Path to the repository or a Repo object. Returns: None """ with open_repo_closing(repo) as repo_obj: repo_obj.get_worktree().configure_for_cone_mode() patterns = ["/*", "!/*/"] # root-level files only sparse_checkout(repo_obj, patterns, force=True, cone=True) def cone_mode_set( repo: str | os.PathLike[str] | Repo, dirs: Sequence[str], force: bool = False ) -> None: """Overwrite the existing 'cone-mode' sparse patterns with a new set of directories. Ensures ``core.sparseCheckout`` and ``core.sparseCheckoutCone`` are enabled. Writes new patterns so that only the specified directories (and top-level files) remain in the working tree, and applies the sparse checkout update. Args: repo: Path to the repository or a Repo object. dirs: List of directory names to include. force: Whether to forcibly discard local modifications (default False). Returns: None """ with open_repo_closing(repo) as repo_obj: repo_obj.get_worktree().configure_for_cone_mode() repo_obj.get_worktree().set_cone_mode_patterns(dirs=dirs) new_patterns = repo_obj.get_worktree().get_sparse_checkout_patterns() # Finally, apply the patterns and update the working tree sparse_checkout(repo_obj, new_patterns, force=force, cone=True) def cone_mode_add( repo: str | os.PathLike[str] | Repo, dirs: Sequence[str], force: bool = False ) -> None: """Add new directories to the existing 'cone-mode' sparse-checkout patterns. Reads the current patterns from ``.git/info/sparse-checkout``, adds pattern lines to include the specified directories, and then performs a sparse checkout to update the working tree accordingly. Args: repo: Path to the repository or a Repo object. dirs: List of directory names to add to the sparse-checkout. force: Whether to forcibly discard local modifications (default False). Returns: None """ with open_repo_closing(repo) as repo_obj: repo_obj.get_worktree().configure_for_cone_mode() # Do not pass base patterns as dirs base_patterns = ["/*", "!/*/"] existing_dirs = [ pat.strip("/") for pat in repo_obj.get_worktree().get_sparse_checkout_patterns() if pat not in base_patterns ] added_dirs = existing_dirs + list(dirs or []) repo_obj.get_worktree().set_cone_mode_patterns(dirs=added_dirs) new_patterns = repo_obj.get_worktree().get_sparse_checkout_patterns() sparse_checkout(repo_obj, patterns=new_patterns, force=force, cone=True) def cone_mode_list(repo: str | os.PathLike[str] | Repo) -> list[str]: """List current sparse-checkout patterns. Args: repo: Path to the repository or a Repo object. Returns: List of sparse-checkout patterns """ with open_repo_closing(repo) as repo_obj: return repo_obj.get_worktree().get_sparse_checkout_patterns() def cone_mode_disable(repo: str | os.PathLike[str] | Repo, force: bool = False) -> None: """Disable sparse checkout and restore all files. This function: 1. Unsets core.sparseCheckout and core.sparseCheckoutCone config 2. Removes the .git/info/sparse-checkout file 3. Restores all files to the working tree Args: repo: Path to the repository or a Repo object. force: Whether to forcibly discard local modifications (default False). Returns: None """ with open_repo_closing(repo) as repo_obj: # Unset sparse checkout config config = repo_obj.get_config() try: del config[(b"core", b"sparseCheckout")] except KeyError: pass try: del config[(b"core", b"sparseCheckoutCone")] except KeyError: pass config.write_to_path() # Remove sparse-checkout file sparse_file = repo_obj.get_worktree()._sparse_checkout_file_path() try: os.remove(sparse_file) except FileNotFoundError: pass # Restore all files by doing a full checkout # Clear all skip-worktree bits and restore all files from HEAD from ..objects import Commit commit = repo_obj[repo_obj.head()] assert isinstance(commit, Commit) from ..index import build_index_from_tree, get_path_element_validator build_index_from_tree( repo_obj.path, repo_obj.index_path(), repo_obj.object_store, commit.tree, validate_path_element=get_path_element_validator(config), ) def cat_file_type(repo: str | os.PathLike[str] | Repo, objectish: str | bytes) -> bytes: """Get the type of a Git object. Args: repo: Path to the repository or a Repo object objectish: Object SHA, reference, or objectish (e.g., 'HEAD', 'main', 'abc123') Returns: Object type as bytes (b'blob', b'tree', b'commit', or b'tag') """ from ..objectspec import parse_object with open_repo_closing(repo) as r: obj = parse_object(r, objectish, config=r.get_config_stack()) return obj.type_name def cat_file_size(repo: str | os.PathLike[str] | Repo, objectish: str | bytes) -> int: """Get the size of a Git object. Args: repo: Path to the repository or a Repo object objectish: Object SHA, reference, or objectish (e.g., 'HEAD', 'main', 'abc123') Returns: Object size in bytes """ from ..objectspec import parse_object with open_repo_closing(repo) as r: obj = parse_object(r, objectish, config=r.get_config_stack()) return len(obj.as_raw_string()) def cat_file_content( repo: str | os.PathLike[str] | Repo, objectish: str | bytes ) -> bytes: """Get the raw content of a Git object. Args: repo: Path to the repository or a Repo object objectish: Object SHA, reference, or objectish (e.g., 'HEAD', 'main', 'abc123') Returns: Raw object content as bytes """ from ..objectspec import parse_object with open_repo_closing(repo) as r: obj = parse_object(r, objectish, config=r.get_config_stack()) return obj.as_raw_string() def hash_object( repo: str | os.PathLike[str] | Repo | None, path: str | os.PathLike[str] | None = None, data: bytes | None = None, object_type: bytes = b"blob", write: bool = False, ) -> bytes: """Compute object ID and optionally write to object store. Args: repo: Path to the repository or a Repo object (required if write=True) path: Path to file to hash (mutually exclusive with data) data: Data to hash (mutually exclusive with path) object_type: Type of object (default: b'blob') write: Whether to write the object to the object store (default: False) Returns: Object SHA as bytes Raises: ValueError: If neither path nor data is provided, or both are provided ValueError: If write=True but repo is None """ from ..objects import Blob if (path is None and data is None) or (path is not None and data is not None): raise ValueError("Exactly one of 'path' or 'data' must be provided") if write and repo is None: raise ValueError("repo is required when write=True") # Only blob type is supported for now if object_type != b"blob": raise NotImplementedError( f"Object type {object_type.decode('utf-8')} not yet supported" ) # Create blob object blob = Blob() if path is not None: with open(path, "rb") as f: blob.data = f.read() else: assert data is not None # Already checked above blob.data = data # Write to object store if requested if write: assert repo is not None # Already checked above with open_repo_closing(repo) as r: r.object_store.add_object(blob) return blob.id def rev_parse(repo: str | os.PathLike[str] | Repo, rev: str | bytes) -> bytes: """Parse a revision string and return the object SHA. Args: repo: Path to the repository or a Repo object rev: Revision string (e.g., 'HEAD', 'main', 'abc123') Returns: Object SHA as bytes """ from ..objectspec import parse_object with open_repo_closing(repo) as r: obj = parse_object(r, rev, config=r.get_config_stack()) return obj.id def update_ref( repo: str | os.PathLike[str] | Repo, ref: str | bytes, new_value: str | bytes | None, old_value: str | bytes | None = None, message: str | bytes | None = None, ) -> None: """Update the object name stored in a ref safely. Args: repo: Path to the repository or a Repo object ref: Name of the ref to update (e.g., 'refs/heads/main', 'HEAD') new_value: New object SHA to set the ref to (None to delete) old_value: Optional old value to verify before updating (for atomic updates) message: Optional message for the reflog Raises: ValueError: If the old value doesn't match """ from ..objectspec import parse_object from ..refs import Ref ref_name = Ref(ref.encode("utf-8") if isinstance(ref, str) else ref) with open_repo_closing(repo) as r: # Parse new and old values to object IDs new_sha: bytes | None if new_value is None: new_sha = None else: new_obj = parse_object(r, new_value, config=r.get_config_stack()) new_sha = new_obj.id old_sha: bytes | None if old_value is not None: old_obj = parse_object(r, old_value, config=r.get_config_stack()) old_sha = old_obj.id else: old_sha = None message_bytes = message.encode("utf-8") if isinstance(message, str) else message # Handle deletion if new_sha is None: if old_sha is not None: # Verify old value before deleting current = r.refs.read_ref(ref_name) if current != old_sha: raise ValueError( f"Ref {ref_name.decode('utf-8')} does not match expected value" ) r.refs.remove_if_equals(ref_name, old_sha, message=message_bytes) else: # Update or create ref if not r.refs.set_if_equals( ref_name, old_sha, new_sha, message=message_bytes ): raise ValueError( f"Ref {ref_name.decode('utf-8')} does not match expected value" ) def mktag(repo: str | os.PathLike[str] | Repo, tag_data: bytes) -> bytes: """Create a tag object from raw tag data. Args: repo: Path to the repository or a Repo object tag_data: Raw tag data in git tag format Returns: Object SHA of the created tag as bytes Raises: ObjectFormatException: If the tag data is invalid """ from ..objects import Tag # Parse the tag data tag = Tag() tag.set_raw_string(tag_data) # Write to object store with open_repo_closing(repo) as r: r.object_store.add_object(tag) return tag.id def show_index( index_path: str | os.PathLike[str], repo: str | os.PathLike[str] | Repo | None = None, ) -> "list[tuple[int, RawObjectID, int | None]]": """Show the contents of a pack index file. Args: index_path: Path to the pack index file repo: Path to the repository or a Repo object (for object format) Returns: List of tuples (offset, sha, crc32) for each entry Raises: FileNotFoundError: If the index file doesn't exist """ from ..pack import load_pack_index with open_repo_closing(repo) as r: with closing(load_pack_index(index_path, r.object_format)) as idx: return [(offset, sha, crc32) for sha, offset, crc32 in idx.iterentries()] def check_mailmap(repo: RepoPath, contact: str | bytes) -> bytes: """Check canonical name and email of contact. Args: repo: Path to the repository contact: Contact name and/or email Returns: Canonical contact data """ with open_repo_closing(repo) as r: from ..mailmap import Mailmap try: mailmap = Mailmap.from_path(os.path.join(r.path, ".mailmap")) except FileNotFoundError: mailmap = Mailmap() contact_bytes = ( contact.encode(DEFAULT_ENCODING) if isinstance(contact, str) else contact ) result = mailmap.lookup(contact_bytes) if isinstance(result, bytes): return result else: # Convert tuple back to bytes format name, email = result if name is None: name = b"" if email is None: email = b"" return name + b" <" + email + b">" def fsck(repo: RepoPath) -> Iterator[tuple[bytes, Exception]]: """Check a repository. This function is shallow-aware and will not report errors for missing parent commits that are beyond the shallow boundary. Args: repo: A path to the repository Returns: Iterator over errors/warnings """ with open_repo_closing(repo) as r: # Get shallow commits for future graph checking shallow = r.get_shallow() # noqa: F841 # TODO(jelmer): check pack files # TODO(jelmer): check graph (excluding commits beyond shallow boundary) # TODO(jelmer): check refs for sha in r.object_store: o = r.object_store[sha] try: o.check() except Exception as e: yield (sha, e) def stash_list( repo: str | os.PathLike[str] | Repo, ) -> Iterator[tuple[int, tuple[bytes, bytes]]]: """List all stashes in a repository.""" with open_repo_closing(repo) as r: from ..stash import Stash stash = Stash.from_repo(r) entries = stash.stashes() # Convert Entry objects to (old_sha, new_sha) tuples return enumerate([(entry.old_sha, entry.new_sha) for entry in entries]) def stash_push(repo: str | os.PathLike[str] | Repo) -> None: """Push a new stash onto the stack.""" with open_repo_closing(repo) as r: from ..stash import Stash stash = Stash.from_repo(r) stash.push(config=r.get_config_stack()) def stash_pop(repo: str | os.PathLike[str] | Repo) -> None: """Pop a stash from the stack.""" with open_repo_closing(repo) as r: from ..stash import Stash stash = Stash.from_repo(r) stash.pop(0, config=r.get_config_stack()) def stash_drop(repo: str | os.PathLike[str] | Repo, index: int) -> None: """Drop a stash from the stack.""" with open_repo_closing(repo) as r: from ..stash import Stash stash = Stash.from_repo(r) stash.drop(index) def ls_files(repo: RepoPath) -> list[bytes]: """List all files in an index.""" with open_repo_closing(repo) as r: return sorted(r.open_index(config=r.get_config_stack())) def find_unique_abbrev( object_store: BaseObjectStore, object_id: str | bytes, min_length: int = 7 ) -> str: """Find the shortest unique abbreviation for an object ID. Args: object_store: Object store to search in object_id: The full object ID to abbreviate min_length: Minimum length of abbreviation (default 7) Returns: The shortest unique prefix of the object ID (at least min_length chars) """ if isinstance(object_id, bytes): hex_id = object_id.decode("ascii") else: hex_id = object_id # Start with minimum length for length in range(min_length, len(hex_id) + 1): prefix = hex_id[:length] matches = 0 # Check if this prefix is unique for obj_id in object_store: if obj_id.decode("ascii").startswith(prefix): matches += 1 if matches > 1: # Not unique, need more characters break if matches == 1: # Found unique prefix return prefix # If we get here, return the full ID return hex_id def describe(repo: str | os.PathLike[str] | Repo, abbrev: int | None = None) -> str: """Describe the repository version. The commit hash is prefixed with a literal "g" (for "git"), matching git's own behaviour. The "g" is not counted towards abbrev. Args: repo: git repository abbrev: number of hex characters of the commit hash to take (not counting the "g" prefix), default is 7 Returns: a string description of the current git revision Examples: "gabcdefg", "v0.1" or "v0.1-5-gabcdefa". """ abbrev_slice = slice(0, abbrev if abbrev is not None else 7) # Get the repository with open_repo_closing(repo) as r: # Get a list of all tags refs = r.get_refs() tags = {} for key, value in refs.items(): key_str = key.decode() obj = r.get_object(value) if "tags" not in key_str: continue _, tag = key_str.rsplit("/", 1) if isinstance(obj, Tag): # Annotated tag case commit = r.get_object(obj.object[1]) else: # Lightweight tag case - obj is already the commit commit = obj if not isinstance(commit, Commit): raise AssertionError( f"Expected Commit object, got {type(commit).__name__}" ) tag_info: list[Any] = [ datetime.datetime(*time.gmtime(commit.commit_time)[:6]), commit.id.decode("ascii"), ] tags[tag] = tag_info # Sort tags by datetime (first element of the value list) sorted_tags = sorted( tags.items(), key=lambda tag_item: tag_item[1][0], reverse=True ) # Get the latest commit latest_commit = r[r.head()] # If there are no tags, return the latest commit if len(sorted_tags) == 0: if abbrev is not None: return "g{}".format(latest_commit.id.decode("ascii")[abbrev_slice]) return f"g{find_unique_abbrev(r.object_store, latest_commit.id)}" # We're now 0 commits from the top commit_count = 0 # Walk through all commits walker = r.get_walker() for entry in walker: # Check if tag commit_id = entry.commit.id.decode("ascii") for tag_item in sorted_tags: tag_name = tag_item[0] tag_commit = tag_item[1][1] if commit_id == tag_commit: if commit_count == 0: return tag_name else: if abbrev is not None: abbrev_hash = latest_commit.id.decode("ascii")[abbrev_slice] else: abbrev_hash = find_unique_abbrev( r.object_store, latest_commit.id ) return f"{tag_name}-{commit_count}-g{abbrev_hash}" commit_count += 1 # Return plain commit if no parent tag can be found if abbrev is not None: return "g{}".format(latest_commit.id.decode("ascii")[abbrev_slice]) return f"g{find_unique_abbrev(r.object_store, latest_commit.id)}" def get_object_by_path( repo: RepoPath, path: str | bytes, committish: str | bytes | Commit | Tag | None = None, ) -> Blob | Tree | Commit | Tag: """Get an object by path. Args: repo: A path to the repository path: Path to look up committish: Commit to look up path in Returns: A `ShaFile` object """ if committish is None: committish = "HEAD" # Get the repository with open_repo_closing(repo) as r: commit = parse_commit(r, committish) base_tree = commit.tree if not isinstance(path, bytes): path = commit_encode(commit, path) (_mode, sha) = tree_lookup_path(r.object_store.__getitem__, base_tree, path) obj = r[sha] assert isinstance(obj, Blob | Tree | Commit | Tag) return obj def write_tree(repo: RepoPath) -> bytes: """Write a tree object from the index. Args: repo: Repository for which to write tree Returns: tree id for the tree that was written """ with open_repo_closing(repo) as r: return r.open_index(config=r.get_config_stack()).commit(r.object_store) def _do_merge( r: Repo, merge_commit_id: ObjectID, no_commit: bool = False, no_ff: bool = False, message: bytes | None = None, author: bytes | None = None, committer: bytes | None = None, env: Mapping[str, str] | None = None, ) -> tuple[ObjectID | None, list[bytes]]: """Internal merge implementation that operates on an open repository. Args: r: Open repository object merge_commit_id: SHA of commit to merge no_commit: If True, do not create a merge commit no_ff: If True, force creation of a merge commit message: Optional merge commit message author: Optional author for merge commit committer: Optional committer for merge commit env: Environment to read the user identity from (defaults to os.environ) Returns: Tuple of (merge_commit_sha, conflicts) where merge_commit_sha is None if no_commit=True or there were conflicts """ from ..graph import find_merge_base from ..merge import recursive_merge # Get HEAD commit try: head_commit_id = r.refs[HEADREF] except KeyError: raise Error("No HEAD reference found") head_commit = r[head_commit_id] assert isinstance(head_commit, Commit), "Expected a Commit object" merge_commit = r[merge_commit_id] assert isinstance(merge_commit, Commit), "Expected a Commit object" # Check if fast-forward is possible merge_bases = find_merge_base(r, [head_commit_id, merge_commit_id]) if not merge_bases: raise Error("No common ancestor found") # Use the first merge base for fast-forward checks base_commit_id = merge_bases[0] # Check if we're trying to merge the same commit if head_commit_id == merge_commit_id: # Already up to date return (None, []) # Check for fast-forward if base_commit_id == head_commit_id and not no_ff: # Fast-forward merge r.refs[HEADREF] = merge_commit_id # Update the working directory changes = tree_changes(r.object_store, head_commit.tree, merge_commit.tree) update_working_tree( r, head_commit.tree, merge_commit.tree, change_iterator=changes, config=r.get_config_stack(), ) return (merge_commit_id, []) if base_commit_id == merge_commit_id: # Already up to date return (None, []) # Perform recursive merge (handles multiple merge bases automatically) gitattributes = r.get_gitattributes() config = r.get_config() merged_tree, conflicts = recursive_merge( r.object_store, merge_bases, head_commit, merge_commit, gitattributes, config ) # Add merged tree to object store r.object_store.add_object(merged_tree) # Update index and working directory changes = tree_changes(r.object_store, head_commit.tree, merged_tree.id) update_working_tree( r, head_commit.tree, merged_tree.id, change_iterator=changes, config=r.get_config_stack(), ) if conflicts or no_commit: # Don't create a commit if there are conflicts or no_commit is True return (None, conflicts) # Create merge commit merge_commit_obj = Commit() merge_commit_obj.tree = merged_tree.id merge_commit_obj.parents = [head_commit_id, merge_commit_id] # Set author/committer if author is None: author = _get_user_identity(_config_stack(r, env=env), kind="AUTHOR", env=env) if committer is None: committer = author merge_commit_obj.author = author merge_commit_obj.committer = committer # Set timestamps timestamp = int(time.time()) timezone = 0 # UTC merge_commit_obj.author_time = timestamp merge_commit_obj.author_timezone = timezone merge_commit_obj.commit_time = timestamp merge_commit_obj.commit_timezone = timezone # Set commit message if message is None: message = f"Merge commit '{merge_commit_id.decode()[:7]}'\n".encode() merge_commit_obj.message = message.encode() if isinstance(message, str) else message # Add commit to object store r.object_store.add_object(merge_commit_obj) # Update HEAD r.refs[HEADREF] = merge_commit_obj.id return (merge_commit_obj.id, []) def _do_octopus_merge( r: Repo, merge_commit_ids: list[ObjectID], no_commit: bool = False, no_ff: bool = False, message: bytes | None = None, author: bytes | None = None, committer: bytes | None = None, env: Mapping[str, str] | None = None, ) -> tuple[ObjectID | None, list[bytes]]: """Internal octopus merge implementation that operates on an open repository. Args: r: Open repository object merge_commit_ids: List of commit SHAs to merge no_commit: If True, do not create a merge commit no_ff: If True, force creation of a merge commit (ignored for octopus) message: Optional merge commit message author: Optional author for merge commit committer: Optional committer for merge commit env: Environment to read the user identity from (defaults to os.environ) Returns: Tuple of (merge_commit_sha, conflicts) where merge_commit_sha is None if no_commit=True or there were conflicts """ from ..graph import find_octopus_base from ..merge import octopus_merge # Get HEAD commit try: head_commit_id = r.refs[HEADREF] except KeyError: raise Error("No HEAD reference found") head_commit = r[head_commit_id] assert isinstance(head_commit, Commit), "Expected a Commit object" # Get all commits to merge other_commits = [] for merge_commit_id in merge_commit_ids: merge_commit = r[merge_commit_id] assert isinstance(merge_commit, Commit), "Expected a Commit object" # Check if we're trying to merge the same commit as HEAD if head_commit_id == merge_commit_id: # Skip this commit, it's already merged continue other_commits.append(merge_commit) # If no commits to merge after filtering, we're already up to date if not other_commits: return (None, []) # If only one commit to merge, use regular merge if len(other_commits) == 1: return _do_merge( r, other_commits[0].id, no_commit, no_ff, message, author, committer, env ) # Find the octopus merge base all_commit_ids = [head_commit_id] + [c.id for c in other_commits] merge_bases = find_octopus_base(r, all_commit_ids) if not merge_bases: raise Error("No common ancestor found") # Check if this is a fast-forward (HEAD is the merge base) # For octopus merges, fast-forward doesn't really apply, so we always create a merge commit # Perform octopus merge gitattributes = r.get_gitattributes() config = r.get_config() merged_tree, conflicts = octopus_merge( r.object_store, merge_bases, head_commit, other_commits, gitattributes, config ) # Add merged tree to object store r.object_store.add_object(merged_tree) # Update index and working directory changes = tree_changes(r.object_store, head_commit.tree, merged_tree.id) update_working_tree( r, head_commit.tree, merged_tree.id, change_iterator=changes, config=r.get_config_stack(), ) if conflicts: # Don't create a commit if there are conflicts # Octopus merge refuses to proceed with conflicts return (None, conflicts) if no_commit: # Don't create a commit if no_commit is True return (None, []) # Create merge commit with multiple parents merge_commit_obj = Commit() merge_commit_obj.tree = merged_tree.id merge_commit_obj.parents = [head_commit_id] + [c.id for c in other_commits] # Set author/committer if author is None: author = _get_user_identity(_config_stack(r, env=env), kind="AUTHOR", env=env) if committer is None: committer = author merge_commit_obj.author = author merge_commit_obj.committer = committer # Set timestamps timestamp = int(time.time()) timezone = 0 # UTC merge_commit_obj.author_time = timestamp merge_commit_obj.author_timezone = timezone merge_commit_obj.commit_time = timestamp merge_commit_obj.commit_timezone = timezone # Set commit message if message is None: # Generate default message for octopus merge branch_names = [] for commit_id in merge_commit_ids: branch_names.append(commit_id.decode()[:7]) message = f"Merge commits {', '.join(branch_names)}\n".encode() merge_commit_obj.message = message.encode() if isinstance(message, str) else message # Add commit to object store r.object_store.add_object(merge_commit_obj) # Update HEAD r.refs[HEADREF] = merge_commit_obj.id return (merge_commit_obj.id, []) def merge( repo: str | os.PathLike[str] | Repo, committish: str | bytes | Commit | Tag | Sequence[str | bytes | Commit | Tag], no_commit: bool = False, no_ff: bool = False, message: bytes | None = None, author: bytes | None = None, committer: bytes | None = None, env: Mapping[str, str] | None = None, ) -> tuple[bytes | None, list[bytes]]: """Merge one or more commits into the current branch. Args: repo: Repository to merge into committish: Commit(s) to merge. Can be a single commit or a sequence of commits. When merging more than two heads, the octopus merge strategy is used. no_commit: If True, do not create a merge commit no_ff: If True, force creation of a merge commit message: Optional merge commit message author: Optional author for merge commit committer: Optional committer for merge commit env: Environment to read the user identity from (defaults to os.environ) Returns: Tuple of (merge_commit_sha, conflicts) where merge_commit_sha is None if no_commit=True or there were conflicts Raises: Error: If there is no HEAD reference or commit cannot be found """ with open_repo_closing(repo) as r: # Handle both single commit and multiple commits if isinstance(committish, list | tuple): # Multiple commits - use octopus merge merge_commit_ids = [] for c in committish: try: merge_commit_ids.append(parse_commit(r, c).id) except KeyError: raise Error( f"Cannot find commit '{c.decode() if isinstance(c, bytes) else c}'" ) if len(merge_commit_ids) == 1: # Only one commit, use regular merge result = _do_merge( r, merge_commit_ids[0], no_commit, no_ff, message, author, committer, env, ) else: # Multiple commits, use octopus merge result = _do_octopus_merge( r, merge_commit_ids, no_commit, no_ff, message, author, committer, env, ) else: # Single commit - use regular merge # Type narrowing: committish is not a sequence in this branch single_committish = cast(str | bytes | Commit | Tag, committish) try: merge_commit_id = parse_commit(r, single_committish).id except KeyError: raise Error( f"Cannot find commit '{single_committish.decode() if isinstance(single_committish, bytes) else single_committish}'" ) result = _do_merge( r, merge_commit_id, no_commit, no_ff, message, author, committer, env ) # Trigger auto GC if needed from ..gc import maybe_auto_gc maybe_auto_gc(r) return result def unpack_objects( pack_path: str | os.PathLike[str], target: str | os.PathLike[str] = "." ) -> int: """Unpack objects from a pack file into the repository. Args: pack_path: Path to the pack file to unpack target: Path to the repository to unpack into Returns: Number of objects unpacked """ from ..pack import Pack with open_repo_closing(target) as r: pack_basename = os.path.splitext(pack_path)[0] with Pack(pack_basename, object_format=r.object_store.object_format) as pack: count = 0 for unpacked in pack.iter_unpacked(): obj = unpacked.sha_file() r.object_store.add_object(obj) count += 1 return count def merge_tree( repo: RepoPath, base_tree: str | bytes | Tree | Commit | Tag | None, our_tree: str | bytes | Tree | Commit | Tag, their_tree: str | bytes | Tree | Commit | Tag, ) -> tuple[bytes, list[bytes]]: """Perform a three-way tree merge without touching the working directory. This is similar to git merge-tree, performing a merge at the tree level without creating commits or updating any references. Args: repo: Repository containing the trees base_tree: Tree-ish of the common ancestor (or None for no common ancestor) our_tree: Tree-ish of our side of the merge their_tree: Tree-ish of their side of the merge Returns: tuple: A tuple of (merged_tree_id, conflicts) where: - merged_tree_id is the SHA-1 of the merged tree - conflicts is a list of paths (as bytes) that had conflicts Raises: KeyError: If any of the tree-ish arguments cannot be resolved """ from ..merge import Merger with open_repo_closing(repo) as r: # Resolve tree-ish arguments to actual trees base = parse_tree(r, base_tree) if base_tree else None ours = parse_tree(r, our_tree) theirs = parse_tree(r, their_tree) # Perform the merge gitattributes = r.get_gitattributes() config = r.get_config() merger = Merger(r.object_store, gitattributes, config) merged_tree, conflicts = merger.merge_trees(base, ours, theirs) # Add the merged tree to the object store r.object_store.add_object(merged_tree) return merged_tree.id, conflicts def cherry( repo: str | os.PathLike[str] | Repo, upstream: str | bytes | None = None, head: str | bytes | None = None, limit: str | bytes | None = None, verbose: bool = False, ) -> list[tuple[str, bytes, bytes | None]]: """Find commits not merged upstream. Args: repo: Repository path or object upstream: Upstream branch (default: tracking branch or @{upstream}) head: Head branch (default: HEAD) limit: Limit commits to those after this ref verbose: Include commit messages in output Returns: List of tuples (status, commit_sha, message) where status is '+' or '-' '+' means commit is not in upstream, '-' means equivalent patch exists upstream message is None unless verbose=True """ from ..patch import commit_patch_id with open_repo_closing(repo) as r: # Resolve upstream if upstream is None: # Try to find tracking branch upstream_found = False head_refs, _ = r.refs.follow(HEADREF) if head_refs: head_ref = head_refs[0] if head_ref.startswith(b"refs/heads/"): config = r.get_config() branch_name = head_ref[len(b"refs/heads/") :] try: upstream_ref = config.get((b"branch", branch_name), b"merge") except KeyError: upstream_ref = None if upstream_ref: try: remote_name = config.get( (b"branch", branch_name), b"remote" ) except KeyError: remote_name = None if remote_name: # Build the tracking branch ref upstream_refname = Ref( b"refs/remotes/" + remote_name + b"/" + upstream_ref.split(b"/")[-1] ) if upstream_refname in r.refs: upstream = upstream_refname upstream_found = True if not upstream_found: # Default to HEAD^ if no tracking branch found head_commit = r[HEADREF] if isinstance(head_commit, Commit) and head_commit.parents: upstream = head_commit.parents[0] else: raise ValueError("Could not determine upstream branch") # Resolve head if head is None: head = b"HEAD" # Convert strings to bytes if isinstance(upstream, str): upstream = upstream.encode("utf-8") if isinstance(head, str): head = head.encode("utf-8") if limit is not None and isinstance(limit, str): limit = limit.encode("utf-8") # Resolve refs to commit IDs assert upstream is not None upstream_obj = r[upstream] head_obj = r[head] upstream_id = upstream_obj.id head_id = head_obj.id # Get limit commit ID if specified limit_id = None if limit is not None: limit_id = r[limit].id # Find all commits reachable from head but not from upstream # This is equivalent to: git rev-list ^upstream head # Get commits from head that are not in upstream walker = r.get_walker([head_id], exclude=[upstream_id]) head_commits = [] for entry in walker: commit = entry.commit # Apply limit if specified if limit_id is not None: # Stop when we reach the limit commit if commit.id == limit_id: break head_commits.append(commit.id) # Compute patch IDs for upstream commits upstream_walker = r.get_walker([upstream_id]) upstream_patch_ids = {} # Maps patch_id -> commit_id for debugging for entry in upstream_walker: commit = entry.commit pid = commit_patch_id(r.object_store, commit.id) upstream_patch_ids[pid] = commit.id # For each head commit, check if equivalent patch exists in upstream results: list[tuple[str, bytes, bytes | None]] = [] for commit_id in reversed(head_commits): # Show oldest first obj = r.object_store[commit_id] assert isinstance(obj, Commit) commit = obj pid = commit_patch_id(r.object_store, commit_id) if pid in upstream_patch_ids: status = "-" else: status = "+" message = None if verbose: message = commit.message.split(b"\n")[0] # First line only results.append((status, commit_id, message)) return results def range_diff( repo: RepoPath, range1: str | bytes | Commit | Tag, range2: str | bytes | Commit | Tag | None = None, base: str | bytes | Commit | Tag | None = None, *, creation_factor: int | None = None, diff_algorithm: str | None = None, outstream: BinaryIO = default_bytes_out_stream, ) -> None: """Compare two commit ranges, like ``git range-diff``. Three argument forms are supported, mirroring git: * ``range_diff(repo, "A..B", "C..D")`` compares the two ranges directly. * ``range_diff(repo, rev1, rev2, base=base)`` compares ``base..rev1`` with ``base..rev2``. * ``range_diff(repo, "rev1...rev2")`` compares ``rev2..rev1`` with ``rev1..rev2`` using their merge base. Revisions may be given as strings or bytes (including ``A..B`` range syntax), or as Commit or Tag objects. Args: repo: Path to repository or a Repo object. range1: First range (``A..B``), or a single revision when ``base`` is given, or ``rev1...rev2`` for the symmetric-difference form. range2: Second range (``C..D``), or a single revision when ``base`` is given. Must be None for the ``rev1...rev2`` form. base: Common base for the ``base rev1 rev2`` form. creation_factor: Percentage controlling when two commits are considered a match (see git's ``--creation-factor``); defaults to git's default. diff_algorithm: Diff algorithm to use ("myers" or "patience"). outstream: Stream to write the rendered range-diff to. """ from ..graph import find_merge_base from ..objectspec import parse_commit_range from ..range_diff import ( DEFAULT_CREATION_FACTOR, format_range_diff, ) from ..range_diff import ( range_diff as _range_diff, ) if creation_factor is None: creation_factor = DEFAULT_CREATION_FACTOR def has_range_syntax(rev: object, sep: bytes) -> bool: if isinstance(rev, str): return sep.decode() in rev if isinstance(rev, bytes): return sep in rev return False with open_repo_closing(repo) as r: if base is not None: # Form: base rev1 rev2 -> base..rev1 and base..rev2 if range2 is None: raise Error("Two revisions are required when a base is given") base_id = parse_commit(r, base).id old_base = base_id new_base = base_id old_tip = parse_commit(r, range1).id new_tip = parse_commit(r, range2).id elif range2 is None: # Form: rev1...rev2 -> rev2..rev1 and rev1..rev2 if not has_range_syntax(range1, b"..."): raise Error("A single argument must be of the form ...") assert isinstance(range1, str | bytes) range1 = range1.encode() if isinstance(range1, str) else range1 left, right = range1.split(b"...", 1) left_id = parse_commit(r, left).id right_id = parse_commit(r, right or b"HEAD").id merge_bases = find_merge_base(r, [left_id, right_id]) if not merge_bases: raise Error("No merge base found for the two revisions") mb = merge_bases[0] old_base = mb old_tip = left_id new_base = mb new_tip = right_id else: # Form: A..B C..D if not isinstance(range1, str | bytes) or not isinstance( range2, str | bytes ): raise Error( "Both arguments must be commit ranges of the form .." ) old_range = parse_commit_range(r, range1) new_range = parse_commit_range(r, range2) if old_range is None or new_range is None: raise Error( "Both arguments must be commit ranges of the form .." ) old_base = old_range[0].id old_tip = old_range[1].id new_base = new_range[0].id new_tip = new_range[1].id entries = _range_diff( r, old_base, old_tip, new_base, new_tip, creation_factor=creation_factor, diff_algorithm=diff_algorithm, ) for line in format_range_diff(entries): outstream.write(line) def cherry_pick( # noqa: D417 repo: str | os.PathLike[str] | Repo, committish: str | bytes | Commit | Tag | None, no_commit: bool = False, continue_: bool = False, abort: bool = False, ) -> bytes | None: r"""Cherry-pick a commit onto the current branch. Args: repo: Repository to cherry-pick into committish: Commit to cherry-pick (can be None only when resuming or aborting) no_commit: If True, do not create a commit after applying changes ``continue_``: Resume an in-progress cherry-pick after resolving conflicts if True abort: Abort an in-progress cherry-pick Returns: The SHA of the newly created commit, or None if no_commit=True or there were conflicts Raises: Error: If there is no HEAD reference, commit cannot be found, or operation fails """ from ..merge import three_way_merge # Validate that committish is provided when needed if not (continue_ or abort) and committish is None: raise ValueError("committish is required when not using --continue or --abort") with open_repo_closing(repo) as r: # Handle abort if abort: # Clean up any cherry-pick state try: os.remove(os.path.join(r.controldir(), "CHERRY_PICK_HEAD")) except FileNotFoundError: pass try: os.remove(os.path.join(r.controldir(), "MERGE_MSG")) except FileNotFoundError: pass # Reset index to HEAD head_commit = r[b"HEAD"] assert isinstance(head_commit, Commit) r.get_worktree().reset_index(head_commit.tree, config=r.get_config_stack()) return None # Handle continue if continue_: # Check if there's a cherry-pick in progress cherry_pick_head_path = os.path.join(r.controldir(), "CHERRY_PICK_HEAD") try: with open(cherry_pick_head_path, "rb") as f: cherry_pick_commit_id = f.read().strip() cherry_pick_commit = r[cherry_pick_commit_id] except FileNotFoundError: raise Error("No cherry-pick in progress") # Check for unresolved conflicts if r.open_index(config=r.get_config_stack()).has_conflicts(): raise Error("Unresolved conflicts remain") # Create the commit tree_id = r.open_index(config=r.get_config_stack()).commit(r.object_store) # Read saved message if any merge_msg_path = os.path.join(r.controldir(), "MERGE_MSG") try: with open(merge_msg_path, "rb") as f: message = f.read() except FileNotFoundError: assert isinstance(cherry_pick_commit, Commit) message = cherry_pick_commit.message assert isinstance(cherry_pick_commit, Commit) new_commit = r.get_worktree().commit( message=message, tree=tree_id, author=cherry_pick_commit.author, author_timestamp=cherry_pick_commit.author_time, author_timezone=cherry_pick_commit.author_timezone, config=r.get_config_stack(), ) # Clean up state files try: os.remove(cherry_pick_head_path) except FileNotFoundError: pass try: os.remove(merge_msg_path) except FileNotFoundError: pass return new_commit # Normal cherry-pick operation # Get current HEAD try: head_commit = r[b"HEAD"] except KeyError: raise Error("No HEAD reference found") # Parse the commit to cherry-pick # committish cannot be None here due to validation above assert committish is not None try: cherry_pick_commit = parse_commit(r, committish) except KeyError: raise Error( f"Cannot find commit '{committish.decode() if isinstance(committish, bytes) else committish}'" ) # Check if commit has parents assert isinstance(cherry_pick_commit, Commit) if not cherry_pick_commit.parents: raise Error("Cannot cherry-pick root commit") # Get parent of cherry-pick commit parent_commit = r[cherry_pick_commit.parents[0]] assert isinstance(parent_commit, Commit) # Perform three-way merge assert isinstance(head_commit, Commit) merged_tree, conflicts = three_way_merge( r.object_store, parent_commit, head_commit, cherry_pick_commit ) # Add merged tree to object store r.object_store.add_object(merged_tree) # Update working tree and index # Reset index to match merged tree r.get_worktree().reset_index(merged_tree.id, config=r.get_config_stack()) # Update working tree from the new index # Allow overwriting because we're applying the merge result assert isinstance(head_commit, Commit) changes = tree_changes(r.object_store, head_commit.tree, merged_tree.id) update_working_tree( r, head_commit.tree, merged_tree.id, change_iterator=changes, allow_overwrite_modified=True, config=r.get_config_stack(), ) if conflicts: # Save state for later continuation with open(os.path.join(r.controldir(), "CHERRY_PICK_HEAD"), "wb") as f: f.write(cherry_pick_commit.id + b"\n") # Save commit message with open(os.path.join(r.controldir(), "MERGE_MSG"), "wb") as f: f.write(cherry_pick_commit.message) raise Error( f"Conflicts in: {', '.join(c.decode('utf-8', 'replace') for c in conflicts)}\n" f"Fix conflicts and run 'dulwich cherry-pick --continue'" ) if no_commit: return None # Create the commit new_commit = r.get_worktree().commit( message=cherry_pick_commit.message, tree=merged_tree.id, author=cherry_pick_commit.author, author_timestamp=cherry_pick_commit.author_time, author_timezone=cherry_pick_commit.author_timezone, config=r.get_config_stack(), ) return new_commit def revert( repo: str | os.PathLike[str] | Repo, commits: str | bytes | Commit | Tag | Sequence[str | bytes | Commit | Tag], no_commit: bool = False, message: str | bytes | None = None, author: bytes | None = None, committer: bytes | None = None, env: Mapping[str, str] | None = None, ) -> bytes | None: """Revert one or more commits. This creates a new commit that undoes the changes introduced by the specified commits. Unlike reset, revert creates a new commit that preserves history. Args: repo: Path to repository or repository object commits: List of commit-ish (SHA, ref, etc.) to revert, or a single commit-ish no_commit: If True, apply changes to index/working tree but don't commit message: Optional commit message (default: "Revert ") author: Optional author for revert commit committer: Optional committer for revert commit env: Environment to read the user identity from (defaults to os.environ) Returns: SHA1 of the new revert commit, or None if no_commit=True Raises: Error: If revert fails due to conflicts or other issues """ from ..merge import three_way_merge # Normalize commits to a list if isinstance(commits, str | bytes | Commit | Tag): commits = [commits] with open_repo_closing(repo) as r: # Convert string refs to bytes commits_to_revert = [] for commit_ref in commits: if isinstance(commit_ref, str): commit_ref = commit_ref.encode("utf-8") commit = parse_commit(r, commit_ref) commits_to_revert.append(commit) # Get current HEAD try: head_commit_id = r.refs[HEADREF] except KeyError: raise Error("No HEAD reference found") head_commit = r[head_commit_id] assert isinstance(head_commit, Commit) current_tree = head_commit.tree # Process commits in order for commit_to_revert in commits_to_revert: # For revert, we want to apply the inverse of the commit # This means using the commit's tree as "base" and its parent as "theirs" if not commit_to_revert.parents: raise Error( f"Cannot revert commit {commit_to_revert.id.decode() if isinstance(commit_to_revert.id, bytes) else commit_to_revert.id} - it has no parents" ) # For simplicity, we only handle commits with one parent (no merge commits) if len(commit_to_revert.parents) > 1: raise Error( f"Cannot revert merge commit {commit_to_revert.id.decode() if isinstance(commit_to_revert.id, bytes) else commit_to_revert.id} - not yet implemented" ) parent_commit = r[commit_to_revert.parents[0]] assert isinstance(parent_commit, Commit) # Perform three-way merge: # - base: the commit we're reverting (what we want to remove) # - ours: current HEAD (what we have now) # - theirs: parent of commit being reverted (what we want to go back to) assert isinstance(commit_to_revert, Commit) head_for_merge = r[head_commit_id] assert isinstance(head_for_merge, Commit) merged_tree, conflicts = three_way_merge( r.object_store, commit_to_revert, # base head_for_merge, # ours parent_commit, # theirs ) if conflicts: # Update working tree with conflicts changes = tree_changes(r.object_store, current_tree, merged_tree.id) update_working_tree( r, current_tree, merged_tree.id, change_iterator=changes, config=r.get_config_stack(), ) conflicted_paths = [c.decode("utf-8", "replace") for c in conflicts] raise Error(f"Conflicts while reverting: {', '.join(conflicted_paths)}") # Add merged tree to object store r.object_store.add_object(merged_tree) # Update working tree changes = tree_changes(r.object_store, current_tree, merged_tree.id) update_working_tree( r, current_tree, merged_tree.id, change_iterator=changes, config=r.get_config_stack(), ) current_tree = merged_tree.id if not no_commit: # Create revert commit revert_commit = Commit() revert_commit.tree = merged_tree.id revert_commit.parents = [head_commit_id] # Set author/committer if author is None: author = _get_user_identity( _config_stack(r, env=env), kind="AUTHOR", env=env ) if committer is None: committer = author revert_commit.author = author revert_commit.committer = committer # Set timestamps timestamp = int(time.time()) timezone = 0 # UTC revert_commit.author_time = timestamp revert_commit.author_timezone = timezone revert_commit.commit_time = timestamp revert_commit.commit_timezone = timezone # Set message if message is None: # Extract original commit subject original_message = commit_to_revert.message if isinstance(original_message, bytes): original_message = original_message.decode("utf-8", "replace") subject = original_message.split("\n")[0] message = f'Revert "{subject}"\n\nThis reverts commit {commit_to_revert.id.decode("ascii")}.'.encode() elif isinstance(message, str): message = message.encode("utf-8") revert_commit.message = message # Add commit to object store r.object_store.add_object(revert_commit) # Update HEAD r.refs[HEADREF] = revert_commit.id head_commit_id = revert_commit.id return head_commit_id if not no_commit else None def gc( repo: RepoPath, auto: bool = False, aggressive: bool = False, prune: bool = True, grace_period: int | None = None, dry_run: bool = False, progress: Callable[[str], None] | None = None, ) -> "GCStats": """Run garbage collection on a repository. Args: repo: Path to the repository or a Repo object auto: If True, only run gc if needed aggressive: If True, use more aggressive settings prune: If True, prune unreachable objects grace_period: Grace period in seconds for pruning. If None, reads gc.pruneExpire from config (default 2 weeks). dry_run: If True, only report what would be done progress: Optional progress callback Returns: GCStats object with garbage collection statistics """ from ..gc import garbage_collect, get_prune_grace_period with open_repo_closing(repo) as r: if grace_period is None: grace_period = get_prune_grace_period(r.get_config()) return garbage_collect( r, auto=auto, aggressive=aggressive, prune=prune, grace_period=grace_period, dry_run=dry_run, progress=progress, ) def prune( repo: RepoPath, grace_period: int | None = None, dry_run: bool = False, progress: Callable[[str], None] | None = None, ) -> None: """Prune/clean up a repository's object store. This removes temporary files that were left behind by interrupted pack operations. Args: repo: Path to the repository or a Repo object grace_period: Grace period in seconds for removing temporary files (default 2 weeks) dry_run: If True, only report what would be done progress: Optional progress callback """ with open_repo_closing(repo) as r: if progress: progress("Pruning temporary files") if not dry_run: r.object_store.prune(grace_period=grace_period) def maintenance_run( repo: RepoPath, tasks: list[str] | None = None, auto: bool = False, progress: Callable[[str], None] | None = None, ) -> "MaintenanceResult": """Run maintenance tasks on a repository. Args: repo: Path to the repository or a Repo object tasks: Optional list of specific task names to run (e.g., ['gc', 'commit-graph', 'pack-refs']) auto: If True, only run tasks if needed progress: Optional progress callback Returns: MaintenanceResult object with task execution results """ from ..maintenance import run_maintenance with open_repo_closing(repo) as r: return run_maintenance(r, tasks=tasks, auto=auto, progress=progress) def maintenance_register(repo: RepoPath) -> None: """Register a repository for background maintenance. This adds the repository to the global maintenance.repo config and sets up recommended configuration for scheduled maintenance. Args: repo: Path to the repository or repository object """ from ..maintenance import register_repository with open_repo_closing(repo) as r: register_repository(r) def maintenance_unregister(repo: RepoPath, force: bool = False) -> None: """Unregister a repository from background maintenance. This removes the repository from the global maintenance.repo config. Args: repo: Path to the repository or repository object force: If True, don't error if repository is not registered """ from ..maintenance import unregister_repository with open_repo_closing(repo) as r: unregister_repository(r, force=force) def count_objects( repo: RepoPath | None = None, verbose: bool = False ) -> CountObjectsResult: """Count unpacked objects and their disk usage. Args: repo: Path to repository or repository object verbose: Whether to return verbose information Returns: CountObjectsResult object with detailed statistics """ with open_repo_closing(repo) as r: object_store = r.object_store # Count loose objects loose_count = 0 loose_size = 0 for sha in object_store._iter_loose_objects(): loose_count += 1 from ..object_store import DiskObjectStore assert isinstance(object_store, DiskObjectStore) path = object_store._get_shafile_path(sha) try: stat_info = os.stat(path) # Git uses disk usage, not file size. st_blocks is always in # 512-byte blocks per POSIX standard st_blocks = getattr(stat_info, "st_blocks", None) if st_blocks is not None: # Available on Linux and macOS loose_size += st_blocks * 512 else: # Fallback for Windows loose_size += stat_info.st_size except FileNotFoundError: # Object may have been removed between iteration and stat pass if not verbose: return CountObjectsResult(count=loose_count, size=loose_size) # Count pack information pack_count = len(object_store.packs) in_pack_count = 0 pack_size = 0 for pack in object_store.packs: in_pack_count += len(pack) # Get pack file size pack_path = pack._data_path try: pack_size += os.path.getsize(pack_path) except FileNotFoundError: pass # Get index file size idx_path = pack._idx_path try: pack_size += os.path.getsize(idx_path) except FileNotFoundError: pass return CountObjectsResult( count=loose_count, size=loose_size, in_pack=in_pack_count, packs=pack_count, size_pack=pack_size, ) def is_interactive_rebase(repo: Repo | str) -> bool: """Check if an interactive rebase is in progress. Args: repo: Repository to check Returns: True if interactive rebase is in progress, False otherwise """ with open_repo_closing(repo) as r: state_manager = r.get_rebase_state_manager() if not state_manager.exists(): return False # Check if todo file exists todo = state_manager.load_todo() return todo is not None def rebase( repo: Repo | str, upstream: bytes | str, onto: bytes | str | None = None, branch: bytes | str | None = None, abort: bool = False, continue_rebase: bool = False, skip: bool = False, interactive: bool = False, edit_todo: bool = False, ) -> list[bytes]: """Rebase commits onto another branch. Args: repo: Repository to rebase in upstream: Upstream branch/commit to rebase onto onto: Specific commit to rebase onto (defaults to upstream) branch: Branch to rebase (defaults to current branch) abort: Abort an in-progress rebase continue_rebase: Continue an in-progress rebase skip: Skip current commit and continue rebase interactive: Start an interactive rebase edit_todo: Edit the todo list of an interactive rebase Returns: List of new commit SHAs created by rebase Raises: Error: If rebase fails or conflicts occur """ # TODO: Avoid importing from ..cli from ..cli import launch_editor from ..rebase import ( RebaseConflict, RebaseError, Rebaser, process_interactive_rebase, start_interactive, ) from ..rebase import ( edit_todo as edit_todo_func, ) with open_repo_closing(repo) as r: rebaser = Rebaser(r) if abort: try: rebaser.abort() return [] except RebaseError as e: raise Error(str(e)) if edit_todo: # Edit the todo list of an interactive rebase try: edit_todo_func(r, launch_editor) print("Todo list updated. Continue with 'rebase --continue'") return [] except RebaseError as e: raise Error(str(e)) if continue_rebase: try: if interactive: # Continue interactive rebase is_complete, pause_reason = process_interactive_rebase( r, editor_callback=launch_editor ) if is_complete: return [c.id for c in rebaser._done] else: if pause_reason == "conflict": raise Error("Conflicts detected. Resolve and continue.") elif pause_reason == "edit": print("Stopped for editing. Make changes and continue.") elif pause_reason == "break": print("Rebase paused at break. Continue when ready.") else: print(f"Rebase paused: {pause_reason}") return [] else: # Continue regular rebase result = rebaser.continue_() if result is None: # Rebase complete return [c.id for c in rebaser._done] elif isinstance(result, tuple) and result[1]: # Still have conflicts raise Error( f"Conflicts in: {', '.join(f.decode('utf-8', 'replace') for f in result[1])}" ) except RebaseError as e: raise Error(str(e)) # Convert string refs to bytes if isinstance(upstream, str): upstream = upstream.encode("utf-8") if isinstance(onto, str): onto = onto.encode("utf-8") if onto else None if isinstance(branch, str): branch = branch.encode("utf-8") if branch else None try: if interactive: # Start interactive rebase todo = start_interactive(r, upstream, onto, branch, launch_editor) # Process the todo list is_complete, pause_reason = process_interactive_rebase( r, todo, editor_callback=launch_editor ) if is_complete: return [c.id for c in rebaser._done] else: if pause_reason == "conflict": raise Error("Conflicts detected. Resolve and continue.") elif pause_reason == "edit": print("Stopped for editing. Make changes and continue.") elif pause_reason == "break": print("Rebase paused at break. Continue when ready.") else: print(f"Rebase paused: {pause_reason}") return [] else: # Regular rebase rebaser.start(upstream, onto, branch) # Continue rebase automatically result = rebaser.continue_() if result is not None: # Conflicts raise RebaseConflict(result[1]) # Return the SHAs of the rebased commits return [c.id for c in rebaser._done] except RebaseConflict as e: raise Error(str(e)) except RebaseError as e: raise Error(str(e)) def annotate( repo: RepoPath, path: str | bytes, committish: str | bytes | Commit | Tag | None = None, ) -> list[tuple[tuple[Commit, TreeEntry], bytes]]: """Annotate the history of a file. :param repo: Path to the repository :param path: Path to annotate :param committish: Commit id to find path in :return: List of ((Commit, TreeChange), line) tuples """ if committish is None: committish = "HEAD" from ..annotate import annotate_lines with open_repo_closing(repo) as r: commit_id = parse_commit(r, committish).id # Ensure path is bytes if isinstance(path, str): path = path.encode() return annotate_lines(r.object_store, commit_id, path) blame = annotate def filter_branch( repo: RepoPath | None = None, branch: str | bytes = "HEAD", *, filter_fn: Callable[[Commit], "CommitData | None"] | None = None, filter_author: Callable[[bytes], bytes | None] | None = None, filter_committer: Callable[[bytes], bytes | None] | None = None, filter_message: Callable[[bytes], bytes | None] | None = None, tree_filter: Callable[[ObjectID, str], ObjectID | None] | None = None, index_filter: Callable[[ObjectID, str], ObjectID | None] | None = None, parent_filter: Callable[[Sequence[ObjectID]], list[ObjectID]] | None = None, commit_filter: Callable[[Commit, ObjectID], ObjectID | None] | None = None, subdirectory_filter: str | bytes | None = None, prune_empty: bool = False, tag_name_filter: Callable[[bytes], bytes | None] | None = None, force: bool = False, keep_original: bool = True, refs: list[bytes] | None = None, ) -> dict[ObjectID, ObjectID]: """Rewrite branch history by creating new commits with filtered properties. This is similar to git filter-branch, allowing you to rewrite commit history by modifying trees, parents, author, committer, or commit messages. Args: repo: Path to repository branch: Branch to rewrite (defaults to HEAD) filter_fn: Optional callable that takes a Commit object and returns a dict of updated fields (author, committer, message, etc.) filter_author: Optional callable that takes author bytes and returns updated author bytes or None to keep unchanged filter_committer: Optional callable that takes committer bytes and returns updated committer bytes or None to keep unchanged filter_message: Optional callable that takes commit message bytes and returns updated message bytes tree_filter: Optional callable that takes (tree_sha, temp_dir) and returns new tree SHA after modifying working directory index_filter: Optional callable that takes (tree_sha, temp_index_path) and returns new tree SHA after modifying index parent_filter: Optional callable that takes parent list and returns modified parent list commit_filter: Optional callable that takes (Commit, tree_sha) and returns new commit SHA or None to skip commit subdirectory_filter: Optional subdirectory path to extract as new root prune_empty: Whether to prune commits that become empty tag_name_filter: Optional callable to rename tags force: Force operation even if branch has been filtered before keep_original: Keep original refs under refs/original/ refs: List of refs to rewrite (defaults to [branch]) Returns: Dict mapping old commit SHAs to new commit SHAs Raises: Error: If branch is already filtered and force is False """ from ..filter_branch import CommitFilter, filter_refs with open_repo_closing(repo) as r: # Parse branch/committish if isinstance(branch, str): branch = branch.encode() # Determine which refs to process if refs is None: if branch == b"HEAD": # Resolve HEAD to actual branch try: resolved = r.refs.follow(HEADREF) if resolved and resolved[0]: # resolved is a list of (refname, sha) tuples resolved_ref = resolved[0][-1] if resolved_ref and resolved_ref != b"HEAD": refs = [resolved_ref] else: # HEAD points directly to a commit refs = [b"HEAD"] else: refs = [b"HEAD"] except SymrefLoop: refs = [b"HEAD"] else: # Convert branch name to full ref if needed if not branch.startswith(b"refs/"): branch = local_branch_name(branch) refs = [branch] # Convert subdirectory filter to bytes if needed if subdirectory_filter: if isinstance(subdirectory_filter, str): subdirectory_filter = subdirectory_filter.encode() else: subdirectory_filter = None # Create commit filter filter_obj = CommitFilter( r.object_store, filter_fn=filter_fn, filter_author=filter_author, filter_committer=filter_committer, filter_message=filter_message, tree_filter=tree_filter, index_filter=index_filter, parent_filter=parent_filter, commit_filter=commit_filter, subdirectory_filter=subdirectory_filter, prune_empty=prune_empty, tag_name_filter=tag_name_filter, ) # Tag callback for renaming tags def rename_tag(old_ref: Ref, new_ref: Ref) -> None: # Copy tag to new name r.refs[new_ref] = r.refs[old_ref] # Delete old tag del r.refs[old_ref] # Filter refs try: return filter_refs( r.refs, r.object_store, refs, filter_obj, keep_original=keep_original, force=force, tag_callback=rename_tag if tag_name_filter else None, ) except ValueError as e: raise Error(str(e)) from e def format_patch( repo: RepoPath | None = None, committish: ObjectID | tuple[ObjectID, ObjectID] | None = None, outstream: TextIO = sys.stdout, outdir: str | os.PathLike[str] | None = None, n: int = 1, stdout: bool = False, version: str | None = None, ) -> list[str]: """Generate patches suitable for git am. Args: repo: Path to repository committish: Commit-ish or commit range to generate patches for. Can be a single commit id, or a tuple of (start, end) commit ids for a range. If None, formats the last n commits from HEAD. outstream: Stream to write to if stdout=True outdir: Directory to write patch files to (default: current directory) n: Number of patches to generate if committish is None stdout: Write patches to stdout instead of files version: Version string to include in patches (default: Dulwich version) Returns: List of patch filenames that were created (empty if stdout=True) """ if outdir is None: outdir = "." filenames = [] with open_repo_closing(repo) as r: # Determine which commits to format commits_to_format = [] if committish is None: # Get the last n commits from HEAD try: walker = r.get_walker() for entry in walker: commits_to_format.append(entry.commit) if len(commits_to_format) >= n: break commits_to_format.reverse() except KeyError: # No HEAD or empty repository pass elif isinstance(committish, tuple): # Handle commit range (start, end) start_commit, end_commit = committish # Extract commit IDs from commit objects if needed start_id = ( start_commit.id if isinstance(start_commit, Commit) else start_commit ) end_id = end_commit.id if isinstance(end_commit, Commit) else end_commit # Walk from end back to start walker = r.get_walker(include=[end_id], exclude=[start_id]) for entry in walker: commits_to_format.append(entry.commit) commits_to_format.reverse() else: # Single commit commit = r.object_store[committish] assert isinstance(commit, Commit) commits_to_format.append(commit) # Generate patches total = len(commits_to_format) for i, commit in enumerate(commits_to_format, 1): assert isinstance(commit, Commit) # Get the parent if commit.parents: parent_id = commit.parents[0] parent = r.object_store[parent_id] assert isinstance(parent, Commit) else: parent = None # Generate the diff from io import BytesIO diff_content = BytesIO() if parent: write_tree_diff( diff_content, r.object_store, parent.tree, commit.tree, ) else: # Initial commit - diff against empty tree write_tree_diff( diff_content, r.object_store, None, commit.tree, ) # Generate patch with commit metadata if stdout: # Get binary stream from TextIO if hasattr(outstream, "buffer"): binary_out: IO[bytes] = outstream.buffer else: # Fallback for non-text streams binary_out = outstream # type: ignore[assignment] write_commit_patch( binary_out, commit, diff_content.getvalue(), (i, total), version=version, ) else: # Generate filename summary = get_summary(commit) filename = os.path.join(outdir, f"{i:04d}-{summary}.patch") with open_nofollow(filename) as f: write_commit_patch( f, commit, diff_content.getvalue(), (i, total), version=version, ) filenames.append(filename) return filenames def _commit_iso_date(commit: Commit) -> str: """Format a commit's committer date like git's ``%ci`` placeholder.""" time_tuple = time.gmtime(commit.commit_time + commit.commit_timezone) time_str = time.strftime("%Y-%m-%d %H:%M:%S", time_tuple) timezone_str = format_timezone(commit.commit_timezone).decode("ascii") return f"{time_str} {timezone_str}" def _request_pull_ref(r: "Repo", local: bytes) -> bytes: """Resolve the ref name to advertise in a pull request. Mirrors git request-pull, which resolves the local ref through a symbolic ref, then a matching head/tag, and finally falls back to the expression itself. """ # A symbolic ref (e.g. HEAD) resolves to the underlying branch. try: target: bytes | None = r.refs.follow(Ref(local))[0][-1] except (KeyError, ValueError): target = None if target is not None and target != local and target.startswith(b"refs/"): return target # An exact branch or tag name. for prefix in (b"refs/heads/", b"refs/tags/"): if Ref(prefix + local) in r.refs: return prefix + local if local.startswith(b"refs/") and Ref(local) in r.refs: return local return local def request_pull( repo: RepoPath, base: str | bytes, url: str, end: str | bytes | None = None, patch: bool = False, outstream: TextIO = sys.stdout, ) -> None: """Generate a summary of pending changes, like git request-pull. Produces a message asking an upstream maintainer to pull a set of changes from a repository, suitable for sending by email. Args: repo: Path to repository base: Commit to start at; must already exist in the upstream history (e.g. a tag or branch the maintainer already has). url: URL of the repository the changes can be pulled from. end: Commit to end at, defaulting to HEAD. May use the ``:`` syntax to advertise a remote ref name that differs from the local one. patch: If True, include the patch text after the summary. outstream: Stream to write the request to. Raises: Error: If there are no commits in common between base and end, or if end resolves to no commits beyond base. """ from ..diffstat import diffstat from ..graph import find_merge_base base_bytes = base.encode() if isinstance(base, str) else base if end is None: local: bytes = b"HEAD" remote: bytes | None = None else: end_bytes = end.encode() if isinstance(end, str) else end if b":" in end_bytes: local, remote = end_bytes.split(b":", 1) else: local, remote = end_bytes, None with open_repo_closing(repo) as r: base_commit = parse_commit(r, base_bytes) head_commit = parse_commit(r, local) merge_bases = find_merge_base(r, [base_commit.id, head_commit.id]) if not merge_bases: raise Error(f"No commits in common between {base!r} and {local.decode()!r}") merge_base = merge_bases[0] # Collect the commits being requested for the shortlog, oldest last # so we can sort and group them like git shortlog. walker = r.get_walker(include=[head_commit.id], exclude=[base_commit.id]) commits = [entry.commit for entry in walker] if not commits: raise Error( f"No commits between {base!r} and {local.decode()!r}; " "nothing to request" ) # Resolve the ref name to advertise. if remote is not None: pretty_remote = shorten_ref_name(remote) else: resolved = _request_pull_ref(r, local) pretty_remote = shorten_ref_name(resolved) # If the local ref resolves to an annotated tag, surface its message. tag_message: str | None = None try: tagged = parse_object(r, local) except KeyError: tagged = None if isinstance(tagged, Tag): tag_message = tagged.message.decode("utf-8", "replace").strip() merge_base_commit = r.object_store[merge_base] assert isinstance(merge_base_commit, Commit) def subject(commit: Commit) -> str: decoded: str = commit.message.decode(commit.encoding or "utf-8", "replace") return decoded.split("\n", 1)[0].strip() outstream.write( "The following changes since commit {}:\n\n".format( merge_base_commit.id.decode("ascii") ) ) outstream.write( f" {subject(merge_base_commit)} ({_commit_iso_date(merge_base_commit)})\n\n" ) outstream.write("are available in the Git repository at:\n\n") outstream.write(f" {url} {pretty_remote.decode('utf-8', 'replace')}\n\n") outstream.write( "for you to fetch changes up to {}:\n\n".format( head_commit.id.decode("ascii") ) ) outstream.write( f" {subject(head_commit)} ({_commit_iso_date(head_commit)})\n\n" ) separator = "-" * 64 if tag_message: outstream.write(separator + "\n") outstream.write(tag_message + "\n") outstream.write(separator + "\n") # Shortlog: group commit subjects by author name. Authors are sorted # case-insensitively; within an author the commits keep chronological # (oldest first) order, like git shortlog. from ..mailmap import Mailmap try: mailmap: Mailmap | None = Mailmap.from_path( os.path.join(r.path, ".mailmap") ) except FileNotFoundError: mailmap = None def author_name(commit: Commit) -> str: ident = commit.author if mailmap is not None: resolved = mailmap.lookup(ident) assert isinstance(resolved, bytes) ident = resolved name: str = ident.decode(commit.encoding or "utf-8", "replace") # Strip the "" portion, keeping just the name. if "<" in name: name = name[: name.index("<")].strip() return name by_author: dict[str, list[str]] = {} for commit in reversed(commits): by_author.setdefault(author_name(commit), []).append(subject(commit)) for author in sorted(by_author, key=str.lower): subjects = by_author[author] outstream.write(f"{author} ({len(subjects)}):\n") for subj in subjects: outstream.write(f" {subj}\n") outstream.write("\n") # Diffstat between the merge base and the head. diff_content = BytesIO() write_tree_diff( diff_content, r.object_store, merge_base_commit.tree, head_commit.tree ) diff_bytes = diff_content.getvalue() stat = diffstat(diff_bytes.splitlines()) outstream.write(stat.decode("utf-8", "replace") + "\n") if patch: outstream.write("\n") if hasattr(outstream, "buffer"): # Flush the text stream so the binary patch is not written # ahead of the already-buffered text output. outstream.flush() outstream.buffer.write(diff_bytes) else: outstream.write(diff_bytes.decode("utf-8", "replace")) def bisect_start( repo: str | os.PathLike[str] | Repo | None = None, bad: str | bytes | Commit | Tag | None = None, good: str | bytes | Commit | Tag | Sequence[str | bytes | Commit | Tag] | None = None, paths: Sequence[bytes] | None = None, no_checkout: bool = False, term_bad: str = "bad", term_good: str = "good", ) -> bytes | None: """Start a new bisect session. Args: repo: Path to repository or a Repo object bad: The bad commit (defaults to HEAD) good: List of good commits or a single good commit paths: Optional paths to limit bisect to no_checkout: If True, don't checkout commits during bisect term_bad: Term to use for bad commits (default: "bad") term_good: Term to use for good commits (default: "good") """ with open_repo_closing(repo) as r: state = BisectState(r) # Convert single good commit to sequence if good is not None and isinstance(good, str | bytes | Commit | Tag): good = [good] # Parse commits bad_sha = parse_commit(r, bad).id if bad else None good_shas = [parse_commit(r, g).id for g in good] if good else None state.start(bad_sha, good_shas, paths, no_checkout, term_bad, term_good) # Return the next commit to test if we have both good and bad if bad_sha and good_shas: next_sha = state._find_next_commit() if next_sha and not no_checkout: # Checkout the next commit old_commit = r[r.head()] assert isinstance(old_commit, Commit) old_tree = old_commit.tree if r.head() else None r.refs[HEADREF] = next_sha commit = r[next_sha] assert isinstance(commit, Commit) changes = tree_changes(r.object_store, old_tree, commit.tree) update_working_tree( r, old_tree, commit.tree, change_iterator=changes, config=r.get_config_stack(), ) return next_sha return None def bisect_bad( repo: str | os.PathLike[str] | Repo | None = None, rev: str | bytes | Commit | Tag | None = None, ) -> bytes | None: """Mark a commit as bad. Args: repo: Path to repository or a Repo object rev: Commit to mark as bad (defaults to HEAD) Returns: The SHA of the next commit to test, or None if bisect is complete """ with open_repo_closing(repo) as r: state = BisectState(r) rev_sha = parse_commit(r, rev).id if rev else None next_sha = state.mark_bad(rev_sha) if next_sha: # Checkout the next commit old_commit = r[r.head()] assert isinstance(old_commit, Commit) old_tree = old_commit.tree if r.head() else None r.refs[HEADREF] = next_sha commit = r[next_sha] assert isinstance(commit, Commit) changes = tree_changes(r.object_store, old_tree, commit.tree) update_working_tree( r, old_tree, commit.tree, change_iterator=changes, config=r.get_config_stack(), ) return next_sha def bisect_good( repo: str | os.PathLike[str] | Repo | None = None, rev: str | bytes | Commit | Tag | None = None, ) -> bytes | None: """Mark a commit as good. Args: repo: Path to repository or a Repo object rev: Commit to mark as good (defaults to HEAD) Returns: The SHA of the next commit to test, or None if bisect is complete """ with open_repo_closing(repo) as r: state = BisectState(r) rev_sha = parse_commit(r, rev).id if rev else None next_sha = state.mark_good(rev_sha) if next_sha: # Checkout the next commit old_commit = r[r.head()] assert isinstance(old_commit, Commit) old_tree = old_commit.tree if r.head() else None r.refs[HEADREF] = next_sha commit = r[next_sha] assert isinstance(commit, Commit) changes = tree_changes(r.object_store, old_tree, commit.tree) update_working_tree( r, old_tree, commit.tree, change_iterator=changes, config=r.get_config_stack(), ) return next_sha def bisect_skip( repo: str | os.PathLike[str] | Repo | None = None, revs: str | bytes | Commit | Tag | Sequence[str | bytes | Commit | Tag] | None = None, ) -> bytes | None: """Skip one or more commits. Args: repo: Path to repository or a Repo object revs: List of commits to skip (defaults to [HEAD]) Returns: The SHA of the next commit to test, or None if bisect is complete """ with open_repo_closing(repo) as r: state = BisectState(r) if revs is None: rev_shas = None else: # Convert single rev to sequence if isinstance(revs, str | bytes | Commit | Tag): revs = [revs] rev_shas = [parse_commit(r, rev).id for rev in revs] next_sha = state.skip(rev_shas) if next_sha: # Checkout the next commit old_commit = r[r.head()] assert isinstance(old_commit, Commit) old_tree = old_commit.tree if r.head() else None r.refs[HEADREF] = next_sha commit = r[next_sha] assert isinstance(commit, Commit) changes = tree_changes(r.object_store, old_tree, commit.tree) update_working_tree( r, old_tree, commit.tree, change_iterator=changes, config=r.get_config_stack(), ) return next_sha def bisect_reset( repo: str | os.PathLike[str] | Repo | None = None, commit: str | bytes | Commit | Tag | None = None, ) -> None: """Reset bisect state and return to original branch/commit. Args: repo: Path to repository or a Repo object commit: Optional commit to reset to (defaults to original branch/commit) """ with open_repo_closing(repo) as r: state = BisectState(r) # Get old tree before reset try: old_commit = r[r.head()] assert isinstance(old_commit, Commit) old_tree = old_commit.tree except KeyError: old_tree = None commit_sha = parse_commit(r, commit).id if commit else None state.reset(commit_sha) # Update working tree to new HEAD try: new_head = r.head() if new_head: new_commit = r[new_head] assert isinstance(new_commit, Commit) changes = tree_changes(r.object_store, old_tree, new_commit.tree) update_working_tree( r, old_tree, new_commit.tree, change_iterator=changes, config=r.get_config_stack(), ) except KeyError: # No HEAD after reset pass def bisect_log(repo: str | os.PathLike[str] | Repo | None = None) -> str: """Get the bisect log. Args: repo: Path to repository or a Repo object Returns: The bisect log as a string """ with open_repo_closing(repo) as r: state = BisectState(r) return state.get_log() def bisect_replay( repo: str | os.PathLike[str] | Repo, log_file: str | os.PathLike[str] | BinaryIO, ) -> None: """Replay a bisect log. Args: repo: Path to repository or a Repo object log_file: Path to the log file or file-like object """ with open_repo_closing(repo) as r: state = BisectState(r) if isinstance(log_file, str | os.PathLike): with open(log_file) as f: log_content = f.read() else: content = log_file.read() log_content = content.decode() if isinstance(content, bytes) else content state.replay(log_content) def reflog( repo: RepoPath | None = None, ref: str | bytes = b"HEAD", all: bool = False ) -> Iterator[Any | tuple[bytes, Any]]: """Show reflog entries for a reference or all references. Args: repo: Path to repository or a Repo object ref: Reference name (defaults to HEAD) all: If True, show reflogs for all refs (ignores ref parameter) Yields: If all=False: ReflogEntry objects If all=True: Tuples of (ref_name, ReflogEntry) for all refs with reflogs """ import os from ..reflog import iter_reflogs if isinstance(ref, str): ref = ref.encode("utf-8") with open_repo_closing(repo) as r: if not all: yield from r.read_reflog(ref) else: logs_dir = os.path.join(r.controldir(), "logs") # Use iter_reflogs to discover all reflogs for ref_bytes in iter_reflogs(logs_dir): # Read the reflog entries for this ref for entry in r.read_reflog(ref_bytes): yield (ref_bytes, entry) def reflog_expire( repo: RepoPath | None = None, ref: str | bytes | None = None, all: bool = False, expire_time: int | None = None, expire_unreachable_time: int | None = None, dry_run: bool = False, ) -> dict[bytes, int]: """Expire reflog entries based on age and reachability. Args: repo: Path to repository or a Repo object ref: Reference name (if not using --all) all: If True, expire reflogs for all refs expire_time: Expire entries older than this timestamp (seconds since epoch) expire_unreachable_time: Expire unreachable entries older than this timestamp dry_run: If True, show what would be expired without making changes Returns: Dictionary mapping ref names to number of expired entries """ import os import time from ..reflog import expire_reflog, iter_reflogs if not all and ref is None: raise ValueError("Must specify either ref or all=True") if isinstance(ref, str): ref = ref.encode("utf-8") # Default expire times if not specified if expire_time is None and expire_unreachable_time is None: # Default: expire entries older than 90 days, unreachable older than 30 days now = int(time.time()) expire_time = now - (90 * 24 * 60 * 60) expire_unreachable_time = now - (30 * 24 * 60 * 60) result = {} with open_repo_closing(repo) as r: # Determine which refs to process refs_to_process: list[bytes] = [] if all: logs_dir = os.path.join(r.controldir(), "logs") refs_to_process = list(iter_reflogs(logs_dir)) else: assert ref is not None # Already checked above refs_to_process = [ref] # Build set of reachable objects if we have unreachable expiration time reachable_objects: set[ObjectID] | None = None if expire_unreachable_time is not None: from ..gc import find_reachable_objects reachable_objects = find_reachable_objects( r.object_store, r.refs, include_reflogs=False ) # Process each ref for ref_name in refs_to_process: reflog_path = r._reflog_path(ref_name) if not os.path.exists(reflog_path): continue # Create reachability checker def is_reachable(sha: bytes) -> bool: if reachable_objects is None: # No unreachable expiration, so assume everything is reachable return True return sha in reachable_objects # Open the reflog file if dry_run: # For dry run, just read and count what would be expired with open(reflog_path, "rb") as f: from ..reflog import read_reflog count = 0 for entry in read_reflog(f): is_obj_reachable = is_reachable(entry.new_sha) should_expire = False if is_obj_reachable and expire_time is not None: if entry.timestamp < expire_time: should_expire = True elif ( not is_obj_reachable and expire_unreachable_time is not None ): if entry.timestamp < expire_unreachable_time: should_expire = True if should_expire: count += 1 result[ref_name] = count else: # Actually expire entries with open(reflog_path, "r+b") as f: # type: ignore[assignment] count = expire_reflog( f, expire_time=expire_time, expire_unreachable_time=expire_unreachable_time, reachable_checker=is_reachable, ) result[ref_name] = count return result def reflog_delete( repo: RepoPath | None = None, ref: str | bytes = b"HEAD", index: int = 0, rewrite: bool = False, ) -> None: """Delete a specific reflog entry. Args: repo: Path to repository or a Repo object ref: Reference name index: Reflog entry index (0 = newest, in Git reflog order) rewrite: If True, rewrite old_sha of subsequent entries to maintain consistency """ import os from ..reflog import drop_reflog_entry if isinstance(ref, str): ref = ref.encode("utf-8") with open_repo_closing(repo) as r: reflog_path = r._reflog_path(ref) if not os.path.exists(reflog_path): raise ValueError(f"No reflog for ref {ref.decode()}") with open(reflog_path, "r+b") as f: drop_reflog_entry(f, index, rewrite=rewrite) def merge_base( repo: RepoPath | None = None, committishes: Sequence[str | bytes] | None = None, all: bool = False, octopus: bool = False, ) -> list[ObjectID]: """Find the best common ancestor(s) between commits. Args: repo: Path to repository committishes: List of commit references (branches, tags, commit IDs) all: If True, return all merge bases, not just one octopus: If True, find merge base of all commits (n-way merge) Returns: List of commit IDs that are merge bases """ from ..graph import find_merge_base, find_octopus_base from ..objectspec import parse_object if committishes is None or len(committishes) < 2: raise ValueError("At least two commits are required") with open_repo_closing(repo) as r: # Resolve committish references to commit IDs commit_ids = [] for committish in committishes: obj = parse_object(r, committish, config=r.get_config_stack()) if not isinstance(obj, Commit): raise ValueError(f"Expected commit, got {obj.type_name.decode()}") commit_ids.append(obj.id) # Find merge base if octopus: result = find_octopus_base(r, commit_ids) else: result = find_merge_base(r, commit_ids) # Return first result only if all=False if not all and result: return [result[0]] return result def is_ancestor( repo: RepoPath | None = None, ancestor: str | bytes | None = None, descendant: str | bytes | None = None, ) -> bool: """Check if one commit is an ancestor of another. Args: repo: Path to repository ancestor: Commit that might be the ancestor descendant: Commit that might be the descendant Returns: True if ancestor is an ancestor of descendant, False otherwise """ from ..graph import find_merge_base from ..objectspec import parse_object if ancestor is None or descendant is None: raise ValueError("Both ancestor and descendant are required") with open_repo_closing(repo) as r: # Resolve committish references to commit IDs ancestor_obj = parse_object(r, ancestor, config=r.get_config_stack()) if not isinstance(ancestor_obj, Commit): raise ValueError(f"Expected commit, got {ancestor_obj.type_name.decode()}") descendant_obj = parse_object(r, descendant, config=r.get_config_stack()) if not isinstance(descendant_obj, Commit): raise ValueError( f"Expected commit, got {descendant_obj.type_name.decode()}" ) # If ancestor is the merge base of (ancestor, descendant), then it's an ancestor merge_bases = find_merge_base(r, [ancestor_obj.id, descendant_obj.id]) return merge_bases == [ancestor_obj.id] def independent_commits( repo: RepoPath | None = None, committishes: Sequence[str | bytes] | None = None, ) -> list[ObjectID]: """Filter commits to only those that are not reachable from others. Args: repo: Path to repository committishes: List of commit references to filter Returns: List of commit IDs that are not ancestors of any other commits in the list """ from ..graph import independent from ..objectspec import parse_object if committishes is None or len(committishes) == 0: return [] with open_repo_closing(repo) as r: # Resolve committish references to commit IDs commit_ids = [] for committish in committishes: obj = parse_object(r, committish, config=r.get_config_stack()) if not isinstance(obj, Commit): raise ValueError(f"Expected commit, got {obj.type_name.decode()}") commit_ids.append(obj.id) # Filter to independent commits return independent(r, commit_ids) def mailsplit( input_path: str | os.PathLike[str] | IO[bytes] | None = None, output_dir: str | os.PathLike[str] = ".", start_number: int = 1, precision: int = 4, keep_cr: bool = False, mboxrd: bool = False, is_maildir: bool = False, ) -> list[str]: r"""Split an mbox file or Maildir into individual message files. This is similar to git mailsplit. Args: input_path: Path to mbox file, Maildir, or file-like object. If None, reads from stdin. output_dir: Directory where individual messages will be written start_number: Starting number for output files (default: 1) precision: Number of digits for output filenames (default: 4) keep_cr: If True, preserve \r in lines ending with \r\n (default: False) mboxrd: If True, treat input as mboxrd format and reverse escaping (default: False) is_maildir: If True, treat input_path as a Maildir (default: False) Returns: List of output file paths that were created Raises: ValueError: If output_dir doesn't exist or input is invalid OSError: If there are issues reading/writing files """ from ..mbox import split_maildir, split_mbox if is_maildir: if input_path is None: raise ValueError("input_path is required for Maildir splitting") if not isinstance(input_path, str | bytes | os.PathLike): raise ValueError("Maildir splitting requires a path, not a file object") # Convert PathLike to str for split_maildir maildir_path: str | bytes = ( os.fspath(input_path) if isinstance(input_path, os.PathLike) else input_path ) out_dir: str | bytes = ( os.fspath(output_dir) if isinstance(output_dir, os.PathLike) else output_dir ) return split_maildir( maildir_path, out_dir, start_number=start_number, precision=precision, keep_cr=keep_cr, ) else: if input_path is None: # Read from stdin input_file: str | bytes | BinaryIO = sys.stdin.buffer else: # Convert PathLike to str if needed if isinstance(input_path, os.PathLike): input_file = os.fspath(input_path) else: # input_path is either str or IO[bytes] here input_file = cast(str | BinaryIO, input_path) out_dir = ( os.fspath(output_dir) if isinstance(output_dir, os.PathLike) else output_dir ) return split_mbox( input_file, out_dir, start_number=start_number, precision=precision, keep_cr=keep_cr, mboxrd=mboxrd, ) def mailinfo( input_path: str | os.PathLike[str] | IO[bytes] | IO[str] | None = None, msg_file: str | os.PathLike[str] | None = None, patch_file: str | os.PathLike[str] | None = None, keep_subject: bool = False, keep_non_patch: bool = False, encoding: str | None = None, scissors: bool = False, message_id: bool = False, ) -> MailinfoResult: """Extract patch information from an email message. This is similar to git mailinfo. Args: input_path: Path to email file or file-like object. If None, reads from stdin. msg_file: Path to write commit message. If None, message not written to file. patch_file: Path to write patch content. If None, patch not written to file. keep_subject: If True, keep subject intact without munging (-k) keep_non_patch: If True, only strip [PATCH] from brackets (-b) encoding: Character encoding to use (default: detect from message) scissors: If True, remove everything before scissors line message_id: If True, include Message-ID in commit message (-m) Returns: MailinfoResult with parsed information Raises: ValueError: If message is malformed or missing required fields OSError: If there are issues reading/writing files Example: >>> result = mailinfo("patch.eml", "msg", "patch") >>> print(f"Author: {result.author_name} <{result.author_email}>") >>> print(f"Subject: {result.subject}") """ from ..mbox import mailinfo as mbox_mailinfo if input_path is None: # Read from stdin input_file: str | bytes | BinaryIO | TextIO = sys.stdin.buffer else: # Convert PathLike to str if needed if isinstance(input_path, os.PathLike): input_file = os.fspath(input_path) else: # input_path is either str or IO[bytes] or IO[str] here input_file = cast(str | BinaryIO | TextIO, input_path) result = mbox_mailinfo( input_file, keep_subject=keep_subject, keep_non_patch=keep_non_patch, encoding=encoding, scissors=scissors, message_id=message_id, ) # Write message to file if requested if msg_file is not None: msg_path = ( os.fspath(msg_file) if isinstance(msg_file, os.PathLike) else msg_file ) with open(msg_path, "w", encoding=encoding or "utf-8") as f: f.write(result.message) if not result.message.endswith("\n"): f.write("\n") # Write patch to file if requested if patch_file is not None: patch_path = ( os.fspath(patch_file) if isinstance(patch_file, os.PathLike) else patch_file ) with open(patch_path, "w", encoding=encoding or "utf-8") as f: f.write(result.patch) return result def rerere(repo: RepoPath | None = None) -> tuple[list[tuple[bytes, str]], list[bytes]]: """Record current conflict resolutions and apply known resolutions. This reads conflicted files from the working tree and records them in the rerere cache. If rerere.autoupdate is enabled and a known resolution exists, it will be automatically applied. Args: repo: Path to the repository Returns: Tuple of: - List of tuples (path, conflict_id) for recorded conflicts - List of paths where resolutions were automatically applied """ from ..rerere import _has_conflict_markers, rerere_auto with open_repo_closing(repo) as r: # Get conflicts from the index (if available) index = r.open_index(config=r.get_config_stack()) conflicts = [] for path, entry in index.items(): if isinstance(entry, ConflictedIndexEntry): conflicts.append(path) # Also scan working tree for files with conflict markers # This is needed because merge() doesn't always create ConflictedIndexEntry if not conflicts: working_tree = r.path for path in index: file_path = os.path.join(working_tree, os.fsdecode(path)) try: with open(file_path, "rb") as f: content = f.read() if _has_conflict_markers(content): conflicts.append(path) except (FileNotFoundError, IsADirectoryError, PermissionError): pass # Record conflicts and apply known resolutions working_tree = r.path return rerere_auto(r, working_tree, conflicts, config=r.get_config_stack()) def rerere_status(repo: RepoPath | None = None) -> list[tuple[str, bool]]: """Get the status of all conflicts in the rerere cache. Args: repo: Path to the repository Returns: List of tuples (conflict_id, has_resolution) """ from ..rerere import RerereCache with open_repo_closing(repo) as r: cache = RerereCache.from_repo(r) return cache.status() def rerere_diff( repo: RepoPath | None = None, conflict_id: str | None = None ) -> list[tuple[str, bytes, bytes | None]]: """Show differences for recorded rerere conflicts. Args: repo: Path to the repository conflict_id: Optional specific conflict ID to show Returns: List of tuples (conflict_id, preimage, postimage) """ from ..rerere import RerereCache with open_repo_closing(repo) as r: cache = RerereCache.from_repo(r) if conflict_id: preimage, postimage = cache.diff(conflict_id) if preimage is not None: return [(conflict_id, preimage, postimage)] return [] # Show all conflicts results = [] for cid, _has_res in cache.status(): preimage, postimage = cache.diff(cid) if preimage is not None: results.append((cid, preimage, postimage)) return results def rerere_forget( repo: RepoPath | None = None, pathspec: str | bytes | None = None ) -> None: """Forget recorded rerere resolutions for a pathspec. Args: repo: Path to the repository pathspec: Path to forget (currently not implemented, forgets all) """ from ..rerere import RerereCache with open_repo_closing(repo) as r: cache = RerereCache.from_repo(r) if pathspec: # TODO: Implement pathspec matching # For now, we need to track which conflict IDs correspond to which paths raise NotImplementedError("Pathspec matching not yet implemented") # Forget all conflicts (this is when called with no pathspec after resolving) cache.clear() def rerere_clear(repo: RepoPath | None = None) -> None: """Clear all recorded rerere resolutions. Args: repo: Path to the repository """ from ..rerere import RerereCache with open_repo_closing(repo) as r: cache = RerereCache.from_repo(r) cache.clear() def rerere_gc(repo: RepoPath | None = None, max_age_days: int = 60) -> None: """Garbage collect old rerere resolutions. Args: repo: Path to the repository max_age_days: Maximum age in days for keeping resolutions """ from ..rerere import RerereCache with open_repo_closing(repo) as r: cache = RerereCache.from_repo(r) cache.gc(max_age_days) def apply_patch( repo: RepoPath | None = None, patch_file: str | bytes | BinaryIO | None = None, cached: bool = False, reverse: bool = False, check: bool = False, strip: int = 1, three_way: bool = False, ) -> None: """Apply a patch to the working tree and/or index. Args: repo: Path to the repository patch_file: Path to patch file or file-like object (stdin if None) cached: Apply patch to index only, not working tree reverse: Apply patch in reverse check: Only check if patch can be applied, don't apply strip: Number of leading path components to strip (default: 1) three_way: Fall back to 3-way merge if patch does not apply cleanly Raises: ValueError: If patch cannot be applied """ with open_repo_closing(repo) as r: # Read patch content if patch_file is None: # Read from stdin import sys patch_content = sys.stdin.buffer.read() elif isinstance(patch_file, str | bytes): # Path to file if isinstance(patch_file, bytes): path = patch_file.decode("utf-8") else: path = patch_file with open(path, "rb") as f: patch_content = f.read() else: # File-like object patch_content = patch_file.read() if isinstance(patch_content, str): patch_content = patch_content.encode("utf-8") # Parse the patch patches = parse_unified_diff(patch_content) if not patches: raise ValueError("No patches found in input") # Apply patches apply_file_patches( r, patches, cached=cached, reverse=reverse, check=check, strip=strip, three_way=three_way, ) def am( repo: RepoPath | None = None, patches: str | bytes | BinaryIO | list[str | bytes | BinaryIO] | None = None, three_way: bool = False, keep_subject: bool = False, keep_non_patch: bool = False, scissors: bool = False, message_id: bool = False, strip: int = 1, committer: bytes | None = None, commit_timestamp: float | None = None, commit_timezone: int | None = None, ) -> list[ObjectID]: """Apply patches from mailbox-style email messages, creating commits. Args: repo: Path to the repository patches: Patch input(s) - file path(s), file-like object(s), or None for stdin. Can be a single mbox file containing multiple messages. three_way: Fall back to 3-way merge if patch does not apply cleanly keep_subject: If True, keep subject intact without munging keep_non_patch: If True, only strip [PATCH] from brackets scissors: If True, remove everything before scissors line message_id: If True, include Message-ID in commit message strip: Number of leading path components to strip (default: 1) committer: Optional committer identity (bytes) commit_timestamp: Optional committer timestamp commit_timezone: Optional committer timezone offset Returns: List of commit SHAs (bytes) created """ import email.message import email.parser import mailbox import tempfile from ..am import am as am_impl # Normalize input to a list if patches is None: import sys inputs: list[str | bytes | BinaryIO] = [sys.stdin.buffer] elif isinstance(patches, list): inputs = patches else: inputs = [patches] # Collect all email messages msgs: list[email.message.Message] = [] parser = email.parser.BytesParser() for inp in inputs: # Read content if isinstance(inp, str | bytes): if isinstance(inp, str): path = inp else: path = inp.decode("utf-8") with open(path, "rb") as f: content = f.read() else: content = inp.read() if isinstance(content, str): content = content.encode("utf-8") # Detect mbox format (multiple messages starting with "From ") if content.startswith(b"From "): # Parse as mbox using mailbox module with tempfile.NamedTemporaryFile(delete=False) as tmp: tmp.write(content) tmp_path = tmp.name try: mbox = mailbox.mbox(tmp_path) try: for mbox_msg in mbox: msgs.append(mbox_msg) finally: mbox.close() finally: import os os.unlink(tmp_path) else: # Parse as single email message msg = parser.parsebytes(content) msgs.append(msg) with open_repo_closing(repo) as r: return am_impl( r, msgs, three_way=three_way, keep_subject=keep_subject, keep_non_patch=keep_non_patch, scissors=scissors, message_id=message_id, strip=strip, committer=committer, commit_timestamp=commit_timestamp, commit_timezone=commit_timezone, ) def am_continue( repo: RepoPath | None = None, committer: bytes | None = None, commit_timestamp: float | None = None, commit_timezone: int | None = None, ) -> list[ObjectID]: """Continue applying patches after resolving a conflict. The user should have resolved conflicts and staged the result. Args: repo: Path to the repository committer: Optional committer identity commit_timestamp: Optional committer timestamp commit_timezone: Optional committer timezone offset Returns: List of commit SHAs created """ from ..am import am_continue as am_continue_impl with open_repo_closing(repo) as r: return am_continue_impl( r, committer=committer, commit_timestamp=commit_timestamp, commit_timezone=commit_timezone, ) def am_skip( repo: RepoPath | None = None, committer: bytes | None = None, commit_timestamp: float | None = None, commit_timezone: int | None = None, ) -> list[ObjectID]: """Skip the current patch and continue with remaining patches. Args: repo: Path to the repository committer: Optional committer identity commit_timestamp: Optional committer timestamp commit_timezone: Optional committer timezone offset Returns: List of commit SHAs created """ from ..am import am_skip as am_skip_impl with open_repo_closing(repo) as r: return am_skip_impl( r, committer=committer, commit_timestamp=commit_timestamp, commit_timezone=commit_timezone, ) def am_abort(repo: RepoPath | None = None) -> None: """Abort the current am and restore the original state. Resets HEAD, index, and working tree to the state before am started. Args: repo: Path to the repository """ from ..am import am_abort as am_abort_impl with open_repo_closing(repo) as r: am_abort_impl(r) def am_quit(repo: RepoPath | None = None) -> None: """Quit the current am without reverting changes. Removes am state but keeps HEAD, index, and working tree as-is. Args: repo: Path to the repository """ from ..am import am_quit as am_quit_impl with open_repo_closing(repo) as r: am_quit_impl(r) def __getattr__(name: str) -> object: if name == "get_user_identity": warnings.warn( "dulwich.porcelain.get_user_identity is deprecated; " "use dulwich.repo.get_user_identity instead.", DeprecationWarning, stacklevel=2, ) from ..repo import get_user_identity return get_user_identity raise AttributeError(f"module {__name__!r} has no attribute {name!r}")