# Example File Systems

pyfuse3 comes with several example file systems in the `examples` directory of the release tarball. For completeness, these examples are also included here.

## Single-file, Read-only File System

(shipped as `examples/lltest.py`)

      1#!/usr/bin/env python3
      2# -*- coding: utf-8 -*-
      3'''
      4hello.py - Example file system for pyfuse3.
      5
      6This program presents a static file system containing a single file.
      7
      8Copyright © 2015 Nikolaus Rath <Nikolaus.org>
      9Copyright © 2015 Gerion Entrup.
     10
     11Permission is hereby granted, free of charge, to any person obtaining a copy of
     12this software and associated documentation files (the "Software"), to deal in
     13the Software without restriction, including without limitation the rights to
     14use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
     15the Software, and to permit persons to whom the Software is furnished to do so.
     16
     17THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
     18IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
     19FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
     20COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
     21IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
     22CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
     23'''
     24
     25import errno
     26import logging
     27import os
     28import stat
     29from argparse import ArgumentParser, Namespace
     30from typing import cast
     31
     32import trio
     33
     34import pyfuse3
     35from pyfuse3 import EntryAttributes, FileHandleT, FileInfo, InodeT, ReaddirToken, RequestContext
     36
     37try:
     38    import faulthandler
     39except ImportError:
     40    pass
     41else:
     42    faulthandler.enable()
     43
     44log = logging.getLogger(__name__)
     45
     46
     47class TestFs(pyfuse3.Operations):
     48    def __init__(self) -> None:
     49        super(TestFs, self).__init__()
     50        self.hello_name = b"message"
     51        self.hello_inode = pyfuse3.ROOT_INODE + 1
     52        self.hello_data = b"hello world\n"
     53
     54    async def getattr(self, inode: InodeT, ctx: RequestContext | None = None) -> EntryAttributes:
     55        entry = EntryAttributes()
     56        if inode == pyfuse3.ROOT_INODE:
     57            entry.st_mode = stat.S_IFDIR | 0o755
     58            entry.st_size = 0
     59        elif inode == self.hello_inode:
     60            entry.st_mode = stat.S_IFREG | 0o644
     61            entry.st_size = len(self.hello_data)
     62        else:
     63            raise pyfuse3.FUSEError(errno.ENOENT)
     64
     65        stamp = int(1438467123.985654 * 1e9)
     66        entry.st_atime_ns = stamp
     67        entry.st_ctime_ns = stamp
     68        entry.st_mtime_ns = stamp
     69        entry.st_gid = os.getgid()
     70        entry.st_uid = os.getuid()
     71        entry.st_ino = inode
     72
     73        return entry
     74
     75    async def lookup(
     76        self, parent_inode: InodeT, name: bytes, ctx: RequestContext
     77    ) -> EntryAttributes:
     78        if parent_inode != pyfuse3.ROOT_INODE or name != self.hello_name:
     79            raise pyfuse3.FUSEError(errno.ENOENT)
     80        return await self.getattr(self.hello_inode, ctx)
     81
     82    async def opendir(self, inode: InodeT, ctx: RequestContext) -> FileHandleT:
     83        if inode != pyfuse3.ROOT_INODE:
     84            raise pyfuse3.FUSEError(errno.ENOENT)
     85        # For simplicity, we use the inode as file handle
     86        return FileHandleT(inode)
     87
     88    async def readdir(self, fh: FileHandleT, start_id: int, token: ReaddirToken) -> None:
     89        assert fh == pyfuse3.ROOT_INODE
     90
     91        # only one entry
     92        if start_id == 0:
     93            pyfuse3.readdir_reply(token, self.hello_name, await self.getattr(self.hello_inode), 1)
     94        return
     95
     96    async def open(self, inode: InodeT, flags: int, ctx: RequestContext) -> FileInfo:
     97        if inode != self.hello_inode:
     98            raise pyfuse3.FUSEError(errno.ENOENT)
     99        if flags & os.O_RDWR or flags & os.O_WRONLY:
    100            raise pyfuse3.FUSEError(errno.EACCES)
    101        # For simplicity, we use the inode as file handle
    102        return FileInfo(fh=FileHandleT(inode))
    103
    104    async def read(self, fh: FileHandleT, off: int, size: int) -> bytes:
    105        assert fh == self.hello_inode
    106        return self.hello_data[off : off + size]
    107
    108
    109def init_logging(debug: bool = False) -> None:
    110    formatter = logging.Formatter(
    111        '%(asctime)s.%(msecs)03d %(threadName)s: [%(name)s] %(message)s',
    112        datefmt="%Y-%m-%d %H:%M:%S",
    113    )
    114    handler = logging.StreamHandler()
    115    handler.setFormatter(formatter)
    116    root_logger = logging.getLogger()
    117    if debug:
    118        handler.setLevel(logging.DEBUG)
    119        root_logger.setLevel(logging.DEBUG)
    120    else:
    121        handler.setLevel(logging.INFO)
    122        root_logger.setLevel(logging.INFO)
    123    root_logger.addHandler(handler)
    124
    125
    126def parse_args() -> Namespace:
    127    '''Parse command line'''
    128
    129    parser = ArgumentParser()
    130
    131    parser.add_argument('mountpoint', type=str, help='Where to mount the file system')
    132    parser.add_argument(
    133        '--debug', action='store_true', default=False, help='Enable debugging output'
    134    )
    135    parser.add_argument(
    136        '--debug-fuse', action='store_true', default=False, help='Enable FUSE debugging output'
    137    )
    138    return parser.parse_args()
    139
    140
    141def main() -> None:
    142    options = parse_args()
    143    init_logging(options.debug)
    144
    145    testfs = TestFs()
    146    fuse_options = set(pyfuse3.default_options)
    147    fuse_options.add('fsname=hello')
    148    if options.debug_fuse:
    149        fuse_options.add('debug')
    150    pyfuse3.init(testfs, options.mountpoint, fuse_options)
    151    try:
    152        trio.run(pyfuse3.main)
    153    except:
    154        pyfuse3.close(unmount=False)
    155        raise
    156
    157    pyfuse3.close()
    158
    159
    160if __name__ == '__main__':
    161    main()

## In-memory File System

