Skip to content

Add more type annotations. #129

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Nikratio merged 1 commit into from
Jan 4, 2026
Merged

Add more type annotations. #129

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 16 additions & 14 deletions examples/hello.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,13 +26,13 @@
import logging
import os
import stat
from argparse import ArgumentParser
from argparse import ArgumentParser, Namespace
from typing import cast

import trio

import pyfuse3
from pyfuse3 import FileHandleT, FileInfo, InodeT
from pyfuse3 import EntryAttributes, FileHandleT, FileInfo, InodeT, ReaddirToken, RequestContext

try:
import faulthandler
Expand All @@ -45,14 +45,14 @@


class TestFs(pyfuse3.Operations):
def __init__(self):
def __init__(self) -> None:
super(TestFs, self).__init__()
self.hello_name = b"message"
self.hello_inode = cast(InodeT, pyfuse3.ROOT_INODE + 1)
self.hello_data = b"hello world\n"

async def getattr(self, inode, ctx=None):
entry = pyfuse3.EntryAttributes()
async def getattr(self, inode: InodeT, ctx: RequestContext | None = None) -> EntryAttributes:
entry = EntryAttributes()
if inode == pyfuse3.ROOT_INODE:
entry.st_mode = stat.S_IFDIR | 0o755
entry.st_size = 0
Expand All @@ -72,39 +72,41 @@ async def getattr(self, inode, ctx=None):

return entry

async def lookup(self, parent_inode, name, ctx=None):
async def lookup(
self, parent_inode: InodeT, name: bytes, ctx: RequestContext
) -> EntryAttributes:
if parent_inode != pyfuse3.ROOT_INODE or name != self.hello_name:
raise pyfuse3.FUSEError(errno.ENOENT)
return await self.getattr(self.hello_inode)
return await self.getattr(self.hello_inode, ctx)

async def opendir(self, inode, ctx):
async def opendir(self, inode: InodeT, ctx: RequestContext) -> FileHandleT:
if inode != pyfuse3.ROOT_INODE:
raise pyfuse3.FUSEError(errno.ENOENT)
# For simplicity, we use the inode as file handle
return FileHandleT(inode)

async def readdir(self, fh, start_id, token):
async def readdir(self, fh: FileHandleT, start_id: int, token: ReaddirToken) -> None:
assert fh == pyfuse3.ROOT_INODE

# only one entry
if start_id == 0:
pyfuse3.readdir_reply(token, self.hello_name, await self.getattr(self.hello_inode), 1)
return

async def open(self, inode, flags, ctx):
async def open(self, inode: InodeT, flags: int, ctx: RequestContext) -> FileInfo:
if inode != self.hello_inode:
raise pyfuse3.FUSEError(errno.ENOENT)
if flags & os.O_RDWR or flags & os.O_WRONLY:
raise pyfuse3.FUSEError(errno.EACCES)
# For simplicity, we use the inode as file handle
return FileInfo(fh=FileHandleT(inode))

async def read(self, fh, off, size):
async def read(self, fh: FileHandleT, off: int, size: int) -> bytes:
assert fh == self.hello_inode
return self.hello_data[off : off + size]


