comparison mercurial/vfs.py @ 51931:992fcf6b2473

typing: add a handful more annotations to `mercurial/vfs.py`

These came out of refactoring into a protocol class, but they can stand on their
own.

The `audit` callback is kinda screwy because the internal lambda and the callable
for `pathutil.pathauditor` have different args and a different return type.  It's
conditionalized where it is called, and can be cleaned up later if desired.

51930:22e1924e9402

from typing import (
    Any,
    BinaryIO,
    Callable,
    Iterable,
    Iterator,
    List,
    MutableMapping,
    Optional,
    Tuple,
    Type,
    TypeVar,
)

from .i18n import _
from . import (
    encoding,

    '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: bool = True,

    def _auditpath(self, path: bytes, mode: bytes) -> None:
        return self.vfs._auditpath(path, mode)

    @property
    def options(self):
        return self.vfs.options

    @options.setter
    def options(self, value):
        self.vfs.options = value

    @property
    def audit(self):
        return self.vfs.audit


class filtervfs(proxyvfs, abstractvfs):
    '''Wrapper vfs for filtering filenames with a function.'''

    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:

51931:992fcf6b2473

from typing import (
    Any,
    BinaryIO,
    Callable,
    Dict,
    Iterable,
    Iterator,
    List,
    MutableMapping,
    Optional,
    Tuple,
    Type,
    TypeVar,
    Union,
)

from .i18n import _
from . import (
    encoding,

    '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.
    """

    audit: Union[pathutil.pathauditor, Callable[[bytes, Optional[bytes]], Any]]
    base: bytes
    createmode: Optional[int]
    options: Dict[bytes, Any]
    _audit: bool
    _trustnlink: Optional[bool]

    def __init__(
        self,
        base: bytes,
        audit: bool = True,

    def _auditpath(self, path: bytes, mode: bytes) -> None:
        return self.vfs._auditpath(path, mode)

    @property
    def options(self) -> Dict[bytes, Any]:
        return self.vfs.options

    @options.setter
    def options(self, value: Dict[bytes, Any]) -> None:
        self.vfs.options = value

    @property
    def audit(self):
        return self.vfs.audit


class filtervfs(proxyvfs, abstractvfs):
    '''Wrapper vfs for filtering filenames with a function.'''

    def __init__(self, vfs: vfs, filter: Callable[[bytes], bytes]) -> None:
        proxyvfs.__init__(self, vfs)
        self._filter = filter

    # TODO: The return type should be BinaryIO
    def __call__(self, path: bytes, *args, **kwargs) -> Any: