Operations

provide.foundation.archive.operations

TODO: Add module docstring.

Classes

ArchiveOperations

Helper class for common archive operation patterns.

Provides convenient methods for common archive formats.

Functions

create_tar_bz2 staticmethod

create_tar_bz2(
    source: Path, output: Path, deterministic: bool = True
) -> Path

Create .tar.bz2 archive in one step.

Parameters:

Name	Type	Description	Default
source	Path	Source file or directory	required
output	Path	Output path (should end with .tar.bz2)	required
deterministic	bool	Create reproducible archive	True

Returns:

Type	Description
Path	Path to created archive

Raises:

Type	Description
ArchiveError	If creation fails

Source code in provide/foundation/archive/operations.py

@staticmethod
def create_tar_bz2(source: Path, output: Path, deterministic: bool = True) -> Path:
    """Create .tar.bz2 archive in one step.

    Args:
        source: Source file or directory
        output: Output path (should end with .tar.bz2)
        deterministic: Create reproducible archive

    Returns:
        Path to created archive

    Raises:
        ArchiveError: If creation fails

    """
    ensure_parent_dir(output)

    chain = OperationChain(
        operations=[ArchiveOperation.TAR, ArchiveOperation.BZIP2],
        operation_config={ArchiveOperation.TAR: {"deterministic": deterministic}},
    )
    return chain.execute(source, output)

create_tar_gz staticmethod

create_tar_gz(
    source: Path, output: Path, deterministic: bool = True
) -> Path

Create .tar.gz archive in one step.

Parameters:

Name	Type	Description	Default
source	Path	Source file or directory	required
output	Path	Output path (should end with .tar.gz)	required
deterministic	bool	Create reproducible archive	True

Returns:

Type	Description
Path	Path to created archive

Raises:

Type	Description
ArchiveError	If creation fails

Source code in provide/foundation/archive/operations.py

@staticmethod
def create_tar_gz(source: Path, output: Path, deterministic: bool = True) -> Path:
    """Create .tar.gz archive in one step.

    Args:
        source: Source file or directory
        output: Output path (should end with .tar.gz)
        deterministic: Create reproducible archive

    Returns:
        Path to created archive

    Raises:
        ArchiveError: If creation fails

    """
    ensure_parent_dir(output)

    chain = OperationChain(
        operations=[ArchiveOperation.TAR, ArchiveOperation.GZIP],
        operation_config={ArchiveOperation.TAR: {"deterministic": deterministic}},
    )
    return chain.execute(source, output)

detect_format staticmethod

detect_format(file: Path) -> list[ArchiveOperation]

Detect archive format and return operation chain.

Parameters:

Name	Type	Description	Default
file	Path	File path to analyze	required

Returns:

Type	Description
list[ArchiveOperation]	List of operations needed to extract

Raises:

Type	Description
ArchiveError	If format cannot be detected

Source code in provide/foundation/archive/operations.py

@staticmethod
def detect_format(file: Path) -> list[ArchiveOperation]:
    """Detect archive format and return operation chain.

    Args:
        file: File path to analyze

    Returns:
        List of operations needed to extract

    Raises:
        ArchiveError: If format cannot be detected

    """
    # Try extension-based detection first
    operations = ArchiveOperations._detect_format_by_extension(file.name)
    if operations is not None:
        return operations

    # Fall back to magic number detection
    operations = ArchiveOperations._detect_format_by_magic(file)
    if operations is not None:
        return operations

    raise ArchiveError(f"Cannot detect format of {file}")

extract_tar_bz2 staticmethod

extract_tar_bz2(archive: Path, output: Path) -> Path

Extract .tar.bz2 archive in one step.

Parameters:

Name	Type	Description	Default
archive	Path	Archive path	required
output	Path	Output directory	required

Returns:

Type	Description
Path	Path to extraction directory

Raises:

Type	Description
ArchiveError	If extraction fails

Source code in provide/foundation/archive/operations.py

