Changeset View
Changeset View
Standalone View
Standalone View
swh/fuse/fs/entry.py
# Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020-2021 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 | ||||
import re | import re | ||||
from stat import S_IFDIR, S_IFLNK, S_IFREG | from stat import S_IFDIR, S_IFLNK, S_IFREG | ||||
from typing import Any, AsyncIterator, Dict, Optional, Pattern, Sequence, Union | from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Optional, Pattern, Union | ||||
# Avoid cycling import | if TYPE_CHECKING: # avoid cyclic import | ||||
Fuse = "Fuse" | from swh.fuse.fuse import Fuse | ||||
class EntryMode(IntEnum): | class EntryMode(IntEnum): | ||||
""" Default entry mode and permissions for the FUSE. | """ Default entry mode and permissions for the FUSE. | ||||
The FUSE mount is always read-only, even if permissions contradict this | The FUSE mount is always read-only, even if permissions contradict this | ||||
statement (in a context of a directory, entries are listed with permissions | statement (in a context of a directory, entries are listed with permissions | ||||
taken from the archive). | taken from the archive). | ||||
▲ Show 20 Lines • Show All 68 Lines • ▼ Show 20 Lines | class FuseDirEntry(FuseEntry): | ||||
async def size(self) -> int: | async def size(self) -> int: | ||||
return 0 | return 0 | ||||
def validate_entry(self, name: str) -> bool: | def validate_entry(self, name: str) -> bool: | ||||
""" Return true if the name matches the directory entries regular | """ Return true if the name matches the directory entries regular | ||||
expression, and false otherwise """ | expression, and false otherwise """ | ||||
if self.ENTRIES_REGEXP: | if self.ENTRIES_REGEXP: | ||||
return re.match(self.ENTRIES_REGEXP, name) | return bool(re.match(self.ENTRIES_REGEXP, name)) | ||||
else: | else: | ||||
return True | return True | ||||
async def compute_entries(self) -> Sequence[FuseEntry]: | async def compute_entries(self): | ||||
ardumont: Is it expected you removed that type annotation here? | |||||
Done Inline ActionsYeah, because putting there the actual type (the one that subclasses must implement) doesn't type properly because the function body doesn't contain a yield. It used to work with previous versions of mypy, but no longer does. I guess there might be better solutions, but this looked good enough for now to me. zack: Yeah, because putting there the actual type (the one that subclasses must implement) doesn't… | |||||
""" Return the child entries of a directory entry """ | """ Return the child entries of a directory entry """ | ||||
raise NotImplementedError | raise NotImplementedError | ||||
async def get_entries(self, offset: int = 0) -> AsyncIterator[FuseEntry]: | async def get_entries(self, offset: int = 0) -> AsyncIterator[FuseEntry]: | ||||
""" Return the child entries of a directory entry using direntry cache """ | """ Return the child entries of a directory entry using direntry cache """ | ||||
cache = self.fuse.cache.direntry.get(self) | cache = self.fuse.cache.direntry.get(self) | ||||
if cache: | if cache: | ||||
entries = cache | entries = cache | ||||
else: | else: | ||||
entries = [x async for x in self.compute_entries()] | entries = [x async for x in self.compute_entries()] | ||||
self.fuse.cache.direntry.set(self, entries) | self.fuse.cache.direntry.set(self, entries) | ||||
# Avoid copy by manual iteration (instead of slicing) and use of a | # Avoid copy by manual iteration (instead of slicing) and use of a | ||||
# generator (instead of returning the full list every time) | # generator (instead of returning the full list every time) | ||||
for i in range(offset, len(entries)): | for i in range(offset, len(entries)): | ||||
yield entries[i] | yield entries[i] | ||||
async def lookup(self, name: str) -> FuseEntry: | async def lookup(self, name: str) -> Optional[FuseEntry]: | ||||
""" Look up a FUSE entry by name """ | """ Look up a FUSE entry by name """ | ||||
async for entry in self.get_entries(): | async for entry in self.get_entries(): | ||||
if entry.name == name: | if entry.name == name: | ||||
return entry | return entry | ||||
return None | return None | ||||
Show All 18 Lines |
Is it expected you removed that type annotation here?