Base

provide.foundation.archive.base

TODO: Add module docstring.

Classes

ArchiveError

ArchiveError(
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any
)

Bases: FoundationError

Base exception for archive-related errors.

Source code in provide/foundation/errors/base.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any,
) -> None:
    self.message = message
    self.code = code or self._default_code()
    self.context = context or {}
    self.context.update(extra_context)
    self.cause = cause
    if cause:
        self.__cause__ = cause
    super().__init__(message)

ArchiveFormatError

ArchiveFormatError(
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any
)

Bases: ArchiveError

Archive format is invalid or corrupted.

Source code in provide/foundation/errors/base.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any,
) -> None:
    self.message = message
    self.code = code or self._default_code()
    self.context = context or {}
    self.context.update(extra_context)
    self.cause = cause
    if cause:
        self.__cause__ = cause
    super().__init__(message)

ArchiveIOError

ArchiveIOError(
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any
)

Bases: ArchiveError

I/O operation failed during archive processing.

Source code in provide/foundation/errors/base.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any,
) -> None:
    self.message = message
    self.code = code or self._default_code()
    self.context = context or {}
    self.context.update(extra_context)
    self.cause = cause
    if cause:
        self.__cause__ = cause
    super().__init__(message)

ArchiveValidationError

ArchiveValidationError(
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any
)

Bases: ArchiveError

Archive validation failed (security checks, malformed paths, etc).

Source code in provide/foundation/errors/base.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any,
) -> None:
    self.message = message
    self.code = code or self._default_code()
    self.context = context or {}
    self.context.update(extra_context)
    self.cause = cause
    if cause:
        self.__cause__ = cause
    super().__init__(message)

BaseArchive

Bases: ABC

Abstract base class for all archive implementations.

This defines the common interface that all archive implementations must follow, ensuring consistency across different archive formats.

Functions

create abstractmethod

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

Create an archive from source path.

Parameters:

Name	Type	Description	Default
source	Path	Source file or directory to archive	required
output	Path	Output archive file path	required

Returns:

Type	Description
Path	Path to the created archive file

Raises:

Type	Description
ArchiveError	If archive creation fails

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

@abstractmethod
def create(self, source: Path, output: Path) -> Path:
    """Create an archive from source path.

    Args:
        source: Source file or directory to archive
        output: Output archive file path

    Returns:
        Path to the created archive file

    Raises:
        ArchiveError: If archive creation fails

    """

extract abstractmethod

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

Extract an archive to output path.

Parameters:

Name	Type	Description	Default
archive	Path	Archive file to extract	required
output	Path	Output directory for extracted contents	required

Returns:

Type	Description
Path	Path to the extraction directory

Raises:

Type	Description
ArchiveError	If extraction fails

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

@abstractmethod
def extract(self, archive: Path, output: Path) -> Path:
    """Extract an archive to output path.

    Args:
        archive: Archive file to extract
        output: Output directory for extracted contents

    Returns:
        Path to the extraction directory

    Raises:
        ArchiveError: If extraction fails

    """

validate abstractmethod

validate(archive: Path) -> bool

Validate that an archive is properly formed.

Parameters:

Name	Type	Description	Default
archive	Path	Archive file to validate	required

Returns:

Type	Description
bool	True if archive is valid, False otherwise

Raises:

Type	Description
ArchiveError	If validation cannot be performed

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

@abstractmethod
def validate(self, archive: Path) -> bool:
    """Validate that an archive is properly formed.

    Args:
        archive: Archive file to validate

    Returns:
        True if archive is valid, False otherwise

    Raises:
        ArchiveError: If validation cannot be performed

    """

BaseCompressor

Bases: ABC

Abstract base class for compression implementations.

Provides common compression/decompression interface for stream, file, and bytes operations. Subclasses must implement the library-specific compression/decompression methods.

Attributes

format_name abstractmethod property

format_name: str

Return the name of the compression format (e.g., 'GZIP', 'BZIP2').

Functions

compress

compress(
    input_stream: BinaryIO, output_stream: BinaryIO
) -> None

Compress data from input stream to output stream.

Parameters:

Name	Type	Description	Default
input_stream	BinaryIO	Input binary stream	required
output_stream	BinaryIO	Output binary stream	required

Raises:

Type	Description
ArchiveError	If compression fails

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

