Skip to content
Snippets Groups Projects
Commit cc9a60050a07 authored by Matt Harbison's avatar Matt Harbison
Browse files

typing: add basic type hints to vfs.py

Again, there's a lot more that could be done, but this sticks to the obviously
correct stuff that is related to primitives or `vfs` objects.  Hopefully this
helps smoke out more path related bytes vs str issues in TortoiseHg.

PyCharm seems smart enough to apply hints from annotated superclass functions,
but pytype isn't (according to the *.pyi file generated), so those are annotated
too.

There was some discussion about changing the default path arg from `None` to
`b''` in order to avoid the more verbose `Optional` declarations.  This would be
more in line with `os.path.join()` (which rejects `None`, but ignores empty
strings), and still not change the behavior for callers still passing `None`
(because the check is `if path` instead of an explicit check for `None`).  But I
didn't want to hold this up while discussing that, so this documents what _is_.
parent c4f07a011714
No related branches found
No related tags found
2 merge requests!485branching: merge default into stable,!283Add basic type hints
......@@ -11,6 +11,10 @@
import stat
import threading
from typing import (
Optional,
)
from .i18n import _
from .pycompat import (
delattr,
......@@ -26,7 +30,7 @@
)
def _avoidambig(path, oldstat):
def _avoidambig(path: bytes, oldstat):
"""Avoid file stat ambiguity forcibly
This function causes copying ``path`` file, if it is owned by
......@@ -60,6 +64,7 @@
'''Prevent instantiation; don't call this from subclasses.'''
raise NotImplementedError('attempted instantiating ' + str(type(self)))
def __call__(self, path, mode=b'rb', **kwargs):
# TODO: type return, which is util.posixfile wrapped by a proxy
def __call__(self, path: bytes, mode: bytes = b'rb', **kwargs):
raise NotImplementedError
......@@ -64,5 +69,5 @@
raise NotImplementedError
def _auditpath(self, path, mode):
def _auditpath(self, path: bytes, mode: bytes):
raise NotImplementedError
......@@ -67,5 +72,5 @@
raise NotImplementedError
def join(self, path, *insidef):
def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
raise NotImplementedError
......@@ -70,6 +75,6 @@
raise NotImplementedError
def tryread(self, path):
def tryread(self, path: bytes) -> bytes:
'''gracefully return an empty string for missing files'''
try:
return self.read(path)
......@@ -77,7 +82,7 @@
pass
return b""
def tryreadlines(self, path, mode=b'rb'):
def tryreadlines(self, path: bytes, mode: bytes = b'rb'):
'''gracefully return an empty array for missing files'''
try:
return self.readlines(path, mode=mode)
......@@ -95,7 +100,7 @@
"""
return self.__call__
def read(self, path):
def read(self, path: bytes) -> bytes:
with self(path, b'rb') as fp:
return fp.read()
......@@ -99,7 +104,7 @@
with self(path, b'rb') as fp:
return fp.read()
def readlines(self, path, mode=b'rb'):
def readlines(self, path: bytes, mode: bytes = b'rb'):
with self(path, mode=mode) as fp:
return fp.readlines()
......@@ -103,7 +108,9 @@
with self(path, mode=mode) as fp:
return fp.readlines()
def write(self, path, data, backgroundclose=False, **kwargs):
def write(
self, path: bytes, data: bytes, backgroundclose=False, **kwargs
) -> int:
with self(path, b'wb', backgroundclose=backgroundclose, **kwargs) as fp:
return fp.write(data)
......@@ -107,7 +114,9 @@
with self(path, b'wb', backgroundclose=backgroundclose, **kwargs) as fp:
return fp.write(data)
def writelines(self, path, data, mode=b'wb', notindexed=False):
def writelines(
self, path: bytes, data: bytes, mode: bytes = b'wb', notindexed=False
) -> None:
with self(path, mode=mode, notindexed=notindexed) as fp:
return fp.writelines(data)
......@@ -111,7 +120,7 @@
with self(path, mode=mode, notindexed=notindexed) as fp:
return fp.writelines(data)
def append(self, path, data):
def append(self, path: bytes, data: bytes) -> int:
with self(path, b'ab') as fp:
return fp.write(data)
......@@ -115,9 +124,9 @@
with self(path, b'ab') as fp:
return fp.write(data)
def basename(self, path):
def basename(self, path: bytes) -> bytes:
"""return base element of a path (as os.path.basename would do)
This exists to allow handling of strange encoding if needed."""
return os.path.basename(path)
......@@ -119,8 +128,8 @@
"""return base element of a path (as os.path.basename would do)
This exists to allow handling of strange encoding if needed."""
return os.path.basename(path)
def chmod(self, path, mode):
def chmod(self, path: bytes, mode: int) -> None:
return os.chmod(self.join(path), mode)
......@@ -125,8 +134,8 @@
return os.chmod(self.join(path), mode)
def dirname(self, path):
def dirname(self, path: bytes) -> bytes:
"""return dirname element of a path (as os.path.dirname would do)
This exists to allow handling of strange encoding if needed."""
return os.path.dirname(path)
......@@ -128,11 +137,11 @@
"""return dirname element of a path (as os.path.dirname would do)
This exists to allow handling of strange encoding if needed."""
return os.path.dirname(path)
def exists(self, path=None):
def exists(self, path: Optional[bytes] = None) -> bool:
return os.path.exists(self.join(path))
def fstat(self, fp):
return util.fstat(fp)
......@@ -134,8 +143,8 @@
return os.path.exists(self.join(path))
def fstat(self, fp):
return util.fstat(fp)
def isdir(self, path=None):
def isdir(self, path: Optional[bytes] = None) -> bool:
return os.path.isdir(self.join(path))
......@@ -140,5 +149,5 @@
return os.path.isdir(self.join(path))
def isfile(self, path=None):
def isfile(self, path: Optional[bytes] = None) -> bool:
return os.path.isfile(self.join(path))
......@@ -143,5 +152,5 @@
return os.path.isfile(self.join(path))
def islink(self, path=None):
def islink(self, path: Optional[bytes] = None) -> bool:
return os.path.islink(self.join(path))
......@@ -146,6 +155,6 @@
return os.path.islink(self.join(path))
def isfileorlink(self, path=None):
def isfileorlink(self, path: Optional[bytes] = None) -> bool:
"""return whether path is a regular file or a symlink
Unlike isfile, this doesn't follow symlinks."""
......@@ -156,7 +165,7 @@
mode = st.st_mode
return stat.S_ISREG(mode) or stat.S_ISLNK(mode)
def _join(self, *paths):
def _join(self, *paths: bytes) -> bytes:
root_idx = 0
for idx, p in enumerate(paths):
if os.path.isabs(p) or p.startswith(self._dir_sep):
......@@ -166,10 +175,10 @@
paths = [p for p in paths if p]
return self._dir_sep.join(paths)
def reljoin(self, *paths):
def reljoin(self, *paths: bytes) -> bytes:
"""join various elements of a path together (as os.path.join would do)
The vfs base is not injected so that path stay relative. This exists
to allow handling of strange encoding if needed."""
return self._join(*paths)
......@@ -170,12 +179,12 @@
"""join various elements of a path together (as os.path.join would do)
The vfs base is not injected so that path stay relative. This exists
to allow handling of strange encoding if needed."""
return self._join(*paths)
def split(self, path):
def split(self, path: bytes):
"""split top-most element of a path (as os.path.split would do)
This exists to allow handling of strange encoding if needed."""
return os.path.split(path)
......@@ -177,8 +186,8 @@
"""split top-most element of a path (as os.path.split would do)
This exists to allow handling of strange encoding if needed."""
return os.path.split(path)
def lexists(self, path=None):
def lexists(self, path: Optional[bytes] = None) -> bool:
return os.path.lexists(self.join(path))
......@@ -183,5 +192,5 @@
return os.path.lexists(self.join(path))
def lstat(self, path=None):
def lstat(self, path: Optional[bytes] = None):
return os.lstat(self.join(path))
......@@ -186,5 +195,5 @@
return os.lstat(self.join(path))
def listdir(self, path=None):
def listdir(self, path: Optional[bytes] = None):
return os.listdir(self.join(path))
......@@ -189,5 +198,5 @@
return os.listdir(self.join(path))
def makedir(self, path=None, notindexed=True):
def makedir(self, path: Optional[bytes] = None, notindexed=True):
return util.makedir(self.join(path), notindexed)
......@@ -192,5 +201,7 @@
return util.makedir(self.join(path), notindexed)
def makedirs(self, path=None, mode=None):
def makedirs(
self, path: Optional[bytes] = None, mode: Optional[int] = None
):
return util.makedirs(self.join(path), mode)
......@@ -195,5 +206,5 @@
return util.makedirs(self.join(path), mode)
def makelock(self, info, path):
def makelock(self, info, path: bytes):
return util.makelock(info, self.join(path))
......@@ -198,5 +209,5 @@
return util.makelock(info, self.join(path))
def mkdir(self, path=None):
def mkdir(self, path: Optional[bytes] = None):
return os.mkdir(self.join(path))
......@@ -201,6 +212,11 @@
return os.mkdir(self.join(path))
def mkstemp(self, suffix=b'', prefix=b'tmp', dir=None):
def mkstemp(
self,
suffix: bytes = b'',
prefix: bytes = b'tmp',
dir: Optional[bytes] = None,
):
fd, name = pycompat.mkstemp(
suffix=suffix, prefix=prefix, dir=self.join(dir)
)
......@@ -210,6 +226,6 @@
else:
return fd, fname
def readdir(self, path=None, stat=None, skip=None):
def readdir(self, path: Optional[bytes] = None, stat=None, skip=None):
return util.listdir(self.join(path), stat, skip)
......@@ -214,5 +230,5 @@
return util.listdir(self.join(path), stat, skip)
def readlock(self, path):
def readlock(self, path: bytes) -> bytes:
return util.readlock(self.join(path))
......@@ -217,6 +233,6 @@
return util.readlock(self.join(path))
def rename(self, src, dst, checkambig=False):
def rename(self, src: bytes, dst: bytes, checkambig=False):
"""Rename from src to dst
checkambig argument is used with util.filestat, and is useful
......@@ -238,6 +254,6 @@
return ret
return util.rename(srcpath, dstpath)
def readlink(self, path):
def readlink(self, path: bytes) -> bytes:
return util.readlink(self.join(path))
......@@ -242,6 +258,6 @@
return util.readlink(self.join(path))
def removedirs(self, path=None):
def removedirs(self, path: Optional[bytes] = None):
"""Remove a leaf directory and all empty intermediate ones"""
return util.removedirs(self.join(path))
......@@ -245,7 +261,7 @@
"""Remove a leaf directory and all empty intermediate ones"""
return util.removedirs(self.join(path))
def rmdir(self, path=None):
def rmdir(self, path: Optional[bytes] = None):
"""Remove an empty directory."""
return os.rmdir(self.join(path))
......@@ -249,7 +265,9 @@
"""Remove an empty directory."""
return os.rmdir(self.join(path))
def rmtree(self, path=None, ignore_errors=False, forcibly=False):
def rmtree(
self, path: Optional[bytes] = None, ignore_errors=False, forcibly=False
):
"""Remove a directory tree recursively
If ``forcibly``, this tries to remove READ-ONLY files, too.
......@@ -272,6 +290,6 @@
self.join(path), ignore_errors=ignore_errors, onerror=onerror
)
def setflags(self, path, l, x):
def setflags(self, path: bytes, l: bool, x: bool):
return util.setflags(self.join(path), l, x)
......@@ -276,5 +294,5 @@
return util.setflags(self.join(path), l, x)
def stat(self, path=None):
def stat(self, path: Optional[bytes] = None):
return os.stat(self.join(path))
......@@ -279,5 +297,5 @@
return os.stat(self.join(path))
def unlink(self, path=None):
def unlink(self, path: Optional[bytes] = None):
return util.unlink(self.join(path))
......@@ -282,6 +300,6 @@
return util.unlink(self.join(path))
def tryunlink(self, path=None):
def tryunlink(self, path: Optional[bytes] = None):
"""Attempt to remove a file, ignoring missing file errors."""
util.tryunlink(self.join(path))
......@@ -285,8 +303,10 @@
"""Attempt to remove a file, ignoring missing file errors."""
util.tryunlink(self.join(path))
def unlinkpath(self, path=None, ignoremissing=False, rmdir=True):
def unlinkpath(
self, path: Optional[bytes] = None, ignoremissing=False, rmdir=True
):
return util.unlinkpath(
self.join(path), ignoremissing=ignoremissing, rmdir=rmdir
)
......@@ -289,7 +309,7 @@
return util.unlinkpath(
self.join(path), ignoremissing=ignoremissing, rmdir=rmdir
)
def utime(self, path=None, t=None):
def utime(self, path: Optional[bytes] = None, t=None):
return os.utime(self.join(path), t)
......@@ -294,6 +314,6 @@
return os.utime(self.join(path), t)
def walk(self, path=None, onerror=None):
def walk(self, path: Optional[bytes] = None, onerror=None):
"""Yield (dirpath, dirs, files) tuple for each directories under path
``dirpath`` is relative one from the root of this vfs. This
......@@ -360,7 +380,7 @@
def __init__(
self,
base,
base: bytes,
audit=True,
cacheaudited=False,
expandpath=False,
......@@ -381,7 +401,7 @@
self.options = {}
@util.propertycache
def _cansymlink(self):
def _cansymlink(self) -> bool:
return util.checklink(self.base)
@util.propertycache
......@@ -393,7 +413,7 @@
return
os.chmod(name, self.createmode & 0o666)
def _auditpath(self, path, mode):
def _auditpath(self, path, mode) -> None:
if self._audit:
if os.path.isabs(path) and path.startswith(self.base):
path = os.path.relpath(path, self.base)
......@@ -404,8 +424,8 @@
def __call__(
self,
path,
mode=b"r",
path: bytes,
mode: bytes = b"r",
atomictemp=False,
notindexed=False,
backgroundclose=False,
......@@ -518,7 +538,7 @@
return fp
def symlink(self, src, dst):
def symlink(self, src: bytes, dst: bytes) -> None:
self.audit(dst)
linkname = self.join(dst)
util.tryunlink(linkname)
......@@ -538,7 +558,7 @@
else:
self.write(dst, src)
def join(self, path, *insidef):
def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
if path:
parts = [self.base, path]
parts.extend(insidef)
......@@ -551,7 +571,7 @@
class proxyvfs(abstractvfs):
def __init__(self, vfs):
def __init__(self, vfs: "vfs"):
self.vfs = vfs
def _auditpath(self, path, mode):
......@@ -569,7 +589,7 @@
class filtervfs(proxyvfs, abstractvfs):
'''Wrapper vfs for filtering filenames with a function.'''
def __init__(self, vfs, filter):
def __init__(self, vfs: "vfs", filter):
proxyvfs.__init__(self, vfs)
self._filter = filter
......@@ -573,6 +593,6 @@
proxyvfs.__init__(self, vfs)
self._filter = filter
def __call__(self, path, *args, **kwargs):
def __call__(self, path: bytes, *args, **kwargs):
return self.vfs(self._filter(path), *args, **kwargs)
......@@ -577,6 +597,6 @@
return self.vfs(self._filter(path), *args, **kwargs)
def join(self, path, *insidef):
def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
if path:
return self.vfs.join(self._filter(self.vfs.reljoin(path, *insidef)))
else:
......@@ -589,6 +609,6 @@
class readonlyvfs(proxyvfs):
'''Wrapper vfs preventing any writing.'''
def __init__(self, vfs):
def __init__(self, vfs: "vfs"):
proxyvfs.__init__(self, vfs)
......@@ -593,7 +613,7 @@
proxyvfs.__init__(self, vfs)
def __call__(self, path, mode=b'r', *args, **kw):
def __call__(self, path: bytes, mode: bytes = b'r', *args, **kw):
if mode not in (b'r', b'rb'):
raise error.Abort(_(b'this vfs is read only'))
return self.vfs(path, mode, *args, **kw)
......@@ -596,8 +616,8 @@
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 join(self, path, *insidef):
def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
return self.vfs.join(path, *insidef)
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment