data_transfer.robocopy

RobocopySettings

Bases: ServiceSettings

Settings for the RobocopyService.

Attributes:

Name	Type	Description
destination	PathLike	The destination path for the data transfer.
log	Optional[PathLike]	The path to the log file.
extra_args	str	Extra arguments to pass to Robocopy.
delete_src	bool	Whether to delete the source files after copying.
overwrite	bool	Whether to overwrite existing files.
force_dir	bool	Whether to create the destination directory if it does not exist.

settings_customise_sources classmethod

settings_customise_sources(
    settings_cls: Type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> Tuple[PydanticBaseSettingsSource, ...]

Customizes the settings sources to include the safe YAML settings source.

Parameters:

Name	Type	Description	Default
settings_cls	Type[BaseSettings]	The settings class.	required
init_settings	PydanticBaseSettingsSource	The initial settings source.	required
env_settings	PydanticBaseSettingsSource	The environment settings source.	required
dotenv_settings	PydanticBaseSettingsSource	The dotenv settings source.	required
file_secret_settings	PydanticBaseSettingsSource	The file secret settings source.	required

Returns:

Type	Description
Tuple[PydanticBaseSettingsSource, ...]	A tuple of settings sources.

Source code in src/clabe/services.py

@classmethod
def settings_customise_sources(
    cls,
    settings_cls: Type[ps.BaseSettings],
    init_settings: ps.PydanticBaseSettingsSource,
    env_settings: ps.PydanticBaseSettingsSource,
    dotenv_settings: ps.PydanticBaseSettingsSource,
    file_secret_settings: ps.PydanticBaseSettingsSource,
) -> t.Tuple[ps.PydanticBaseSettingsSource, ...]:
    """
    Customizes the settings sources to include the safe YAML settings source.

    Args:
        settings_cls: The settings class.
        init_settings: The initial settings source.
        env_settings: The environment settings source.
        dotenv_settings: The dotenv settings source.
        file_secret_settings: The file secret settings source.

    Returns:
        A tuple of settings sources.
    """
    return (
        init_settings,
        _SafeYamlSettingsSource(settings_cls),
        env_settings,
        dotenv_settings,
        file_secret_settings,
    )

RobocopyService

RobocopyService(
    source: PathLike,
    settings: RobocopySettings,
    *,
    ui_helper: Optional[UiHelper] = None,
)

Bases: DataTransfer[RobocopySettings]

A data transfer service that uses the Robocopy command-line utility to copy files between source and destination directories.

This service provides a wrapper around the Windows Robocopy utility with configurable options for file copying, logging, and directory management.

Attributes:

Name	Type	Description
source	PathLike	Source directory or file path
_settings	RobocopySettings	Service settings containing destination, options, etc.
_ui_helper	UiHelper	UI helper for user prompts

Example

# Basic file copying:
settings = RobocopySettings(destination="D:/backup/experiment1")
service = RobocopyService(
    source="C:/data/experiment1",
    settings=settings
)
service.transfer()

# Copy with custom options:
settings = RobocopySettings(
    destination="D:/backup/experiment1",
    delete_src=True,
    overwrite=True,
    log="copy_log.txt",
    extra_args="/E /DCOPY:DAT /R:50 /W:5"
)
service = RobocopyService(
    source="C:/data/experiment1",
    settings=settings
)
if service.validate():
    service.transfer()

Initializes the RobocopyService.

Parameters:

Name	Type	Description	Default
source	PathLike	The source directory or file to copy	required
settings	RobocopySettings	RobocopySettings containing destination and options	required
ui_helper	Optional[UiHelper]	UI helper for user prompts. Default is None	None

Example

# Initialize with basic parameters:
settings = RobocopySettings(destination="D:/destination")
service = RobocopyService("C:/source", settings)

# Initialize with logging and move operation:
settings = RobocopySettings(
    destination="D:/archive/data",
    log="transfer.log",
    delete_src=True,
    extra_args="/E /COPY:DAT /R:10"
)
service = RobocopyService("C:/temp/data", settings)

Source code in src/clabe/data_transfer/robocopy.py

def __init__(
    self,
    source: PathLike,
    settings: RobocopySettings,
    *,
    ui_helper: Optional[ui.UiHelper] = None,
):
    """
    Initializes the RobocopyService.

    Args:
        source: The source directory or file to copy
        settings: RobocopySettings containing destination and options
        ui_helper: UI helper for user prompts. Default is None

    Example:
        ```python
        # Initialize with basic parameters:
        settings = RobocopySettings(destination="D:/destination")
        service = RobocopyService("C:/source", settings)

        # Initialize with logging and move operation:
        settings = RobocopySettings(
            destination="D:/archive/data",
            log="transfer.log",
            delete_src=True,
            extra_args="/E /COPY:DAT /R:10"
        )
        service = RobocopyService("C:/temp/data", settings)
        ```
    """

    self.source = source
    self._settings = settings
    self._ui_helper = ui_helper or ui.DefaultUIHelper()

settings property

settings: TSettings

Returns the settings for the data transfer service.

transfer

transfer() -> None

Executes the data transfer using Robocopy.

Processes source-destination mappings and executes Robocopy commands for each pair, handling logging and error reporting.

Source code in src/clabe/data_transfer/robocopy.py

def transfer(
    self,
) -> None:
    """
    Executes the data transfer using Robocopy.

    Processes source-destination mappings and executes Robocopy commands
    for each pair, handling logging and error reporting.
    """

    # Loop through each source-destination pair and call robocopy'
    logger.info("Starting robocopy transfer service.")
    src_dist = self._solve_src_dst_mapping(self.source, self._settings.destination)
    if src_dist is None:
        raise ValueError("Source and destination should be provided.")

    for src, dst in src_dist.items():
        dst = Path(dst)
        src = Path(src)
        try:
            command = ["robocopy", f"{src.as_posix()}", f"{dst.as_posix()}", self._settings.extra_args]
            if self._settings.log:
                command.append(f'/LOG:"{Path(dst) / self._settings.log}"')
            if self._settings.delete_src:
                command.append("/MOV")
            if self._settings.overwrite:
                command.append("/IS")
            if self._settings.force_dir:
                makedirs(dst, exist_ok=True)
            cmd = " ".join(command)
            logger.info("Running Robocopy command: %s", cmd)
            with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) as process:
                if process.stdout:
                    for line in process.stdout:
                        logger.info(line.strip())
            _ = process.wait()
            logger.info("Successfully copied from %s to %s:\n", src, dst)
        except subprocess.CalledProcessError as e:
            logger.error("Error copying from %s to %s:\n%s", src, dst, e.stdout)

