Worktree sync - `vcspull._internal.worktree_sync`¶

Warning

Be careful with these! Internal APIs are not covered by version policies. They can break or be removed between minor versions!

If you need an internal API stabilized please file an issue.

Core worktree synchronization logic for vcspull.

class vcspull._internal.worktree_sync.WorktreeAction[source]¶

Bases: Enum

Actions that can be taken on a worktree during sync.

CREATE = 'create'¶: Worktree doesn’t exist, will be created.

UPDATE = 'update'¶: Branch worktree exists, will pull latest.

UNCHANGED = 'unchanged'¶: Tag/commit worktree exists, already at target.

BLOCKED = 'blocked'¶: Worktree has uncommitted changes (safety).

ERROR = 'error'¶: Operation failed (ref not found, permission, etc.).

class vcspull._internal.worktree_sync.WorktreeCheck[source]¶

Bases: object

A single check performed during worktree planning.

name: str¶

‘validate_config’, ‘ref_exists’, ‘worktree_exists’, ‘is_dirty’.

Type:: Check name

passed: bool¶: Whether the check passed.

detail: str¶: Human-readable description of the check result.

exception: WorktreeError | None = None¶: Typed exception if the check failed.

__init__(name, passed, detail, exception=None)[source]¶

Parameters:

name (str)
passed (bool)
detail (str)
exception (WorktreeError | None)

Return type:

None

class vcspull._internal.worktree_sync.WorktreePlanEntry[source]¶

Bases: object

Planning information for a single worktree operation.

worktree_path: Path¶: Absolute path where the worktree will be/is located.

ref_type: str¶

‘tag’, ‘branch’, or ‘commit’.

Type:: Type of reference

ref_value: str¶: The actual tag name, branch name, or commit SHA.

action: WorktreeAction¶: What action will be/was taken.

detail: str | None = None¶: Human-readable explanation of the action.

error: str | None = None¶: Error message if action is ERROR.

exists: bool = False¶: Whether the worktree currently exists.

is_dirty: bool = False¶: Whether the worktree has uncommitted changes.

current_ref: str | None = None¶: Current HEAD reference if worktree exists.

checks: list[WorktreeCheck]¶: Ordered audit trail of checks performed during planning.

__init__(worktree_path, ref_type, ref_value, action, detail=None, error=None, exists=False, is_dirty=False, current_ref=None, checks=<factory>)[source]¶

Parameters:

worktree_path (Path)
ref_type (str)
ref_value (str)
action (WorktreeAction)
detail (str | None)
error (str | None)
exists (bool)
is_dirty (bool)
current_ref (str | None)
checks (list[WorktreeCheck])

Return type:

None

class vcspull._internal.worktree_sync.WorktreeSyncResult[source]¶

Bases: object

Result of a worktree sync operation.

__init__(entries=<factory>, created=0, updated=0, unchanged=0, blocked=0, errors=0)[source]¶

Parameters:

entries (list[WorktreePlanEntry])
created (int)
updated (int)
unchanged (int)
blocked (int)
errors (int)

Return type:

None

entries: list[WorktreePlanEntry]¶: List of worktree plan entries.

created: int = 0¶: Number of worktrees created.

updated: int = 0¶: Number of worktrees updated.

unchanged: int = 0¶: Number of worktrees left unchanged.

blocked: int = 0¶: Number of worktrees blocked due to dirty state.

errors: int = 0¶: Number of worktrees that encountered errors.

vcspull._internal.worktree_sync._get_ref_type_and_value(wt_config)[source]¶

Extract the reference type and value from worktree config.

Return type:: tuple[str, str] | None
Returns:: Tuple of (ref_type, ref_value) or None if invalid config.
Return type:: tuple[str, str] | None
Parameters:: wt_config (WorktreeConfigDict)

Examples

>>> _get_ref_type_and_value({"dir": "../v1", "tag": "v1.0.0"})
('tag', 'v1.0.0')
>>> _get_ref_type_and_value({"dir": "../dev", "branch": "develop"})
('branch', 'develop')
>>> _get_ref_type_and_value({"dir": "../abc", "commit": "abc123"})
('commit', 'abc123')
>>> _get_ref_type_and_value({"dir": "../empty"}) is None
True
>>> multi = {"dir": "../multi", "tag": "v1", "branch": "main"}
>>> _get_ref_type_and_value(multi) is None
True
>>> _get_ref_type_and_value({"dir": "../wt", "tag": ""}) is None
True

vcspull._internal.worktree_sync.validate_worktree_config(wt_config)[source]¶

Validate a worktree configuration dictionary.

Return type:: None
Parameters:: wt_config (WorktreeConfigDict) – The worktree configuration to validate.
Raises:: WorktreeConfigError – If the configuration is invalid.

Examples