def init_logging(debug=False):
def init_logging(debug: bool = False) -> None:
formatter = logging.Formatter(
'%(asctime)s.%(msecs)03d %(threadName)s: [%(name)s] %(message)s',
datefmt="%Y-%m-%d %H:%M:%S",
Expand All @@ -121,7 +123,7 @@ def init_logging(debug=False):
root_logger.addHandler(handler)


def parse_args():
def parse_args() -> Namespace:
'''Parse command line'''

parser = ArgumentParser()
Expand All @@ -136,7 +138,7 @@ def parse_args():
return parser.parse_args()


def main():
def main() -> None:
options = parse_args()
init_logging(options.debug)

Expand Down
32 changes: 17 additions & 15 deletions examples/hello_asyncio.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,12 +27,12 @@
import logging
import os
import stat
from argparse import ArgumentParser
from argparse import ArgumentParser, Namespace
from typing import cast

import pyfuse3
import pyfuse3.asyncio
from pyfuse3 import FileHandleT, FileInfo, InodeT
from pyfuse3 import EntryAttributes, FileHandleT, FileInfo, InodeT, ReaddirToken, RequestContext

try:
import faulthandler
Expand All @@ -46,14 +46,14 @@


class TestFs(pyfuse3.Operations):
def __init__(self):
def __init__(self) -> None:
super(TestFs, self).__init__()
self.hello_name = b"message"
self.hello_inode = cast(InodeT, pyfuse3.ROOT_INODE + 1)
self.hello_data = b"hello world\n"

async def getattr(self, inode, ctx=None):
entry = pyfuse3.EntryAttributes()
async def getattr(self, inode: InodeT, ctx: RequestContext | None = None) -> EntryAttributes:
entry = EntryAttributes()
if inode == pyfuse3.ROOT_INODE:
entry.st_mode = stat.S_IFDIR | 0o755
entry.st_size = 0
Expand All @@ -73,26 +73,28 @@ async def getattr(self, inode, ctx=None):

return entry

async def lookup(self, parent_inode, name, ctx=None):
async def lookup(
self, parent_inode: InodeT, name: bytes, ctx: RequestContext
) -> EntryAttributes:
if parent_inode != pyfuse3.ROOT_INODE or name != self.hello_name:
raise pyfuse3.FUSEError(errno.ENOENT)
return await self.getattr(self.hello_inode)
return await self.getattr(self.hello_inode, ctx)

async def opendir(self, inode, ctx):
async def opendir(self, inode: InodeT, ctx: RequestContext) -> FileHandleT:
if inode != pyfuse3.ROOT_INODE:
raise pyfuse3.FUSEError(errno.ENOENT)
# For simplicity, we use the inode as file handle
return FileHandleT(inode)

async def readdir(self, fh, start_id, token):
async def readdir(self, fh: FileHandleT, start_id: int, token: ReaddirToken) -> None:
assert fh == pyfuse3.ROOT_INODE

# only one entry
if start_id == 0:
pyfuse3.readdir_reply(token, self.hello_name, await self.getattr(self.hello_inode), 1)
return

async def setxattr(self, inode, name, value, ctx):
async def setxattr(self, inode: InodeT, name: bytes, value: bytes, ctx: RequestContext) -> None:
if inode != pyfuse3.ROOT_INODE or name != b'command':
raise pyfuse3.FUSEError(errno.ENOTSUP)

Expand All @@ -101,20 +103,20 @@ async def setxattr(self, inode, name, value, ctx):
else:
raise pyfuse3.FUSEError(errno.EINVAL)

async def open(self, inode, flags, ctx):
async def open(self, inode: InodeT, flags: int, ctx: RequestContext) -> FileInfo:
if inode != self.hello_inode:
raise pyfuse3.FUSEError(errno.ENOENT)
if flags & os.O_RDWR or flags & os.O_WRONLY:
raise pyfuse3.FUSEError(errno.EACCES)
# For simplicity, we use the inode as file handle
return FileInfo(fh=FileHandleT(inode))

async def read(self, fh, off, size):
async def read(self, fh: FileHandleT, off: int, size: int) -> bytes:
assert fh == self.hello_inode
return self.hello_data[off : off + size]


def init_logging(debug=False):
def init_logging(debug: bool = False) -> None:
formatter = logging.Formatter(
'%(asctime)s.%(msecs)03d %(threadName)s: [%(name)s] %(message)s',
datefmt="%Y-%m-%d %H:%M:%S",
Expand All @@ -131,7 +133,7 @@ def init_logging(debug=False):
root_logger.addHandler(handler)


def parse_args():
def parse_args() -> Namespace:
'''Parse command line'''

parser = ArgumentParser()
Expand All @@ -146,7 +148,7 @@ def parse_args():
return parser.parse_args()


def main():
def main() -> None:
options = parse_args()
init_logging(options.debug)

Expand Down
Loading