libvcs.sync.base

Foundation for the sync classes. Add your own VCS — or extend libvcs — by subclassing BaseSync.

Foundational tools to set up a VCS manager in libvcs.sync.

class libvcs.sync.base.SyncError
class[source]
class[source]
class libvcs.sync.base.SyncError

Bases: object

An error encountered during a sync step.

Examples

>>> error = SyncError(step="fetch", message="remote not found")
>>> error.step
'fetch'
>>> error.message
'remote not found'
>>> error.exception is None
True
step: str
attribute
attribute
step: str
message: str
attribute
attribute
message: str
exception: Exception | None = None
attribute
attribute
exception: Exception | None = None
__init__(step, message, exception=None)
method[source]
method[source]
__init__(step, message, exception=None)
Parameters:
Return type:

None

class libvcs.sync.base.SyncResult
class[source]
class[source]
class libvcs.sync.base.SyncResult

Bases: object

Result of a repository synchronization.

Examples

>>> result = SyncResult()
>>> result.ok
True
>>> bool(result)
True

>>> result = SyncResult()
>>> result.add_error(step="fetch", message="remote not found")
>>> result.ok
False
>>> bool(result)
False
>>> result.errors[0].step
'fetch'
ok: bool = True
attribute
attribute
ok: bool = True
errors: list[SyncError]
attribute
attribute
errors: list[SyncError]
add_error(step, message, exception=None)
method[source]
method[source]
add_error(step, message, exception=None)

Record an error and mark the result as failed.

Parameters:

  • step (str) – Name of the sync step that failed (e.g. "fetch", "checkout").

  • message (str) – Human-readable description of the error.

  • exception (Exception or None, optional) – The underlying exception, if available.

Return type:

None

__init__(ok=True, errors=[])
method[source]
method[source]
__init__(ok=True, errors=[])
Parameters:
Return type:

None

class libvcs.sync.base.VCSLocation
class[source]
class[source]
class libvcs.sync.base.VCSLocation

Bases: NamedTuple

Generic VCS Location (URL and optional revision).

url: str
attribute
attribute
url: str

Alias for field number 0

rev: str | None
attribute
attribute
rev: str | None

Alias for field number 1

static __new__(_cls, url, rev)
staticmethod method
staticmethod method
static __new__(_cls, url, rev)

Create new instance of VCSLocation(url, rev)

Parameters:
_asdict()
method[source]
method[source]
_asdict()

Return a new dict which maps field names to their values.

_field_defaults = {}
attribute
attribute
_field_defaults = {}
_fields = ('url', 'rev')
attribute
attribute
_fields = ('url', 'rev')
classmethod _make(iterable)
classmethod method[source]
classmethod method[source]
classmethod _make(iterable)

Make a new VCSLocation object from a sequence or iterable

_replace(**kwds)
method[source]
method[source]
_replace(**kwds)

Return a new VCSLocation object replacing specified fields with new values

libvcs.sync.base.convert_pip_url(pip_url)
function[source]
function[source]
libvcs.sync.base.convert_pip_url(pip_url)

Parse pip URL via libvcs.sync.base.BaseSync.url.

Parameters:

pip_url (str)

Return type:

VCSLocation

class libvcs.sync.base.BaseSync
class[source]
class[source]
class libvcs.sync.base.BaseSync

Bases: object

Base class for repositories.

log_in_real_time = None
attribute
attribute
log_in_real_time = None

Log command output to buffer

bin_name: str = ''
attribute
attribute
bin_name: str = ''

VCS app name, e.g. ‘git’

schemes: tuple[str, ...] = ()
attribute
attribute
schemes: tuple[str, ...] = ()

List of supported schemes to register in urlparse.uses_netloc

__init__(*, url, path, progress_callback=None, **kwargs)
method[source]
method[source]
__init__(*, url, path, progress_callback=None, **kwargs)

Initialize a tool to manage a local VCS Checkout, Clone, Copy, or Work tree.

Parameters:

  • progress_callback (func) –

    Retrieve live progress from sys.stderr (useful for certain vcs commands like git pull. Use progress_callback:

    >>> import os
>>> import sys
>>> def progress_cb(output, timestamp):
...     sys.stdout.write(output)
...     sys.stdout.flush()
>>> class Project(BaseSync):
...     bin_name = 'git'
...     def obtain(self, *args, **kwargs):
...         self.ensure_dir()
...         self.run(
...             ['clone', '--progress', self.url, self.path],
...             log_in_real_time=True
...         )
>>> r = Project(
...     url=f'file://{create_git_remote_repo()}',
...     path=str(tmp_path),
...     progress_callback=progress_cb
... )
>>> r.obtain()
Cloning into '...'...
remote: Enumerating objects: ...
remote: Counting objects: ...% (...)...
...
remote: Total ... (delta 0), reused 0 (delta 0), pack-reused 0
...
Receiving objects: ...% (...)...
...
>>> assert r.path.exists()
>>> assert pathlib.Path(r.path / '.git').exists()

  • url (str)

  • path (StrPath)

  • kwargs (Any)

Return type:

None

progress_callback
attribute
attribute
progress_callback

Callback for run updates

path: Path
attribute
attribute
path: Path

Directory to check out

log: CmdLoggingAdapter
attribute
attribute
log: CmdLoggingAdapter

Logging attribute

property repo_name: str
property
property
property repo_name: str

Return the short name of a repo checkout.

classmethod from_pip_url(pip_url, **kwargs)
classmethod method[source]
classmethod method[source]
classmethod from_pip_url(pip_url, **kwargs)

Create synchronization object from pip-style URL.

Parameters:
Return type:

BaseSync

run(cmd, cwd=None, check_returncode=True, log_in_real_time=None, *args, **kwargs)
method[source]
method[source]
run(cmd, cwd=None, check_returncode=True, log_in_real_time=None, *args, **kwargs)

Return combined stderr/stdout from a command.

This method will also prefix the VCS command bin_name. By default runs using the cwd libvcs.sync.base.BaseSync.path of the repo.

Parameters:

  • cwd (str) – dir command is run from, defaults to libvcs.sync.base.BaseSync.path.

  • check_returncode (bool) – Indicate whether a CommandError should be raised if return code is different from 0.

  • cmd (_CMD)

  • log_in_real_time (bool | None)

  • args (Any)

  • kwargs (Any)

Returns:

combined stdout/stderr in a big string, newlines retained

Return type:

str

ensure_dir(*args, **kwargs)
method[source]
method[source]
ensure_dir(*args, **kwargs)

Assure destination path exists. If not, create directories.

Parameters:
Return type:

bool

update_repo(*args, **kwargs)
method[source]
method[source]
update_repo(*args, **kwargs)

Pull latest changes to here from remote repository.

Returns:

Result of the sync operation, with any errors recorded.

Return type:

SyncResult

Parameters:
obtain(*args, **kwargs)
method[source]
method[source]
obtain(*args, **kwargs)

Checkout initial VCS repository or working copy from remote repository.

Parameters:
Return type:

None