Docserver overhaul: Gitiles away from me.

chrome/common/extensions/docs/server2/local_git_file_system.py

+# Copyright 2015 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+
+from itertools import izip
+import local_git_util
+import posixpath
+
+from docs_server_utils import StringIdentity
+from environment import IsTest
+from file_system import FileNotFoundError, FileSystem, StatInfo
+from future import Future
+from local_file_system import LocalFileSystem
+from path_util import AssertIsValid, IsDirectory, ToDirectory
+from third_party.json_schema_compiler.memoize import memoize
+
+
+class LocalGitFileSystem(FileSystem):
+  '''Class to fetch filesystem data from this script's local git repository.
+  '''
+  @classmethod
+  def Create(cls, branch='master', commit=None):
+    if IsTest():
+      return LocalFileSystem.Create('')
+    return LocalGitFileSystem(branch, commit)
+
+  def __init__(self, branch, pinned_commit):
+    self._branch = branch
+    self._pinned_commit = pinned_commit
+    if self._pinned_commit:
+      commit = self._pinned_commit
+    else:
+      if branch != 'master':
+        commit = 'branch-heads/%s' % branch
+      else:
+        commit = 'origin/master'
+    try:
+      self._commit = local_git_util.ParseRevision(commit)
+    except ImportError:
+      # We ignore ImportErrors here. It means we're running in AppEngine, so
+      # this doesn't need to work anyway.
+      pass
+
+  def Read(self, paths, skip_not_found=False):
+
+    def get_entry_name(entry):
+      if entry['type'] == 'tree':
+        return entry['name'] + '/'
+      return entry['name']
+
+    def read_path(path):
+      try:
+        if IsDirectory(path):
+          return [get_entry_name(e)
+                  for e in local_git_util.ListDir(path, self._commit)]
+        else:
+          return local_git_util.ReadFile(path, self._commit)
+      except FileNotFoundError as e:
+        if skip_not_found:
+          return None
+        raise e
+
+    results = dict((path, read_path(path)) for path in paths)
+    return Future(value=dict((k, v) for k, v in results.iteritems()
+                             if v is not None))
+
+  def Refresh(self):
+    return Future(value=())
+
+  def GetCommitID(self):
+    '''Returns a future that resolves to the commit ID for this file system's
+    revision.
+    '''
+    return Future(value=self._commit)
+
+  def GetPreviousCommitID(self):
+    '''Returns a future that resolves to the parent commit ID of this file
+    system's revision.
+    '''
+    return Future(value=local_git_util.GetParentRevision(self._commit))
+
+  def StatAsync(self, path):
+
+    def get_child_versions(path):
+      return dict((e['name'], e['id'])
+                  for e in local_git_util.ListDir(path, self._commit))
+
+    def get_file_version(dir, filename):
+      try:
+        return next(e['id'] for e in local_git_util.ListDir(dir, self._commit)
+                    if e['name'] == filename)
+      except StopIteration:
+        raise FileNotFoundError('%s not found in revision %s' %
+                                (path, self._commit))
+
+    dir, filename = posixpath.split(path)
+    if path == '':
+      version = local_git_util.GetRootTree(self._commit)
+      child_versions = get_child_versions('')
+    elif IsDirectory(path):
+      parent_dir, stat_dir = posixpath.split(dir)
+      version = get_file_version(parent_dir, stat_dir)
+      child_versions = get_child_versions(dir)
+    else:
+      version = get_file_version(dir, filename)
+      child_versions = None
+
+    #print 'Accessing local git for stat on %s (%s)' % (path, version)
+    return Future(value=StatInfo(version, child_versions))
+
+  def GetIdentity(self):
+    if self._branch == 'master':
+      # A master FS always carries the same identity even if pinned to a commit.
+      str_id = 'master'
+    elif self._pinned_commit is not None:
+      str_id = self._pinned_commit
+    else:
+      str_id = 'branch-heads/%s' % self._branch
+    return '@'.join((self.__class__.__name__, str_id))
+
+  def GetVersion(self):
+    return self._pinned_commit