mercurial-scm/hg: mercurial/vfs.py comparison

comparison mercurial/vfs.py @ 51888:fa9e8a6521c1

typing: manually add type annotations to `mercurial/vfs.py` This isn't everything, but hopefully it's close enough to hack on a protocol class.

author	Matt Harbison <matt_harbison@yahoo.com>
date	Fri, 20 Sep 2024 20:16:12 -0400
parents	ad83e4f9b40e
children	22e1924e9402

comparison

equal deleted inserted replaced

-:ad83e4f9b40e
+:fa9e8a6521c1
 import threading
 import typing
 from typing import (
 Any,
+BinaryIO,
+Callable,
 Iterable,
 Iterator,
 List,
+MutableMapping,
 Optional,
 Tuple,
 Type,
 TypeVar,
 )
 pycompat,
 util,
 )
 if typing.TYPE_CHECKING:
+from . import (
+ui as uimod,
+)
 _Tbackgroundfilecloser = TypeVar(
 '_Tbackgroundfilecloser', bound='backgroundfilecloser'
 )
 _Tclosewrapbase = TypeVar('_Tclosewrapbase', bound='closewrapbase')
+_OnErrorFn = Callable[[Exception], Optional[object]]
-def _avoidambig(path: bytes, oldstat) -> None:
+def _avoidambig(path: bytes, oldstat: util.filestat) -> None:
 """Avoid file stat ambiguity forcibly
 This function causes copying ``path`` file, if it is owned by
 another (see issue5418 and issue5584 for detail).
 """
 @abc.abstractmethod
 def __call__(self, path: bytes, mode: bytes = b'rb', **kwargs) -> Any:
 ...
 @abc.abstractmethod
-def _auditpath(self, path: bytes, mode: bytes) -> Any:
+def _auditpath(self, path: bytes, mode: bytes) -> None:
 ...
 @abc.abstractmethod
 def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
 ...
 return self.read(path)
 except FileNotFoundError:
 pass
 return b""
-def tryreadlines(self, path: bytes, mode: bytes = b'rb') -> Any:
+def tryreadlines(self, path: bytes, mode: bytes = b'rb') -> List[bytes]:
 '''gracefully return an empty array for missing files'''
 try:
 return self.readlines(path, mode=mode)
 except FileNotFoundError:
 pass
 def read(self, path: bytes) -> bytes:
 with self(path, b'rb') as fp:
 return fp.read()
-def readlines(self, path: bytes, mode: bytes = b'rb') -> Any:
+def readlines(self, path: bytes, mode: bytes = b'rb') -> List[bytes]:
 with self(path, mode=mode) as fp:
 return fp.readlines()
 def write(
-self, path: bytes, data: bytes, backgroundclose=False, **kwargs
+self, path: bytes, data: bytes, backgroundclose: bool = False, **kwargs
 ) -> int:
 with self(path, b'wb', backgroundclose=backgroundclose, **kwargs) as fp:
 return fp.write(data)
 def writelines(
 self,
 path: bytes,
 data: Iterable[bytes],
 mode: bytes = b'wb',
-notindexed=False,
+notindexed: bool = False,
 ) -> None:
 with self(path, mode=mode, notindexed=notindexed) as fp:
 return fp.writelines(data)
 def append(self, path: bytes, data: bytes) -> int:
 return os.path.dirname(path)
 def exists(self, path: Optional[bytes] = None) -> bool:
 return os.path.exists(self.join(path))
-def fstat(self, fp) -> os.stat_result:
+def fstat(self, fp: BinaryIO) -> os.stat_result:
 return util.fstat(fp)
 def isdir(self, path: Optional[bytes] = None) -> bool:
 return os.path.isdir(self.join(path))
 def makedirs(
 self, path: Optional[bytes] = None, mode: Optional[int] = None
 ) -> None:
 return util.makedirs(self.join(path), mode)
-def makelock(self, info, path: bytes) -> None:
+def makelock(self, info: bytes, path: bytes) -> None:
 return util.makelock(info, self.join(path))
 def mkdir(self, path: Optional[bytes] = None) -> None:
 return os.mkdir(self.join(path))
 if dir:
 return fd, os.path.join(dir, fname)
 else:
 return fd, fname
+# TODO: This doesn't match osutil.listdir().  stat=False in pure;
+#  non-optional bool in cext.  'skip' is bool if we trust cext, or bytes
+#  going by how pure uses it.  Also, cext returns a custom stat structure.
+#  from cext.osutil.pyi:
+#
+#     path: bytes, st: bool, skip: Optional[bool]
 def readdir(
 self, path: Optional[bytes] = None, stat=None, skip=None
 ) -> Any:
 return util.listdir(self.join(path), stat, skip)
 def readlock(self, path: bytes) -> bytes:
 return util.readlock(self.join(path))