validate

validate() -> bool

Validates whether the Robocopy command is available on the system.

Returns:

Type	Description
bool	True if Robocopy is available, False otherwise

Source code in src/clabe/data_transfer/robocopy.py

def validate(self) -> bool:
    """
    Validates whether the Robocopy command is available on the system.

    Returns:
        True if Robocopy is available, False otherwise
    """
    if not _HAS_ROBOCOPY:
        logger.warning("Robocopy command is not available on this system.")
        return False
    return True

prompt_input

prompt_input() -> bool

Prompts the user to confirm whether to trigger the Robocopy transfer.

Returns:

Type	Description
bool	True if the user confirms, False otherwise

Example

# Interactive transfer confirmation:
settings = RobocopySettings(destination="D:/backup")
service = RobocopyService("C:/data", settings)
if service.prompt_input():
    service.transfer()
    # User confirmed, transfer proceeds
else:
    print("Transfer cancelled by user")

Source code in src/clabe/data_transfer/robocopy.py

def prompt_input(self) -> bool:
    """
    Prompts the user to confirm whether to trigger the Robocopy transfer.

    Returns:
        True if the user confirms, False otherwise

    Example:
        ```python
        # Interactive transfer confirmation:
        settings = RobocopySettings(destination="D:/backup")
        service = RobocopyService("C:/data", settings)
        if service.prompt_input():
            service.transfer()
            # User confirmed, transfer proceeds
        else:
            print("Transfer cancelled by user")
        ```
    """
    return self._ui_helper.prompt_yes_no_question("Would you like to trigger robocopy (Y/N)?")

build_runner

build_runner(*args, **kwargs) -> Callable[[Launcher], Any]

Builds a runner function for the service.

Subclasses must implement this method to return a callable that can be executed by the launcher.

Source code in src/clabe/services.py

def build_runner(self, *args, **kwargs) -> Callable[[Launcher], Any]:
    """
    Builds a runner function for the service.

    Subclasses must implement this method to return a callable that can be executed by the launcher.
    """
    return lambda launcher: None