def compress(self, input_stream: BinaryIO, output_stream: BinaryIO) -> None:
    """Compress data from input stream to output stream.

    Args:
        input_stream: Input binary stream
        output_stream: Output binary stream

    Raises:
        ArchiveError: If compression fails

    """
    try:
        self._compress_stream(input_stream, output_stream)
        log.debug(f"Compressed data with {self.format_name} level {self.level}")
    except (OSError, ValueError) as e:
        raise ArchiveError(f"Failed to compress with {self.format_name}: {e}") from e

compress_bytes

compress_bytes(data: bytes) -> bytes

Compress bytes data.

Parameters:

Name	Type	Description	Default
data	bytes	Input bytes	required

Returns:

Type	Description
bytes	Compressed bytes

Raises:

Type	Description
ArchiveError	If compression fails

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

def compress_bytes(self, data: bytes) -> bytes:
    """Compress bytes data.

    Args:
        data: Input bytes

    Returns:
        Compressed bytes

    Raises:
        ArchiveError: If compression fails

    """
    try:
        return self._compress_bytes_impl(data)
    except (OSError, ValueError) as e:
        raise ArchiveError(f"Failed to compress bytes: {e}") from e

compress_file

compress_file(input_path: Path, output_path: Path) -> Path

Compress a file.

Parameters:

Name	Type	Description	Default
input_path	Path	Input file path	required
output_path	Path	Output file path	required

Returns:

Type	Description
Path	Path to compressed file

Raises:

Type	Description
ArchiveError	If compression fails

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

def compress_file(self, input_path: Path, output_path: Path) -> Path:
    """Compress a file.

    Args:
        input_path: Input file path
        output_path: Output file path

    Returns:
        Path to compressed file

    Raises:
        ArchiveError: If compression fails

    """
    try:
        ensure_parent_dir(output_path)

        with input_path.open("rb") as f_in, output_path.open("wb") as f_out:
            self._compress_stream(f_in, f_out)

        log.debug(f"Compressed {input_path} to {output_path}")
        return output_path

    except (OSError, ValueError) as e:
        raise ArchiveError(f"Failed to compress file: {e}") from e

decompress

decompress(
    input_stream: BinaryIO, output_stream: BinaryIO
) -> None

Decompress data from input stream to output stream.

Parameters:

Name	Type	Description	Default
input_stream	BinaryIO	Input binary stream (compressed)	required
output_stream	BinaryIO	Output binary stream	required

Raises:

Type	Description
ArchiveError	If decompression fails

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

def decompress(self, input_stream: BinaryIO, output_stream: BinaryIO) -> None:
    """Decompress data from input stream to output stream.

    Args:
        input_stream: Input binary stream (compressed)
        output_stream: Output binary stream

    Raises:
        ArchiveError: If decompression fails

    """
    try:
        self._decompress_stream(input_stream, output_stream)
        log.debug(f"Decompressed {self.format_name} data")
    except (OSError, ValueError) as e:
        raise ArchiveError(f"Failed to decompress {self.format_name}: {e}") from e

decompress_bytes

decompress_bytes(data: bytes) -> bytes

Decompress bytes data.

Parameters:

Name	Type	Description	Default
data	bytes	Compressed bytes	required

Returns:

Type	Description
bytes	Decompressed bytes

Raises:

Type	Description
ArchiveError	If decompression fails

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

def decompress_bytes(self, data: bytes) -> bytes:
    """Decompress bytes data.

    Args:
        data: Compressed bytes

    Returns:
        Decompressed bytes

    Raises:
        ArchiveError: If decompression fails

    """
    try:
        return self._decompress_bytes_impl(data)
    except (OSError, ValueError) as e:
        raise ArchiveError(f"Failed to decompress bytes: {e}") from e

decompress_file

decompress_file(
    input_path: Path, output_path: Path
) -> Path

Decompress a file.

Parameters:

Name	Type	Description	Default
input_path	Path	Input file path (compressed)	required
output_path	Path	Output file path	required

Returns:

Type	Description
Path	Path to decompressed file

Raises:

Type	Description
ArchiveError	If decompression fails

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

def decompress_file(self, input_path: Path, output_path: Path) -> Path:
    """Decompress a file.

    Args:
        input_path: Input file path (compressed)
        output_path: Output file path

    Returns:
        Path to decompressed file

    Raises:
        ArchiveError: If decompression fails

    """
    try:
        ensure_parent_dir(output_path)

        with input_path.open("rb") as f_in, output_path.open("wb") as f_out:
            self._decompress_stream(f_in, f_out)

        log.debug(f"Decompressed {input_path} to {output_path}")
        return output_path

    except (OSError, ValueError) as e:
        raise ArchiveError(f"Failed to decompress file: {e}") from e

Functions