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 | Dict[PathLike, 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. Supports both single source-destination pairs and multiple mappings via a dictionary.

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 | Dict[PathLike, PathLike]	The source directory/file to copy, or a dict mapping sources to destinations	required
settings	RobocopySettings	RobocopySettings containing options	required

Example

# Single source-destination:
settings = RobocopySettings(destination="D:/destination")
service = RobocopyService("C:/source", settings)

# Multiple source-destination mappings:
service = RobocopyService({
    "C:/data1": "D:/backup1",
    "C:/data2": "D:/backup2",
}, settings)

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

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

    Args:
        source: The source directory/file to copy, or a dict mapping sources to destinations
        settings: RobocopySettings containing options

    Example:
        ```python
        # Single source-destination:
        settings = RobocopySettings(destination="D:/destination")
        service = RobocopyService("C:/source", settings)

        # Multiple source-destination mappings:
        service = RobocopyService({
            "C:/data1": "D:/backup1",
            "C:/data2": "D:/backup2",
        }, 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

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()
        ```
    """
    logger.info("Starting robocopy transfer service.")
    self.run()
    logger.info("Robocopy transfer completed.")

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.

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

def run(self: ExecutableApp, executor_kwargs: Optional[dict[str, Any]] = None) -> CommandResult:
    """Execute the command using a local executor and return the result."""
    executor = LocalExecutor(**(executor_kwargs or {}))
    return self.command.execute(executor)

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.

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

async def run_async(self: ExecutableApp, executor_kwargs: Optional[dict[str, Any]] = None) -> CommandResult:
    """Execute the command asynchronously using a local executor and return the result."""
    executor = AsyncLocalExecutor(**(executor_kwargs or {}))
    return await self.command.execute_async(executor)