data_transfer.robocopy

RobocopyExitCode

Bases: IntFlag

Bitmask flags returned by Robocopy as its exit code.

Robocopy ORs these bits together, so a single exit code can encode multiple conditions simultaneously (e.g. FILES_COPIED | EXTRA_FILES == 3). Codes whose only set bits are in the 0–4 range (i.e. values 0–7) are informational successes. Any code with bit 3 (COPY_FAILURES) or bit 4 (FATAL_ERROR) set indicates an actual problem.

References

https://learn.microsoft.com/en-us/troubleshoot/windows-server/backup-and-storage/return-codes-used-robocopy-utility

NO_CHANGE class-attribute instance-attribute

NO_CHANGE = 0

No files were copied; the destination was already up to date.

FILES_COPIED class-attribute instance-attribute

FILES_COPIED = 1

One or more files were copied successfully.

EXTRA_FILES class-attribute instance-attribute

EXTRA_FILES = 2

Extra files or directories were found in the destination.

MISMATCHED class-attribute instance-attribute

MISMATCHED = 4

Mismatched files or directories were detected.

COPY_FAILURES class-attribute instance-attribute

COPY_FAILURES = 8

One or more files could not be copied (retry limit exceeded).

FATAL_ERROR class-attribute instance-attribute

FATAL_ERROR = 16

Serious error; Robocopy did not copy any files.

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: type[ps.BaseSettings],
    init_settings: ps.PydanticBaseSettingsSource,
    env_settings: ps.PydanticBaseSettingsSource,
    dotenv_settings: ps.PydanticBaseSettingsSource,
    file_secret_settings: ps.PydanticBaseSettingsSource,
) -> 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()
        ```
    """
    self.run()

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: dict[str, Any] | None = None,
) -> CommandResult

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

Parameters:

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

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

@runnable
def run(self, executor_kwargs: dict[str, Any] | None = 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: dict[str, Any] | None = None,
) -> CommandResult

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

Parameters:

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

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

@runnable
async def run_async(self, executor_kwargs: dict[str, Any] | None = 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 {})))