Function extract_range

recoverpy/lib/storage/byte_range_reader.py:14–57

(
    source_path: str,
    offset: int,
    length: int,
    output_path: str,
    *,
    chunk_size: int = 1024 * 1024,
)

def extract_range(
    source_path: str,
    offset: int,
    length: int,
    output_path: str,
    *,
    chunk_size: int = 1024 * 1024,
) -> None:
    _validate_range(offset, length, chunk_size)
    source_fd = _open_source(source_path)
    position = offset
    remaining = length

    try:
        with open(output_path, "wb") as output_file:
            while remaining > 0:
                to_read = min(chunk_size, remaining)
                chunk = _safe_pread(source_fd, to_read, position, source_path)
                if not chunk:
                    break
                output_file.write(chunk)
                read_len = len(chunk)
                remaining -= read_len
                position += read_len
    except PermissionError as error:
        raise BlockExtractionError(
            f"Cannot write output file {output_path}: {error}",
            f"Permission denied: cannot write {output_path}.",
        ) from error
    except OSError as error:
        raise BlockExtractionError(
            f"Cannot write output file {output_path}: {error}",
            f"Cannot write output file {output_path}.",
        ) from error
    finally:
        os.close(source_fd)

    if remaining > 0:
        # Partial reads at EOF/device end are treated as explicit failures so
        # callers never mistake truncated output for a successful extraction.
        raise BlockExtractionError(
            "Requested range exceeds available data.",
            f"Cannot read requested range from {source_path}: reached end of file/device.",
        )


def read_block(source_path: str, block_size: int, block_index: int) -> bytes: