Changeset View
Changeset View
Standalone View
Standalone View
swh/fuse/fs/entry.py
# Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
from __future__ import annotations | from __future__ import annotations | ||||
from dataclasses import dataclass, field | from dataclasses import dataclass, field | ||||
from enum import IntEnum | from enum import IntEnum | ||||
from pathlib import Path | from pathlib import Path | ||||
from stat import S_IFDIR, S_IFLNK, S_IFREG | from stat import S_IFDIR, S_IFLNK, S_IFREG | ||||
from typing import Any, AsyncIterator, Union | from typing import Any, Union | ||||
# Avoid cycling import | # Avoid cycling import | ||||
Fuse = "Fuse" | Fuse = "Fuse" | ||||
class EntryMode(IntEnum): | class EntryMode(IntEnum): | ||||
""" Default entry mode and permissions for the FUSE. | """ Default entry mode and permissions for the FUSE. | ||||
Show All 22 Lines | class FuseEntry: | ||||
mode: int | mode: int | ||||
depth: int | depth: int | ||||
fuse: Fuse | fuse: Fuse | ||||
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 size(self) -> int: | |||||
""" Return the size (in bytes) of an entry """ | |||||
raise NotImplementedError | |||||
def get_relative_root_path(self) -> str: | |||||
return "../" * (self.depth - 1) | |||||
def create_child(self, constructor: Any, **kwargs) -> FuseEntry: | |||||
return constructor(depth=self.depth + 1, fuse=self.fuse, **kwargs) | |||||
class FuseFileEntry(FuseEntry): | |||||
""" FUSE virtual file entry """ | |||||
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 | raise NotImplementedError | ||||
async def size(self) -> int: | |||||
""" Return the size of a file entry """ | |||||
class FuseDirEntry(FuseEntry): | |||||
""" FUSE virtual directory entry """ | |||||
async def size(self) -> int: | |||||
return 0 | return 0 | ||||
async def __aiter__(self) -> AsyncIterator[FuseEntry]: | async def __aiter__(self): | ||||
""" Return the child entries of a directory entry """ | """ Return the child entries of a directory entry """ | ||||
yield None | raise NotImplementedError | ||||
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: | async for entry in self: | ||||
if entry.name == name: | if entry.name == name: | ||||
return entry | return entry | ||||
return None | return None | ||||
def get_target(self) -> Union[str, bytes, Path]: | |||||
""" Return the path target of a symlink entry """ | |||||
return None | @dataclass | ||||
class FuseSymlinkEntry(FuseEntry): | |||||
""" FUSE virtual symlink entry | |||||
def get_relative_root_path(self) -> str: | Attributes: | ||||
return "../" * (self.depth - 1) | target: path to symlink target | ||||
""" | |||||
def create_child(self, constructor: Any, **kwargs) -> FuseEntry: | mode: int = field(init=False, default=int(EntryMode.SYMLINK)) | ||||
return constructor(depth=self.depth + 1, fuse=self.fuse, **kwargs) | target: Union[str, bytes, Path] | ||||
async def size(self) -> int: | |||||
return len(str(self.target)) | |||||
def get_target(self) -> Union[str, bytes, Path]: | |||||
""" Return the path target of a symlink entry """ | |||||
return self.target |