sync

Low-level functionality for synchronizing across different machines

abspath_from_uri(uri)

Extract and unquote the path component of a URI to turn it into an unambiguous absolute pathlib.Path

h/t https://stackoverflow.com/a/61922504

Parameters:

Name	Type	Description	Default
uri	ParseResult	The parsed URI to extract the path from	required

Returns:

Type	Description
Path	The path part of the URI as a Path

Source code in enderchest/sync/utils.py

def abspath_from_uri(uri: ParseResult) -> Path:
    """Extract and unquote the path component of a URI to turn it into an
    unambiguous absolute `pathlib.Path`

    h/t https://stackoverflow.com/a/61922504

    Parameters
    ----------
    uri : ParseResult
        The parsed URI to extract the path from

    Returns
    -------
    Path
        The path part of the URI as a Path
    """
    host = "{0}{0}{mnt}{0}".format(os.path.sep, mnt=uri.netloc)
    return Path(os.path.abspath(os.path.join(host, url2pathname(unquote(uri.path)))))

get_default_netloc()

Compile a netloc from environment variables, etc.

Returns:

Type	Description
str	The default netloc, which is {user}@{hostname}

Source code in enderchest/sync/utils.py

def get_default_netloc() -> str:
    """Compile a netloc from environment variables, etc.

    Returns
    -------
    str
        The default netloc, which is {user}@{hostname}
    """
    return f"{getpass.getuser()}@{socket.gethostname()}".lower()

pull(remote_uri, local_path, exclude=None, dry_run=False, **kwargs)

Pull all upstream changes from a remote into the specified location

Parameters:

Name	Type	Description	Default
remote_uri	ParseResult	The URI for the remote resource to pull	required
local_path	Path	The local destination	required
exclude	list of str	Any patterns that should be excluded from the sync	None
dry_run	bool	Whether to only simulate this sync (report the operations to be performed but not actually perform them). Default is False.	False
**kwargs		Any additional options to pass into the sync command	{}

Source code in enderchest/sync/__init__.py

def pull(
    remote_uri: ParseResult,
    local_path: Path,
    exclude: Collection[str] | None = None,
    dry_run: bool = False,
    **kwargs,
) -> None:
    """Pull all upstream changes from a remote into the specified location

    Parameters
    ----------
    remote_uri : ParseResult
        The URI for the remote resource to pull
    local_path : Path
        The local destination
    exclude : list of str, optional
        Any patterns that should be excluded from the sync
    dry_run : bool, optional
        Whether to only simulate this sync (report the operations to be performed
        but not actually perform them). Default is False.
    **kwargs
        Any additional options to pass into the sync command
    """
    try:
        protocol = importlib.import_module(f"{__package__}.{remote_uri.scheme.lower()}")
        protocol.pull(remote_uri, local_path, exclude or (), dry_run, **kwargs)
    except ModuleNotFoundError as not_installed:  # pragma: no cover
        raise NotImplementedError(
            f"Protocol {remote_uri.scheme} is not currently implemented"
        ) from not_installed

push(local_path, remote_uri, exclude=None, dry_run=False, **kwargs)

Push all local changes in the specified directory into the specified remote

Parameters:

Name	Type	Description	Default
local_path	Path	The local path to push	required
remote_uri	ParseResult	The URI for the remote destination	required
exclude	list of str	Any patterns that should be excluded from the sync	None
dry_run	bool	Whether to only simulate this sync (report the operations to be performed but not actually perform them). Default is False.	False
**kwargs		Any additional options to pass into the sync command	{}

Source code in enderchest/sync/__init__.py

def push(
    local_path: Path,
    remote_uri: ParseResult,
    exclude: Collection[str] | None = None,
    dry_run: bool = False,
    **kwargs,
) -> None:
    """Push all local changes in the specified directory into the specified remote

    Parameters
    ----------
    local_path
        The local path to push
    remote_uri : ParseResult
        The URI for the remote destination
    exclude : list of str, optional
        Any patterns that should be excluded from the sync
    dry_run : bool, optional
        Whether to only simulate this sync (report the operations to be performed
        but not actually perform them). Default is False.
    **kwargs
        Any additional options to pass into the sync command
    """
    try:
        protocol = importlib.import_module(f"{__package__}.{remote_uri.scheme.lower()}")
        protocol.push(local_path, remote_uri, exclude or (), dry_run, **kwargs)
    except ModuleNotFoundError as not_installed:
        raise NotImplementedError(
            f"Protocol {remote_uri.scheme} is not currently implemented"
        ) from not_installed

remote_file(uri)

Grab a file from a remote filesystem by its URI and read its contents

Parameters:

Name	Type	Description	Default
uri	parsed URI	The URI of the file to read	required

Yields:

Type	Description
Path	A path to a local (temp) copy of the file

Source code in enderchest/sync/__init__.py

@contextmanager
def remote_file(uri: ParseResult) -> Generator[Path, None, None]:
    """Grab a file from a remote filesystem by its URI and read its contents

    Parameters
    ----------
    uri : parsed URI
        The URI of the file to read

    Yields
    ------
    Path
        A path to a local (temp) copy of the file
    """
    with TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
        pull(uri, Path(tmpdir), verbosity=-1)
        yield Path(tmpdir) / Path(uri.path).name

render_remote(alias, uri)

Render a remote to a descriptive string

Parameters:

Name	Type	Description	Default
alias	str	The name of the remote	required
uri	ParseResult	The parsed URI for the remote	required

Returns:

Type	Description
str	{uri_string} [({alias})]} (alias is omitted if it's the same as the URI's hostname)

Source code in enderchest/sync/utils.py

def render_remote(alias: str, uri: ParseResult) -> str:
    """Render a remote to a descriptive string

    Parameters
    ----------
    alias : str
        The name of the remote
    uri : ParseResult
        The parsed URI for the remote

    Returns
    -------
    str
        `{uri_string} [({alias})]}`
            (alias is omitted if it's the same as the URI's hostname)
    """
    uri_string = uri.geturl()

    if uri.hostname != alias:
        uri_string += f" ({alias})"
    return uri_string

uri_to_ssh(uri)

Convert a URI to an SSH address

Parameters:

Name	Type	Description	Default
uri	ParseResult	The URI to convert	required

Returns:

Type	Description
str	The SSH-format address

Source code in enderchest/sync/utils.py

def uri_to_ssh(uri: ParseResult) -> str:
    """Convert a URI to an SSH address

    Parameters
    ----------
    uri: ParseResult
        The URI to convert

    Returns
    -------
    str
        The SSH-format address
    """
    return "{user}{host}:{path}".format(
        user=f"{uri.username}@" if uri.username else "",
        host=(uri.hostname or "localhost") + (f":{uri.port}" if uri.port else ""),
        path=uri.path,
    )