Changeset View
Changeset View
Standalone View
Standalone View
swh/fuse/fs/entry.py
Show First 20 Lines • Show All 45 Lines • ▼ Show 20 Lines | class FuseEntry: | ||||
inode: int = field(init=False) | inode: int = field(init=False) | ||||
def __post_init__(self): | def __post_init__(self): | ||||
self.inode = self.fuse._alloc_inode(self) | self.inode = self.fuse._alloc_inode(self) | ||||
async def get_content(self) -> bytes: | async def get_content(self) -> bytes: | ||||
""" Return the content of a file entry """ | """ Return the content of a file entry """ | ||||
return None | if isinstance(self, FuseFileEntry): | ||||
return await self.get_content() | |||||
else: | |||||
raise ValueError("Cannot get raw content from a non-file entry") | |||||
async def size(self) -> int: | async def size(self) -> int: | ||||
""" Return the size of a file entry """ | """ Return the size of a file entry """ | ||||
if isinstance(self, FuseFileEntry) or isinstance(self, FuseSymlinkEntry): | |||||
return await self.size() | |||||
else: | |||||
return 0 | return 0 | ||||
async def __aiter__(self) -> AsyncIterator[FuseEntry]: | async def __aiter__(self) -> AsyncIterator[FuseEntry]: | ||||
""" Return the child entries of a directory entry """ | """ Return the child entries of a directory entry """ | ||||
yield None | if isinstance(self, FuseDirEntry): | ||||
yield await self.__aiter__() | |||||
else: | |||||
raise ValueError("Cannot iterate over a non-directory entry") | |||||
async def lookup(self, name: str) -> FuseEntry: | async def lookup(self, name: str) -> FuseEntry: | ||||
""" Look up a FUSE entry by name """ | """ Look up a FUSE entry by name """ | ||||
async for entry in self: | if isinstance(self, FuseDirEntry): | ||||
if entry.name == name: | return await self.lookup() | ||||
return entry | else: | ||||
return None | raise ValueError("Cannot lookup a non-directory entry") | ||||
def get_target(self) -> Union[str, bytes, Path]: | def get_target(self) -> Union[str, bytes, Path]: | ||||
""" Return the path target of a symlink entry """ | """ Return the path target of a symlink entry """ | ||||
return None | if isinstance(self, FuseSymlinkEntry): | ||||
return self.get_target() | |||||
else: | |||||
raise ValueError("Cannot get target from a non-symlink entry") | |||||
def get_relative_root_path(self) -> str: | def get_relative_root_path(self) -> str: | ||||
return "../" * (self.depth - 1) | return "../" * (self.depth - 1) | ||||
def create_child(self, constructor: Any, **kwargs) -> FuseEntry: | def create_child(self, constructor: Any, **kwargs) -> FuseEntry: | ||||
return constructor(depth=self.depth + 1, fuse=self.fuse, **kwargs) | return constructor(depth=self.depth + 1, fuse=self.fuse, **kwargs) | ||||
class FuseFileEntry(FuseEntry): | |||||
""" FUSE virtual file entry """ | |||||
async def get_content(self) -> bytes: | |||||
raise NotImplementedError | |||||
async def size(self) -> int: | |||||
raise NotImplementedError | |||||
class FuseDirEntry(FuseEntry): | |||||
""" FUSE virtual directory entry """ | |||||
async def __aiter__(self) -> AsyncIterator[FuseEntry]: | |||||
raise NotImplementedError | |||||
async def lookup(self, name: str) -> FuseEntry: | |||||
async for entry in self: | |||||
if entry.name == name: | |||||
return entry | |||||
return None | |||||
@dataclass | |||||
class FuseSymlinkEntry(FuseEntry): | |||||
""" FUSE virtual symlink entry | |||||
Attributes: | |||||
target: path to symlink target | |||||
""" | |||||
mode: int = field(init=False, default=int(EntryMode.SYMLINK)) | |||||
target: Union[str, bytes, Path] | |||||
async def size(self) -> int: | |||||
return len(str(self.target)) | |||||
def get_target(self) -> Union[str, bytes, Path]: | |||||
return self.target |