-def rename(self, src: bytes, dst: bytes, checkambig=False) -> None:
+def rename(self, src: bytes, dst: bytes, checkambig: bool = False) -> None:
 """Rename from src to dst
 checkambig argument is used with util.filestat, and is useful
 only if destination file is guarded by any lock
 (e.g. repo.lock or repo.wlock).
 def rmdir(self, path: Optional[bytes] = None) -> None:
 """Remove an empty directory."""
 return os.rmdir(self.join(path))
 def rmtree(
-self, path: Optional[bytes] = None, ignore_errors=False, forcibly=False
+self,
+path: Optional[bytes] = None,
+ignore_errors: bool = False,
+forcibly: bool = False,
 ) -> None:
 """Remove a directory tree recursively
 If ``forcibly``, this tries to remove READ-ONLY files, too.
 """
 if forcibly:
-def onexc(function, path, excinfo):
+def onexc(function, path: bytes, excinfo):
 if function is not os.remove:
 raise
 # read-only files cannot be unlinked under Windows
 s = os.stat(path)
 if (s.st_mode & stat.S_IWRITE) != 0:
 def tryunlink(self, path: Optional[bytes] = None) -> bool:
 """Attempt to remove a file, ignoring missing file errors."""
 return util.tryunlink(self.join(path))
 def unlinkpath(
-self, path: Optional[bytes] = None, ignoremissing=False, rmdir=True
+self,
+path: Optional[bytes] = None,
+ignoremissing: bool = False,
+rmdir: bool = True,
 ) -> None:
 return util.unlinkpath(
 self.join(path), ignoremissing=ignoremissing, rmdir=rmdir
 )
-def utime(self, path: Optional[bytes] = None, t=None) -> None:
+# TODO: could be Tuple[float, float] too.
+def utime(
+self, path: Optional[bytes] = None, t: Optional[Tuple[int, int]] = None
+) -> None:
 return os.utime(self.join(path), t)
 def walk(
-self, path: Optional[bytes] = None, onerror=None
+self, path: Optional[bytes] = None, onerror: Optional[_OnErrorFn] = None
 ) -> Iterator[Tuple[bytes, List[bytes], List[bytes]]]:
 """Yield (dirpath, dirs, files) tuple for each directory under path
 ``dirpath`` is relative one from the root of this vfs. This
 uses ``os.sep`` as path separator, even you specify POSIX
 for dirpath, dirs, files in os.walk(self.join(path), onerror=onerror):
 yield (dirpath[prefixlen:], dirs, files)
 @contextlib.contextmanager
 def backgroundclosing(
-self, ui, expectedcount=-1
+self, ui: uimod.ui, expectedcount: int = -1
 ) -> Iterator[Optional[backgroundfilecloser]]:
 """Allow files to be closed asynchronously.
 When this context manager is active, ``backgroundclose`` can be passed
 to ``__call__``/``open`` to result in the file possibly being closed
 finally:
 vfs._backgroundfilecloser = (
 None  # pytype: disable=attribute-error
 )
-def register_file(self, path) -> None:
+def register_file(self, path: bytes) -> None:
 """generic hook point to lets fncache steer its stew"""
 class vfs(abstractvfs):
 """Operate files relative to a base directory
 'cacheaudited' should be enabled only if (a) vfs object is short-lived, or
 (b) the base directory is managed by hg and considered sort-of append-only.
 See pathutil.pathauditor() for details.
 """
+createmode: Optional[int]
 def __init__(
 self,
 base: bytes,
-audit=True,
+audit: bool = True,
-cacheaudited=False,
+cacheaudited: bool = False,
-expandpath=False,
+expandpath: bool = False,
-realpath=False,
+realpath: bool = False,
 ) -> None:
 if expandpath:
 base = util.expandpath(base)
 if realpath:
 base = os.path.realpath(base)
 @util.propertycache
 def _cansymlink(self) -> bool:
 return util.checklink(self.base)
 @util.propertycache
-def _chmod(self):
+def _chmod(self) -> bool:
 return util.checkexec(self.base)
-def _fixfilemode(self, name) -> None:
+def _fixfilemode(self, name: bytes) -> None:
 if self.createmode is None or not self._chmod:
 return
 os.chmod(name, self.createmode & 0o666)
-def _auditpath(self, path, mode) -> None:
+def _auditpath(self, path: bytes, mode: bytes) -> None:
 if self._audit:
 if os.path.isabs(path) and path.startswith(self.base):
 path = os.path.relpath(path, self.base)
 r = util.checkosfilename(path)
 if r:
 raise error.Abort(b"%s: %r" % (r, path))
 self.audit(path, mode=mode)
 def isfileorlink_checkdir(
-self, dircache, path: Optional[bytes] = None
+self,
+dircache: MutableMapping[bytes, bool],
+path: Optional[bytes] = None,
 ) -> bool:
 """return True if the path is a regular file or a symlink and
 the directories along the path are "normal", that is
 not symlinks or nested hg repositories.
 Ignores the `_audit` setting, and checks the directories regardless.
 `dircache` is used to cache the directory checks.
 """
+# TODO: Should be a None check on 'path', or shouldn't default to None
+#  because of the immediate call to util.localpath().
 try:
 for prefix in pathutil.finddirs_rev_noroot(util.localpath(path)):
 if prefix in dircache:
 res = dircache[prefix]
 else:
 def __call__(
 self,
 path: bytes,
 mode: bytes = b"rb",
-atomictemp=False,
+atomictemp: bool = False,
-notindexed=False,
+notindexed: bool = False,
-backgroundclose=False,
+backgroundclose: bool = False,
-checkambig=False,
+checkambig: bool = False,
-auditpath=True,
+auditpath: bool = True,
-makeparentdirs=True,
+makeparentdirs: bool = True,
-) -> Any:
+) -> Any:  # TODO: should be BinaryIO if util.atomictempfile can be coersed
 """Open ``path`` file, which is relative to vfs root.
 By default, parent directories are created as needed. Newly created
 directories are marked as "not to be indexed by the content indexing
 service", if ``notindexed`` is specified for "write" mode access.
 opener: Type[vfs] = vfs
 class proxyvfs(abstractvfs, abc.ABC):
