hgext/lfs/blobstore.py

51880:8f3cbea2547c

import hashlib
import json
import os
import re
import socket

from mercurial.i18n import _
from mercurial.node import hex

from mercurial import (

# 64 bytes for SHA256
_lfsre = re.compile(br'\A[a-f0-9]{64}\Z')


class lfsvfs(vfsmod.vfs):
    def join(self, path):
        """split the path at first two characters, like: XX/XXXXX..."""
        if not _lfsre.match(path):
            raise error.ProgrammingError(b'unexpected lfs path: %s' % path)
        return super(lfsvfs, self).join(path[0:2], path[2:])

    def walk(self, path=None, onerror=None):
        """Yield (dirpath, [], oids) tuple for blobs under path

        Oids only exist in the root of this vfs, so dirpath is always ''.

class nullvfs(lfsvfs):
    def __init__(self):
        pass

    def exists(self, oid):
        return False

    def read(self, oid):
        # store.read() calls into here if the blob doesn't exist in its
        # self.vfs.  Raise the same error as a normal vfs when asked to read a

        )

    def walk(self, path=None, onerror=None):
        return (b'', [], [])

    def write(self, oid, data):
        pass


class lfsuploadfile(httpconnectionmod.httpsendfile):
    """a file-like object that supports keepalive."""

51881:d62887764687

import hashlib
import json
import os
import re
import socket

from typing import (
    Optional,
)

from mercurial.i18n import _
from mercurial.node import hex

from mercurial import (

# 64 bytes for SHA256
_lfsre = re.compile(br'\A[a-f0-9]{64}\Z')


class lfsvfs(vfsmod.vfs):
    def join(self, path: Optional[bytes], *insidef: bytes) -> bytes:
        """split the path at first two characters, like: XX/XXXXX..."""
        if not _lfsre.match(path):
            raise error.ProgrammingError(b'unexpected lfs path: %s' % path)
        return super(lfsvfs, self).join(path[0:2], path[2:], *insidef)

    def walk(self, path=None, onerror=None):
        """Yield (dirpath, [], oids) tuple for blobs under path

        Oids only exist in the root of this vfs, so dirpath is always ''.

class nullvfs(lfsvfs):
    def __init__(self):
        pass

    def exists(self, path: Optional[bytes] = None) -> bool:
        return False

    def read(self, oid):
        # store.read() calls into here if the blob doesn't exist in its
        # self.vfs.  Raise the same error as a normal vfs when asked to read a

        )

    def walk(self, path=None, onerror=None):
        return (b'', [], [])

    def write(self, *args, **kwargs) -> int:
        return 0


class lfsuploadfile(httpconnectionmod.httpsendfile):
    """a file-like object that supports keepalive."""