7 files changed, 718 insertions, 0 deletions
diff --git a/webapp/codereview/internal/__init__.py b/webapp/codereview/internal/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/webapp/codereview/internal/__init__.py
diff --git a/webapp/codereview/internal/admin_service.py b/webapp/codereview/internal/admin_service.py
new file mode 100644
index 0000000000..a331663c7f
--- /dev/null
+++ b/webapp/codereview/internal/admin_service.py
@@ -0,0 +1,49 @@
+# Copyright 2008 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import datetime
+import logging
+
+from google.appengine.ext import db
+
+from codereview import models
+from codereview.models import gql
+
+from admin_pb2 import AdminService
+from sync_project_pb2 import *
+from util import u, InternalAPI
+
+class AdminServiceImp(AdminService, InternalAPI):
+  def SyncProject(self, rpc_controller, req, done):
+    rsp = SyncProjectResponse()
+
+    proj = models.Project.get_project_for_name(req.project_name)
+    if proj is None:
+      proj = models.Project(name = req.project_name)
+      proj.put()
+
+    really_exists = set()
+    for bs in req.branch:
+      branch = models.Branch.get_or_insert_branch(proj, bs.branch_name)
+      really_exists.add(branch.name)
+
+    to_delete = []
+    for b in list(gql(models.Branch, 'WHERE project = :1', proj)):
+      if b.name not in really_exists:
+        to_delete += gql(models.BuildAttempt, 'WHERE branch = :1', b)
+
+    if to_delete:
+      db.delete(to_delete)
+
+    done(rsp)
diff --git a/webapp/codereview/internal/build_service.py b/webapp/codereview/internal/build_service.py
new file mode 100644
index 0000000000..9f8bb0c1f8
--- /dev/null
+++ b/webapp/codereview/internal/build_service.py
@@ -0,0 +1,110 @@
+# Copyright 2008 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import datetime
+import logging
+
+from google.appengine.ext import db
+
+from codereview import models
+from codereview.models import gql
+
+from build_pb2 import BuildService
+from submit_build_pb2 import *
+from post_build_result_pb2 import *
+from prune_builds_pb2 import *
+from util import InternalAPI, automatic_retry
+
+MAX_PRUNE = 100
+
+class BuildServiceImp(BuildService, InternalAPI):
+  @automatic_retry
+  def SubmitBuild(self, rpc_controller, req, done):
+    rsp = SubmitBuildResponse()
+
+    branch = db.get(db.Key(req.branch_key))
+    new_changes = [db.Key(c) for c in req.new_change]
+
+    build = models.BuildAttempt(
+              branch = branch,
+              revision_id = req.revision_id,
+              new_changes = new_changes)
+    build.put()
+
+    rsp.build_id = build.key().id()
+    done(rsp)
+
+  @automatic_retry
+  def PostBuildResult(self, rpc_controller, req, done):
+    rsp = PostBuildResultResponse()
+
+    if req.build_status == PostBuildResultRequest.SUCCESS:
+      ok = True
+    elif req.build_status == PostBuildResultRequest.BUILD_FAILED:
+      ok = False
+    else:
+      raise InvalidValueError, req.build_status
+
+    build = models.BuildAttempt.get_by_id(req.build_id)
+    if not build.finished:
+      build.finished = True
+      build.success = ok
+      build.put()
+
+    branch = build.branch
+    project = branch.project
+
+    rsp.dest_project_name = str(project.name)
+    rsp.dest_project_key = str(project.key())
+
+    rsp.dest_branch_name = str(branch.name)
+    rsp.dest_branch_key = str(branch.key())
+
+    rsp.revision_id = str(build.revision_id)
+    for patchset_key in build.new_changes:
+      rsp.new_change.append(str(patchset_key))
+
+    done(rsp)
+
+  @automatic_retry
+  def PruneBuilds(self, rpc_controller, req, done):
+    rsp = PruneBuildsResponse()
+
+    build_list = []
+
+    for m in [_AgedSuccess,
+              _AgedFailed,
+              ]:
+      build_list.extend(m())
+      if len(build_list) >= MAX_PRUNE:
+        break
+
+    if build_list:
+      db.delete(build_list)
+      rsp.status_code = PruneBuildsResponse.BUILDS_PRUNED
+    else:
+      rsp.status_code = PruneBuildsResponse.QUEUE_EMPTY
+    done(rsp)
+
+def _AgedSuccess():
+  aged = datetime.datetime.now() - datetime.timedelta(days=2)
+  return gql(models.BuildAttempt,
+             'WHERE success = :1 AND started <= :2',
+             True, aged).fetch(MAX_PRUNE)
+
+def _AgedFailed():
+  aged = datetime.datetime.now() - datetime.timedelta(days=7)
+  return gql(models.BuildAttempt,
+             'WHERE success = :1 AND started <= :2',
+             False, aged).fetch(MAX_PRUNE)
diff --git a/webapp/codereview/internal/bundle_store_service.py b/webapp/codereview/internal/bundle_store_service.py
new file mode 100644
index 0000000000..7cd26cb7cd
--- /dev/null
+++ b/webapp/codereview/internal/bundle_store_service.py
@@ -0,0 +1,158 @@
+# Copyright 2008 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import datetime
+import logging
+
+from google.appengine.ext import db
+
+from codereview.models import gql
+from codereview.git_models import *
+
+from bundle_store_pb2 import BundleStoreService
+from next_received_bundle_pb2 import *
+from update_received_bundle_pb2 import *
+from prune_bundles_pb2 import *
+from util import InternalAPI, u, commit_to_revision, automatic_retry
+
+MAX_PRUNE = 10
+MAX_SEGS_PER_PRUNE = 10
+
+class BundleStoreServiceImp(BundleStoreService, InternalAPI):
+  @automatic_retry
+  def NextReceivedBundle(self, rpc_controller, req, done):
+    rsp = NextReceivedBundleResponse()
+
+    rb = ReceivedBundle.lock_next_new()
+    if rb:
+      rsp.status_code = NextReceivedBundleResponse.BUNDLE_AVAILABLE
+      rsp.bundle_key = str(rb.key())
+      rsp.dest_project = str(rb.dest_project.name)
+      rsp.dest_project_key = str(rb.dest_project.key())
+      rsp.dest_branch_key = str(rb.dest_branch.key())
+      rsp.owner = str(rb.owner.email())
+      rsp.n_segments = rb.n_segments
+      seg = rb.get_segment(1)
+      if seg:
+        rsp.bundle_data = seg.bundle_data
+    else:
+      rsp.status_code = NextReceivedBundleResponse.QUEUE_EMPTY
+    done(rsp)
+
+  @automatic_retry
+  def BundleSegment(self, rpc_controller, req, done):
+    rsp = BundleSegmentResponse()
+
+    rb = db.get(db.Key(req.bundle_key))
+    if not rb:
+      rsp.status_code = BundleSegmentResponse.UNKNOWN_BUNDLE
+      done(rsp)
+      return
+
+    seg = rb.get_segment(req.segment_id)
+    if seg:
+      rsp.status_code = BundleSegmentResponse.DATA
+      rsp.bundle_data = seg.bundle_data
+    else:
+      rsp.status_code = BundleSegmentResponse.UNKNOWN_SEGMENT
+    done(rsp)
+
+  @automatic_retry
+  def UpdateReceivedBundle(self, rpc_controller, req, done):
+    rsp = UpdateReceivedBundleResponse()
+
+    old_state = ReceivedBundle.STATE_UNPACKING
+    sc = req.status_code
+    if UpdateReceivedBundleRequest.UNPACKED_OK == sc:
+      new_state = ReceivedBundle.STATE_UNPACKED
+      err_msg = None
+    elif UpdateReceivedBundleRequest.SUSPEND_BUNDLE == sc:
+      new_state = ReceivedBundle.STATE_SUSPENDED
+      err_msg = req.error_details
+    else:
+      new_state = ReceivedBundle.STATE_INVALID
+      err_msg = req.error_details
+
+    try:
+      ReceivedBundle.update_state(req.bundle_key, old_state, new_state, err_msg)
+      rsp.status_code = UpdateReceivedBundleResponse.UPDATED
+    except InvalidBundleId, err:
+      logging.warn("Invalid bundle id %s: %s" % (req.bundle_key, err))
+      rsp.status_code = UpdateReceivedBundleResponse.INVALID_BUNDLE
+    except InvalidBundleState, err:
+      logging.warn("Invalid bundle state %s: %s" % (req.bundle_key, err))
+      rsp.status_code = UpdateReceivedBundleResponse.INVALID_STATE
+    done(rsp)
+
+  @automatic_retry
+  def PruneBundles(self, rpc_controller, req, done):
+    rsp = PruneBundlesResponse()
+
+    rb_list = []
+    to_rm = []
+
+    for m in [_AgedUploading,
+              _AgedInvalid,
+              _AgedSuspended,
+              _AgedUnpacking,
+              _AgedUnpacked,
+              ]:
+      rb_list.extend(m())
+      if len(rb_list) >= MAX_PRUNE:
+        break
+
+    for rb in rb_list:
+      segs = gql(ReceivedBundleSegment,
+                'WHERE ANCESTOR IS :1',
+                rb).fetch(MAX_SEGS_PER_PRUNE)
+      if len(segs) < MAX_SEGS_PER_PRUNE:
+        to_rm.append(rb)
+      to_rm.extend(segs)
+
+    if to_rm:
+      db.delete(to_rm)
+      rsp.status_code = PruneBundlesResponse.BUNDLES_PRUNED
+    else:
+      rsp.status_code = PruneBundlesResponse.QUEUE_EMPTY
+    done(rsp)
+
+def _AgedUploading():
+  aged = datetime.datetime.now() - datetime.timedelta(days=7)
+  return gql(ReceivedBundle,
+             'WHERE state = :1 AND created <= :2',
+             ReceivedBundle.STATE_UPLOADING, aged).fetch(MAX_PRUNE)
+
+def _AgedInvalid():
+  aged = datetime.datetime.now() - datetime.timedelta(days=2)
+  return gql(ReceivedBundle,
+             'WHERE state = :1 AND created <= :2',
+             ReceivedBundle.STATE_INVALID, aged).fetch(MAX_PRUNE)
+
+def _AgedSuspended():
+  aged = datetime.datetime.now() - datetime.timedelta(days=7)
+  return gql(ReceivedBundle,
+             'WHERE state = :1 AND created <= :2',
+             ReceivedBundle.STATE_SUSPENDED, aged).fetch(MAX_PRUNE)
+
+def _AgedUnpacking():
+  aged = datetime.datetime.now() - datetime.timedelta(days=7)
+  return gql(ReceivedBundle,
+             'WHERE state = :1 AND created <= :2',
+             ReceivedBundle.STATE_UNPACKING, aged).fetch(MAX_PRUNE)
+
+def _AgedUnpacked():
+  aged = datetime.datetime.now() - datetime.timedelta(hours=1)
+  return gql(ReceivedBundle,
+             'WHERE state = :1 AND created <= :2',
+             ReceivedBundle.STATE_UNPACKED, aged).fetch(MAX_PRUNE)
diff --git a/webapp/codereview/internal/change_service.py b/webapp/codereview/internal/change_service.py
new file mode 100644
index 0000000000..1ce26dccb9
--- /dev/null
+++ b/webapp/codereview/internal/change_service.py
@@ -0,0 +1,165 @@
+# Copyright 2008 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from google.appengine.api import users
+from google.appengine.ext import db
+
+from codereview import models
+
+from change_pb2 import ChangeService
+from complete_patchset_pb2 import *
+from upload_patchset_file_pb2 import *
+from submit_change_pb2 import *
+from util import *
+
+class ChangeServiceImp(ChangeService, InternalAPI):
+
+  @automatic_retry
+  def SubmitChange(self, rpc_controller, req, done):
+    rsp = SubmitChangeResponse()
+
+    user = users.User(req.owner)
+    branch = db.get(db.Key(req.dest_branch_key))
+    if not branch:
+      rsp.status_code = SubmitChangeResponse.UNKNOWN_BRANCH
+      done(rsp)
+
+    rev = commit_to_revision(branch.project, req.commit)
+    if rev.patchset:
+      rsp.status_code = SubmitChangeResponse.PATCHSET_EXISTS
+      done(rsp)
+
+    subject = u(req.commit.subject)[0:100]
+
+    def trans():
+      change = models.Change(
+                 subject = subject,
+                 description = u(req.commit.message),
+                 owner = user,
+                 dest_project = branch.project,
+                 dest_branch = branch,
+                 n_patchsets = 1)
+      change.put()
+
+      patchset = models.PatchSet(
+                   change = change,
+                   owner = user,
+                   parent = change,
+                   revision = rev,
+                   id = 1)
+      patchset.put()
+      return (change, patchset)
+    change, patchset = db.run_in_transaction(trans)
+
+    if rev.link_patchset(patchset):
+      rsp.status_code = SubmitChangeResponse.CREATED
+      rsp.change_id = change.key().id()
+      rsp.patchset_id = patchset.id
+      rsp.patchset_key = str(patchset.key())
+    else:
+      db.delete([change, patchset])
+      rsp.status_code = SubmitChangeResponse.PATCHSET_EXISTS
+    done(rsp)
+
+  @automatic_retry
+  def UploadPatchsetFile(self, rpc_controller, req, done):
+    rsp = UploadPatchsetFileResponse()
+
+    patchset = db.get(db.Key(req.patchset_key))
+    if not patchset:
+      rsp.status_code = UploadPatchsetFileResponse.UNKNOWN_PATCHSET
+      done(rsp)
+      return
+
+    if patchset.complete or patchset.change.closed:
+      rsp.status_code = UploadPatchsetFileResponse.CLOSED
+      done(rsp)
+      return
+
+    if UploadPatchsetFileRequest.ADD == req.status:
+      status = 'A'
+    elif UploadPatchsetFileRequest.MODIFY == req.status:
+      status = 'M'
+    elif UploadPatchsetFileRequest.DELETE == req.status:
+      status = 'D'
+    else:
+      status = '?'
+
+    try:
+      if req.base_id:
+        old_data = models.DeltaContent.create_content(
+                     id = req.base_id,
+                     text_z = req.base_z)
+        new_data = models.DeltaContent.create_content(
+                     id = req.final_id,
+                     text_z = req.patch_z,
+                     base = old_data)
+        if new_data.text_z == req.patch_z:
+          diff_data = new_data
+        else:
+          diff_data = models.DeltaContent.create_patch(
+                        id = req.patch_id,
+                        text_z = req.patch_z)
+      else:
+        old_data = None
+        new_data = None
+        diff_data = models.DeltaContent.create_patch(
+                      id = req.patch_id,
+                      text_z = req.patch_z)
+    except models.DeltaPatchingException:
+      logging.error("Patch error on change %s, patch set %s, file %s"
+        % (patchset.change.key().id(),
+           str(patchset.id),
+           u(req.file_name))
+      )
+      rsp.status_code = UploadPatchsetFileResponse.PATCHING_ERROR
+      done(rsp)
+      return
+
+    patch = models.Patch.get_or_insert_patch(
+              patchset = patchset,
+              filename = u(req.file_name),
+              status = status,
+              n_comments = 0,
+              old_data = old_data,
+              new_data = new_data,
+              diff_data = diff_data)
+
+    if old_data:
+      models.CachedDeltaContent.get(old_data.key())
+      models.CachedDeltaContent.get(new_data.key())
+      if diff_data != new_data:
+        models.CachedDeltaContent.get(diff_data.key())
+    else:
+      models.CachedDeltaContent.get(diff_data.key())
+
+    rsp.status_code = UploadPatchsetFileResponse.CREATED
+    done(rsp)
+
+  @automatic_retry
+  def CompletePatchset(serlf, rpc_controller, req, done):
+    rsp = CompletePatchsetResponse()
+
+    patchset = db.get(db.Key(req.patchset_key))
+    if not patchset.complete:
+      patchset.complete = True
+      patchset.put()
+
+      if len(req.compressed_filenames) > 0:
+        models.PatchSetFilenames.store_compressed(
+          patchset,
+          req.compressed_filenames)
+    done(rsp)
diff --git a/webapp/codereview/internal/merge_service.py b/webapp/codereview/internal/merge_service.py
new file mode 100644
index 0000000000..bf9c721f29
--- /dev/null
+++ b/webapp/codereview/internal/merge_service.py
@@ -0,0 +1,161 @@
+# Copyright 2008 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from google.appengine.ext import db
+
+from codereview import models
+from codereview.models import gql
+
+from merge_pb2 import MergeService
+from pending_merge_pb2 import *
+from post_merge_result_pb2 import *
+from post_branch_update_pb2 import *
+from util import InternalAPI, automatic_retry
+
+from codereview import email
+
+def _send_clean_merge_email(http_request, change):
+  if not change.emailed_clean_merge:
+    email.send_change_message(http_request, change,
+                              "mails/clean_merge.txt", None)
+    change.emailed_clean_merge = True
+  
+def _send_missing_dependency_merge_email(http_request, change):
+  if not change.emailed_clean_merge:
+    email.send_change_message(http_request, change,
+                              "mails/missing_dependency.txt", None)
+    change.emailed_missing_dependency = True
+
+def _send_path_conflict_email(http_request, change):
+  if not change.emailed_clean_merge:
+    email.send_change_message(http_request, change,
+                              "mails/path_conflict.txt", None)
+    change.emailed_path_conflict = True
+
+
+class InvalidBranchStatusError(Exception):
+  """The branch cannot be updated in this way at this time."""
+
+class MergeServiceImp(MergeService, InternalAPI):
+  @automatic_retry
+  def NextPendingMerge(self, rpc_controller, req, done):
+
+    patchsets = []
+    while not patchsets:
+      branch = gql(models.Branch,
+                  "WHERE status = 'NEEDS_MERGE'"
+                  " ORDER BY merge_submitted").get()
+      if branch is None:
+        break
+      patchsets = branch.begin_merge()
+
+    rsp = PendingMergeResponse()
+    if patchsets:
+      first = patchsets[0].change
+      rsp.status_code = PendingMergeResponse.MERGE_READY
+      rsp.dest_project_name = str(first.dest_project.name)
+      rsp.dest_project_key = str(first.dest_project.key())
+      rsp.dest_branch_name = str(first.dest_branch.name)
+      rsp.dest_branch_key = str(first.dest_branch.key())
+      for ps in patchsets:
+        pmi = rsp.change.add()
+        pmi.patchset_key = str(ps.key())
+        pmi.revision_id = str(ps.revision.id)
+    else:
+      rsp.status_code = PendingMergeResponse.QUEUE_EMPTY
+    done(rsp)
+
+  @automatic_retry
+  def PostMergeResult(self, rpc_controller, req, done):
+    rsp = PostMergeResultResponse()
+
+    success = []
+    fail = []
+    defer = []
+
+    for ri in req.change:
+      sc = ri.status_code
+      ps = db.get(db.Key(ri.patchset_key))
+
+      if ps.change.merged:
+        success.append(ps)
+        continue
+
+      def chg_trans(key):
+        change = db.get(key)
+        if change.merge_patchset.key() != ps.key():
+          return False
+
+        if sc == MergeResultItem.CLEAN_MERGE:
+          pass
+
+        elif sc == MergeResultItem.ALREADY_MERGED:
+          change.merged = True
+          change.closed = True
+          change.put()
+
+        elif sc == MergeResultItem.MISSING_DEPENDENCY:
+          pass
+
+        elif sc == MergeResultItem.PATH_CONFLICT:
+          change.unsubmit_merge()
+          change.put()
+
+        return True
+      if db.run_in_transaction(chg_trans, ps.change.key()):
+        if sc == MergeResultItem.CLEAN_MERGE:
+          _send_clean_merge_email(self.http_request, ps.change)
+          ps.change.put()
+
+        elif sc == MergeResultItem.ALREADY_MERGED:
+          success.append(ps)
+
+        elif sc == MergeResultItem.MISSING_DEPENDENCY:
+          _send_missing_dependency_merge_email(self.http_request, ps.change)
+          ps.change.put()
+          defer.append(ps)
+
+        elif sc == MergeResultItem.PATH_CONFLICT:
+          _send_path_conflict_email(self.http_request, ps.change)
+          ps.change.put()
+          fail.append(ps)
+      else:
+        fail.append(ps)
+
+    branch = db.get(db.Key(req.dest_branch_key))
+    branch.finish_merge(success, fail, defer)
+
+    done(rsp)
+
+  @automatic_retry
+  def PostBranchUpdate(self, rpc_controller, req, done):
+    rsp = PostBranchUpdateResponse()
+
+    branch = db.get(db.Key(req.branch_key))
+    merged = [db.get(db.Key(c_key)) for c_key in req.new_change]
+
+    branch.merged(merged)
+
+    for ps in merged:
+      def trans(key):
+        change = db.get(key)
+        if change.merge_patchset.key() == ps.key():
+          change.merged = True
+          change.closed = True
+          change.put()
+      db.run_in_transaction(trans, ps.change.key())
+
+    done(rsp)
diff --git a/webapp/codereview/internal/util.py b/webapp/codereview/internal/util.py
new file mode 100644
index 0000000000..267fe29348
--- /dev/null
+++ b/webapp/codereview/internal/util.py
@@ -0,0 +1,75 @@
+# Copyright 2008 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import datetime
+
+from google.appengine.ext import db
+from codereview import models
+from codereview.need_retry_pb2 import RetryRequestLaterResponse
+
+class InternalAPI(object):
+  """
+    Marker superclass for service implementations which should be
+    restricted to only role accounts executing batch processing.
+  """
+  def __init__(self):
+    pass
+
+def u(str):
+  """Decode the UTF-8 byte sequence into a unicode object."""
+  return str.decode('utf_8')
+
+def commit_to_revision(proj, commit):
+  """Converts a GitCommit into a RevisionId data store object.
+  """
+  p_a = commit.author
+  p_c = commit.committer
+
+  return models.RevisionId.get_or_insert_revision(
+          project = proj,
+          id = commit.id,
+          ancestors = [p for p in commit.parent_id],
+          message = db.Text(u(commit.message)),
+
+          author_name = u(p_a.name),
+          author_email = db.Email(u(p_a.email)),
+          author_when = datetime.datetime.utcfromtimestamp(p_a.when),
+          author_tz = p_a.tz,
+
+          committer_name = u(p_c.name),
+          committer_email = db.Email(u(p_c.email)),
+          committer_when = datetime.datetime.utcfromtimestamp(p_c.when),
+          committer_tz = p_a.tz,
+          )
+
+def automatic_retry(func):
+  """Decorator that catches data store errors and sends a retry response."""
+  def retry_wrapper(self, rpc_controller, req, done):
+    try:
+      func(self, rpc_controller, req, done)
+
+    except db.InternalError:
+      rsp = RetryRequestLaterResponse()
+      done(rsp)
+
+    except db.Timeout:
+      rsp = RetryRequestLaterResponse()
+      done(rsp)
+
+    except db.TransactionFailedError:
+      rsp = RetryRequestLaterResponse()
+      done(rsp)
+
+  return retry_wrapper
+