data_transfer.robocopy

RobocopySettings

Bases: ServiceSettings

Settings for the RobocopyService.

Configuration for Robocopy file transfer including destination, logging, and copy options.

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, ...]	Tuple[PydanticBaseSettingsSource, ...]: A tuple of settings sources

Source code in src/clabe/services.py

@classmethod
def settings_customise_sources(
    cls,
    settings_cls: t.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:
        Tuple[PydanticBaseSettingsSource, ...]: A tuple of settings sources
    """
    return (
        init_settings,
        *(
            _SafeYamlSettingsSource(settings_cls, yaml_file=p, yaml_config_section=cls.__yml_section__)
            for p in KNOWN_CONFIG_FILES
        ),
        env_settings,
        dotenv_settings,
        file_secret_settings,
    )

RobocopyService

RobocopyService(
    source: PathLike, settings: RobocopySettings
)

Bases: DataTransfer[RobocopySettings], _DefaultExecutorMixin, ExecutableApp

A data transfer service that uses Robocopy to copy files between directories.

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

Attributes:

Name	Type	Description
command	Command[CommandResult]	The robocopy command to be executed

Methods:

Name	Description
transfer	Executes the Robocopy file transfer
validate	Validates the Robocopy service configuration

Initializes the RobocopyService.

Parameters:

Name	Type	Description	Default
source	PathLike	The source root directory to copy from	required
settings	RobocopySettings	RobocopySettings containing destination and options	required

Example

settings = RobocopySettings(
    destination="D:/destination",
    exclude_dirs=["__pycache__", ".git"],
    exclude_files=["*.pyc"],
)
service = RobocopyService("C:/source", settings)

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

def __init__(
    self,
    source: PathLike,
    settings: RobocopySettings,
):
    """
    Initializes the RobocopyService.

    Args:
        source: The source root directory to copy from
        settings: RobocopySettings containing destination and options

    Example:
        ```python
        settings = RobocopySettings(
            destination="D:/destination",
            exclude_dirs=["__pycache__", ".git"],
            exclude_files=["*.pyc"],
        )
        service = RobocopyService("C:/source", settings)
        ```
    """
    self.source = source
    self._settings = settings
    self._command = self._build_command()

command property

command: Command[CommandResult]

Returns the robocopy command to be executed.

settings property

settings: TSettings

Returns the settings for the data transfer service.

Returns:

Name	Type	Description
TSettings	TSettings	The service settings

transfer

transfer() -> None

Executes the data transfer using Robocopy.

Uses the command executor pattern to run robocopy with configured settings.

Example

settings = RobocopySettings(destination="D:/backup")
service = RobocopyService("C:/data", settings)
service.transfer()

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

@runnable(name="Transfer (robocopy)", notify="Transferring data (robocopy)…")
def transfer(self) -> None:
    """
    Executes the data transfer using Robocopy.

    Uses the command executor pattern to run robocopy with configured settings.

    Example:
        ```python
        settings = RobocopySettings(destination="D:/backup")
        service = RobocopyService("C:/data", settings)
        service.transfer()
        ```
    """
    try:
        self.run()
    except CommandError as e:
        if e.exit_code > _ROBOCOPY_SUCCESS_MAX:
            raise
        logger.debug("Robocopy exited with code %d (informational success).", e.exit_code)

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

run

run(
    executor_kwargs: Optional[dict[str, Any]] = None,
) -> CommandResult

Execute the command using a local executor and return the result.

Parameters:

Name	Type	Description	Default
executor_kwargs	Optional[dict[str, Any]]	Keyword arguments forwarded to the local executor.	None

Source code in src/clabe/apps/_executors.py

@runnable
def run(self, executor_kwargs: Optional[dict[str, Any]] = None) -> CommandResult:
    """Execute the command using a local executor and return the result.

    Args:
        executor_kwargs: Keyword arguments forwarded to the local executor.
    """
    return self.command.execute(LocalExecutor(**(executor_kwargs or {})))

run_async async

run_async(
    executor_kwargs: Optional[dict[str, Any]] = None,
) -> CommandResult

Execute the command asynchronously using a local executor and return the result.

Parameters:

Name	Type	Description	Default
executor_kwargs	Optional[dict[str, Any]]	Keyword arguments forwarded to the local executor.	None

Source code in src/clabe/apps/_executors.py

@runnable
async def run_async(self, executor_kwargs: Optional[dict[str, Any]] = None) -> CommandResult:
    """Execute the command asynchronously using a local executor and return the result.

    Args:
        executor_kwargs: Keyword arguments forwarded to the local executor.
    """
    return await self.command.execute_async(AsyncLocalExecutor(**(executor_kwargs or {})))