>>> validate_worktree_config({"dir": "../v1", "tag": "v1.0.0"})
>>> validate_worktree_config({"dir": "../dev", "branch": "develop"})
>>> validate_worktree_config({"tag": "v1.0.0"})  # Missing dir
Traceback (most recent call last):
    ...
vcspull.exc.WorktreeConfigError: Worktree config missing required 'dir' field
>>> validate_worktree_config({"dir": "../proj"})  # No ref
Traceback (most recent call last):
    ...
vcspull.exc.WorktreeConfigError: Worktree config must specify one of: ...

vcspull._internal.worktree_sync._is_worktree_dirty(worktree_path)[source]¶

Check if a worktree has uncommitted changes.

Returns True if dirty or if the check fails (fail-safe). This prevents destructive operations when the dirty state is unknown.

Return type:: bool
Parameters:: worktree_path (pathlib.Path) – Path to the worktree directory.
Returns:: True if the worktree has uncommitted changes or if the check fails.
Return type:: bool

Examples

>>> import pathlib
>>> _is_worktree_dirty(pathlib.Path("/nonexistent/path"))
True

vcspull._internal.worktree_sync._ref_exists(repo_path, ref, ref_type)[source]¶

Check if a reference exists in the repository.

Return type:

bool

Parameters:

repo_path (pathlib.Path) – Path to the main repository.
ref (str) – The reference to check.
ref_type (str) – Type of reference: ‘tag’, ‘branch’, or ‘commit’.

Returns:

True if the reference exists.

Return type:

bool

Examples

>>> import pathlib
>>> _ref_exists(pathlib.Path("/nonexistent/repo"), "v1.0.0", "tag")
False
>>> _ref_exists(pathlib.Path("/nonexistent/repo"), "main", "branch")
False

vcspull._internal.worktree_sync._get_worktree_head(worktree_path)[source]¶

Get the current HEAD reference of a worktree.

Return type:: str | None
Parameters:: worktree_path (pathlib.Path) – Path to the worktree.
Returns:: The HEAD reference or None if unable to determine.
Return type:: str | None

Examples

>>> import pathlib
>>> _get_worktree_head(pathlib.Path("/nonexistent/worktree")) is None
True

vcspull._internal.worktree_sync._worktree_exists(repo_path, worktree_path)[source]¶

Check if a worktree is registered in the repository.

Return type:

bool

Parameters:

repo_path (pathlib.Path) – Path to the main repository.
worktree_path (pathlib.Path) – Path to check for worktree.

Returns:

True if the worktree exists and is registered.

Return type:

bool

Examples

>>> import pathlib
>>> _worktree_exists(pathlib.Path("/repo"), pathlib.Path("/nonexistent"))
False
>>> repo_dir = tmp_path / "repo"
>>> repo_dir.mkdir()
>>> _worktree_exists(repo_dir, tmp_path / "missing")
False

vcspull._internal.worktree_sync._resolve_worktree_path(wt_config, workspace_root)[source]¶

Resolve the worktree path from config.

Return type:

Path

Parameters:

wt_config (WorktreeConfigDict) – Worktree configuration.
workspace_root (pathlib.Path) – The workspace root directory.

Returns:

Absolute path for the worktree.

Return type:

pathlib.Path

Examples

>>> import pathlib
>>> workspace = pathlib.Path("/home/user/code")
>>> wt = {"dir": "../sibling", "tag": "v1.0.0"}
>>> _resolve_worktree_path(wt, workspace)
PosixPath('/home/user/sibling')
>>> wt_abs = {"dir": "/tmp/worktree", "tag": "v1.0.0"}
>>> _resolve_worktree_path(wt_abs, workspace)
PosixPath('/tmp/worktree')

vcspull._internal.worktree_sync.plan_worktree_sync(repo_path, worktrees_config, workspace_root)[source]¶

Plan worktree sync operations without executing them.

Return type:

list[WorktreePlanEntry]

Parameters:

repo_path (pathlib.Path) – Path to the main repository.
worktrees_config (list[WorktreeConfigDict]) – List of worktree configurations.
workspace_root (pathlib.Path) – The workspace root directory for resolving relative paths.

Returns:

List of planned operations.

Return type:

list[WorktreePlanEntry]

Examples

>>> import pathlib
>>> entries = plan_worktree_sync(
...     pathlib.Path("/nonexistent/repo"),
...     [{"dir": "../wt", "tag": "v1.0.0"}],
...     pathlib.Path("/nonexistent"),
... )
>>> len(entries)
1
>>> entries[0].action == WorktreeAction.ERROR
True

vcspull._internal.worktree_sync.sync_worktree(repo_path, wt_config, workspace_root, *, dry_run=False)[source]¶

Synchronize a single worktree.

Return type:

WorktreePlanEntry

Parameters:

repo_path (pathlib.Path) – Path to the main repository.
wt_config (WorktreeConfigDict) – Worktree configuration.
workspace_root (pathlib.Path) – The workspace root directory.
dry_run (bool) – If True, only plan without executing.

Returns:

Result of the sync operation.

Return type:

WorktreePlanEntry

Examples

>>> import pathlib
>>> entry = sync_worktree(
...     pathlib.Path("/nonexistent/repo"),
...     {"dir": "../wt", "tag": "v1.0.0"},
...     pathlib.Path("/nonexistent"),
...     dry_run=True,
... )
>>> entry.action == WorktreeAction.ERROR
True

vcspull._internal.worktree_sync._create_worktree(repo_path, worktree_path, ref_type, ref_value, wt_config)[source]¶

Create a new worktree.

Return type:

None

Parameters:

repo_path (pathlib.Path) – Path to the main repository.
worktree_path (pathlib.Path) – Path for the new worktree.
ref_type (str) – Type of reference: ‘tag’, ‘branch’, or ‘commit’.
ref_value (str) – The reference value.
wt_config (WorktreeConfigDict) – Full worktree configuration.

Raises:

subprocess.CalledProcessError – If the git command fails (e.g., ref not found).
FileNotFoundError – If the repository path does not exist.

Examples

This function requires a valid git repository. When called with an invalid path, it raises FileNotFoundError:

>>> import pathlib
>>> _create_worktree(
...     pathlib.Path("/nonexistent"),
...     pathlib.Path("/tmp/wt"),
...     "tag",
...     "v1.0.0",
...     {"dir": "../wt", "tag": "v1.0.0"},
... )
Traceback (most recent call last):
    ...
FileNotFoundError: ...

vcspull._internal.worktree_sync._update_worktree(worktree_path, branch)[source]¶

Update a branch worktree by pulling latest changes.

Verifies the worktree is on the expected branch before pulling. If the worktree is on a different branch, checks out the expected branch first.

Return type:

None

Parameters:

worktree_path (pathlib.Path) – Path to the worktree.
branch (str) – The expected branch name.

Raises:

subprocess.CalledProcessError – If the git command fails.
FileNotFoundError – If the worktree path does not exist.

Examples

This function requires a valid git worktree. When called with an invalid path, it raises FileNotFoundError:

>>> import pathlib
>>> _update_worktree(pathlib.Path("/nonexistent"), "main")
Traceback (most recent call last):
    ...
FileNotFoundError: ...

vcspull._internal.worktree_sync.sync_all_worktrees(repo_path, worktrees_config, workspace_root, *, dry_run=False)[source]¶

Synchronize all worktrees for a repository.

Return type:

WorktreeSyncResult

Parameters:

repo_path (pathlib.Path) – Path to the main repository.
worktrees_config (list[WorktreeConfigDict]) – List of worktree configurations.
workspace_root (pathlib.Path) – The workspace root directory.
dry_run (bool) – If True, only plan without executing.

Returns:

Summary of all sync operations.

Return type:

WorktreeSyncResult

Examples

>>> import pathlib
>>> result = sync_all_worktrees(
...     pathlib.Path("/nonexistent/repo"),
...     [{"dir": "../wt", "tag": "v1.0.0"}],
...     pathlib.Path("/nonexistent"),
...     dry_run=True,
... )
>>> result.errors
1
>>> len(result.entries)
1

vcspull._internal.worktree_sync.list_existing_worktrees(repo_path)[source]¶

List all existing worktrees for a repository.

Return type:: list[Path]
Parameters:: repo_path (pathlib.Path) – Path to the main repository.
Returns:: List of worktree paths.
Return type:: list[pathlib.Path]

Examples

>>> import pathlib
>>> list_existing_worktrees(pathlib.Path("/nonexistent/repo"))
[]

vcspull._internal.worktree_sync.prune_worktrees(repo_path, config_worktrees, workspace_root, *, dry_run=False)[source]¶

Remove worktrees that are not in the configuration.

Return type:

list[Path]

Parameters:

repo_path (pathlib.Path) – Path to the main repository.
config_worktrees (list[WorktreeConfigDict]) – List of configured worktrees.
workspace_root (pathlib.Path) – The workspace root directory.
dry_run (bool) – If True, only report what would be pruned.

Returns:

List of worktree paths that were (or would be) pruned.

Return type:

list[pathlib.Path]

Examples

>>> import pathlib
>>> prune_worktrees(
...     pathlib.Path("/nonexistent/repo"),
...     [],
...     pathlib.Path("/nonexistent"),
...     dry_run=True,
... )
[]

Worktree sync - vcspull._internal.worktree_sync¶

Worktree sync - `vcspull._internal.worktree_sync`¶