libvcs.sync.base
#
Base objects / classes for projects.
Adding your own VCS / Extending libvcs can be done through subclassing BaseSync
.
Foundational tools to set up a VCS manager in libvcs.sync.
- class libvcs.sync.base.VCSLocation(url, rev)[source]#
Bases:
NamedTuple
Generic VCS Location (URL and optional revision).
Create new instance of VCSLocation(url, rev)
- _field_defaults = {}#
- _fields = ('url', 'rev')#
- libvcs.sync.base.convert_pip_url(pip_url)[source]#
Parse pip URL via libvcs.sync.base.BaseSync.url.
- Return type:
- Parameters:
pip_url (str) β
- class libvcs.sync.base.BaseSync(*, url, dir, progress_callback=None, **kwargs)[source]#
Bases:
object
Base class for repositories.
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 likegit pull
. Useprogress_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.dir], ... log_in_real_time=True ... ) >>> r = Project( ... url=f'file://{create_git_remote_repo()}', ... dir=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.dir.exists() >>> assert pathlib.Path(r.dir / '.git').exists()
url (str) β
kwargs (Any) β
- log_in_real_time = None#
Log command output to buffer
- __init__(*, url, dir, progress_callback=None, **kwargs)[source]#
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 likegit pull
. Useprogress_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.dir], ... log_in_real_time=True ... ) >>> r = Project( ... url=f'file://{create_git_remote_repo()}', ... dir=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.dir.exists() >>> assert pathlib.Path(r.dir / '.git').exists()
url (str) β
kwargs (Any) β
- Return type:
None
- progress_callback#
Callback for run updates
-
log:
CmdLoggingAdapter
# Logging attribute
- classmethod from_pip_url(pip_url, **kwargs)[source]#
Create synchronization object from pip-style URL.
- run(cmd, cwd=None, check_returncode=True, log_in_real_time=None, *args, **kwargs)[source]#
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.dir of the repo.
- Parameters:
cwd (str) β dir command is run from, defaults to libvcs.sync.base.BaseSync.dir.
check_returncode (bool) β Indicate whether a
CommandError
should be raised if return code is different from 0.cmd (str | bytes | PathLike[str] | PathLike[bytes] | Sequence[str | bytes | PathLike[str] | PathLike[bytes]]) β
log_in_real_time (bool | None) β
args (Any) β
kwargs (Any) β
- Returns:
combined stdout/stderr in a big string, newlines retained
- Return type: