some seekable tweaks

Author: Fallen-Breath

prime_backup/action/helpers/chunk_io.py

@@ -5,7 +5,7 @@
 from prime_backup.compressors import Compressor
 from prime_backup.types.chunk_info import ChunkInfo
 from prime_backup.utils.bypass_io import BypassReader
-from prime_backup.utils.io_types import SupportsReadBytes
+from prime_backup.utils.io_types import SupportsReadBytes, SupportsReadAndSeek
 
 
 class ChunkIO:
@@ -18,7 +18,7 @@ def __get_pack_id(self) -> int:
 		return self.chunk.pack_entry.pack_id
 
 	@contextlib.contextmanager
-	def open_raw(self) -> Generator[SupportsReadBytes, None, None]:
+	def open_raw(self) -> Generator[SupportsReadAndSeek, None, None]:
 		with PackReader.open_entry(self.__get_pack_id(), self.chunk.pack_entry.offset, self.chunk.stored_size) as reader:
 			yield reader
 

prime_backup/action/helpers/pack_reader.py

@@ -1,15 +1,18 @@
 import contextlib
+import os
 from typing import BinaryIO, Generator
 
+from typing_extensions import Final
+
 from prime_backup.utils import pack_utils
-from prime_backup.utils.io_types import SupportsReadBytes
+from prime_backup.utils.io_types import SupportsReadAndSeek
 
 
 class PackEntryReader:
 	def __init__(self, file_obj: BinaryIO, offset: int, length: int):
-		self.__file_obj = file_obj
-		self.__offset = offset
-		self.__length = length
+		self.__file_obj: Final[BinaryIO] = file_obj
+		self.__offset: Final[int] = offset
+		self.__length: Final[int] = length
 		self.__position = 0
 		self.__file_obj.seek(offset)
 
@@ -23,11 +26,33 @@ def read(self, size: int = -1) -> bytes:
 		self.__position += len(data)
 		return data
 
+	def seekable(self) -> bool:
+		return True
+
+	def seek(self, offset: int, whence: int = 0):
+		if whence == os.SEEK_SET:
+			position = offset
+		elif whence == os.SEEK_CUR:
+			position = self.__position + offset
+		elif whence == os.SEEK_END:
+			position = self.__length + offset
+		else:
+			raise ValueError('invalid whence {}'.format(whence))
+
+		if position < 0:
+			raise ValueError('negative seek position {}'.format(position))
+		if position > self.__length:
+			position = self.__length
+
+		self.__file_obj.seek(self.__offset + position)
+		self.__position = position
+		return self.__position
+
 
 class PackReader:
 	@classmethod
 	@contextlib.contextmanager
-	def open_entry(cls, pack_id: int, offset: int, length: int) -> Generator[SupportsReadBytes, None, None]:
+	def open_entry(cls, pack_id: int, offset: int, length: int) -> Generator[SupportsReadAndSeek, None, None]:
 		pack_path = pack_utils.get_pack_path(pack_id)
 		with open(pack_path, 'rb') as fh:
 			yield PackEntryReader(fh, offset, length)

prime_backup/cli/fuse/file.py

@@ -4,7 +4,7 @@
 from contextlib import AbstractContextManager
 from io import BytesIO
 from pathlib import Path
-from typing import IO, cast, List
+from typing import List, cast
 
 from typing_extensions import Optional, override
 
@@ -17,6 +17,7 @@
 from prime_backup.types.blob_info import BlobInfo
 from prime_backup.types.chunk_info import OffsetChunkInfo, ChunkInfo
 from prime_backup.utils import blob_utils
+from prime_backup.utils.io_types import SupportsReadBytes, SupportsReadAndSeek
 
 
 class _FileReader(ABC):
@@ -33,17 +34,17 @@ def close(self):
 
 
 class _SingleFileReader(_FileReader):
-	def __init__(self, file_ctx: AbstractContextManager[IO[bytes]]):
+	def __init__(self, file_ctx: AbstractContextManager[SupportsReadBytes]):
 		self.file_ctx = file_ctx
 		self.file_obj = file_ctx.__enter__()
-		self.file_seekable = self.file_obj.seekable()
+		self.file_seekable = hasattr(self.file_obj, 'seekable') and hasattr(self.file_obj, 'seek') and self.file_obj.seekable()
 		self.offset = 0
 
 	@override
 	def read(self, size: int, offset: int) -> bytes:
 		if offset != self.offset:
 			if self.file_seekable:
-				self.file_obj.seek(offset)
+				cast(SupportsReadAndSeek, self.file_obj).seek(offset)
 				self.offset = offset
 			else:
 				raise self.NoSequenceRead()
@@ -62,15 +63,15 @@ def create_from_file(cls, file_path: Path, compress_method: CompressMethod) -> '
 			return _SingleFileReader(open(file_path, 'rb'))
 		else:
 			decompressed_stream = Compressor.create(compress_method).open_decompressed(file_path)
-			return _SingleFileReader(cast(AbstractContextManager[IO[bytes]], decompressed_stream))
+			return _SingleFileReader(decompressed_stream)
 
 	@classmethod
 	def create_from_blob(cls, blob: BlobInfo) -> '_SingleFileReader':
 		return _SingleFileReader.create_from_file(blob_utils.get_blob_path(blob.hash), blob.compress)
 
 	@classmethod
 	def create_from_chunk(cls, chunk: ChunkInfo) -> '_SingleFileReader':
-		return _SingleFileReader(cast(AbstractContextManager[IO[bytes]], ChunkIO(chunk).open_decompressed()))
+		return _SingleFileReader(ChunkIO(chunk).open_decompressed())
 
 
 class _MultiFileReader(_FileReader):

prime_backup/utils/io_types.py

@@ -6,6 +6,14 @@ def read(self, length: int = -1) -> bytes:
 		...
 
 
+class SupportsReadAndSeek(SupportsReadBytes, Protocol):
+	def seekable(self) -> bool:
+		...
+
+	def seek(self, offset: int, whence: int = 0):
+		...
+
+
 class SupportsWriteBytes(Protocol):
 	def write(self, s: bytes) -> int:
 		...