mercurial/vfs.py

51883:1edac12af730

# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.

from __future__ import annotations

import contextlib
import os
import shutil
import stat
import threading

        # advance mtime (see issue5418)
        util.rename(util.mktempcopy(path), path)
        checkandavoid()


class abstractvfs:
    """Abstract base class; cannot be instantiated"""

    # default directory separator for vfs
    #
    # Other vfs code always use `/` and this works fine because python file API
    # abstract the use of `/` and make it work transparently. For consistency
    # vfs will always use `/` when joining. This avoid some confusion in
    # encoded vfs (see issue6546)
    _dir_sep = b'/'

    def __init__(self, *args, **kwargs):
        '''Prevent instantiation; don't call this from subclasses.'''
        raise NotImplementedError('attempted instantiating ' + str(type(self)))

    # TODO: type return, which is util.posixfile wrapped by a proxy
    def __call__(self, path: bytes, mode: bytes = b'rb', **kwargs):
        raise NotImplementedError

    def _auditpath(self, path: bytes, mode: bytes):
        raise NotImplementedError

    def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
        raise NotImplementedError

    def tryread(self, path: bytes) -> bytes:
        '''gracefully return an empty string for missing files'''
        try:
            return self.read(path)

opener = vfs


class proxyvfs(abstractvfs):
    def __init__(self, vfs: "vfs"):
        self.vfs = vfs

    @property
    def createmode(self):

    def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
        return self.vfs.join(path, *insidef)


class closewrapbase:
    """Base class of wrapper, which hooks closing

    Do not instantiate outside of the vfs layer.
    """

    def __enter__(self):
        self._origfh.__enter__()
        return self

    def __exit__(self, exc_type, exc_value, exc_tb):
        raise NotImplementedError('attempted instantiating ' + str(type(self)))

    def close(self):
        raise NotImplementedError('attempted instantiating ' + str(type(self)))


class delayclosedfile(closewrapbase):
    """Proxy for a file object whose close is delayed.

51884:f79f98733a5b

# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.

from __future__ import annotations

import abc
import contextlib
import os
import shutil
import stat
import threading

        # advance mtime (see issue5418)
        util.rename(util.mktempcopy(path), path)
        checkandavoid()


class abstractvfs(abc.ABC):
    """Abstract base class; cannot be instantiated"""

    # default directory separator for vfs
    #
    # Other vfs code always use `/` and this works fine because python file API
    # abstract the use of `/` and make it work transparently. For consistency
    # vfs will always use `/` when joining. This avoid some confusion in
    # encoded vfs (see issue6546)
    _dir_sep = b'/'

    # TODO: type return, which is util.posixfile wrapped by a proxy
    @abc.abstractmethod
    def __call__(self, path: bytes, mode: bytes = b'rb', **kwargs):
        ...

    @abc.abstractmethod
    def _auditpath(self, path: bytes, mode: bytes):
        ...

    @abc.abstractmethod
    def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
        ...

    def tryread(self, path: bytes) -> bytes:
        '''gracefully return an empty string for missing files'''
        try:
            return self.read(path)

opener = vfs


class proxyvfs(abstractvfs, abc.ABC):
    def __init__(self, vfs: "vfs"):
        self.vfs = vfs

    @property
    def createmode(self):

    def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
        return self.vfs.join(path, *insidef)


class closewrapbase(abc.ABC):
    """Base class of wrapper, which hooks closing

    Do not instantiate outside of the vfs layer.
    """

    def __enter__(self):
        self._origfh.__enter__()
        return self

    @abc.abstractmethod
    def __exit__(self, exc_type, exc_value, exc_tb):
        ...

    @abc.abstractmethod
    def close(self):
        ...


class delayclosedfile(closewrapbase):
    """Proxy for a file object whose close is delayed.