Source code for swh.fuse.fs.entry

# Copyright (C) 2020-2021  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information

from __future__ import annotations

from dataclasses import dataclass, field
from enum import IntEnum
from pathlib import Path
import re
from stat import S_IFDIR, S_IFLNK, S_IFREG
from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Optional, Pattern, Union

if TYPE_CHECKING:  # avoid cyclic import
    from swh.fuse.fuse import Fuse

[docs] class EntryMode(IntEnum): """Default entry mode and permissions for the FUSE. The FUSE mount is always read-only, even if permissions contradict this statement (in a context of a directory, entries are listed with permissions taken from the archive). """ RDONLY_FILE = S_IFREG | 0o444 RDONLY_DIR = S_IFDIR | 0o555 RDONLY_LNK = S_IFLNK | 0o444 # `cache/` sub-directories need the write permission in order to invalidate # cached artifacts using `rm {SWHID}` RDWR_DIR = S_IFDIR | 0o755
[docs] @dataclass class FuseEntry: """Main wrapper class to manipulate virtual FUSE entries""" name: str """entry filename""" mode: int """entry permission mode""" depth: int fuse: Fuse """internal reference to the main FUSE class""" inode: int = field(init=False) """unique integer identifying the entry""" file_info_attrs: Dict[str, Any] = field(init=False, default_factory=dict) def __post_init__(self): self.inode = self.fuse._alloc_inode(self) # By default, let the kernel cache previously accessed data self.file_info_attrs["keep_cache"] = True
[docs] async def size(self) -> int: """Return the size (in bytes) of an entry""" raise NotImplementedError
[docs] def get_relative_root_path(self) -> str: return "../" * (self.depth - 1)
[docs] def create_child(self, constructor: Any, **kwargs) -> FuseEntry: return constructor(depth=self.depth + 1, fuse=self.fuse, **kwargs)
[docs] @dataclass class FuseFileEntry(FuseEntry): """FUSE virtual file entry"""
[docs] async def get_content(self) -> bytes: """Return the content of a file entry""" raise NotImplementedError
[docs] async def size(self) -> int: return len(await self.get_content())
[docs] @dataclass class FuseDirEntry(FuseEntry): """FUSE virtual directory entry""" ENTRIES_REGEXP: Optional[Pattern] = field(init=False, default=None)
[docs] async def size(self) -> int: return 0
[docs] def validate_entry(self, name: str) -> bool: """Return true if the name matches the directory entries regular expression, and false otherwise""" if self.ENTRIES_REGEXP: return bool(re.match(self.ENTRIES_REGEXP, name)) else: return True
[docs] async def compute_entries(self): """Return the child entries of a directory entry""" raise NotImplementedError
[docs] async def get_entries(self, offset: int = 0) -> AsyncIterator[FuseEntry]: """Return the child entries of a directory entry using direntry cache""" cache = self.fuse.cache.direntry.get(self) if cache: entries = cache else: entries = [x async for x in self.compute_entries()] self.fuse.cache.direntry.set(self, entries) # Avoid copy by manual iteration (instead of slicing) and use of a # generator (instead of returning the full list every time) for i in range(offset, len(entries)): yield entries[i]
[docs] async def lookup(self, name: str) -> Optional[FuseEntry]: """Look up a FUSE entry by name""" async for entry in self.get_entries(): if == name: return entry return None
[docs] @dataclass class FuseSymlinkEntry(FuseEntry): """FUSE virtual symlink entry""" mode: int = field(init=False, default=int(EntryMode.RDONLY_LNK)) target: Union[str, bytes, Path] """path to symlink target"""
[docs] async def size(self) -> int: return len(str(
[docs] def get_target(self) -> Union[str, bytes, Path]: """Return the path target of a symlink entry""" return