@staticmethod
def extract_tar_bz2(archive: Path, output: Path) -> Path:
    """Extract .tar.bz2 archive in one step.

    Args:
        archive: Archive path
        output: Output directory

    Returns:
        Path to extraction directory

    Raises:
        ArchiveError: If extraction fails

    """
    output.mkdir(parents=True, exist_ok=True)

    chain = OperationChain(operations=[ArchiveOperation.TAR, ArchiveOperation.BZIP2])
    return chain.reverse(archive, output)

extract_tar_gz staticmethod

extract_tar_gz(archive: Path, output: Path) -> Path

Extract .tar.gz archive in one step.

Parameters:

Name	Type	Description	Default
archive	Path	Archive path	required
output	Path	Output directory	required

Returns:

Type	Description
Path	Path to extraction directory

Raises:

Type	Description
ArchiveError	If extraction fails

Source code in provide/foundation/archive/operations.py

@staticmethod
def extract_tar_gz(archive: Path, output: Path) -> Path:
    """Extract .tar.gz archive in one step.

    Args:
        archive: Archive path
        output: Output directory

    Returns:
        Path to extraction directory

    Raises:
        ArchiveError: If extraction fails

    """
    output.mkdir(parents=True, exist_ok=True)

    chain = OperationChain(operations=[ArchiveOperation.TAR, ArchiveOperation.GZIP])
    return chain.reverse(archive, output)

OperationChain

Chain multiple archive operations together.

Enables complex operations like tar.gz, tar.bz2, etc. Operations are executed in order for creation, reversed for extraction.

Functions

execute

execute(source: Path, output: Path) -> Path

Execute operation chain on source.

Parameters:

Name	Type	Description	Default
source	Path	Source file or directory	required
output	Path	Final output path	required

Returns:

Type	Description
Path	Path to final output

Raises:

Type	Description
ArchiveError	If any operation fails

Source code in provide/foundation/archive/operations.py

def execute(self, source: Path, output: Path) -> Path:
    """Execute operation chain on source.

    Args:
        source: Source file or directory
        output: Final output path

    Returns:
        Path to final output

    Raises:
        ArchiveError: If any operation fails

    """
    current = source
    temp_files = []

    try:
        for i, op in enumerate(self.operations):
            # Determine output for this operation
            if i == len(self.operations) - 1:
                # Last operation, use final output
                next_output = output
            else:
                # Intermediate operation, use temp file
                suffix = self._get_suffix_for_operation(op)
                # Use Foundation's temp_file with cleanup=False so we manage it
                with temp_file(suffix=suffix, cleanup=False) as temp_path:
                    next_output = temp_path
                temp_files.append(next_output)

            # Execute operation
            current = self._execute_operation(op, current, next_output)
            log.debug(f"Executed operation '{op}': {current}")

        return current

    except Exception as e:
        raise ArchiveError(f"Operation chain failed: {e}") from e
    finally:
        # Clean up temp files using Foundation's safe file operations
        for temp in temp_files:
            safe_delete(temp, missing_ok=True)

reverse

reverse(source: Path, output: Path) -> Path

Reverse operation chain (extract/decompress).

Parameters:

Name	Type	Description	Default
source	Path	Source archive	required
output	Path	Final output path	required

Returns:

Type	Description
Path	Path to final output

Raises:

Type	Description
ArchiveError	If any operation fails

Source code in provide/foundation/archive/operations.py

def reverse(self, source: Path, output: Path) -> Path:
    """Reverse operation chain (extract/decompress).

    Args:
        source: Source archive
        output: Final output path

    Returns:
        Path to final output

    Raises:
        ArchiveError: If any operation fails

    """
    # Operations are the same when reversed; the _execute_operation
    # method will handle whether to create or extract based on context
    reversed_chain = OperationChain(
        operations=list(reversed(self.operations)), operation_config=self.operation_config
    )
    return reversed_chain.execute(source, output)

Functions