-def __init__(self, vfs: "vfs"):
+def __init__(self, vfs: vfs) -> None:
 self.vfs = vfs
 @property
-def createmode(self):
+def createmode(self) -> Optional[int]:
 return self.vfs.createmode
-def _auditpath(self, path, mode) -> None:
+def _auditpath(self, path: bytes, mode: bytes) -> None:
 return self.vfs._auditpath(path, mode)
 @property
 def options(self):
 return self.vfs.options
 class filtervfs(proxyvfs, abstractvfs):
 '''Wrapper vfs for filtering filenames with a function.'''
-def __init__(self, vfs: "vfs", filter):
+def __init__(self, vfs: vfs, filter) -> None:
 proxyvfs.__init__(self, vfs)
 self._filter = filter
+# TODO: The return type should be BinaryIO
 def __call__(self, path: bytes, *args, **kwargs) -> Any:
 return self.vfs(self._filter(path), *args, **kwargs)
 def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
 if path:
 class readonlyvfs(proxyvfs):
 '''Wrapper vfs preventing any writing.'''
-def __init__(self, vfs: "vfs"):
+def __init__(self, vfs: vfs) -> None:
 proxyvfs.__init__(self, vfs)
+# TODO: The return type should be BinaryIO
 def __call__(self, path: bytes, mode: bytes = b'rb', *args, **kw) -> Any:
 if mode not in (b'r', b'rb'):
 raise error.Abort(_(b'this vfs is read only'))
 return self.vfs(path, mode, *args, **kw)
 """
 def __init__(self, fh) -> None:
 object.__setattr__(self, '_origfh', fh)
-def __getattr__(self, attr) -> Any:
+def __getattr__(self, attr: str) -> Any:
 return getattr(self._origfh, attr)
-def __setattr__(self, attr, value) -> None:
+def __setattr__(self, attr: str, value: Any) -> None:
 return setattr(self._origfh, attr, value)
-def __delattr__(self, attr) -> None:
+def __delattr__(self, attr: str) -> None:
 return delattr(self._origfh, attr)
 def __enter__(self: _Tclosewrapbase) -> _Tclosewrapbase:
 self._origfh.__enter__()
 return self
 class backgroundfilecloser:
 """Coordinates background closing of file handles on multiple threads."""
-def __init__(self, ui, expectedcount=-1) -> None:
+def __init__(self, ui: uimod.ui, expectedcount: int = -1) -> None:
 self._running = False
 self._entered = False
 self._threads = []
 self._threadexception = None

Mercurial > public > mercurial-scm > hg

comparison mercurial/vfs.py @ 51888:fa9e8a6521c1