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, Union | from typing import Any, List, Union | ||||
from swh.model.identifiers import SWHID | |||||
# 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 30 Lines | class FuseEntry: | ||||
async def size(self) -> int: | async def size(self) -> int: | ||||
""" Return the size (in bytes) of an entry """ | """ Return the size (in bytes) of an entry """ | ||||
raise NotImplementedError | raise NotImplementedError | ||||
def get_relative_root_path(self) -> str: | def get_relative_root_path(self) -> str: | ||||
return "../" * (self.depth - 1) | return "../" * (self.depth - 1) | ||||
def get_sharded_name(self, swhid: SWHID) -> str: | |||||
sharding = self.fuse.conf["sharding"] | |||||
if sharding["depth"] <= 0: | |||||
return str(swhid) | |||||
else: | |||||
basename = swhid.object_id | |||||
name, i = "", 0 | |||||
for _ in range(sharding["depth"]): | |||||
name += basename[i : i + sharding["length"]] | |||||
name += "/" | |||||
i += sharding["length"] | |||||
# Always keep the full SWHID as the path basename (otherwise we | |||||
# loose the SWHID object type information) | |||||
name += str(swhid) | |||||
return name | |||||
def get_sharded_next_prefixes( | |||||
self, current_prefix: str, entries: List[SWHID] | |||||
) -> List[SWHID]: | |||||
prefix_len = len(current_prefix) | |||||
next_prefixes = set() | |||||
for swhid in entries: | |||||
next_prefix = swhid.object_id[ | |||||
prefix_len : prefix_len + self.fuse.conf["sharding"]["length"] | |||||
] | |||||
next_prefixes.add(next_prefix) | |||||
return list(next_prefixes) | |||||
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): | class FuseFileEntry(FuseEntry): | ||||
""" FUSE virtual file entry """ | """ FUSE virtual file entry """ | ||||
async def get_content(self) -> bytes: | async def get_content(self) -> bytes: | ||||
Show All 17 Lines | 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 | ||||
# @dataclass | |||||
# class FuseDirEntrySharded(FuseDirEntry): | |||||
# """ TODO """ | |||||
# | |||||
# entries: List[FuseEntry] | |||||
# sharding_depth: int = field(default=0) | |||||
# prefix: str = field(default="") | |||||
# mode: int = field(init=False, default=int(EntryMode.RDONLY_DIR)) | |||||
# | |||||
# async def __aiter__(self): | |||||
# if self.sharding_depth == self.fuse.conf["sharding"]["depth"]: | |||||
# for entry in self.entries: | |||||
# entry.depth = self.depth | |||||
# yield entry | |||||
# else: | |||||
# subdirs = {} | |||||
# for entry in self.entries: | |||||
# name = self.get_sharded_name(entry.swhid) | |||||
# prefix_len = len(self.prefix) | |||||
# | |||||
# next_prefix = name[ | |||||
# prefix_len : prefix_len + self.fuse.conf["sharding"]["length"] | |||||
# ] | |||||
# subdirs.setdefault(next_prefix, []).append(entry) | |||||
# | |||||
# for subdir, subentries in subdirs.items(): | |||||
# yield self.create_child( | |||||
# FuseDirEntrySharded, | |||||
# name=subdir, | |||||
# prefix=f"{self.prefix}{subdir}/", | |||||
# entries=subentries, | |||||
# sharding_depth=self.sharding_depth + 1, | |||||
# ) | |||||
@dataclass | @dataclass | ||||
class FuseSymlinkEntry(FuseEntry): | class FuseSymlinkEntry(FuseEntry): | ||||
""" FUSE virtual symlink entry | """ FUSE virtual symlink entry | ||||
Attributes: | Attributes: | ||||
target: path to symlink target | target: path to symlink target | ||||
""" | """ | ||||
Show All 10 Lines |