Data Structures

class EntryAttributes:
    """
    Attributes of directory entries.

    Most attributes correspond to stat struct elements.
    Used to return file/directory metadata from various operations.
    Can be pickled for caching/serialization.
    """

    st_ino: InodeT
    generation: int
    entry_timeout: Union[float, int]
    attr_timeout: Union[float, int]
    st_mode: ModeT
    st_nlink: int
    st_uid: int
    st_gid: int
    st_rdev: int
    st_size: int
    st_blksize: int
    st_blocks: int
    st_atime_ns: int
    st_ctime_ns: int
    st_mtime_ns: int
    st_birthtime_ns: int

    def __init__(self) -> None: ...
    def __getstate__(self) -> StatDict: ...
    def __setstate__(self, state: StatDict) -> None: ...

import pyfuse3
import stat
import os
import time

entry = pyfuse3.EntryAttributes()
entry.st_ino = 42
entry.st_mode = stat.S_IFDIR | 0o755
entry.st_nlink = 2  # . and ..
entry.st_uid = os.getuid()
entry.st_gid = os.getgid()
entry.st_size = 4096
entry.st_blksize = 4096
entry.st_blocks = 8  # 4096 / 512

# Set timestamps (nanoseconds since epoch)
now_ns = int(time.time() * 1e9)
entry.st_atime_ns = now_ns
entry.st_mtime_ns = now_ns
entry.st_ctime_ns = now_ns
entry.st_birthtime_ns = now_ns  # 0 on Linux

# Set cache timeouts
entry.entry_timeout = 1.0  # Cache entry for 1 second
entry.attr_timeout = 1.0   # Cache attributes for 1 second

import time

# Current time in nanoseconds
now_ns = time.time_ns()  # Python 3.7+
# OR
now_ns = int(time.time() * 1_000_000_000)

# Convert from seconds to nanoseconds
seconds = 1234567890
nanoseconds = seconds * 1_000_000_000

# Convert from nanoseconds to seconds
nanoseconds = 1234567890000000000
seconds = nanoseconds / 1_000_000_000

import stat

# File types (mutually exclusive)
stat.S_IFREG   # Regular file (0o100000)
stat.S_IFDIR   # Directory (0o040000)
stat.S_IFLNK   # Symbolic link (0o120000)
stat.S_IFCHR   # Character device (0o020000)
stat.S_IFBLK   # Block device (0o060000)
stat.S_IFIFO   # Named pipe (FIFO) (0o010000)
stat.S_IFSOCK  # Socket (0o140000)

# Extract file type
file_type = entry.st_mode & stat.S_IFMT

# Check file type
if stat.S_ISREG(entry.st_mode):
    # Regular file
    pass
elif stat.S_ISDIR(entry.st_mode):
    # Directory
    pass
elif stat.S_ISLNK(entry.st_mode):
    # Symbolic link
    pass

# Set file type and permissions
entry.st_mode = stat.S_IFREG | 0o644
entry.st_mode = stat.S_IFDIR | 0o755
entry.st_mode = stat.S_IFLNK | 0o777  # Symlinks always 0o777

import math

# Simple: assume all bytes allocated
entry.st_size = 10000
entry.st_blocks = math.ceil(entry.st_size / 512)

# More accurate: account for block size
block_size = 4096
blocks_used = math.ceil(entry.st_size / block_size)
entry.st_blocks = blocks_used * (block_size // 512)

# Sparse files: fewer blocks than size would indicate
entry.st_size = 1_000_000_000  # 1 GB
entry.st_blocks = 0  # Sparse file, no blocks allocated

# No caching (always re-query)
entry.entry_timeout = 0
entry.attr_timeout = 0

# Short caching (frequently changing data)
entry.entry_timeout = 1.0  # 1 second
entry.attr_timeout = 1.0

# Long caching (rarely changing data)
entry.entry_timeout = 3600.0  # 1 hour
entry.attr_timeout = 3600.0

# Infinite caching (immutable data)
entry.entry_timeout = float('inf')
entry.attr_timeout = float('inf')

import pickle

entry = pyfuse3.EntryAttributes()
# ... set attributes ...

# Serialize
data = pickle.dumps(entry)

# Deserialize
restored = pickle.loads(data)

class FileInfo:
    """
    Options and data for open files.

    Returned by open() and create() handlers.
    Controls caching behavior for the open file.
    """

    fh: FileHandleT
    direct_io: bool
    keep_cache: bool
    nonseekable: bool

    def __init__(
        self,
        fh: FileHandleT = 0,
        direct_io: bool = False,
        keep_cache: bool = True,
        nonseekable: bool = False
    ) -> None: ...

import pyfuse3

class MyFS(pyfuse3.Operations):
    def __init__(self):
        super().__init__()
        self.next_fh = 1
        self.open_files = {}

    async def open(self, inode, flags, ctx):
        # Allocate file handle
        fh = self.next_fh
        self.next_fh += 1
        self.open_files[fh] = {'inode': inode, 'flags': flags}

        # Return FileInfo with options
        fi = pyfuse3.FileInfo(
            fh=fh,
            direct_io=False,      # Enable caching
            keep_cache=True,      # Keep cache on open
            nonseekable=False     # File supports seeking
        )
        return fi

    async def create(self, parent_inode, name, mode, flags, ctx):
        # ... create inode ...
        
        fh = self.next_fh
        self.next_fh += 1
        self.open_files[fh] = {'inode': new_inode, 'flags': flags}

        fi = pyfuse3.FileInfo(fh=fh)
        entry = await self.getattr(new_inode, ctx)
        return (fi, entry)

async def open(self, inode, flags, ctx):
    fh = self.allocate_fh(inode, flags)
    
    # Frequently changing file: bypass cache
    if self.is_volatile(inode):
        return pyfuse3.FileInfo(fh=fh, direct_io=True, keep_cache=False)
    
    # File modified externally: invalidate cache
    if self.modified_externally(inode):
        return pyfuse3.FileInfo(fh=fh, keep_cache=False)
    
    # Normal file: use cache
    return pyfuse3.FileInfo(fh=fh, direct_io=False, keep_cache=True)

async def open(self, inode, flags, ctx):
    fh = self.allocate_fh(inode)
    
    # Check if file is a pipe or socket
    attr = await self.getattr(inode, ctx)
    is_nonseekable = stat.S_ISFIFO(attr.st_mode) or stat.S_ISSOCK(attr.st_mode)
    
    return pyfuse3.FileInfo(
        fh=fh,
        nonseekable=is_nonseekable,
        direct_io=is_nonseekable  # Usually want direct_io too
    )

class RequestContext:
    """
    Information about the request caller.

    Passed to most Operations methods.
    Cannot be pickled.
    All properties are read-only.
    """

    @property
    def uid(self) -> int: ...  # User ID

    @property
    def pid(self) -> int: ...  # Process ID

    @property
    def gid(self) -> int: ...  # Group ID

    @property
    def umask(self) -> int: ...  # File mode creation mask

    def __getstate__(self) -> None: ...  # Raises PicklingError

import pyfuse3
import stat
import os

async def create(self, parent_inode, name, mode, flags, ctx):
    # Apply umask to mode
    final_mode = mode & ~ctx.umask
    
    entry = pyfuse3.EntryAttributes()
    entry.st_mode = stat.S_IFREG | final_mode
    
    # Use caller's uid/gid for new file
    entry.st_uid = ctx.uid
    entry.st_gid = ctx.gid
    
    # ... create file ...
    return (fi, entry)

import pyfuse3
import os

async def check_access(self, inode, mode, ctx):
    """Check if ctx has permission for mode on inode."""
    entry = await self.getattr(inode, ctx)
    
    # Owner permissions
    if ctx.uid == entry.st_uid:
        owner_perms = (entry.st_mode >> 6) & 0o7
        if (mode & owner_perms) == mode:
            return True
    
    # Group permissions
    if ctx.gid == entry.st_gid:
        group_perms = (entry.st_mode >> 3) & 0o7
        if (mode & group_perms) == mode:
            return True
    
    # Check supplementary groups
    sup_groups = pyfuse3.get_sup_groups(ctx.pid)
    if entry.st_gid in sup_groups:
        group_perms = (entry.st_mode >> 3) & 0o7
        if (mode & group_perms) == mode:
            return True
    
    # Others permissions
    other_perms = entry.st_mode & 0o7
    return (mode & other_perms) == mode

import stat

async def mknod(self, parent_inode, name, mode, rdev, ctx):
    # Apply umask
    final_mode = mode & ~ctx.umask
    
    entry = pyfuse3.EntryAttributes()
    entry.st_mode = final_mode
    # ... rest of creation ...

async def mkdir(self, parent_inode, name, mode, ctx):
    # Apply umask
    final_mode = (stat.S_IFDIR | mode) & ~ctx.umask
    
    entry = pyfuse3.EntryAttributes()
    entry.st_mode = final_mode
    # ... rest of creation ...

class SetattrFields:
    """
    Indicates which attributes to update in setattr.

    Passed to Operations.setattr().
    Cannot be pickled.
    All properties are read-only booleans.
    """

    @property
    def update_atime(self) -> bool: ...

    @property
    def update_mtime(self) -> bool: ...

    @property
    def update_ctime(self) -> bool: ...

    @property
    def update_mode(self) -> bool: ...

    @property
    def update_uid(self) -> bool: ...

    @property
    def update_gid(self) -> bool: ...

    @property
    def update_size(self) -> bool: ...

    def __init__(self) -> None: ...
    def __getstate__(self) -> None: ...  # Raises PicklingError

import time
import pyfuse3

async def setattr(self, inode, attr, fields, fh, ctx):
    # Get current attributes
    entry = await self.getattr(inode, ctx)

    # Update only specified fields
    if fields.update_mode:
        # Update permissions only (not file type)
        new_mode = attr.st_mode & 0o7777
        entry.st_mode = (entry.st_mode & ~0o7777) | new_mode

    if fields.update_uid:
        entry.st_uid = attr.st_uid

    if fields.update_gid:
        entry.st_gid = attr.st_gid

    if fields.update_size:
        # Truncate or extend file
        old_size = entry.st_size
        new_size = attr.st_size
        if new_size < old_size:
            # Truncate
            self.truncate_file(inode, new_size)
        elif new_size > old_size:
            # Extend (with zeros)
            self.extend_file(inode, new_size)
        entry.st_size = new_size

    if fields.update_atime:
        entry.st_atime_ns = attr.st_atime_ns

    if fields.update_mtime:
        entry.st_mtime_ns = attr.st_mtime_ns

    # Always update ctime when metadata changes
    entry.st_ctime_ns = time.time_ns()

    # Save changes
    self.save_inode(inode, entry)

    return entry

# chmod: updates mode
if fields.update_mode and not any([
    fields.update_size,
    fields.update_uid,
    fields.update_gid,
    fields.update_atime,
    fields.update_mtime
]):
    # Just chmod
    pass

# chown: updates uid/gid
if fields.update_uid or fields.update_gid:
    # chown
    pass

# truncate: updates size
if fields.update_size:
    # truncate/extend
    pass

# touch/utime: updates times
if fields.update_atime or fields.update_mtime:
    # touch/utime
    pass

class StatvfsData:
    """
    Filesystem statistics.

    Attributes correspond to statvfs struct elements.
    Returned by Operations.statfs().
    Can be pickled for caching.
    """

    f_bsize: int
    f_frsize: int
    f_blocks: int
    f_bfree: int
    f_bavail: int
    f_files: int
    f_ffree: int
    f_favail: int
    f_namemax: int

    def __init__(self) -> None: ...
    def __getstate__(self) -> StatDict: ...
    def __setstate__(self, state: StatDict) -> None: ...

import pyfuse3

async def statfs(self, ctx):
    stat = pyfuse3.StatvfsData()
    
    # Block size
    stat.f_bsize = 4096
    stat.f_frsize = 4096
    
    # Total space: 1 GB
    stat.f_blocks = 1_000_000_000 // 4096
    
    # Free space: 500 MB
    free_blocks = 500_000_000 // 4096
    stat.f_bfree = free_blocks
    stat.f_bavail = free_blocks  # All free to users
    
    # Inodes
    stat.f_files = 100_000
    stat.f_ffree = 50_000
    stat.f_favail = 50_000
    
    # Maximum filename length
    stat.f_namemax = 255
    
    return stat

async def statfs(self, ctx):
    stat = pyfuse3.StatvfsData()
    
    # Block size
    block_size = 4096
    stat.f_bsize = block_size
    stat.f_frsize = block_size
    
    # Calculate total space
    total_bytes = self.calculate_total_space()
    stat.f_blocks = total_bytes // block_size
    
    # Calculate free space
    used_bytes = self.calculate_used_space()
    free_bytes = total_bytes - used_bytes
    free_blocks = free_bytes // block_size
    stat.f_bfree = free_blocks
    
    # Reserve 5% for root
    reserved_blocks = stat.f_blocks // 20
    stat.f_bavail = max(0, free_blocks - reserved_blocks)
    
    # Inode statistics
    total_inodes = len(self.all_inodes)
    free_inodes = self.max_inodes - total_inodes
    stat.f_files = self.max_inodes
    stat.f_ffree = free_inodes
    stat.f_favail = free_inodes
    
    stat.f_namemax = 255
    
    return stat

class ReaddirToken:
    """
    Token for readdir operations.

    Passed to Operations.readdir() and used with readdir_reply().
    Not directly instantiated by users.
    Treated as opaque by filesystem implementations.
    """
    pass

async def readdir(self, fh, start_id, token):
    entries = self.get_directory_entries(fh)

    for i, (name, inode) in enumerate(entries):
        if i < start_id:
            continue

        attr = await self.getattr(inode, None)

        # Pass token to readdir_reply
        if not pyfuse3.readdir_reply(token, name, attr, i + 1):
            # Buffer full, stop
            break

StatDict = Mapping[str, int]  # Stat dictionary for pickling

import pyfuse3
import stat
import time

class InodeManager:
    def __init__(self):
        self.inodes = {}
        self.next_inode = pyfuse3.ROOT_INODE + 1
    
    def create_entry(self, mode, uid, gid):
        """Create new entry attributes."""
        entry = pyfuse3.EntryAttributes()
        entry.st_ino = self.next_inode
        self.next_inode += 1
        
        entry.st_mode = mode
        entry.st_nlink = 1
        entry.st_uid = uid
        entry.st_gid = gid
        entry.st_size = 0
        entry.st_blksize = 4096
        entry.st_blocks = 0
        
        now_ns = time.time_ns()
        entry.st_atime_ns = now_ns
        entry.st_mtime_ns = now_ns
        entry.st_ctime_ns = now_ns
        
        entry.entry_timeout = 1.0
        entry.attr_timeout = 1.0
        
        self.inodes[entry.st_ino] = entry
        return entry

import time

def current_time_ns():
    """Get current time in nanoseconds."""
    return time.time_ns()

def update_times(entry, atime=False, mtime=False, ctime=True):
    """Update entry timestamps."""
    now_ns = current_time_ns()
    if atime:
        entry.st_atime_ns = now_ns
    if mtime:
        entry.st_mtime_ns = now_ns
    if ctime:
        entry.st_ctime_ns = now_ns

import math

def update_size(entry, new_size):
    """Update size and blocks."""
    entry.st_size = new_size
    
    # Calculate blocks (512-byte units)
    block_size = 4096
    blocks_used = math.ceil(new_size / block_size)
    entry.st_blocks = blocks_used * (block_size // 512)

# Fast-changing data: short timeouts
entry.entry_timeout = 0.1
entry.attr_timeout = 0.1

# Stable data: long timeouts
entry.entry_timeout = 300.0
entry.attr_timeout = 300.0

# Immutable data: infinite timeouts
entry.entry_timeout = float('inf')
entry.attr_timeout = float('inf')

class EfficientFileHandles:
    def __init__(self):
        self.next_fh = 1
        self.free_fhs = []
        self.open_files = {}
    
    def allocate_fh(self, inode):
        """Allocate file handle, reusing freed handles."""
        if self.free_fhs:
            fh = self.free_fhs.pop()
        else:
            fh = self.next_fh
            self.next_fh += 1
        
        self.open_files[fh] = inode
        return fh
    
    def release_fh(self, fh):
        """Release file handle for reuse."""
        if fh in self.open_files:
            del self.open_files[fh]
            self.free_fhs.append(fh)