Changeset View
Changeset View
Standalone View
Standalone View
swh/fuse/fs/artifact.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 dataclasses import dataclass | from dataclasses import dataclass | ||||
| from pathlib import Path | from pathlib import Path | ||||
| from typing import Any, AsyncIterator, List | from typing import Any, List, Sequence | ||||
| import urllib.parse | import urllib.parse | ||||
| from swh.fuse.fs.entry import ( | from swh.fuse.fs.entry import ( | ||||
| EntryMode, | EntryMode, | ||||
| FuseDirEntry, | FuseDirEntry, | ||||
| FuseEntry, | FuseEntry, | ||||
| FuseFileEntry, | FuseFileEntry, | ||||
| FuseSymlinkEntry, | FuseSymlinkEntry, | ||||
| ▲ Show 20 Lines • Show All 47 Lines • ▼ Show 20 Lines | class Directory(FuseDirEntry): | ||||
| entry metadata. | entry metadata. | ||||
| Note that the FUSE mount is read-only, no matter what the permissions say. | Note that the FUSE mount is read-only, no matter what the permissions say. | ||||
| So it is possible that, in the context of a directory, a file is presented | So it is possible that, in the context of a directory, a file is presented | ||||
| as writable, whereas actually writing to it will fail with `EPERM`. """ | as writable, whereas actually writing to it will fail with `EPERM`. """ | ||||
| swhid: SWHID | swhid: SWHID | ||||
| async def __aiter__(self) -> AsyncIterator[FuseEntry]: | async def compute_entries(self) -> Sequence[FuseEntry]: | ||||
seirl: You should probably return `Iterator[]` here, and only use the covariant `Sequence[]` type in… | |||||
| metadata = await self.fuse.get_metadata(self.swhid) | metadata = await self.fuse.get_metadata(self.swhid) | ||||
| for entry in metadata: | for entry in metadata: | ||||
| name = entry["name"] | name = entry["name"] | ||||
| swhid = entry["target"] | swhid = entry["target"] | ||||
| mode = ( | mode = ( | ||||
| # Archived permissions for directories are always set to | # Archived permissions for directories are always set to | ||||
| # 0o040000 so use a read-only permission instead | # 0o040000 so use a read-only permission instead | ||||
| int(EntryMode.RDONLY_DIR) | int(EntryMode.RDONLY_DIR) | ||||
| ▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines | class Revision(FuseDirEntry): | ||||
| - `history`: a virtual directory listing all its revision ancestors, sorted | - `history`: a virtual directory listing all its revision ancestors, sorted | ||||
| in reverse topological order. Each entry is a symlink pointing into | in reverse topological order. Each entry is a symlink pointing into | ||||
| `archive/SWHID`. | `archive/SWHID`. | ||||
| - `meta.json`: metadata for the current node, as a symlink pointing to the | - `meta.json`: metadata for the current node, as a symlink pointing to the | ||||
| relevant `meta/<SWHID>.json` file """ | relevant `meta/<SWHID>.json` file """ | ||||
| swhid: SWHID | swhid: SWHID | ||||
| async def __aiter__(self) -> AsyncIterator[FuseEntry]: | async def compute_entries(self) -> Sequence[FuseEntry]: | ||||
| metadata = await self.fuse.get_metadata(self.swhid) | metadata = await self.fuse.get_metadata(self.swhid) | ||||
| directory = metadata["directory"] | directory = metadata["directory"] | ||||
| parents = metadata["parents"] | parents = metadata["parents"] | ||||
| root_path = self.get_relative_root_path() | root_path = self.get_relative_root_path() | ||||
| yield self.create_child( | yield self.create_child( | ||||
| FuseSymlinkEntry, | FuseSymlinkEntry, | ||||
| Show All 26 Lines | |||||
| @dataclass | @dataclass | ||||
| class RevisionParents(FuseDirEntry): | class RevisionParents(FuseDirEntry): | ||||
| """ Revision virtual `parents/` directory """ | """ Revision virtual `parents/` directory """ | ||||
| parents: List[SWHID] | parents: List[SWHID] | ||||
| async def __aiter__(self) -> AsyncIterator[FuseEntry]: | async def compute_entries(self) -> Sequence[FuseEntry]: | ||||
| root_path = self.get_relative_root_path() | root_path = self.get_relative_root_path() | ||||
| for i, parent in enumerate(self.parents): | for i, parent in enumerate(self.parents): | ||||
| yield self.create_child( | yield self.create_child( | ||||
| FuseSymlinkEntry, | FuseSymlinkEntry, | ||||
| name=str(i + 1), | name=str(i + 1), | ||||
| target=Path(root_path, f"archive/{parent}"), | target=Path(root_path, f"archive/{parent}"), | ||||
| ) | ) | ||||
| @dataclass | @dataclass | ||||
| class RevisionHistory(FuseDirEntry): | class RevisionHistory(FuseDirEntry): | ||||
| """ Revision virtual `history/` directory """ | """ Revision virtual `history/` directory """ | ||||
| swhid: SWHID | swhid: SWHID | ||||
| async def __aiter__(self) -> AsyncIterator[FuseEntry]: | async def compute_entries(self) -> Sequence[FuseEntry]: | ||||
| history = await self.fuse.get_history(self.swhid) | history = await self.fuse.get_history(self.swhid) | ||||
| root_path = self.get_relative_root_path() | root_path = self.get_relative_root_path() | ||||
| for swhid in history: | for swhid in history: | ||||
| yield self.create_child( | yield self.create_child( | ||||
| FuseSymlinkEntry, | FuseSymlinkEntry, | ||||
| name=str(swhid), | name=str(swhid), | ||||
| target=Path(root_path, f"archive/{swhid}"), | target=Path(root_path, f"archive/{swhid}"), | ||||
| ) | ) | ||||
| Show All 26 Lines | async def find_root_directory(self, swhid: SWHID) -> SWHID: | ||||
| elif swhid.object_type == REVISION: | elif swhid.object_type == REVISION: | ||||
| metadata = await self.fuse.get_metadata(swhid) | metadata = await self.fuse.get_metadata(swhid) | ||||
| return metadata["directory"] | return metadata["directory"] | ||||
| elif swhid.object_type == DIRECTORY: | elif swhid.object_type == DIRECTORY: | ||||
| return swhid | return swhid | ||||
| else: | else: | ||||
| return None | return None | ||||
| async def __aiter__(self) -> AsyncIterator[FuseEntry]: | async def compute_entries(self) -> Sequence[FuseEntry]: | ||||
| metadata = await self.fuse.get_metadata(self.swhid) | metadata = await self.fuse.get_metadata(self.swhid) | ||||
| root_path = self.get_relative_root_path() | root_path = self.get_relative_root_path() | ||||
| yield self.create_child( | yield self.create_child( | ||||
| FuseSymlinkEntry, | FuseSymlinkEntry, | ||||
| name="meta.json", | name="meta.json", | ||||
| target=Path(root_path, f"meta/{self.swhid}.json"), | target=Path(root_path, f"meta/{self.swhid}.json"), | ||||
| ) | ) | ||||
| Show All 40 Lines | class Snapshot(FuseDirEntry): | ||||
| Snapshot nodes are represented on the file-system as directories with one | Snapshot nodes are represented on the file-system as directories with one | ||||
| entry for each branch in the snapshot. Each entry is a symlink pointing into | entry for each branch in the snapshot. Each entry is a symlink pointing into | ||||
| `archive/` to the branch target SWHID. Branch names are URL encoded (hence | `archive/` to the branch target SWHID. Branch names are URL encoded (hence | ||||
| '/' are replaced with '%2F'). """ | '/' are replaced with '%2F'). """ | ||||
| swhid: SWHID | swhid: SWHID | ||||
| async def __aiter__(self) -> AsyncIterator[FuseEntry]: | async def compute_entries(self) -> Sequence[FuseEntry]: | ||||
| metadata = await self.fuse.get_metadata(self.swhid) | metadata = await self.fuse.get_metadata(self.swhid) | ||||
| root_path = self.get_relative_root_path() | root_path = self.get_relative_root_path() | ||||
| for branch_name, branch_meta in metadata.items(): | for branch_name, branch_meta in metadata.items(): | ||||
| # Mangle branch name to create a valid UNIX filename | # Mangle branch name to create a valid UNIX filename | ||||
| name = urllib.parse.quote_plus(branch_name) | name = urllib.parse.quote_plus(branch_name) | ||||
| yield self.create_child( | yield self.create_child( | ||||
| FuseSymlinkEntry, | FuseSymlinkEntry, | ||||
| Show All 12 Lines | |||||
You should probably return Iterator[] here, and only use the covariant Sequence[] type in the base class.