(shipped as `examples/tmpfs.py`)

      1#!/usr/bin/env python3
      2# -*- coding: utf-8 -*-
      3'''
      4tmpfs.py - Example file system for pyfuse3.
      5
      6This file system stores all data in memory.
      7
      8Copyright © 2013 Nikolaus Rath <Nikolaus.org>
      9
     10Permission is hereby granted, free of charge, to any person obtaining a copy of
     11this software and associated documentation files (the "Software"), to deal in
     12the Software without restriction, including without limitation the rights to
     13use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
     14the Software, and to permit persons to whom the Software is furnished to do so.
     15
     16THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
     17IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
     18FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
     19COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
     20IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
     21CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
     22'''
     23
     24import errno
     25import logging
     26import os
     27import sqlite3
     28import stat
     29from argparse import ArgumentParser, Namespace
     30from collections import defaultdict
     31from time import time
     32from typing import Any, cast
     33
     34import trio
     35
     36import pyfuse3
     37from pyfuse3 import (
     38    EntryAttributes,
     39    FileHandleT,
     40    FileInfo,
     41    FUSEError,
     42    InodeT,
     43    ReaddirToken,
     44    RequestContext,
     45    SetattrFields,
     46    StatvfsData,
     47)
     48
     49try:
     50    import faulthandler
     51except ImportError:
     52    pass
     53else:
     54    faulthandler.enable()
     55
     56log = logging.getLogger()
     57
     58
     59class Operations(pyfuse3.Operations):
     60    '''An example filesystem that stores all data in memory
     61
     62    This is a very simple implementation with terrible performance.
     63    Don't try to store significant amounts of data. Also, there are
     64    some other flaws that have not been fixed to keep the code easier
     65    to understand:
     66
     67    * atime, mtime and ctime are not updated
     68    * generation numbers are not supported
     69    * lookup counts are not maintained
     70    '''
     71
     72    enable_writeback_cache = True
     73
     74    def __init__(self) -> None:
     75        super(Operations, self).__init__()
     76        self.db: sqlite3.Connection = sqlite3.connect(':memory:')
     77        self.db.text_factory = str
     78        self.db.row_factory = sqlite3.Row
     79        self.cursor: sqlite3.Cursor = self.db.cursor()
     80        self.inode_open_count: defaultdict[InodeT, int] = defaultdict(int)
     81        self.init_tables()
     82
     83    def init_tables(self) -> None:
     84        '''Initialize file system tables'''
     85
     86        self.cursor.execute("""
     87        CREATE TABLE inodes (
     88            id        INTEGER PRIMARY KEY,
     89            uid       INT NOT NULL,
     90            gid       INT NOT NULL,
     91            mode      INT NOT NULL,
     92            mtime_ns  INT NOT NULL,
     93            atime_ns  INT NOT NULL,
     94            ctime_ns  INT NOT NULL,
     95            target    BLOB(256) ,
     96            size      INT NOT NULL DEFAULT 0,
     97            rdev      INT NOT NULL DEFAULT 0,
     98            data      BLOB
     99        )
    100        """)
    101
    102        self.cursor.execute("""
    103        CREATE TABLE contents (
    104            rowid     INTEGER PRIMARY KEY AUTOINCREMENT,
    105            name      BLOB(256) NOT NULL,
    106            inode     INT NOT NULL REFERENCES inodes(id),
    107            parent_inode INT NOT NULL REFERENCES inodes(id),
    108
    109            UNIQUE (name, parent_inode)
    110        )""")
    111
    112        # Insert root directory
    113        now_ns = int(time() * 1e9)
    114        self.cursor.execute(
    115            "INSERT INTO inodes (id,mode,uid,gid,mtime_ns,atime_ns,ctime_ns) "
    116            "VALUES (?,?,?,?,?,?,?)",
    117            (
    118                pyfuse3.ROOT_INODE,
    119                stat.S_IFDIR
    120                | stat.S_IRUSR
    121                | stat.S_IWUSR
    122                | stat.S_IXUSR
    123                | stat.S_IRGRP
    124                | stat.S_IXGRP
    125                | stat.S_IROTH
    126                | stat.S_IXOTH,
    127                os.getuid(),
    128                os.getgid(),
    129                now_ns,
    130                now_ns,
    131                now_ns,
    132            ),
    133        )
    134        self.cursor.execute(
    135            "INSERT INTO contents (name, parent_inode, inode) VALUES (?,?,?)",
    136            (b'..', pyfuse3.ROOT_INODE, pyfuse3.ROOT_INODE),
    137        )
    138
    139    def get_row(self, *a: Any, **kw: Any) -> sqlite3.Row:
    140        self.cursor.execute(*a, **kw)
    141        try:
    142            row = next(self.cursor)
    143        except StopIteration:
    144            raise NoSuchRowError()
    145        try:
    146            next(self.cursor)
    147        except StopIteration:
    148            pass
    149        else:
    150            raise NoUniqueValueError()
    151
    152        return row
    153
    154    async def lookup(
    155        self, parent_inode: InodeT, name: bytes, ctx: RequestContext
    156    ) -> EntryAttributes:
    157        if name == b'.':
    158            inode = parent_inode
    159        elif name == b'..':
    160            inode = self.get_row("SELECT * FROM contents WHERE inode=?", (parent_inode,))[
    161                'parent_inode'
    162            ]
    163        else:
    164            try:
    165                inode = self.get_row(
    166                    "SELECT * FROM contents WHERE name=? AND parent_inode=?", (name, parent_inode)
    167                )['inode']
    168            except NoSuchRowError:
    169                raise (pyfuse3.FUSEError(errno.ENOENT))
    170
    171        return await self.getattr(InodeT(inode), ctx)
    172
    173    async def getattr(self, inode: InodeT, ctx: RequestContext | None = None) -> EntryAttributes:
    174        try:
    175            row = self.get_row("SELECT * FROM inodes WHERE id=?", (inode,))
    176        except NoSuchRowError:
    177            raise (pyfuse3.FUSEError(errno.ENOENT))
    178
    179        entry = EntryAttributes()
    180        entry.st_ino = inode
    181        entry.generation = 0
    182        entry.entry_timeout = 300
    183        entry.attr_timeout = 300
    184        entry.st_mode = row['mode']
    185        entry.st_nlink = self.get_row("SELECT COUNT(inode) FROM contents WHERE inode=?", (inode,))[
    186            0
    187        ]
    188        entry.st_uid = row['uid']
    189        entry.st_gid = row['gid']
    190        entry.st_rdev = row['rdev']
    191        entry.st_size = row['size']
    192
    193        entry.st_blksize = 512
    194        entry.st_blocks = 1
    195        entry.st_atime_ns = row['atime_ns']
    196        entry.st_mtime_ns = row['mtime_ns']
    197        entry.st_ctime_ns = row['ctime_ns']
    198
    199        return entry
    200
    201    async def readlink(self, inode: InodeT, ctx: RequestContext) -> bytes:
    202        return self.get_row('SELECT * FROM inodes WHERE id=?', (inode,))['target']
    203
    204    async def opendir(self, inode: InodeT, ctx: RequestContext) -> FileHandleT:
    205        # For simplicity, we use the inode as file handle
    206        return FileHandleT(inode)
    207
    208    async def readdir(self, fh: FileHandleT, start_id: int, token: ReaddirToken) -> None:
    209        if start_id == 0:
    210            off = -1
    211        else:
    212            off = start_id
    213
    214        cursor2 = self.db.cursor()
    215        cursor2.execute(
    216            "SELECT * FROM contents WHERE parent_inode=? AND rowid > ? ORDER BY rowid", (fh, off)
    217        )
    218
    219        for row in cursor2:
    220            pyfuse3.readdir_reply(
    221                token, row['name'], await self.getattr(InodeT(row['inode'])), row['rowid']
    222            )
    223
    224    async def unlink(self, parent_inode: InodeT, name: bytes, ctx: RequestContext) -> None:
    225        entry = await self.lookup(parent_inode, name, ctx)
    226
    227        if stat.S_ISDIR(entry.st_mode):
    228            raise pyfuse3.FUSEError(errno.EISDIR)
    229
    230        self._remove(parent_inode, name, entry)
    231
    232    async def rmdir(self, parent_inode: InodeT, name: bytes, ctx: RequestContext) -> None:
    233        entry = await self.lookup(parent_inode, name, ctx)
    234
    235        if not stat.S_ISDIR(entry.st_mode):
    236            raise pyfuse3.FUSEError(errno.ENOTDIR)
    237
    238        self._remove(parent_inode, name, entry)
    239
    240    def _remove(self, parent_inode: InodeT, name: bytes, entry: EntryAttributes) -> None:
    241        if (
    242            self.get_row("SELECT COUNT(inode) FROM contents WHERE parent_inode=?", (entry.st_ino,))[
    243                0
    244            ]
    245            > 0
    246        ):
    247            raise pyfuse3.FUSEError(errno.ENOTEMPTY)
    248
    249        self.cursor.execute(
    250            "DELETE FROM contents WHERE name=? AND parent_inode=?", (name, parent_inode)
    251        )
    252
    253        if entry.st_nlink == 1 and entry.st_ino not in self.inode_open_count:
    254            self.cursor.execute("DELETE FROM inodes WHERE id=?", (entry.st_ino,))
    255
    256    async def symlink(
    257        self, parent_inode: InodeT, name: bytes, target: bytes, ctx: RequestContext
    258    ) -> EntryAttributes:
    259        mode = (
    260            stat.S_IFLNK
    261            | stat.S_IRUSR
    262            | stat.S_IWUSR
    263            | stat.S_IXUSR
    264            | stat.S_IRGRP
    265            | stat.S_IWGRP
    266            | stat.S_IXGRP
    267            | stat.S_IROTH
    268            | stat.S_IWOTH
    269            | stat.S_IXOTH
    270        )
    271        return await self._create(parent_inode, name, mode, ctx, target=target)
    272
    273    async def rename(
    274        self,
    275        parent_inode_old: InodeT,
    276        name_old: bytes,
    277        parent_inode_new: InodeT,
    278        name_new: bytes,
    279        flags: int,
    280        ctx: RequestContext,
    281    ) -> None:
    282        if flags != 0:
    283            raise FUSEError(errno.EINVAL)
    284
    285        entry_old = await self.lookup(parent_inode_old, name_old, ctx)
    286
    287        entry_new = None
    288        try:
    289            entry_new = await self.lookup(
    290                parent_inode_new,
    291                name_new if isinstance(name_new, bytes) else name_new.encode(),
    292                ctx,
    293            )
    294        except pyfuse3.FUSEError as exc:
    295            if exc.errno != errno.ENOENT:
    296                raise
    297
    298        if entry_new is not None:
    299            self._replace(
    300                parent_inode_old, name_old, parent_inode_new, name_new, entry_old, entry_new
    301            )
    302        else:
    303            self.cursor.execute(
    304                "UPDATE contents SET name=?, parent_inode=? WHERE name=? AND parent_inode=?",
    305                (name_new, parent_inode_new, name_old, parent_inode_old),
    306            )
    307
    308    def _replace(
    309        self,
    310        parent_inode_old: InodeT,
    311        name_old: bytes,
    312        parent_inode_new: InodeT,
    313        name_new: bytes,
    314        entry_old: EntryAttributes,
    315        entry_new: EntryAttributes,
    316    ) -> None:
    317        if (
    318            self.get_row(
    319                "SELECT COUNT(inode) FROM contents WHERE parent_inode=?", (entry_new.st_ino,)
    320            )[0]
    321            > 0
    322        ):
    323            raise pyfuse3.FUSEError(errno.ENOTEMPTY)
    324
    325        self.cursor.execute(
    326            "UPDATE contents SET inode=? WHERE name=? AND parent_inode=?",
    327            (entry_old.st_ino, name_new, parent_inode_new),
    328        )
    329        self.db.execute(
    330            'DELETE FROM contents WHERE name=? AND parent_inode=?', (name_old, parent_inode_old)
    331        )
    332
    333        if entry_new.st_nlink == 1 and entry_new.st_ino not in self.inode_open_count:
    334            self.cursor.execute("DELETE FROM inodes WHERE id=?", (entry_new.st_ino,))
    335
    336    async def link(
    337        self, inode: InodeT, new_parent_inode: InodeT, new_name: bytes, ctx: RequestContext
    338    ) -> EntryAttributes:
    339        entry_p = await self.getattr(new_parent_inode, ctx)
    340        if entry_p.st_nlink == 0:
    341            log.warning(
    342                'Attempted to create entry %s with unlinked parent %d', new_name, new_parent_inode
    343            )
    344            raise FUSEError(errno.EINVAL)
    345
    346        self.cursor.execute(
    347            "INSERT INTO contents (name, inode, parent_inode) VALUES(?,?,?)",
    348            (new_name, inode, new_parent_inode),
    349        )
    350
    351        return await self.getattr(inode, ctx)
    352
    353    async def setattr(
    354        self,
    355        inode: InodeT,
    356        attr: EntryAttributes,
    357        fields: SetattrFields,
    358        fh: FileHandleT | None,
    359        ctx: RequestContext,
    360    ) -> EntryAttributes:
    361        if fields.update_size:
    362            data = self.get_row('SELECT data FROM inodes WHERE id=?', (inode,))[0]
    363            if data is None:
    364                data = b''
    365            if len(data) < attr.st_size:
    366                data = data + b'\0' * (attr.st_size - len(data))
    367            else:
    368                data = data[: attr.st_size]
    369            self.cursor.execute(
    370                'UPDATE inodes SET data=?, size=? WHERE id=?',
    371                (memoryview(data), attr.st_size, inode),
    372            )
    373        if fields.update_mode:
    374            self.cursor.execute('UPDATE inodes SET mode=? WHERE id=?', (attr.st_mode, inode))
    375
    376        if fields.update_uid:
    377            self.cursor.execute('UPDATE inodes SET uid=? WHERE id=?', (attr.st_uid, inode))
    378
    379        if fields.update_gid:
    380            self.cursor.execute('UPDATE inodes SET gid=? WHERE id=?', (attr.st_gid, inode))
    381
    382        if fields.update_atime:
    383            self.cursor.execute(
    384                'UPDATE inodes SET atime_ns=? WHERE id=?', (attr.st_atime_ns, inode)
    385            )
    386
    387        if fields.update_mtime:
    388            self.cursor.execute(
    389                'UPDATE inodes SET mtime_ns=? WHERE id=?', (attr.st_mtime_ns, inode)
    390            )
    391
    392        if fields.update_ctime:
    393            self.cursor.execute(
    394                'UPDATE inodes SET ctime_ns=? WHERE id=?', (attr.st_ctime_ns, inode)
    395            )
    396        else:
    397            self.cursor.execute(
    398                'UPDATE inodes SET ctime_ns=? WHERE id=?', (int(time() * 1e9), inode)
    399            )
    400
    401        return await self.getattr(inode, ctx)
    402
    403    async def mknod(
    404        self, parent_inode: InodeT, name: bytes, mode: int, rdev: int, ctx: RequestContext
    405    ) -> EntryAttributes:
    406        return await self._create(parent_inode, name, mode, ctx, rdev=rdev)
    407
    408    async def mkdir(
    409        self, parent_inode: InodeT, name: bytes, mode: int, ctx: RequestContext
    410    ) -> EntryAttributes:
    411        return await self._create(parent_inode, name, mode, ctx)
    412
    413    async def statfs(self, ctx: RequestContext) -> StatvfsData:
    414        stat_ = StatvfsData()
    415
    416        stat_.f_bsize = 512
    417        stat_.f_frsize = 512
    418
    419        size = self.get_row('SELECT SUM(size) FROM inodes')[0]
    420        stat_.f_blocks = size // stat_.f_frsize
    421        stat_.f_bfree = max(size // stat_.f_frsize, 1024)
    422        stat_.f_bavail = stat_.f_bfree
    423
    424        inodes = self.get_row('SELECT COUNT(id) FROM inodes')[0]
    425        stat_.f_files = inodes
    426        stat_.f_ffree = max(inodes, 100)
    427        stat_.f_favail = stat_.f_ffree
    428
    429        return stat_
    430
    431    async def open(self, inode: InodeT, flags: int, ctx: RequestContext) -> FileInfo:
    432        self.inode_open_count[inode] += 1
    433
    434        # For simplicity, we use the inode as file handle
    435        return FileInfo(fh=FileHandleT(inode))
    436
    437    async def access(self, inode: InodeT, mode: int, ctx: RequestContext) -> bool:
    438        # Yeah, could be a function and has unused arguments
    439        # pylint: disable=R0201,W0613
    440        return True
    441
    442    async def create(
    443        self, parent_inode: InodeT, name: bytes, mode: int, flags: int, ctx: RequestContext
    444    ) -> tuple[FileInfo, EntryAttributes]:
    445        # pylint: disable=W0612
    446        entry = await self._create(parent_inode, name, mode, ctx)
    447        self.inode_open_count[entry.st_ino] += 1
    448        # For simplicity, we use the inode as file handle
    449        return (FileInfo(fh=FileHandleT(entry.st_ino)), entry)
    450
    451    async def _create(
    452        self,
    453        parent_inode: InodeT,
    454        name: bytes,
    455        mode: int,
    456        ctx: RequestContext,
    457        rdev: int = 0,
    458        target: bytes | None = None,
    459    ) -> EntryAttributes:
    460        if (await self.getattr(parent_inode, ctx)).st_nlink == 0:
    461            log.warning('Attempted to create entry %s with unlinked parent %d', name, parent_inode)
    462            raise FUSEError(errno.EINVAL)
    463
    464        now_ns = int(time() * 1e9)
    465        self.cursor.execute(
    466            'INSERT INTO inodes (uid, gid, mode, mtime_ns, atime_ns, '
    467            'ctime_ns, target, rdev) VALUES(?, ?, ?, ?, ?, ?, ?, ?)',
    468            (ctx.uid, ctx.gid, mode, now_ns, now_ns, now_ns, target, rdev),
    469        )
    470
    471        inode = cast(InodeT, self.cursor.lastrowid)
    472        self.db.execute(
    473            "INSERT INTO contents(name, inode, parent_inode) VALUES(?,?,?)",
    474            (name, inode, parent_inode),
    475        )
    476        return await self.getattr(inode, ctx)
    477
    478    async def read(self, fh: FileHandleT, off: int, size: int) -> bytes:
    479        data = self.get_row('SELECT data FROM inodes WHERE id=?', (fh,))[0]
    480        if data is None:
    481            data = b''
    482        return data[off : off + size]
    483
    484    async def write(self, fh: FileHandleT, off: int, buf: bytes) -> int:
    485        data = self.get_row('SELECT data FROM inodes WHERE id=?', (fh,))[0]
    486        if data is None:
    487            data = b''
    488        data = data[:off] + buf + data[off + len(buf) :]
    489
    490        self.cursor.execute(
    491            'UPDATE inodes SET data=?, size=? WHERE id=?', (memoryview(data), len(data), fh)
    492        )
    493        return len(buf)
    494
    495    async def release(self, fh: FileHandleT) -> None:
    496        inode = fh
    497        self.inode_open_count[inode] -= 1
    498
    499        if self.inode_open_count[inode] == 0:
    500            del self.inode_open_count[inode]
    501            if (await self.getattr(inode)).st_nlink == 0:
    502                self.cursor.execute("DELETE FROM inodes WHERE id=?", (inode,))
    503
    504
    505class NoUniqueValueError(Exception):
    506    def __str__(self) -> str:
    507        return 'Query generated more than 1 result row'
    508
    509
    510class NoSuchRowError(Exception):
    511    def __str__(self) -> str:
    512        return 'Query produced 0 result rows'
    513
    514
    515def init_logging(debug: bool = False) -> None:
    516    formatter = logging.Formatter(
    517        '%(asctime)s.%(msecs)03d %(threadName)s: [%(name)s] %(message)s',
    518        datefmt="%Y-%m-%d %H:%M:%S",
    519    )
    520    handler = logging.StreamHandler()
    521    handler.setFormatter(formatter)
    522    root_logger = logging.getLogger()
    523    if debug:
    524        handler.setLevel(logging.DEBUG)
    525        root_logger.setLevel(logging.DEBUG)
    526    else:
    527        handler.setLevel(logging.INFO)
    528        root_logger.setLevel(logging.INFO)
    529    root_logger.addHandler(handler)
    530
    531
    532def parse_args() -> Namespace:
    533    '''Parse command line'''
    534
    535    parser = ArgumentParser()
    536
    537    parser.add_argument('mountpoint', type=str, help='Where to mount the file system')
    538    parser.add_argument(
    539        '--debug', action='store_true', default=False, help='Enable debugging output'
    540    )
    541    parser.add_argument(
    542        '--debug-fuse', action='store_true', default=False, help='Enable FUSE debugging output'
    543    )
    544
    545    return parser.parse_args()
    546
    547
    548if __name__ == '__main__':
    549    options = parse_args()
    550    init_logging(options.debug)
    551    operations = Operations()
    552
    553    fuse_options = set(pyfuse3.default_options)
    554    fuse_options.add('fsname=tmpfs')
    555    fuse_options.discard('default_permissions')
    556    if options.debug_fuse:
    557        fuse_options.add('debug')
    558    pyfuse3.init(operations, options.mountpoint, fuse_options)
    559
    560    try:
    561        trio.run(pyfuse3.main)
    562    except:
    563        pyfuse3.close(unmount=False)
    564        raise
    565
    566    pyfuse3.close()

## Passthrough / Overlay File System

(shipped as `examples/passthroughfs.py`)

      1#!/usr/bin/env python3
      2'''
      3passthroughfs.py - Example file system for pyfuse3
      4
      5This file system mirrors the contents of a specified directory tree.
      6
      7Caveats:
      8
      9 * Inode generation numbers are not passed through but set to zero.
     10
     11 * Block size (st_blksize) and number of allocated blocks (st_blocks) are not
     12   passed through.
     13
     14 * Performance for large directories is not good, because the directory
     15   is always read completely.
     16
     17 * There may be a way to break-out of the directory tree.
     18
     19 * The readdir implementation is not fully POSIX compliant. If a directory
     20   contains hardlinks and is modified during a readdir call, readdir()
     21   may return some of the hardlinked files twice or omit them completely.
     22
     23 * If you delete or rename files in the underlying file system, the
     24   passthrough file system will get confused.
     25
     26Copyright ©  Nikolaus Rath <Nikolaus.org>
     27
     28Permission is hereby granted, free of charge, to any person obtaining a copy of
     29this software and associated documentation files (the "Software"), to deal in
     30the Software without restriction, including without limitation the rights to
     31use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
     32the Software, and to permit persons to whom the Software is furnished to do so.
     33
     34THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
     35IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
     36FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
     37COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
     38IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
     39CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
     40'''
     41
     42import errno
     43import faulthandler
     44import logging
     45import os
     46import stat as stat_m
     47import sys
     48from argparse import ArgumentParser, Namespace
     49from collections import defaultdict
     50from collections.abc import Sequence
     51from os import fsdecode, fsencode
     52
     53import trio
     54
     55import pyfuse3
     56from pyfuse3 import (
     57    EntryAttributes,
     58    FileHandleT,
     59    FileInfo,
     60    FUSEError,
     61    InodeT,
     62    ReaddirToken,
     63    RequestContext,
     64    SetattrFields,
     65    StatvfsData,
     66)
     67
     68faulthandler.enable()
     69
     70log = logging.getLogger(__name__)
     71
     72
     73class Operations(pyfuse3.Operations):
     74    def __init__(self, source: str, enable_writeback_cache: bool = False) -> None:
     75        super().__init__()
     76        self.enable_writeback_cache = enable_writeback_cache
     77        self._inode_path_map: dict[InodeT, str | set[str]] = {pyfuse3.ROOT_INODE: source}
     78        self._lookup_cnt: defaultdict[InodeT, int] = defaultdict(lambda: 0)
     79        self._fd_inode_map: dict[int, InodeT] = dict()
     80        self._inode_fd_map: dict[InodeT, int] = dict()
     81        self._fd_open_count: dict[int, int] = dict()
     82
     83    def _inode_to_path(self, inode: InodeT) -> str:
     84        try:
     85            val = self._inode_path_map[inode]
     86        except KeyError:
     87            raise FUSEError(errno.ENOENT)
     88
     89        if isinstance(val, set):
     90            # In case of hardlinks, pick any path
     91            val = next(iter(val))
     92        return val
     93
     94    def _add_path(self, inode: InodeT, path: str) -> None:
     95        log.debug('_add_path for %d, %s', inode, path)
     96        self._lookup_cnt[inode] += 1
     97
     98        # With hardlinks, one inode may map to multiple paths.
     99        if inode not in self._inode_path_map:
    100            self._inode_path_map[inode] = path
    101            return
    102
    103        val = self._inode_path_map[inode]
    104        if isinstance(val, set):
    105            val.add(path)
    106        elif val != path:
    107            self._inode_path_map[inode] = {path, val}
    108
    109    async def forget(self, inode_list: Sequence[tuple[InodeT, int]]) -> None:
    110        for inode, nlookup in inode_list:
    111            if self._lookup_cnt[inode] > nlookup:
    112                self._lookup_cnt[inode] -= nlookup
    113                continue
    114            log.debug('forgetting about inode %d', inode)
    115            assert inode not in self._inode_fd_map
    116            del self._lookup_cnt[inode]
    117            try:
    118                del self._inode_path_map[inode]
    119            except KeyError:  # may have been deleted
    120                pass
    121
    122    async def lookup(
    123        self, parent_inode: InodeT, name: bytes, ctx: RequestContext
    124    ) -> EntryAttributes:
    125        name_str = fsdecode(name)
    126        log.debug('lookup for %s in %d', name_str, parent_inode)
    127        path = os.path.join(self._inode_to_path(parent_inode), name_str)
    128        attr = self._getattr(path=path)
    129        if name_str != '.' and name_str != '..':
    130            self._add_path(InodeT(attr.st_ino), path)
    131        return attr
    132
    133    async def getattr(self, inode: InodeT, ctx: RequestContext | None = None) -> EntryAttributes:
    134        if inode in self._inode_fd_map:
    135            return self._getattr(fd=self._inode_fd_map[inode])
    136        else:
    137            return self._getattr(path=self._inode_to_path(inode))
    138
    139    def _getattr(self, path: str | None = None, fd: int | None = None) -> EntryAttributes:
    140        assert fd is None or path is None
    141        assert not (fd is None and path is None)
    142        try:
    143            if fd is None:
    144                assert path is not None
    145                stat = os.lstat(path)
    146            else:
    147                stat = os.fstat(fd)
    148        except OSError as exc:
    149            assert exc.errno is not None
    150            raise FUSEError(exc.errno)
    151
    152        entry = EntryAttributes()
    153        for attr in (
    154            'st_ino',
    155            'st_mode',
    156            'st_nlink',
    157            'st_uid',
    158            'st_gid',
    159            'st_rdev',
    160            'st_size',
    161            'st_atime_ns',
    162            'st_mtime_ns',
    163            'st_ctime_ns',
    164        ):
    165            setattr(entry, attr, getattr(stat, attr))
    166        entry.generation = 0
    167        entry.entry_timeout = 0
    168        entry.attr_timeout = 0
    169        entry.st_blksize = 512
    170        entry.st_blocks = (entry.st_size + entry.st_blksize - 1) // entry.st_blksize
    171
    172        return entry
    173
    174    async def readlink(self, inode: InodeT, ctx: RequestContext) -> bytes:
    175        path = self._inode_to_path(inode)
    176        try:
    177            target = os.readlink(path)
    178        except OSError as exc:
    179            assert exc.errno is not None
    180            raise FUSEError(exc.errno)
    181        return fsencode(target)
    182
    183    async def opendir(self, inode: InodeT, ctx: RequestContext) -> FileHandleT:
    184        # For simplicity, we use the inode as file handle
    185        return FileHandleT(inode)
    186
    187    async def readdir(self, fh: FileHandleT, start_id: int, token: ReaddirToken) -> None:
    188        path = self._inode_to_path(InodeT(fh))
    189        log.debug('reading %s', path)
    190        entries: list[tuple[InodeT, str, EntryAttributes]] = []
    191        for name in os.listdir(path):
    192            if name == '.' or name == '..':
    193                continue
    194            attr = self._getattr(path=os.path.join(path, name))
    195            entries.append((InodeT(attr.st_ino), name, attr))
    196
    197        log.debug('read %d entries, starting at %d', len(entries), start_id)
    198
    199        # This is not fully posix compatible. If there are hardlinks
    200        # (two names with the same inode), we don't have a unique
    201        # offset to start in between them. Note that we cannot simply
    202        # count entries, because then we would skip over entries
    203        # (or return them more than once) if the number of directory
    204        # entries changes between two calls to readdir().
    205        for ino, name, attr in sorted(entries):
    206            if ino <= start_id:
    207                continue
    208            if not pyfuse3.readdir_reply(token, fsencode(name), attr, ino):
    209                break
    210            self._add_path(attr.st_ino, os.path.join(path, name))
    211
    212    async def unlink(self, parent_inode: InodeT, name: bytes, ctx: RequestContext) -> None:
    213        name_str = fsdecode(name)
    214        parent = self._inode_to_path(parent_inode)
    215        path = os.path.join(parent, name_str)
    216        try:
    217            inode = os.lstat(path).st_ino
    218            os.unlink(path)
    219        except OSError as exc:
    220            assert exc.errno is not None
    221            raise FUSEError(exc.errno)
    222        if inode in self._lookup_cnt:
    223            self._forget_path(InodeT(inode), path)
    224
    225    async def rmdir(self, parent_inode: InodeT, name: bytes, ctx: RequestContext) -> None:
    226        name_str = fsdecode(name)
    227        parent = self._inode_to_path(parent_inode)
    228        path = os.path.join(parent, name_str)
    229        try:
    230            inode = os.lstat(path).st_ino
    231            os.rmdir(path)
    232        except OSError as exc:
    233            assert exc.errno is not None
    234            raise FUSEError(exc.errno)
    235        if inode in self._lookup_cnt:
    236            self._forget_path(InodeT(inode), path)
    237
    238    def _forget_path(self, inode: InodeT, path: str) -> None:
    239        log.debug('forget %s for %d', path, inode)
    240        val = self._inode_path_map[inode]
    241        if isinstance(val, set):
    242            val.remove(path)
    243            if len(val) == 1:
    244                self._inode_path_map[inode] = next(iter(val))
    245        else:
    246            del self._inode_path_map[inode]
    247
    248    async def symlink(
    249        self, parent_inode: InodeT, name: bytes, target: bytes, ctx: RequestContext
    250    ) -> EntryAttributes:
    251        name_str = fsdecode(name)
    252        target_str = fsdecode(target)
    253        parent = self._inode_to_path(parent_inode)
    254        path = os.path.join(parent, name_str)
    255        try:
    256            os.symlink(target_str, path)
    257            os.lchown(path, ctx.uid, ctx.gid)
    258        except OSError as exc:
    259            assert exc.errno is not None
    260            raise FUSEError(exc.errno)
    261        inode = InodeT(os.lstat(path).st_ino)
    262        self._add_path(inode, path)
    263        return await self.getattr(inode, ctx)
    264
    265    async def rename(
    266        self,
    267        parent_inode_old: InodeT,
    268        name_old: bytes,
    269        parent_inode_new: InodeT,
    270        name_new: bytes,
    271        flags: int,
    272        ctx: RequestContext,
    273    ) -> None:
    274        if flags != 0:
    275            raise FUSEError(errno.EINVAL)
    276
    277        name_old_str = fsdecode(name_old)
    278        name_new_str = fsdecode(name_new)
    279        parent_old = self._inode_to_path(parent_inode_old)
    280        parent_new = self._inode_to_path(parent_inode_new)
    281        path_old = os.path.join(parent_old, name_old_str)
    282        path_new = os.path.join(parent_new, name_new_str)
    283        try:
    284            os.rename(path_old, path_new)
    285            inode = os.lstat(path_new).st_ino
    286        except OSError as exc:
    287            assert exc.errno is not None
    288            raise FUSEError(exc.errno)
    289        if inode not in self._lookup_cnt:
    290            return
    291
    292        val = self._inode_path_map[inode]
    293        if isinstance(val, set):
    294            assert len(val) > 1
    295            val.add(path_new)
    296            val.remove(path_old)
    297        else:
    298            assert val == path_old
    299            self._inode_path_map[inode] = path_new
    300
    301    async def link(
    302        self, inode: InodeT, new_parent_inode: InodeT, new_name: bytes, ctx: RequestContext
    303    ) -> EntryAttributes:
    304        new_name_str = fsdecode(new_name)
    305        parent = self._inode_to_path(new_parent_inode)
    306        path = os.path.join(parent, new_name_str)
    307        try:
    308            os.link(self._inode_to_path(inode), path, follow_symlinks=False)
    309        except OSError as exc:
    310            assert exc.errno is not None
    311            raise FUSEError(exc.errno)
    312        self._add_path(inode, path)
    313        return await self.getattr(inode, ctx)
    314
    315    async def setattr(
    316        self,
    317        inode: InodeT,
    318        attr: EntryAttributes,
    319        fields: SetattrFields,
    320        fh: FileHandleT | None,
    321        ctx: RequestContext,
    322    ) -> EntryAttributes:
    323        try:
    324            if fields.update_size:
    325                if fh is None:
    326                    os.truncate(self._inode_to_path(inode), attr.st_size)
    327                else:
    328                    os.ftruncate(fh, attr.st_size)
    329
    330            if fields.update_mode:
    331                # Under Linux, chmod always resolves symlinks so we should
    332                # actually never get a setattr() request for a symbolic
    333                # link.
    334                assert not stat_m.S_ISLNK(attr.st_mode)
    335                if fh is None:
    336                    os.chmod(self._inode_to_path(inode), stat_m.S_IMODE(attr.st_mode))
    337                else:
    338                    os.fchmod(fh, stat_m.S_IMODE(attr.st_mode))
    339
    340            if fields.update_uid and fields.update_gid:
    341                if fh is None:
    342                    os.chown(
    343                        self._inode_to_path(inode), attr.st_uid, attr.st_gid, follow_symlinks=False
    344                    )
    345                else:
    346                    os.fchown(fh, attr.st_uid, attr.st_gid)
    347
    348            elif fields.update_uid:
    349                if fh is None:
    350                    os.chown(self._inode_to_path(inode), attr.st_uid, -1, follow_symlinks=False)
    351                else:
    352                    os.fchown(fh, attr.st_uid, -1)
    353
    354            elif fields.update_gid:
    355                if fh is None:
    356                    os.chown(self._inode_to_path(inode), -1, attr.st_gid, follow_symlinks=False)
    357                else:
    358                    os.fchown(fh, -1, attr.st_gid)
    359
    360            if fields.update_atime and fields.update_mtime:
    361                if fh is None:
    362                    os.utime(
    363                        self._inode_to_path(inode),
    364                        None,
    365                        follow_symlinks=False,
    366                        ns=(attr.st_atime_ns, attr.st_mtime_ns),
    367                    )
    368                else:
    369                    os.utime(fh, None, ns=(attr.st_atime_ns, attr.st_mtime_ns))
    370            elif fields.update_atime or fields.update_mtime:
    371                # We can only set both values, so we first need to retrieve the
    372                # one that we shouldn't be changing.
    373                if fh is None:
    374                    path = self._inode_to_path(inode)
    375                    oldstat = os.stat(path, follow_symlinks=False)
    376                else:
    377                    oldstat = os.fstat(fh)
    378                if not fields.update_atime:
    379                    attr.st_atime_ns = oldstat.st_atime_ns
    380                else:
    381                    attr.st_mtime_ns = oldstat.st_mtime_ns
    382                if fh is None:
    383                    os.utime(
    384                        path,  # pyright: ignore[reportPossiblyUnboundVariable]
    385                        None,
    386                        follow_symlinks=False,
    387                        ns=(attr.st_atime_ns, attr.st_mtime_ns),
    388                    )
    389                else:
    390                    os.utime(fh, None, ns=(attr.st_atime_ns, attr.st_mtime_ns))
    391
    392        except OSError as exc:
    393            assert exc.errno is not None
    394            raise FUSEError(exc.errno)
    395
    396        return await self.getattr(inode, ctx)
    397
    398    async def mknod(
    399        self, parent_inode: InodeT, name: bytes, mode: int, rdev: int, ctx: RequestContext
    400    ) -> EntryAttributes:
    401        path = os.path.join(self._inode_to_path(parent_inode), fsdecode(name))
    402        try:
    403            os.mknod(path, mode=(mode & ~ctx.umask), device=rdev)
    404            os.chown(path, ctx.uid, ctx.gid)
    405        except OSError as exc:
    406            assert exc.errno is not None
    407            raise FUSEError(exc.errno)
    408        attr = self._getattr(path=path)
    409        self._add_path(attr.st_ino, path)
    410        return attr
    411
    412    async def mkdir(
    413        self, parent_inode: InodeT, name: bytes, mode: int, ctx: RequestContext
    414    ) -> EntryAttributes:
    415        path = os.path.join(self._inode_to_path(parent_inode), fsdecode(name))
    416        try:
    417            os.mkdir(path, mode=(mode & ~ctx.umask))
    418            os.chown(path, ctx.uid, ctx.gid)
    419        except OSError as exc:
    420            assert exc.errno is not None
    421            raise FUSEError(exc.errno)
    422        attr = self._getattr(path=path)
    423        self._add_path(attr.st_ino, path)
    424        return attr
    425
    426    async def statfs(self, ctx: RequestContext) -> StatvfsData:
    427        root = self._inode_path_map[pyfuse3.ROOT_INODE]
    428        assert isinstance(root, str)
    429        stat_ = StatvfsData()
    430        try:
    431            statfs = os.statvfs(root)
    432        except OSError as exc:
    433            assert exc.errno is not None
    434            raise FUSEError(exc.errno)
    435        for attr in (
    436            'f_bsize',
    437            'f_frsize',
    438            'f_blocks',
    439            'f_bfree',
    440            'f_bavail',
    441            'f_files',
    442            'f_ffree',
    443            'f_favail',
    444        ):
    445            setattr(stat_, attr, getattr(statfs, attr))
    446        stat_.f_namemax = statfs.f_namemax - (len(root) + 1)
    447        return stat_
    448
    449    async def open(self, inode: InodeT, flags: int, ctx: RequestContext) -> FileInfo:
    450        if inode in self._inode_fd_map:
    451            fd = self._inode_fd_map[inode]
    452            self._fd_open_count[fd] += 1
    453            return FileInfo(fh=FileHandleT(fd))
    454        assert flags & os.O_CREAT == 0
    455        try:
    456            fd = os.open(self._inode_to_path(inode), flags)
    457        except OSError as exc:
    458            assert exc.errno is not None
    459            raise FUSEError(exc.errno)
    460        self._inode_fd_map[inode] = fd
    461        self._fd_inode_map[fd] = inode
    462        self._fd_open_count[fd] = 1
    463        return FileInfo(fh=fd)
    464
    465    async def create(
    466        self, parent_inode: InodeT, name: bytes, mode: int, flags: int, ctx: RequestContext
    467    ) -> tuple[FileInfo, EntryAttributes]:
    468        path = os.path.join(self._inode_to_path(parent_inode), fsdecode(name))
    469        try:
    470            fd = os.open(path, flags | os.O_CREAT | os.O_TRUNC)
    471        except OSError as exc:
    472            assert exc.errno is not None
    473            raise FUSEError(exc.errno)
    474        attr = self._getattr(fd=fd)
    475        self._add_path(attr.st_ino, path)
    476        self._inode_fd_map[attr.st_ino] = fd
    477        self._fd_inode_map[fd] = attr.st_ino
    478        self._fd_open_count[fd] = 1
    479        return (FileInfo(fh=fd), attr)
    480
    481    async def read(self, fh: FileHandleT, off: int, size: int) -> bytes:
    482        os.lseek(fh, off, os.SEEK_SET)
    483        return os.read(fh, size)
    484
    485    async def write(self, fh: FileHandleT, off: int, buf: bytes) -> int:
    486        os.lseek(fh, off, os.SEEK_SET)
    487        return os.write(fh, buf)
    488
    489    async def release(self, fh: FileHandleT) -> None:
    490        if self._fd_open_count[fh] > 1:
    491            self._fd_open_count[fh] -= 1
    492            return
    493
    494        del self._fd_open_count[fh]
    495        inode = self._fd_inode_map[fh]
    496        del self._inode_fd_map[inode]
    497        del self._fd_inode_map[fh]
    498        try:
    499            os.close(fh)
    500        except OSError as exc:
    501            assert exc.errno is not None
    502            raise FUSEError(exc.errno)
    503
    504
    505def init_logging(debug: bool = False) -> None:
    506    formatter = logging.Formatter(
    507        '%(asctime)s.%(msecs)03d %(threadName)s: [%(name)s] %(message)s',
    508        datefmt="%Y-%m-%d %H:%M:%S",
    509    )
    510    handler = logging.StreamHandler()
    511    handler.setFormatter(formatter)
    512    root_logger = logging.getLogger()
    513    if debug:
    514        handler.setLevel(logging.DEBUG)
    515        root_logger.setLevel(logging.DEBUG)
    516    else:
    517        handler.setLevel(logging.INFO)
    518        root_logger.setLevel(logging.INFO)
    519    root_logger.addHandler(handler)
    520
    521
    522def parse_args(args: list[str]) -> Namespace:
    523    '''Parse command line'''
    524
    525    parser = ArgumentParser()
    526
    527    parser.add_argument('source', type=str, help='Directory tree to mirror')
    528    parser.add_argument('mountpoint', type=str, help='Where to mount the file system')
    529    parser.add_argument(
    530        '--debug', action='store_true', default=False, help='Enable debugging output'
    531    )
    532    parser.add_argument(
    533        '--debug-fuse', action='store_true', default=False, help='Enable FUSE debugging output'
    534    )
    535    parser.add_argument(
    536        '--enable-writeback-cache',
    537        action='store_true',
    538        default=False,
    539        help='Enable writeback cache (default: disabled)',
    540    )
    541
    542    return parser.parse_args(args)
    543
    544
    545def main() -> None:
    546    options = parse_args(sys.argv[1:])
    547    init_logging(options.debug)
    548    operations = Operations(options.source, enable_writeback_cache=options.enable_writeback_cache)
    549
    550    log.debug('Mounting...')
    551    fuse_options = set(pyfuse3.default_options)
    552    fuse_options.add('fsname=passthroughfs')
    553    if options.debug_fuse:
    554        fuse_options.add('debug')
    555    pyfuse3.init(operations, options.mountpoint, fuse_options)
    556
    557    try:
    558        log.debug('Entering main loop..')
    559        trio.run(pyfuse3.main)
    560    except:
    561        pyfuse3.close(unmount=False)
    562        raise
    563
    564    log.debug('Unmounting..')
    565    pyfuse3.close()
    566
    567
    568if __name__ == '__main__':
    569    main()