diff --git a/api/v3/converters.py b/api/v3/converters.py
new file mode 100644
index 0000000..60aebd7
--- /dev/null
+++ b/api/v3/converters.py
@@ -0,0 +1,1979 @@
+# Copyright 2020 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 __future__ import print_function
+from __future__ import division
+from __future__ import absolute_import
+
+import collections
+import itertools
+import logging
+import time
+
+from google.protobuf import timestamp_pb2
+
+from api import resource_name_converters as rnc
+from api.v3.api_proto import feature_objects_pb2
+from api.v3.api_proto import issues_pb2
+from api.v3.api_proto import issue_objects_pb2
+from api.v3.api_proto import project_objects_pb2
+from api.v3.api_proto import user_objects_pb2
+
+from framework import exceptions
+from framework import filecontent
+from framework import framework_bizobj
+from framework import framework_constants
+from framework import framework_helpers
+from proto import tracker_pb2
+from project import project_helpers
+from tracker import attachment_helpers
+from tracker import field_helpers
+from tracker import tracker_bizobj as tbo
+from tracker import tracker_helpers
+
+Choice = project_objects_pb2.FieldDef.EnumTypeSettings.Choice
+
+# Ingest/convert dicts for ApprovalStatus.
+_V3_APPROVAL_STATUS = issue_objects_pb2.ApprovalValue.ApprovalStatus.Value
+_APPROVAL_STATUS_INGEST = {
+  _V3_APPROVAL_STATUS('APPROVAL_STATUS_UNSPECIFIED'): None,
+  _V3_APPROVAL_STATUS('NOT_SET'): tracker_pb2.ApprovalStatus.NOT_SET,
+  _V3_APPROVAL_STATUS('NEEDS_REVIEW'): tracker_pb2.ApprovalStatus.NEEDS_REVIEW,
+  _V3_APPROVAL_STATUS('NA'): tracker_pb2.ApprovalStatus.NA,
+  _V3_APPROVAL_STATUS('REVIEW_REQUESTED'):
+      tracker_pb2.ApprovalStatus.REVIEW_REQUESTED,
+  _V3_APPROVAL_STATUS('REVIEW_STARTED'):
+      tracker_pb2.ApprovalStatus.REVIEW_STARTED,
+  _V3_APPROVAL_STATUS('NEED_INFO'): tracker_pb2.ApprovalStatus.NEED_INFO,
+  _V3_APPROVAL_STATUS('APPROVED'): tracker_pb2.ApprovalStatus.APPROVED,
+  _V3_APPROVAL_STATUS('NOT_APPROVED'): tracker_pb2.ApprovalStatus.NOT_APPROVED,
+}
+_APPROVAL_STATUS_CONVERT = {
+  val: key for key, val in _APPROVAL_STATUS_INGEST.items()}
+
+
+class Converter(object):
+  """Class to manage converting objects between the API and backend layer."""
+
+  def __init__(self, mc, services):
+    # type: (MonorailContext, Services) -> Converter
+    """Create a Converter with the given MonorailContext and Services.
+
+    Args:
+      mc: MonorailContext object containing the MonorailConnection to the DB
+            and the requester's AuthData object.
+      services: Services object for connections to backend services.
+    """
+    self.cnxn = mc.cnxn
+    self.user_auth = mc.auth
+    self.services = services
+
+  # Hotlists
+
+  def ConvertHotlist(self, hotlist):
+    # type: (proto.feature_objects_pb2.Hotlist)
+    #    -> api_proto.feature_objects_pb2.Hotlist
+    """Convert a protorpc Hotlist into a protoc Hotlist."""
+
+    hotlist_resource_name = rnc.ConvertHotlistName(hotlist.hotlist_id)
+    members_by_id = rnc.ConvertUserNames(
+        hotlist.owner_ids + hotlist.editor_ids)
+    default_columns = self._ComputeIssuesListColumns(hotlist.default_col_spec)
+    if hotlist.is_private:
+      hotlist_privacy = feature_objects_pb2.Hotlist.HotlistPrivacy.Value(
+          'PRIVATE')
+    else:
+      hotlist_privacy = feature_objects_pb2.Hotlist.HotlistPrivacy.Value(
+          'PUBLIC')
+
+    return feature_objects_pb2.Hotlist(
+        name=hotlist_resource_name,
+        display_name=hotlist.name,
+        owner=members_by_id.get(hotlist.owner_ids[0]),
+        editors=[
+            members_by_id.get(editor_id) for editor_id in hotlist.editor_ids
+        ],
+        summary=hotlist.summary,
+        description=hotlist.description,
+        default_columns=default_columns,
+        hotlist_privacy=hotlist_privacy)
+
+  def ConvertHotlists(self, hotlists):
+    # type: (Sequence[proto.feature_objects_pb2.Hotlist])
+    #    -> Sequence[api_proto.feature_objects_pb2.Hotlist]
+    """Convert protorpc Hotlists into protoc Hotlists."""
+    return [self.ConvertHotlist(hotlist) for hotlist in hotlists]
+
+  def ConvertHotlistItems(self, hotlist_id, items):
+    # type: (int, Sequence[proto.features_pb2.HotlistItem]) ->
+    #     Sequence[api_proto.feature_objects_pb2.Hotlist]
+    """Convert a Sequence of protorpc HotlistItems into a Sequence of protoc
+       HotlistItems.
+
+    Args:
+      hotlist_id: ID of the Hotlist the items belong to.
+      items: Sequence of HotlistItem protorpc objects.
+
+    Returns:
+      Sequence of protoc HotlistItems in the same order they are given in
+        `items`.
+      In the rare event that any issues in `items` are not found, they will be
+        omitted from the result.
+    """
+    issue_ids = [item.issue_id for item in items]
+    # Converting HotlistItemNames and IssueNames both require looking up the
+    # issues in the hotlist. However, we want to keep the code clean and
+    # readable so we keep the two processes separate.
+    resource_names_dict = rnc.ConvertHotlistItemNames(
+        self.cnxn, hotlist_id, issue_ids, self.services)
+    issue_names_dict = rnc.ConvertIssueNames(
+        self.cnxn, issue_ids, self.services)
+    adders_by_id = rnc.ConvertUserNames([item.adder_id for item in items])
+
+    # Filter out items whose issues were not found.
+    found_items = [
+        item for item in items if resource_names_dict.get(item.issue_id) and
+        issue_names_dict.get(item.issue_id)
+    ]
+    if len(items) != len(found_items):
+      found_ids = [item.issue_id for item in found_items]
+      missing_ids = [iid for iid in issue_ids if iid not in found_ids]
+      logging.info('HotlistItem issues %r not found' % missing_ids)
+
+    # Generate user friendly ranks (0, 1, 2, 3,...) that are exposed to API
+    # clients, instead of using padded ranks (1, 11, 21, 31,...).
+    sorted_ranks = sorted(item.rank for item in found_items)
+    friendly_ranks_dict = {
+        rank: friendly_rank for friendly_rank, rank in enumerate(sorted_ranks)
+    }
+
+    api_items = []
+    for item in found_items:
+      api_item = feature_objects_pb2.HotlistItem(
+          name=resource_names_dict.get(item.issue_id),
+          issue=issue_names_dict.get(item.issue_id),
+          rank=friendly_ranks_dict[item.rank],
+          adder=adders_by_id.get(item.adder_id),
+          note=item.note)
+      if item.date_added:
+        api_item.create_time.FromSeconds(item.date_added)
+      api_items.append(api_item)
+
+    return api_items
+
+  # Issues
+
+  def _ConvertComponentValues(self, issue):
+    # proto.tracker_pb2.Issue ->
+    #     Sequence[api_proto.issue_objects_pb2.Issue.ComponentValue]
+    """Convert the status string on issue into a ComponentValue."""
+    component_values = []
+    component_ids = itertools.chain(
+        issue.component_ids, issue.derived_component_ids)
+    ids_to_names = rnc.ConvertComponentDefNames(
+        self.cnxn, component_ids, issue.project_id, self.services)
+
+    for component_id in issue.component_ids:
+      if component_id in ids_to_names:
+        component_values.append(
+            issue_objects_pb2.Issue.ComponentValue(
+                component=ids_to_names[component_id],
+                derivation=issue_objects_pb2.Derivation.Value(
+                    'EXPLICIT')))
+    for derived_component_id in issue.derived_component_ids:
+      if derived_component_id in ids_to_names:
+        component_values.append(
+            issue_objects_pb2.Issue.ComponentValue(
+                component=ids_to_names[derived_component_id],
+                derivation=issue_objects_pb2.Derivation.Value('RULE')))
+
+    return component_values
+
+  def _ConvertStatusValue(self, issue):
+    # proto.tracker_pb2.Issue -> api_proto.issue_objects_pb2.Issue.StatusValue
+    """Convert the status string on issue into a StatusValue."""
+    derivation = issue_objects_pb2.Derivation.Value(
+        'DERIVATION_UNSPECIFIED')
+    if issue.status:
+      derivation = issue_objects_pb2.Derivation.Value('EXPLICIT')
+    else:
+      derivation = issue_objects_pb2.Derivation.Value('RULE')
+    return issue_objects_pb2.Issue.StatusValue(
+        status=issue.status or issue.derived_status, derivation=derivation)
+
+  def _ConvertAmendments(self, amendments, user_display_names):
+    # type: (Sequence[proto.tracker_pb2.Amendment], Mapping[int, str]) ->
+    #     Sequence[api_proto.issue_objects_pb2.Comment.Amendment]
+    """Convert protorpc Amendments to protoc Amendments.
+
+    Args:
+      amendments: the amendments to convert
+      user_display_names: map from user_id to display name for all users
+          involved in the amendments.
+
+    Returns:
+      The converted amendments.
+    """
+    results = []
+    for amendment in amendments:
+      field_name = tbo.GetAmendmentFieldName(amendment)
+      new_value = tbo.AmendmentString_New(amendment, user_display_names)
+      results.append(
+          issue_objects_pb2.Comment.Amendment(
+              field_name=field_name,
+              new_or_delta_value=new_value,
+              old_value=amendment.oldvalue))
+    return results
+
+  def _ConvertAttachments(self, attachments, project_name):
+    # type: (Sequence[proto.tracker_pb2.Attachment], str) ->
+    #     Sequence[api_proto.issue_objects_pb2.Comment.Attachment]
+    """Convert protorpc Attachments to protoc Attachments."""
+    results = []
+    for attach in attachments:
+      if attach.deleted:
+        state = issue_objects_pb2.IssueContentState.Value('DELETED')
+        size, thumbnail_uri, view_uri, download_uri = None, None, None, None
+      else:
+        state = issue_objects_pb2.IssueContentState.Value('ACTIVE')
+        size = attach.filesize
+        download_uri = attachment_helpers.GetDownloadURL(attach.attachment_id)
+        view_uri = attachment_helpers.GetViewURL(
+            attach, download_uri, project_name)
+        thumbnail_uri = attachment_helpers.GetThumbnailURL(attach, download_uri)
+      results.append(
+          issue_objects_pb2.Comment.Attachment(
+              filename=attach.filename,
+              state=state,
+              size=size,
+              media_type=attach.mimetype,
+              thumbnail_uri=thumbnail_uri,
+              view_uri=view_uri,
+              download_uri=download_uri))
+    return results
+
+  def ConvertComments(self, issue_id, comments):
+    # type: (int, Sequence[proto.tracker_pb2.IssueComment])
+    #     -> Sequence[api_proto.issue_objects_pb2.Comment]
+    """Convert protorpc IssueComments from issue into protoc Comments."""
+    issue = self.services.issue.GetIssue(self.cnxn, issue_id)
+    users_by_id = self.services.user.GetUsersByIDs(
+        self.cnxn, tbo.UsersInvolvedInCommentList(comments))
+    (user_display_names,
+     _user_display_emails) = framework_bizobj.CreateUserDisplayNamesAndEmails(
+         self.cnxn, self.services, self.user_auth, users_by_id.values())
+    comment_names_dict = rnc.CreateCommentNames(
+        issue.local_id, issue.project_name,
+        [comment.sequence for comment in comments])
+    approval_ids = [
+        comment.approval_id
+        for comment in comments
+        if comment.approval_id is not None  # In case of a 0 approval_id.
+    ]
+    approval_ids_to_names = rnc.ConvertApprovalDefNames(
+        self.cnxn, approval_ids, issue.project_id, self.services)
+
+    converted_comments = []
+    for comment in comments:
+      if comment.is_spam:
+        state = issue_objects_pb2.IssueContentState.Value('SPAM')
+      elif comment.deleted_by:
+        state = issue_objects_pb2.IssueContentState.Value('DELETED')
+      else:
+        state = issue_objects_pb2.IssueContentState.Value('ACTIVE')
+      comment_type = issue_objects_pb2.Comment.Type.Value('COMMENT')
+      if comment.is_description:
+        comment_type = issue_objects_pb2.Comment.Type.Value('DESCRIPTION')
+      converted_attachments = self._ConvertAttachments(
+          comment.attachments, issue.project_name)
+      converted_amendments = self._ConvertAmendments(
+          comment.amendments, user_display_names)
+      converted_comment = issue_objects_pb2.Comment(
+          name=comment_names_dict[comment.sequence],
+          state=state,
+          type=comment_type,
+          create_time=timestamp_pb2.Timestamp(seconds=comment.timestamp),
+          attachments=converted_attachments,
+          amendments=converted_amendments)
+      if comment.content:
+        converted_comment.content = comment.content
+      if comment.user_id:
+        converted_comment.commenter = rnc.ConvertUserName(comment.user_id)
+      if comment.inbound_message:
+        converted_comment.inbound_message = comment.inbound_message
+      if comment.approval_id and comment.approval_id in approval_ids_to_names:
+        converted_comment.approval = approval_ids_to_names[comment.approval_id]
+      converted_comments.append(converted_comment)
+    return converted_comments
+
+  def ConvertIssue(self, issue):
+    # type: (proto.tracker_pb2.Issue) -> api_proto.issue_objects_pb2.Issue
+    """Convert a protorpc Issue into a protoc Issue."""
+    issues = self.ConvertIssues([issue])
+    if len(issues) < 1:
+      raise exceptions.NoSuchIssueException()
+    if len(issues) > 1:
+      logging.warning('More than one converted issue returned: %s', issues)
+    return issues[0]
+
+  def ConvertIssues(self, issues):
+    # type: (Sequence[proto.tracker_pb2.Issue]) ->
+    #     Sequence[api_proto.issue_objects_pb2.Issue]
+    """Convert protorpc Issues into protoc Issues."""
+    issue_ids = [issue.issue_id for issue in issues]
+    issue_names_dict = rnc.ConvertIssueNames(
+        self.cnxn, issue_ids, self.services)
+    found_issues = [
+        issue for issue in issues if issue.issue_id in issue_names_dict
+    ]
+    converted_issues = []
+    for issue in found_issues:
+      status = self._ConvertStatusValue(issue)
+      content_state = issue_objects_pb2.IssueContentState.Value(
+          'STATE_UNSPECIFIED')
+      if issue.is_spam:
+        content_state = issue_objects_pb2.IssueContentState.Value('SPAM')
+      elif issue.deleted:
+        content_state = issue_objects_pb2.IssueContentState.Value('DELETED')
+      else:
+        content_state = issue_objects_pb2.IssueContentState.Value('ACTIVE')
+
+      owner = None
+      # Explicit values override values derived from rules.
+      if issue.owner_id:
+        owner = issue_objects_pb2.Issue.UserValue(
+            derivation=issue_objects_pb2.Derivation.Value('EXPLICIT'),
+            user=rnc.ConvertUserName(issue.owner_id))
+      elif issue.derived_owner_id:
+        owner = issue_objects_pb2.Issue.UserValue(
+            derivation=issue_objects_pb2.Derivation.Value('RULE'),
+            user=rnc.ConvertUserName(issue.derived_owner_id))
+
+      cc_users = []
+      for cc_user_id in issue.cc_ids:
+        cc_users.append(
+            issue_objects_pb2.Issue.UserValue(
+                derivation=issue_objects_pb2.Derivation.Value('EXPLICIT'),
+                user=rnc.ConvertUserName(cc_user_id)))
+      for derived_cc_user_id in issue.derived_cc_ids:
+        cc_users.append(
+            issue_objects_pb2.Issue.UserValue(
+                derivation=issue_objects_pb2.Derivation.Value('RULE'),
+                user=rnc.ConvertUserName(derived_cc_user_id)))
+
+      labels = self.ConvertLabels(
+          issue.labels, issue.derived_labels, issue.project_id)
+      components = self._ConvertComponentValues(issue)
+      non_approval_fvs = self._GetNonApprovalFieldValues(
+          issue.field_values, issue.project_id)
+      field_values = self.ConvertFieldValues(
+          non_approval_fvs, issue.project_id, issue.phases)
+      field_values.extend(
+          self.ConvertEnumFieldValues(
+              issue.labels, issue.derived_labels, issue.project_id))
+      related_issue_ids = (
+          [issue.merged_into] + issue.blocked_on_iids + issue.blocking_iids)
+      issue_names_by_ids = rnc.ConvertIssueNames(
+          self.cnxn, related_issue_ids, self.services)
+      merged_into_issue_ref = None
+      if issue.merged_into and issue.merged_into in issue_names_by_ids:
+        merged_into_issue_ref = issue_objects_pb2.IssueRef(
+            issue=issue_names_by_ids[issue.merged_into])
+      if issue.merged_into_external:
+        merged_into_issue_ref = issue_objects_pb2.IssueRef(
+            ext_identifier=issue.merged_into_external)
+
+      blocked_on_issue_refs = [
+          issue_objects_pb2.IssueRef(issue=issue_names_by_ids[iid])
+          for iid in issue.blocked_on_iids
+          if iid in issue_names_by_ids
+      ]
+      blocked_on_issue_refs.extend(
+          issue_objects_pb2.IssueRef(
+              ext_identifier=blocked_on.ext_issue_identifier)
+          for blocked_on in issue.dangling_blocked_on_refs)
+
+      blocking_issue_refs = [
+          issue_objects_pb2.IssueRef(issue=issue_names_by_ids[iid])
+          for iid in issue.blocking_iids
+          if iid in issue_names_by_ids
+      ]
+      blocking_issue_refs.extend(
+          issue_objects_pb2.IssueRef(
+              ext_identifier=blocking.ext_issue_identifier)
+          for blocking in issue.dangling_blocking_refs)
+      # All other timestamps were set when the issue was created.
+      close_time = None
+      if issue.closed_timestamp:
+        close_time = timestamp_pb2.Timestamp(seconds=issue.closed_timestamp)
+
+      phases = self._ComputePhases(issue.phases)
+
+      result = issue_objects_pb2.Issue(
+          name=issue_names_dict[issue.issue_id],
+          summary=issue.summary,
+          state=content_state,
+          status=status,
+          reporter=rnc.ConvertUserName(issue.reporter_id),
+          owner=owner,
+          cc_users=cc_users,
+          labels=labels,
+          components=components,
+          field_values=field_values,
+          merged_into_issue_ref=merged_into_issue_ref,
+          blocked_on_issue_refs=blocked_on_issue_refs,
+          blocking_issue_refs=blocking_issue_refs,
+          create_time=timestamp_pb2.Timestamp(seconds=issue.opened_timestamp),
+          close_time=close_time,
+          modify_time=timestamp_pb2.Timestamp(seconds=issue.modified_timestamp),
+          component_modify_time=timestamp_pb2.Timestamp(
+              seconds=issue.component_modified_timestamp),
+          status_modify_time=timestamp_pb2.Timestamp(
+              seconds=issue.status_modified_timestamp),
+          owner_modify_time=timestamp_pb2.Timestamp(
+              seconds=issue.owner_modified_timestamp),
+          star_count=issue.star_count,
+          phases=phases)
+      # TODO(crbug.com/monorail/5857): Set attachment_count unconditionally
+      # after the underlying source of negative attachment counts has been
+      # resolved and database has been repaired.
+      if issue.attachment_count >= 0:
+        result.attachment_count = issue.attachment_count
+      converted_issues.append(result)
+    return converted_issues
+
+  def IngestAttachmentUploads(self, attachment_uploads):
+    # type: (Sequence[api_proto.issues_pb2.AttachmentUpload] ->
+    #     Sequence[framework_helpers.AttachmentUpload])
+    """Ingests protoc AttachmentUploads into framework_helpers.AttachUploads."""
+    ingested_uploads = []
+    with exceptions.ErrorAggregator(exceptions.InputException) as err_agg:
+      for up in attachment_uploads:
+        if not up.filename or not up.content:
+          err_agg.AddErrorMessage(
+              'Uploaded atachment missing filename or content')
+        mimetype = filecontent.GuessContentTypeFromFilename(up.filename)
+        ingested_uploads.append(
+            framework_helpers.AttachmentUpload(
+                up.filename, up.content, mimetype))
+
+    return ingested_uploads
+
+  def IngestIssueDeltas(self, issue_deltas):
+    # type: (Sequence[api_proto.issues_pb2.IssueDelta]) ->
+    #     Sequence[Tuple[int, proto.tracker_pb2.IssueDelta]]
+    """Ingests protoc IssueDeltas, into protorpc IssueDeltas.
+
+    Args:
+      issue_deltas: the protoc IssueDeltas to ingest.
+
+    Returns:
+      A list of (issue_id, tracker_pb2.IssueDelta) tuples that contain
+      values found in issue_deltas, ignoring all OUTPUT_ONLY and masked
+      fields.
+
+    Raises:
+      InputException: if any fields in the approval_deltas were invalid.
+      NoSuchProjectException: if any parent projects are not found.
+      NoSuchIssueException: if any issues are not found.
+      NoSuchComponentException: if any components are not found.
+    """
+    issue_names = [delta.issue.name for delta in issue_deltas]
+    issue_ids = rnc.IngestIssueNames(self.cnxn, issue_names, self.services)
+    issues_dict, misses = self.services.issue.GetIssuesDict(
+        self.cnxn, issue_ids)
+    if misses:
+      logging.info(
+          'Issues not found for supposedly valid issue_ids: %r' % misses)
+      raise ValueError('Could not fetch some issues.')
+    configs_by_pid = self.services.config.GetProjectConfigs(
+        self.cnxn, {issue.project_id for issue in issues_dict.values()})
+
+    with exceptions.ErrorAggregator(exceptions.InputException) as err_agg:
+      for api_delta in issue_deltas:
+        if not api_delta.HasField('update_mask'):
+          err_agg.AddErrorMessage(
+              '`update_mask` must be set for {} delta.', api_delta.issue.name)
+        elif not api_delta.update_mask.IsValidForDescriptor(
+            issue_objects_pb2.Issue.DESCRIPTOR):
+          err_agg.AddErrorMessage(
+              'Invalid `update_mask` for {} delta.', api_delta.issue.name)
+
+    ingested = []
+    for iid, api_delta in zip(issue_ids, issue_deltas):
+      delta = tracker_pb2.IssueDelta()
+
+      # Check non-repeated fields before MergeMessage because in an object
+      # where fields are not set and with a FieldMask applied, there is no
+      # way to tell if empty fields were explicitly listed or not listed
+      # in the FieldMask.
+      paths_set = set(api_delta.update_mask.paths)
+      if (not paths_set.isdisjoint({'status', 'status.status'}) and
+          api_delta.issue.status.status):
+        delta.status = api_delta.issue.status.status
+      elif 'status.status' in paths_set and not api_delta.issue.status.status:
+        delta.status = ''
+
+      if (not paths_set.isdisjoint({'owner', 'owner.user'}) and
+          api_delta.issue.owner.user):
+        delta.owner_id = rnc.IngestUserName(
+              self.cnxn, api_delta.issue.owner.user, self.services)
+      elif 'owner.user' in paths_set and not api_delta.issue.owner.user:
+        delta.owner_id = framework_constants.NO_USER_SPECIFIED
+
+      if 'summary' in paths_set:
+        if api_delta.issue.summary:
+          delta.summary = api_delta.issue.summary
+        else:
+          delta.summary = ''
+
+      merge_ref = api_delta.issue.merged_into_issue_ref
+      if 'merged_into_issue_ref' in paths_set:
+        if (api_delta.issue.merged_into_issue_ref.issue or
+            api_delta.issue.merged_into_issue_ref.ext_identifier):
+          ingested_ref = self._IngestIssueRef(merge_ref)
+          if isinstance(ingested_ref, tracker_pb2.DanglingIssueRef):
+            delta.merged_into_external = ingested_ref.ext_issue_identifier
+          else:
+            delta.merged_into = ingested_ref
+      elif 'merged_into_issue_ref.issue' in paths_set:
+        if api_delta.issue.merged_into_issue_ref.issue:
+          delta.merged_into = self._IngestIssueRef(merge_ref)
+        else:
+          delta.merged_into = 0
+      elif 'merged_into_issue_ref.ext_identifier' in paths_set:
+        if api_delta.issue.merged_into_issue_ref.ext_identifier:
+          ingested_ref = self._IngestIssueRef(merge_ref)
+          delta.merged_into_external = ingested_ref.ext_issue_identifier
+        else:
+          delta.merged_into_external = ''
+
+      filtered_api_issue = issue_objects_pb2.Issue()
+      api_delta.update_mask.MergeMessage(
+          api_delta.issue,
+          filtered_api_issue,
+          replace_message_field=True,
+          replace_repeated_field=True)
+
+      cc_names = [name for name in api_delta.ccs_remove] + [
+          user_value.user for user_value in filtered_api_issue.cc_users
+      ]
+      cc_ids = rnc.IngestUserNames(self.cnxn, cc_names, self.services)
+      delta.cc_ids_remove = cc_ids[:len(api_delta.ccs_remove)]
+      delta.cc_ids_add = cc_ids[len(api_delta.ccs_remove):]
+
+      comp_names = [component for component in api_delta.components_remove] + [
+          c_value.component for c_value in filtered_api_issue.components
+      ]
+      project_comp_ids = rnc.IngestComponentDefNames(
+          self.cnxn, comp_names, self.services)
+      comp_ids = [comp_id for (_pid, comp_id) in project_comp_ids]
+      delta.comp_ids_remove = comp_ids[:len(api_delta.components_remove)]
+      delta.comp_ids_add = comp_ids[len(api_delta.components_remove):]
+
+      # Added to delta below, after ShiftEnumFieldsIntoLabels.
+      labels_add = [value.label for value in filtered_api_issue.labels]
+      labels_remove = [label for label in api_delta.labels_remove]
+
+      config = configs_by_pid[issues_dict[iid].project_id]
+      fvs_add, add_enums = self._IngestFieldValues(
+          filtered_api_issue.field_values, config)
+      fvs_remove, remove_enums = self._IngestFieldValues(
+          api_delta.field_vals_remove, config)
+      field_helpers.ShiftEnumFieldsIntoLabels(
+          labels_add, labels_remove, add_enums, remove_enums, config)
+      delta.field_vals_add = fvs_add
+      delta.field_vals_remove = fvs_remove
+      delta.labels_add = labels_add
+      delta.labels_remove = labels_remove
+      assert len(add_enums) == 0  # ShiftEnumFieldsIntoLabels clears all enums.
+      assert len(remove_enums) == 0
+
+      blocked_on_iids_rm, blocked_on_dangling_rm = self._IngestIssueRefs(
+          api_delta.blocked_on_issues_remove)
+      delta.blocked_on_remove = blocked_on_iids_rm
+      delta.ext_blocked_on_remove = [
+          ref.ext_issue_identifier for ref in blocked_on_dangling_rm
+      ]
+
+      blocked_on_iids_add, blocked_on_dangling_add = self._IngestIssueRefs(
+          filtered_api_issue.blocked_on_issue_refs)
+      delta.blocked_on_add = blocked_on_iids_add
+      delta.ext_blocked_on_add = [
+          ref.ext_issue_identifier for ref in blocked_on_dangling_add
+      ]
+
+      blocking_iids_rm, blocking_dangling_rm = self._IngestIssueRefs(
+          api_delta.blocking_issues_remove)
+      delta.blocking_remove = blocking_iids_rm
+      delta.ext_blocking_remove = [
+          ref.ext_issue_identifier for ref in blocking_dangling_rm
+      ]
+
+      blocking_iids_add, blocking_dangling_add = self._IngestIssueRefs(
+          filtered_api_issue.blocking_issue_refs)
+      delta.blocking_add = blocking_iids_add
+      delta.ext_blocking_add = [
+          ref.ext_issue_identifier for ref in blocking_dangling_add
+      ]
+
+      ingested.append((iid, delta))
+
+    return ingested
+
+  def IngestApprovalDeltas(self, approval_deltas, setter_id):
+    # type: (Sequence[api_proto.issues_pb2.ApprovalDelta], int) ->
+    #     Sequence[Tuple[int, int, proto.tracker_pb2.ApprovalDelta]]
+    """Ingests protoc ApprovalDeltas into protorpc ApprovalDeltas.
+
+    Args:
+      approval_deltas: the protoc ApprovalDeltas to ingest.
+      setter_id: The ID for the user setting the deltas.
+
+    Returns:
+      Sequence of (issue_id, approval_id, ApprovalDelta) tuples in the order
+      provided. The ApprovalDeltas ignore all OUTPUT_ONLY and masked fields.
+      The tuples are "delta_specifications;" they identify one requested change.
+
+    Raises:
+      InputException: if any fields in the approval_delta protos were invalid.
+      NoSuchProjectException: if the parent project of any ApprovalValue isn't
+          found.
+      NoSuchIssueException: if the issue of any ApprovalValue isn't found.
+      NoSuchUserException: if any user value was provided with an invalid email.
+          Note that users specified by ID are not checked for existence.
+    """
+    delta_specifications = []
+    set_on = int(time.time())  # Use the same timestamp for all deltas.
+    for approval_delta in approval_deltas:
+      approval_name = approval_delta.approval_value.name
+      # TODO(crbug/monorail/8173): Aggregate errors.
+      project_id, issue_id, approval_id = rnc.IngestApprovalValueName(
+          self.cnxn, approval_name, self.services)
+
+      if not approval_delta.HasField('update_mask'):
+        raise exceptions.InputException(
+            '`update_mask` must be set for %s delta.' % approval_name)
+      elif not approval_delta.update_mask.IsValidForDescriptor(
+          issue_objects_pb2.ApprovalValue.DESCRIPTOR):
+        raise exceptions.InputException(
+            'Invalid `update_mask` for %s delta.' % approval_name)
+      filtered_value = issue_objects_pb2.ApprovalValue()
+      approval_delta.update_mask.MergeMessage(
+          approval_delta.approval_value,
+          filtered_value,
+          replace_message_field=True,
+          replace_repeated_field=True)
+      status = _APPROVAL_STATUS_INGEST[filtered_value.status]
+      # Approvers
+      # No autocreate.
+      # A user may try to remove all existing approvers [a, b] and add another
+      # approver [c]. If they mis-type `c` and we auto-create `c` instead of
+      # raising error, this would cause the ApprovalValue to be editable by no
+      # one but site admins.
+      approver_ids_add = rnc.IngestUserNames(
+          self.cnxn, filtered_value.approvers, self.services, autocreate=False)
+      approver_ids_remove = rnc.IngestUserNames(
+          self.cnxn,
+          approval_delta.approvers_remove,
+          self.services,
+          autocreate=False)
+
+      # Field Values.
+      config = self.services.config.GetProjectConfig(self.cnxn, project_id)
+      approval_fds_by_id = {
+          fd.field_id: fd
+          for fd in config.field_defs
+          if fd.field_type is tracker_pb2.FieldTypes.APPROVAL_TYPE
+      }
+      if approval_id not in approval_fds_by_id:
+        raise exceptions.InputException(
+            'Approval not found in project for %s' % approval_name)
+
+      sub_fvs_add, add_enums = self._IngestFieldValues(
+          filtered_value.field_values, config, approval_id_filter=approval_id)
+      sub_fvs_remove, remove_enums = self._IngestFieldValues(
+          approval_delta.field_vals_remove,
+          config,
+          approval_id_filter=approval_id)
+      labels_add = []
+      labels_remove = []
+      field_helpers.ShiftEnumFieldsIntoLabels(
+          labels_add, labels_remove, add_enums, remove_enums, config)
+      assert len(add_enums) == 0  # ShiftEnumFieldsIntoLabels clears all enums.
+      assert len(remove_enums) == 0
+      delta = tbo.MakeApprovalDelta(
+          status,
+          setter_id,
+          approver_ids_add,
+          approver_ids_remove,
+          sub_fvs_add,
+          sub_fvs_remove, [],
+          labels_add,
+          labels_remove,
+          set_on=set_on)
+      delta_specifications.append((issue_id, approval_id, delta))
+    return delta_specifications
+
+  def IngestIssue(self, issue, project_id):
+    # type: (api_proto.issue_objects_pb2.Issue, int) -> proto.tracker_pb2.Issue
+    """Ingest a protoc Issue into a protorpc Issue.
+
+    Args:
+      issue: the protoc issue to ingest.
+      project_id: The project into which we're ingesting `issue`.
+
+    Returns:
+      protorpc version of issue, ignoring all OUTPUT_ONLY fields.
+
+    Raises:
+      InputException: if any fields in the 'issue' proto were invalid.
+      NoSuchProjectException: if 'project_id' is not found.
+    """
+    # Get config first. We can't ingest the issue if the project isn't found.
+    config = self.services.config.GetProjectConfig(self.cnxn, project_id)
+    ingestedDict = {
+      'project_id': project_id,
+      'summary': issue.summary
+    }
+    with exceptions.ErrorAggregator(exceptions.InputException) as err_agg:
+      self._ExtractOwner(issue, ingestedDict, err_agg)
+
+      # Extract ccs.
+      try:
+        ingestedDict['cc_ids'] = rnc.IngestUserNames(
+            self.cnxn, [cc.user for cc in issue.cc_users], self.services,
+            autocreate=True)
+      except exceptions.InputException as e:
+        err_agg.AddErrorMessage('Error ingesting cc_users: {}', e)
+
+      # Extract status.
+      if issue.HasField('status') and issue.status.status:
+        ingestedDict['status'] = issue.status.status
+      else:
+        err_agg.AddErrorMessage('Status is required when creating an issue')
+
+      # Extract components.
+      try:
+        project_comp_ids = rnc.IngestComponentDefNames(
+            self.cnxn, [cv.component for cv in issue.components], self.services)
+        ingestedDict['component_ids'] = [
+            comp_id for (_pid, comp_id) in project_comp_ids]
+      except (exceptions.InputException, exceptions.NoSuchProjectException,
+              exceptions.NoSuchComponentException) as e:
+        err_agg.AddErrorMessage('Error ingesting components: {}', e)
+
+      # Extract labels and field values.
+      ingestedDict['labels'] = [lv.label for lv in issue.labels]
+      try:
+        ingestedDict['field_values'], enums = self._IngestFieldValues(
+            issue.field_values, config)
+        field_helpers.ShiftEnumFieldsIntoLabels(
+            ingestedDict['labels'], [], enums, [], config)
+        assert len(
+            enums) == 0  # ShiftEnumFieldsIntoLabels must clear all enums.
+      except exceptions.InputException as e:
+        err_agg.AddErrorMessage(e.message)
+
+      # Ingest merged, blocking/blocked_on.
+      self._ExtractIssueRefs(issue, ingestedDict, err_agg)
+    return tracker_pb2.Issue(**ingestedDict)
+
+  def _IngestFieldValues(self, field_values, config, approval_id_filter=None):
+    # type: (Sequence[api_proto.issue_objects.FieldValue],
+    #     proto.tracker_pb2.ProjectIssueConfig, Optional[int]) ->
+    #     Tuple[Sequence[proto.tracker_pb2.FieldValue],
+    #         Mapping[int, Sequence[str]]]
+    """Returns protorpc FieldValues for the given protoc FieldValues.
+
+    Raises exceptions if any field could not be parsed for any reasons such as
+        unsupported field type, non-existent field, field from different
+        projects, or fields with mismatched parent approvals.
+
+    Args:
+      field_values: protoc FieldValues to ingest.
+      config: ProjectIssueConfig for the FieldValues we're ingesting.
+      approval_id_filter: an approval_id, including any FieldValues that does
+          not have this approval as a parent will trigger InputException.
+
+    Returns:
+      A pair 1) Ingested FieldValues. 2) A mapping of field ids to values
+      for ENUM_TYPE fields in 'field_values.'
+
+    Raises:
+      InputException: if any fields_values could not be parsed for any reasons
+          such as unsupported field type, non-existent field, or field from
+          different projects.
+    """
+    fds_by_id = {fd.field_id: fd for fd in config.field_defs}
+    enums = {}
+    ingestedFieldValues = []
+    with exceptions.ErrorAggregator(exceptions.InputException) as err_agg:
+      for fv in field_values:
+        try:
+          project_id, fd_id = rnc.IngestFieldDefName(
+              self.cnxn, fv.field, self.services)
+          fd = fds_by_id[fd_id]
+          # Raise if field does not belong to approval_id_filter (if provided).
+          if (approval_id_filter is not None and
+              fd.approval_id != approval_id_filter):
+            approval_name = rnc.ConvertApprovalDefNames(
+                self.cnxn, [approval_id_filter], project_id,
+                self.services)[approval_id_filter]
+            err_agg.AddErrorMessage(
+                'Field {} does not belong to approval {}', fv.field,
+                approval_name)
+            continue
+          if fd.field_type == tracker_pb2.FieldTypes.ENUM_TYPE:
+            enums.setdefault(fd_id, []).append(fv.value)
+          else:
+            ingestedFieldValues.append(self._IngestFieldValue(fv, fd))
+        except (exceptions.InputException, exceptions.NoSuchProjectException,
+                exceptions.NoSuchFieldDefException, ValueError) as e:
+          err_agg.AddErrorMessage(
+              'Could not ingest value ({}) for FieldDef ({}): {}', fv.value,
+              fv.field, e)
+        except exceptions.NoSuchUserException as e:
+          err_agg.AddErrorMessage(
+              'User ({}) not found when ingesting user field: {}', fv.value,
+              fv.field)
+        except KeyError as e:
+          err_agg.AddErrorMessage('Field {} is not in this project', fv.field)
+    return ingestedFieldValues, enums
+
+  def _IngestFieldValue(self, field_value, field_def):
+    # type: (api_proto.issue_objects.FieldValue, proto.tracker_pb2.FieldDef) ->
+    #     proto.tracker_pb2.FieldValue
+    """Ingest a protoc FieldValue into a protorpc FieldValue.
+
+    Args:
+      field_value: protoc FieldValue to ingest.
+      field_def: protorpc FieldDef associated with 'field_value'.
+          BOOL_TYPE and APPROVAL_TYPE are ignored.
+          Enum values are not allowed. They must be ingested as labels.
+
+    Returns:
+      Ingested protorpc FieldValue.
+
+    Raises:
+      InputException if 'field_def' is USER_TYPE and 'field_value' does not
+          have a valid formatted resource name.
+      NoSuchUserException if specified user in field does not exist.
+      ValueError if 'field_value' could not be parsed for 'field_def'.
+    """
+    assert field_def.field_type != tracker_pb2.FieldTypes.ENUM_TYPE
+    if field_def.field_type == tracker_pb2.FieldTypes.USER_TYPE:
+      return self._ParseOneUserFieldValue(field_value.value, field_def.field_id)
+    fv = field_helpers.ParseOneFieldValue(
+        self.cnxn, self.services.user, field_def, field_value.value)
+    # ParseOneFieldValue currently ignores parsing errors, although it has TODOs
+    # to raise them.
+    if not fv:
+      raise ValueError('Could not parse %s' % field_value.value)
+    return fv
+
+  def _ParseOneUserFieldValue(self, value, field_id):
+    # type: (str, int) -> proto.tracker_pb2.FieldValue
+    """Replacement for the obsolete user parsing in ParseOneFieldValue."""
+    user_id = rnc.IngestUserName(self.cnxn, value, self.services)
+    return tbo.MakeFieldValue(field_id, None, None, user_id, None, None, False)
+
+  def _ExtractOwner(self, issue, ingestedDict, err_agg):
+    # type: (api_proto.issue_objects_pb2.Issue, Dict[str, Any], ErrorAggregator)
+    #     -> None
+    """Fills 'owner' into `ingestedDict`, if it can be extracted."""
+    if issue.HasField('owner'):
+      try:
+        # Unlike for cc's, we require owner be an existing user, thus call we
+        # do not autocreate.
+        ingestedDict['owner_id'] = rnc.IngestUserName(
+            self.cnxn, issue.owner.user, self.services, autocreate=False)
+      except exceptions.InputException as e:
+        err_agg.AddErrorMessage(
+            'Error ingesting owner ({}): {}', issue.owner.user, e)
+      except exceptions.NoSuchUserException as e:
+        err_agg.AddErrorMessage(
+            'User ({}) not found when ingesting owner', e)
+    else:
+      ingestedDict['owner_id'] = framework_constants.NO_USER_SPECIFIED
+
+  def _ExtractIssueRefs(self, issue, ingestedDict, err_agg):
+    # type: (api_proto.issue_objects_pb2.Issue, Dict[str, Any], ErrorAggregator)
+    #     -> None
+    """Fills issue relationships into `ingestedDict` from `issue`."""
+    if issue.HasField('merged_into_issue_ref'):
+      try:
+        merged_into_ref = self._IngestIssueRef(issue.merged_into_issue_ref)
+        if isinstance(merged_into_ref, tracker_pb2.DanglingIssueRef):
+          ingestedDict['merged_into_external'] = (
+              merged_into_ref.ext_issue_identifier)
+        else:
+          ingestedDict['merged_into'] = merged_into_ref
+      except exceptions.InputException as e:
+        err_agg.AddErrorMessage(
+            'Error ingesting ref {}: {}', issue.merged_into_issue_ref, e)
+    try:
+      iids, dangling_refs = self._IngestIssueRefs(issue.blocked_on_issue_refs)
+      ingestedDict['blocked_on_iids'] = iids
+      ingestedDict['dangling_blocked_on_refs'] = dangling_refs
+    except exceptions.InputException as e:
+      err_agg.AddErrorMessage(e.message)
+    try:
+      iids, dangling_refs = self._IngestIssueRefs(issue.blocking_issue_refs)
+      ingestedDict['blocking_iids'] = iids
+      ingestedDict['dangling_blocking_refs'] = dangling_refs
+    except exceptions.InputException as e:
+      err_agg.AddErrorMessage(e.message)
+
+  def _IngestIssueRefs(self, issue_refs):
+    # type: (api_proto.issue_objects.IssueRf) ->
+    #     Tuple[Sequence[int], Sequence[tracker_pb2.DanglingIssueRef]]
+    """Given protoc IssueRefs, returns issue_ids and DanglingIssueRefs."""
+    issue_ids = []
+    external_refs = []
+    with exceptions.ErrorAggregator(exceptions.InputException) as err_agg:
+      for ref in issue_refs:
+        try:
+          ingested_ref = self._IngestIssueRef(ref)
+          if isinstance(ingested_ref, tracker_pb2.DanglingIssueRef):
+            external_refs.append(ingested_ref)
+          else:
+            issue_ids.append(ingested_ref)
+        except (exceptions.InputException, exceptions.NoSuchIssueException,
+                exceptions.NoSuchProjectException) as e:
+          err_agg.AddErrorMessage('Error ingesting ref {}: {}', ref, e)
+
+    return issue_ids, external_refs
+
+  def _IngestIssueRef(self, issue_ref):
+    # type: (api_proto.issue_objects.IssueRef) ->
+    #     Union[int, tracker_pb2.DanglingIssueRef]
+    """Given a protoc IssueRef, returns an issue id or DanglingIssueRef."""
+    if issue_ref.issue and issue_ref.ext_identifier:
+      raise exceptions.InputException(
+        'IssueRefs MUST NOT have both `issue` and `ext_identifier`')
+    if issue_ref.issue:
+      return rnc.IngestIssueName(self.cnxn, issue_ref.issue, self.services)
+    if issue_ref.ext_identifier:
+      # TODO(crbug.com/monorail/7208): Handle ingestion/conversion of CodeSite
+      # refs. We may be able to avoid ever needing to ingest them.
+      return tracker_pb2.DanglingIssueRef(
+          ext_issue_identifier=issue_ref.ext_identifier
+        )
+    raise exceptions.InputException(
+        'IssueRefs MUST have one of `issue` and `ext_identifier`')
+
+  def IngestIssuesListColumns(self, issues_list_columns):
+    # type: (Sequence[proto.issue_objects_pb2.IssuesListColumn] -> str
+    """Ingest a list of protoc IssueListColumns and returns a string."""
+    return ' '.join([col.column for col in issues_list_columns])
+
+  def _ComputeIssuesListColumns(self, columns):
+    # type: (string) -> Sequence[api_proto.issue_objects_pb2.IssuesListColumn]
+    """Convert string representation of columns to protoc IssuesListColumns"""
+    return [
+        issue_objects_pb2.IssuesListColumn(column=col)
+        for col in columns.split()
+    ]
+
+  def IngestNotifyType(self, notify):
+    # type: (issue_pb.NotifyType) -> bool
+    """Ingest a NotifyType to boolean."""
+    if (notify == issues_pb2.NotifyType.Value('NOTIFY_TYPE_UNSPECIFIED') or
+        notify == issues_pb2.NotifyType.Value('EMAIL')):
+      return True
+    elif notify == issues_pb2.NotifyType.Value('NO_NOTIFICATION'):
+      return False
+
+  # Users
+
+  def ConvertUser(self, user):
+    # type: (protorpc.User) -> api_proto.user_objects_pb2.User
+    """Convert a protorpc User into a protoc User.
+
+    Args:
+      user: protorpc User object.
+
+    Returns:
+      The protoc User object.
+    """
+    return self.ConvertUsers([user.user_id])[user.user_id]
+
+
+  # TODO(crbug/monorail/7238): Make this take in a full User object and
+  # return a Sequence, rather than a map, after hotlist users are converted.
+  def ConvertUsers(self, user_ids):
+    # type: (Sequence[int]) -> Map(int, api_proto.user_objects_pb2.User)
+    """Convert list of protorpc Users into list of protoc Users.
+
+    Args:
+      user_ids: List of User IDs.
+
+    Returns:
+      Dict of User IDs to User protos for given user_ids that could be found.
+    """
+    user_ids_to_names = {}
+
+    # Get display names
+    users_by_id = self.services.user.GetUsersByIDs(self.cnxn, user_ids)
+    (display_names_by_id,
+     display_emails_by_id) = framework_bizobj.CreateUserDisplayNamesAndEmails(
+         self.cnxn, self.services, self.user_auth, users_by_id.values())
+
+    for user_id, user in users_by_id.items():
+      name = rnc.ConvertUserNames([user_id]).get(user_id)
+
+      display_name = display_names_by_id.get(user_id)
+      display_email = display_emails_by_id.get(user_id)
+      availability = framework_helpers.GetUserAvailability(user)
+      availability_message, _availability_status = availability
+
+      user_ids_to_names[user_id] = user_objects_pb2.User(
+          name=name,
+          display_name=display_name,
+          email=display_email,
+          availability_message=availability_message)
+
+    return user_ids_to_names
+
+  def ConvertProjectStars(self, user_id, projects):
+    # type: (int, Collection[protorpc.Project]) ->
+    #     Collection[api_proto.user_objects_pb2.ProjectStar]
+    """Convert list of protorpc Projects into protoc ProjectStars.
+
+    Args:
+      user_id: The user the ProjectStar is associated with.
+      projects: All starred projects.
+
+    Returns:
+      List of ProjectStar messages.
+    """
+    api_project_stars = []
+    for proj in projects:
+      name = rnc.ConvertProjectStarName(
+          self.cnxn, user_id, proj.project_id, self.services)
+      star = user_objects_pb2.ProjectStar(name=name)
+      api_project_stars.append(star)
+    return api_project_stars
+
+  # Field Defs
+
+  def ConvertFieldDefs(self, field_defs, project_id):
+    # type: (Sequence[proto.tracker_pb2.FieldDef], int) ->
+    #     Sequence[api_proto.project_objects_pb2.FieldDef]
+    """Convert sequence of protorpc FieldDefs to protoc FieldDefs.
+
+    Args:
+      field_defs: List of protorpc FieldDefs
+      project_id: ID of the Project that is ancestor to all given
+        `field_defs`.
+
+    Returns:
+      Sequence of protoc FieldDef in the same order they are given in
+      `field_defs`. In the event any field_def or the referenced approval_id
+      in `field_defs` is not found, they will be omitted from the result.
+    """
+    field_ids = [fd.field_id for fd in field_defs]
+    resource_names_dict = rnc.ConvertFieldDefNames(
+        self.cnxn, field_ids, project_id, self.services)
+    parent_approval_ids = [
+        fd.approval_id for fd in field_defs if fd.approval_id is not None
+    ]
+    approval_names_dict = rnc.ConvertApprovalDefNames(
+        self.cnxn, parent_approval_ids, project_id, self.services)
+
+    api_fds = []
+    for fd in field_defs:
+      # Skip over approval fields, they have their separate ApprovalDef
+      if fd.field_type == tracker_pb2.FieldTypes.APPROVAL_TYPE:
+        continue
+      if fd.field_id not in resource_names_dict:
+        continue
+
+      name = resource_names_dict.get(fd.field_id)
+      display_name = fd.field_name
+      docstring = fd.docstring
+      field_type = self._ConvertFieldDefType(fd.field_type)
+      applicable_issue_type = fd.applicable_type
+      admins = rnc.ConvertUserNames(fd.admin_ids).values()
+      editors = rnc.ConvertUserNames(fd.editor_ids).values()
+      traits = self._ComputeFieldDefTraits(fd)
+      approval_parent = approval_names_dict.get(fd.approval_id)
+
+      enum_settings = None
+      if field_type == project_objects_pb2.FieldDef.Type.Value('ENUM'):
+        enum_settings = project_objects_pb2.FieldDef.EnumTypeSettings(
+            choices=self._GetEnumFieldChoices(fd))
+
+      int_settings = None
+      if field_type == project_objects_pb2.FieldDef.Type.Value('INT'):
+        int_settings = project_objects_pb2.FieldDef.IntTypeSettings(
+            min_value=fd.min_value, max_value=fd.max_value)
+
+      str_settings = None
+      if field_type == project_objects_pb2.FieldDef.Type.Value('STR'):
+        str_settings = project_objects_pb2.FieldDef.StrTypeSettings(
+            regex=fd.regex)
+
+      user_settings = None
+      if field_type == project_objects_pb2.FieldDef.Type.Value('USER'):
+        user_settings = project_objects_pb2.FieldDef.UserTypeSettings(
+            role_requirements=self._ConvertRoleRequirements(fd.needs_member),
+            notify_triggers=self._ConvertNotifyTriggers(fd.notify_on),
+            grants_perm=fd.grants_perm,
+            needs_perm=fd.needs_perm)
+
+      date_settings = None
+      if field_type == project_objects_pb2.FieldDef.Type.Value('DATE'):
+        date_settings = project_objects_pb2.FieldDef.DateTypeSettings(
+            date_action=self._ConvertDateAction(fd.date_action))
+
+      api_fd = project_objects_pb2.FieldDef(
+          name=name,
+          display_name=display_name,
+          docstring=docstring,
+          type=field_type,
+          applicable_issue_type=applicable_issue_type,
+          admins=admins,
+          traits=traits,
+          approval_parent=approval_parent,
+          enum_settings=enum_settings,
+          int_settings=int_settings,
+          str_settings=str_settings,
+          user_settings=user_settings,
+          date_settings=date_settings,
+          editors=editors)
+      api_fds.append(api_fd)
+    return api_fds
+
+  def _ConvertDateAction(self, date_action):
+    # type: (proto.tracker_pb2.DateAction) ->
+    #     api_proto.project_objects_pb2.FieldDef.DateTypeSettings.DateAction
+    """Convert protorpc DateAction to protoc
+       FieldDef.DateTypeSettings.DateAction"""
+    if date_action == tracker_pb2.DateAction.NO_ACTION:
+      return project_objects_pb2.FieldDef.DateTypeSettings.DateAction.Value(
+          'NO_ACTION')
+    elif date_action == tracker_pb2.DateAction.PING_OWNER_ONLY:
+      return project_objects_pb2.FieldDef.DateTypeSettings.DateAction.Value(
+          'NOTIFY_OWNER')
+    elif date_action == tracker_pb2.DateAction.PING_PARTICIPANTS:
+      return project_objects_pb2.FieldDef.DateTypeSettings.DateAction.Value(
+          'NOTIFY_PARTICIPANTS')
+    else:
+      raise ValueError('Unsupported DateAction Value')
+
+  def _ConvertRoleRequirements(self, needs_member):
+    # type: (bool) ->
+    #     api_proto.project_objects_pb2.FieldDef.
+    #     UserTypeSettings.RoleRequirements
+    """Convert protorpc RoleRequirements to protoc
+       FieldDef.UserTypeSettings.RoleRequirements"""
+
+    proto_user_settings = project_objects_pb2.FieldDef.UserTypeSettings
+    if needs_member:
+      return proto_user_settings.RoleRequirements.Value('PROJECT_MEMBER')
+    else:
+      return proto_user_settings.RoleRequirements.Value('NO_ROLE_REQUIREMENT')
+
+  def _ConvertNotifyTriggers(self, notify_trigger):
+    # type: (proto.tracker_pb2.NotifyTriggers) ->
+    #     api_proto.project_objects_pb2.FieldDef.UserTypeSettings.NotifyTriggers
+    """Convert protorpc NotifyTriggers to protoc
+       FieldDef.UserTypeSettings.NotifyTriggers"""
+    if notify_trigger == tracker_pb2.NotifyTriggers.NEVER:
+      return project_objects_pb2.FieldDef.UserTypeSettings.NotifyTriggers.Value(
+          'NEVER')
+    elif notify_trigger == tracker_pb2.NotifyTriggers.ANY_COMMENT:
+      return project_objects_pb2.FieldDef.UserTypeSettings.NotifyTriggers.Value(
+          'ANY_COMMENT')
+    else:
+      raise ValueError('Unsupported NotifyTriggers Value')
+
+  def _ConvertFieldDefType(self, field_type):
+    # type: (proto.tracker_pb2.FieldTypes) ->
+    #     api_proto.project_objects_pb2.FieldDef.Type
+    """Convert protorpc FieldType to protoc FieldDef.Type
+
+    Args:
+      field_type: protorpc FieldType
+
+    Returns:
+      Corresponding protoc FieldDef.Type
+
+    Raises:
+      ValueError if input `field_type` has no suitable supported FieldDef.Type,
+      or input `field_type` is not a recognized enum option.
+    """
+    if field_type == tracker_pb2.FieldTypes.ENUM_TYPE:
+      return project_objects_pb2.FieldDef.Type.Value('ENUM')
+    elif field_type == tracker_pb2.FieldTypes.INT_TYPE:
+      return project_objects_pb2.FieldDef.Type.Value('INT')
+    elif field_type == tracker_pb2.FieldTypes.STR_TYPE:
+      return project_objects_pb2.FieldDef.Type.Value('STR')
+    elif field_type == tracker_pb2.FieldTypes.USER_TYPE:
+      return project_objects_pb2.FieldDef.Type.Value('USER')
+    elif field_type == tracker_pb2.FieldTypes.DATE_TYPE:
+      return project_objects_pb2.FieldDef.Type.Value('DATE')
+    elif field_type == tracker_pb2.FieldTypes.URL_TYPE:
+      return project_objects_pb2.FieldDef.Type.Value('URL')
+    else:
+      raise ValueError(
+          'Unsupported tracker_pb2.FieldType enum. Boolean types '
+          'are unsupported and approval types are found in ApprovalDefs')
+
+  def _ComputeFieldDefTraits(self, field_def):
+    # type: (proto.tracker_pb2.FieldDef) ->
+    #     Sequence[api_proto.project_objects_pb2.FieldDef.Traits]
+    """Compute sequence of FieldDef.Traits for a given protorpc FieldDef."""
+    trait_protos = []
+    if field_def.is_required:
+      trait_protos.append(project_objects_pb2.FieldDef.Traits.Value('REQUIRED'))
+    if field_def.is_niche:
+      trait_protos.append(
+          project_objects_pb2.FieldDef.Traits.Value('DEFAULT_HIDDEN'))
+    if field_def.is_multivalued:
+      trait_protos.append(
+          project_objects_pb2.FieldDef.Traits.Value('MULTIVALUED'))
+    if field_def.is_phase_field:
+      trait_protos.append(project_objects_pb2.FieldDef.Traits.Value('PHASE'))
+    if field_def.is_restricted_field:
+      trait_protos.append(
+          project_objects_pb2.FieldDef.Traits.Value('RESTRICTED'))
+    return trait_protos
+
+  def _GetEnumFieldChoices(self, field_def):
+    # type: (proto.tracker_pb2.FieldDef) ->
+    #     Sequence[Choice]
+    """Get sequence of choices for an enum field
+
+    Args:
+      field_def: protorpc FieldDef
+
+    Returns:
+      Sequence of valid Choices for enum field `field_def`.
+
+    Raises:
+      ValueError if input `field_def` is not an enum type field.
+    """
+    if field_def.field_type != tracker_pb2.FieldTypes.ENUM_TYPE:
+      raise ValueError('Cannot get value from label for non-enum-type field')
+
+    config = self.services.config.GetProjectConfig(
+        self.cnxn, field_def.project_id)
+    value_docstr_tuples = tracker_helpers._GetEnumFieldValuesAndDocstrings(
+        field_def, config)
+
+    return [
+        Choice(value=value, docstring=docstring)
+        for value, docstring in value_docstr_tuples
+    ]
+
+  # Field Values
+
+  def _GetNonApprovalFieldValues(self, field_values, project_id):
+    # type: (Sequence[proto.tracker_pb2.FieldValue], int) ->
+    #     Sequence[proto.tracker_pb2.FieldValue]
+    """Filter out field values that belong to an approval field."""
+    config = self.services.config.GetProjectConfig(self.cnxn, project_id)
+    approval_fd_ids = set(
+        [fd.field_id for fd in config.field_defs if fd.approval_id])
+
+    return [fv for fv in field_values if fv.field_id not in approval_fd_ids]
+
+  def ConvertFieldValues(self, field_values, project_id, phases):
+    # type: (Sequence[proto.tracker_pb2.FieldValue], int,
+    #     Sequence[proto.tracker_pb2.Phase]) ->
+    #     Sequence[api_proto.issue_objects_pb2.FieldValue]
+    """Convert sequence of field_values to protoc FieldValues.
+
+    This method does not handle enum_type fields.
+
+    Args:
+      field_values: List of FieldValues
+      project_id: ID of the Project that is ancestor to all given
+        `field_values`.
+      phases: List of Phases
+
+    Returns:
+      Sequence of protoc FieldValues in the same order they are given in
+      `field_values`. In the event any field_values in `field_values` are not
+      found, they will be omitted from the result.
+    """
+    phase_names_by_id = {phase.phase_id: phase.name for phase in phases}
+    field_ids = [fv.field_id for fv in field_values]
+    resource_names_dict = rnc.ConvertFieldDefNames(
+        self.cnxn, field_ids, project_id, self.services)
+
+    api_fvs = []
+    for fv in field_values:
+      if fv.field_id not in resource_names_dict:
+        continue
+
+      name = resource_names_dict.get(fv.field_id)
+      value = self._ComputeFieldValueString(fv)
+      derivation = self._ComputeFieldValueDerivation(fv)
+      phase = phase_names_by_id.get(fv.phase_id)
+      api_item = issue_objects_pb2.FieldValue(
+          field=name, value=value, derivation=derivation, phase=phase)
+      api_fvs.append(api_item)
+
+    return api_fvs
+
+  def _ComputeFieldValueString(self, field_value):
+    # type: (proto.tracker_pb2.FieldValue) -> str
+    """Convert a FieldValue's value to a string."""
+    if field_value is None:
+      raise exceptions.InputException('No FieldValue specified')
+    elif field_value.int_value is not None:
+      return str(field_value.int_value)
+    elif field_value.str_value is not None:
+      return field_value.str_value
+    elif field_value.user_id is not None:
+      return rnc.ConvertUserNames([field_value.user_id
+                                  ]).get(field_value.user_id)
+    elif field_value.date_value is not None:
+      return str(field_value.date_value)
+    elif field_value.url_value is not None:
+      return field_value.url_value
+    else:
+      raise exceptions.InputException('FieldValue must have at least one value')
+
+  def _ComputeFieldValueDerivation(self, field_value):
+    # type: (proto.tracker_pb2.FieldValue) ->
+    #     api_proto.issue_objects_pb2.Issue.Derivation
+    """Convert a FieldValue's 'derived' to a protoc Issue.Derivation.
+
+    Args:
+      field_value: protorpc FieldValue
+
+    Returns:
+      Issue.Derivation of given `field_value`
+    """
+    if field_value.derived:
+      return issue_objects_pb2.Derivation.Value('RULE')
+    else:
+      return issue_objects_pb2.Derivation.Value('EXPLICIT')
+
+  # Approval Def
+
+  def ConvertApprovalDefs(self, approval_defs, project_id):
+    # type: (Sequence[proto.tracker_pb2.ApprovalDef], int) ->
+    #     Sequence[api_proto.project_objects_pb2.ApprovalDef]
+    """Convert sequence of protorpc ApprovalDefs to protoc ApprovalDefs.
+
+    Args:
+      approval_defs: List of protorpc ApprovalDefs
+      project_id: ID of the Project the approval_defs belong to.
+
+    Returns:
+      Sequence of protoc ApprovalDefs in the same order they are given in
+      in `approval_defs`. In the event any approval_def in `approval_defs`
+      are not found, they will be omitted from the result.
+    """
+    approval_ids = set([ad.approval_id for ad in approval_defs])
+    resource_names_dict = rnc.ConvertApprovalDefNames(
+        self.cnxn, approval_ids, project_id, self.services)
+
+    # Get matching field defs, needed to fill out protoc ApprovalDefs
+    config = self.services.config.GetProjectConfig(self.cnxn, project_id)
+    fd_by_id = {}
+    for fd in config.field_defs:
+      if (fd.field_type == tracker_pb2.FieldTypes.APPROVAL_TYPE and
+          fd.field_id in approval_ids):
+        fd_by_id[fd.field_id] = fd
+
+    all_users = tbo.UsersInvolvedInApprovalDefs(
+        approval_defs, fd_by_id.values())
+    user_resource_names_dict = rnc.ConvertUserNames(all_users)
+
+    api_ads = []
+    for ad in approval_defs:
+      if (ad.approval_id not in resource_names_dict or
+          ad.approval_id not in fd_by_id):
+        continue
+      matching_fd = fd_by_id.get(ad.approval_id)
+      name = resource_names_dict.get(ad.approval_id)
+      display_name = matching_fd.field_name
+      docstring = matching_fd.docstring
+      survey = ad.survey
+      approvers = [
+          user_resource_names_dict.get(approver_id)
+          for approver_id in ad.approver_ids
+      ]
+      admins = [
+          user_resource_names_dict.get(admin_id)
+          for admin_id in matching_fd.admin_ids
+      ]
+
+      api_ad = project_objects_pb2.ApprovalDef(
+          name=name,
+          display_name=display_name,
+          docstring=docstring,
+          survey=survey,
+          approvers=approvers,
+          admins=admins)
+      api_ads.append(api_ad)
+    return api_ads
+
+  def ConvertApprovalValues(self, approval_values, field_values, phases,
+                            issue_id=None, project_id=None):
+    # type: (Sequence[proto.tracker_pb2.ApprovalValue],
+    #     Sequence[proto.tracker_pb2.FieldValue],
+    #     Sequence[proto.tracker_pb2.Phase], Optional[int], Optional[int]) ->
+    #     Sequence[api_proto.issue_objects_pb2.ApprovalValue]
+    """Convert sequence of approval_values to protoc ApprovalValues.
+
+    `approval_values` may belong to a template or an issue. If they belong to a
+    template, `project_id` should be given for the project the template is in.
+    If these are issue `approval_values` `issue_id` should be given`.
+    So, one of `issue_id` or `project_id` must be provided.
+    If both are given, we ignore `project_id` and assume the `approval_values`
+    belong to an issue.
+
+    Args:
+      approval_values: List of ApprovalValues.
+      field_values: List of FieldValues that may belong to the approval_values.
+      phases: List of Phases that may be associated with the approval_values.
+      issue_id: ID of the Issue that the `approval_values` belong to.
+      project_id: ID of the Project that the `approval_values`
+        template belongs to.
+
+    Returns:
+      Sequence of protoc ApprovalValues in the same order they are given in
+      in `approval_values`. In the event any approval definitions in
+      `approval_values` are not found, they will be omitted from the result.
+
+    Raises:
+      InputException if neither `issue_id` nor `project_id` is given.
+    """
+
+    approval_ids = [av.approval_id for av in approval_values]
+    resource_names_dict = {}
+    if issue_id is not None:
+      # Only issue approval_values have resource names.
+      resource_names_dict = rnc.ConvertApprovalValueNames(
+          self.cnxn, issue_id, self.services)
+      project_id = self.services.issue.GetIssue(self.cnxn, issue_id).project_id
+    elif project_id is None:
+      raise exceptions.InputException(
+          'One  `issue_id` or `project_id` must be given.')
+
+    phase_names_by_id = {phase.phase_id: phase.name for phase in phases}
+    ad_names_dict = rnc.ConvertApprovalDefNames(
+        self.cnxn, approval_ids, project_id, self.services)
+
+    # Organize the field values by the approval values they are
+    # associated with.
+    config = self.services.config.GetProjectConfig(self.cnxn, project_id)
+    fds_by_id = {fd.field_id: fd for fd in config.field_defs}
+    fvs_by_parent_approvals = collections.defaultdict(list)
+    for fv in field_values:
+      fd = fds_by_id.get(fv.field_id)
+      if fd and fd.approval_id:
+        fvs_by_parent_approvals[fd.approval_id].append(fv)
+
+    api_avs = []
+    for av in approval_values:
+      # We only skip missing approval names if we are converting issue approval
+      # values.
+      if issue_id is not None and av.approval_id not in resource_names_dict:
+        continue
+
+      name = resource_names_dict.get(av.approval_id)
+      approval_def = ad_names_dict.get(av.approval_id)
+      approvers = rnc.ConvertUserNames(av.approver_ids).values()
+      status = self._ComputeApprovalValueStatus(av.status)
+      setter = rnc.ConvertUserName(av.setter_id)
+      phase = phase_names_by_id.get(av.phase_id)
+
+      field_values = self.ConvertFieldValues(
+          fvs_by_parent_approvals[av.approval_id], project_id, phases)
+
+      api_item = issue_objects_pb2.ApprovalValue(
+          name=name,
+          approval_def=approval_def,
+          approvers=approvers,
+          status=status,
+          setter=setter,
+          field_values=field_values,
+          phase=phase)
+      if av.set_on:
+        api_item.set_time.FromSeconds(av.set_on)
+      api_avs.append(api_item)
+
+    return api_avs
+
+  def _ComputeApprovalValueStatus(self, status):
+    # type: (proto.tracker_pb2.ApprovalStatus) ->
+    #     api_proto.issue_objects_pb2.Issue.ApprovalStatus
+    """Convert a protorpc ApprovalStatus to a protoc Issue.ApprovalStatus."""
+    try:
+      return _APPROVAL_STATUS_CONVERT[status]
+    except KeyError:
+      raise ValueError('Unrecognized tracker_pb2.ApprovalStatus enum')
+
+  # Projects
+
+  def ConvertIssueTemplates(self, project_id, templates):
+    # type: (int, Sequence[proto.tracker_pb2.TemplateDef]) ->
+    #     Sequence[api_proto.project_objects_pb2.IssueTemplate]
+    """Convert a Sequence of TemplateDefs to protoc IssueTemplates.
+
+    Args:
+      project_id: ID of the Project the templates belong to.
+      templates: Sequence of TemplateDef protorpc objects.
+
+    Returns:
+      Sequence of protoc IssueTemplate in the same order they are given in
+      `templates`. In the rare event that any templates are not found,
+      they will be omitted from the result.
+    """
+    api_templates = []
+
+    resource_names_dict = rnc.ConvertTemplateNames(
+        self.cnxn, project_id, [template.template_id for template in templates],
+        self.services)
+
+    for template in templates:
+      if template.template_id not in resource_names_dict:
+        continue
+      name = resource_names_dict.get(template.template_id)
+      summary_must_be_edited = template.summary_must_be_edited
+      template_privacy = self._ComputeTemplatePrivacy(template)
+      default_owner = self._ComputeTemplateDefaultOwner(template)
+      component_required = template.component_required
+      admins = rnc.ConvertUserNames(template.admin_ids).values()
+      issue = self._FillIssueFromTemplate(template, project_id)
+      approval_values = self.ConvertApprovalValues(
+          template.approval_values, template.field_values, template.phases,
+          project_id=project_id)
+      api_templates.append(
+          project_objects_pb2.IssueTemplate(
+              name=name,
+              display_name=template.name,
+              issue=issue,
+              approval_values=approval_values,
+              summary_must_be_edited=summary_must_be_edited,
+              template_privacy=template_privacy,
+              default_owner=default_owner,
+              component_required=component_required,
+              admins=admins))
+
+    return api_templates
+
+  def _FillIssueFromTemplate(self, template, project_id):
+    # type: (proto.tracker_pb2.TemplateDef, int) ->
+    #     api_proto.issue_objects_pb2.Issue
+    """Convert a TemplateDef to its embedded protoc Issue.
+
+    IssueTemplate does not set the following fields:
+      name
+      reporter
+      cc_users
+      blocked_on_issue_refs
+      blocking_issue_refs
+      create_time
+      close_time
+      modify_time
+      component_modify_time
+      status_modify_time
+      owner_modify_time
+      attachment_count
+      star_count
+
+    Args:
+      template: TemplateDef protorpc objects.
+      project_id: ID of the Project the template belongs to.
+
+    Returns:
+      protoc Issue filled with data from given `template`.
+    """
+    summary = template.summary
+    state = issue_objects_pb2.IssueContentState.Value('ACTIVE')
+    status = issue_objects_pb2.Issue.StatusValue(
+        status=template.status,
+        derivation=issue_objects_pb2.Derivation.Value('EXPLICIT'))
+    owner = None
+    if template.owner_id is not None:
+      owner = issue_objects_pb2.Issue.UserValue(
+          user=rnc.ConvertUserNames([template.owner_id]).get(template.owner_id))
+    labels = self.ConvertLabels(template.labels, [], project_id)
+    components_dict = rnc.ConvertComponentDefNames(
+        self.cnxn, template.component_ids, project_id, self.services)
+    components = []
+    for component_resource_name in components_dict.values():
+      components.append(
+          issue_objects_pb2.Issue.ComponentValue(
+              component=component_resource_name,
+              derivation=issue_objects_pb2.Derivation.Value('EXPLICIT')))
+    non_approval_fvs = self._GetNonApprovalFieldValues(
+        template.field_values, project_id)
+    field_values = self.ConvertFieldValues(
+        non_approval_fvs, project_id, template.phases)
+    field_values.extend(
+        self.ConvertEnumFieldValues(template.labels, [], project_id))
+    phases = self._ComputePhases(template.phases)
+
+    filled_issue = issue_objects_pb2.Issue(
+        summary=summary,
+        state=state,
+        status=status,
+        owner=owner,
+        labels=labels,
+        components=components,
+        field_values=field_values,
+        phases=phases)
+    return filled_issue
+
+  def _ComputeTemplatePrivacy(self, template):
+    # type: (proto.tracker_pb2.TemplateDef) ->
+    #     api_proto.project_objects_pb2.IssueTemplate.TemplatePrivacy
+    """Convert a protorpc TemplateDef to its protoc TemplatePrivacy."""
+    if template.members_only:
+      return project_objects_pb2.IssueTemplate.TemplatePrivacy.Value(
+          'MEMBERS_ONLY')
+    else:
+      return project_objects_pb2.IssueTemplate.TemplatePrivacy.Value('PUBLIC')
+
+  def _ComputeTemplateDefaultOwner(self, template):
+    # type: (proto.tracker_pb2.TemplateDef) ->
+    #     api_proto.project_objects_pb2.IssueTemplate.DefaultOwner
+    """Convert a protorpc TemplateDef to its protoc DefaultOwner."""
+    if template.owner_defaults_to_member:
+      return project_objects_pb2.IssueTemplate.DefaultOwner.Value(
+          'PROJECT_MEMBER_REPORTER')
+    else:
+      return project_objects_pb2.IssueTemplate.DefaultOwner.Value(
+          'DEFAULT_OWNER_UNSPECIFIED')
+
+  def _ComputePhases(self, phases):
+    # type: (proto.tracker_pb2.TemplateDef) -> Sequence[str]
+    """Convert a protorpc TemplateDef to its sorted string phases."""
+    sorted_phases = sorted(phases, key=lambda phase: phase.rank)
+    return [phase.name for phase in sorted_phases]
+
+  def ConvertLabels(self, labels, derived_labels, project_id):
+    # type: (Sequence[str], Sequence[str], int) ->
+    #     Sequence[api_proto.issue_objects_pb2.Issue.LabelValue]
+    """Convert string labels to LabelValues for non-enum-field labels
+
+    Args:
+      labels: Sequence of string labels
+      project_id: ID of the Project these labels belong to.
+
+    Return:
+      Sequence of protoc IssueValues for given `labels` that
+      do not represent enum field values.
+    """
+    config = self.services.config.GetProjectConfig(self.cnxn, project_id)
+    non_fd_labels, non_fd_der_labels = tbo.ExplicitAndDerivedNonMaskedLabels(
+        labels, derived_labels, config)
+    api_labels = []
+    for label in non_fd_labels:
+      api_labels.append(
+          issue_objects_pb2.Issue.LabelValue(
+              label=label,
+              derivation=issue_objects_pb2.Derivation.Value('EXPLICIT')))
+    for label in non_fd_der_labels:
+      api_labels.append(
+          issue_objects_pb2.Issue.LabelValue(
+              label=label,
+              derivation=issue_objects_pb2.Derivation.Value('RULE')))
+    return api_labels
+
+  def ConvertEnumFieldValues(self, labels, derived_labels, project_id):
+    # type: (Sequence[str], Sequence[str], int) ->
+    #     Sequence[api_proto.issue_objects_pb2.FieldValue]
+    """Convert string labels to FieldValues for enum-field labels
+
+    Args:
+      labels: Sequence of string labels
+      project_id: ID of the Project these labels belong to.
+
+    Return:
+      Sequence of protoc FieldValues only for given `labels` that
+      represent enum field values.
+    """
+    config = self.services.config.GetProjectConfig(self.cnxn, project_id)
+    enum_ids_by_name = {
+        fd.field_name.lower(): fd.field_id
+        for fd in config.field_defs
+        if fd.field_type is tracker_pb2.FieldTypes.ENUM_TYPE
+        and not fd.is_deleted
+    }
+    resource_names_dict = rnc.ConvertFieldDefNames(
+        self.cnxn, enum_ids_by_name.values(), project_id, self.services)
+
+    api_fvs = []
+
+    labels_by_prefix = tbo.LabelsByPrefix(labels, enum_ids_by_name.keys())
+    for lower_field_name, values in labels_by_prefix.items():
+      field_id = enum_ids_by_name.get(lower_field_name)
+      resource_name = resource_names_dict.get(field_id)
+      if not resource_name:
+        continue
+      api_fvs.extend(
+          [
+              issue_objects_pb2.FieldValue(
+                  field=resource_name,
+                  value=value,
+                  derivation=issue_objects_pb2.Derivation.Value(
+                      'EXPLICIT')) for value in values
+          ])
+
+    der_labels_by_prefix = tbo.LabelsByPrefix(
+        derived_labels, enum_ids_by_name.keys())
+    for lower_field_name, values in der_labels_by_prefix.items():
+      field_id = enum_ids_by_name.get(lower_field_name)
+      resource_name = resource_names_dict.get(field_id)
+      if not resource_name:
+        continue
+      api_fvs.extend(
+          [
+              issue_objects_pb2.FieldValue(
+                  field=resource_name,
+                  value=value,
+                  derivation=issue_objects_pb2.Derivation.Value('RULE'))
+              for value in values
+          ])
+
+    return api_fvs
+
+  def ConvertProject(self, project):
+    # type: (proto.project_pb2.Project) ->
+    #     api_proto.project_objects_pb2.Project
+    """Convert a protorpc Project to its protoc Project."""
+
+    return project_objects_pb2.Project(
+        name=rnc.ConvertProjectName(
+            self.cnxn, project.project_id, self.services),
+        display_name=project.project_name,
+        summary=project.summary,
+        thumbnail_url=project_helpers.GetThumbnailUrl(project.logo_gcs_id))
+
+  def ConvertProjects(self, projects):
+    # type: (Sequence[proto.project_pb2.Project]) ->
+    #     Sequence[api_proto.project_objects_pb2.Project]
+    """Convert a Sequence of protorpc Projects to protoc Projects."""
+    return [self.ConvertProject(proj) for proj in projects]
+
+  def ConvertProjectConfig(self, project_config):
+    # type: (proto.tracker_pb2.ProjectIssueConfig) ->
+    #     api_proto.project_objects_pb2.ProjectConfig
+    """Convert protorpc ProjectIssueConfig to protoc ProjectConfig."""
+    project = self.services.project.GetProject(
+        self.cnxn, project_config.project_id)
+    project_grid_config = project_objects_pb2.ProjectConfig.GridViewConfig(
+        default_x_attr=project_config.default_x_attr,
+        default_y_attr=project_config.default_y_attr)
+    template_names = rnc.ConvertTemplateNames(
+        self.cnxn, project_config.project_id, [
+            project_config.default_template_for_developers,
+            project_config.default_template_for_users
+        ], self.services)
+    return project_objects_pb2.ProjectConfig(
+        name=rnc.ConvertProjectConfigName(
+            self.cnxn, project_config.project_id, self.services),
+        exclusive_label_prefixes=project_config.exclusive_label_prefixes,
+        member_default_query=project_config.member_default_query,
+        default_sort=project_config.default_sort_spec,
+        default_columns=self._ComputeIssuesListColumns(
+            project_config.default_col_spec),
+        project_grid_config=project_grid_config,
+        member_default_template=template_names.get(
+            project_config.default_template_for_developers),
+        non_members_default_template=template_names.get(
+            project_config.default_template_for_users),
+        revision_url_format=project.revision_url_format,
+        custom_issue_entry_url=project_config.custom_issue_entry_url)
+
+  def CreateProjectMember(self, cnxn, project_id, user_id, role):
+    # type: (MonorailContext, int, int, str) ->
+    #     api_proto.project_objects_pb2.ProjectMember
+    """Creates a ProjectMember object from specified parameters.
+
+    Args:
+      cnxn: MonorailConnection object.
+      project_id: ID of the Project the User is a member of.
+      user_id: ID of the user who is a member.
+      role: str specifying the user's role based on a ProjectRole value.
+
+    Return:
+      A protoc ProjectMember object.
+    """
+    name = rnc.ConvertProjectMemberName(
+        cnxn, project_id, user_id, self.services)
+    return project_objects_pb2.ProjectMember(
+        name=name,
+        role=project_objects_pb2.ProjectMember.ProjectRole.Value(role))
+
+  def ConvertLabelDefs(self, label_defs, project_id):
+    # type: (Sequence[proto.tracker_pb2.LabelDef], int) ->
+    #     Sequence[api_proto.project_objects_pb2.LabelDef]
+    """Convert protorpc LabelDefs to protoc LabelDefs"""
+    resource_names_dict = rnc.ConvertLabelDefNames(
+        self.cnxn, [ld.label for ld in label_defs], project_id, self.services)
+
+    api_lds = []
+    for ld in label_defs:
+      state = project_objects_pb2.LabelDef.LabelDefState.Value('ACTIVE')
+      if ld.deprecated:
+        state = project_objects_pb2.LabelDef.LabelDefState.Value('DEPRECATED')
+      api_lds.append(
+          project_objects_pb2.LabelDef(
+              name=resource_names_dict.get(ld.label),
+              value=ld.label,
+              docstring=ld.label_docstring,
+              state=state))
+    return api_lds
+
+  def ConvertStatusDefs(self, status_defs, project_id):
+    # type: (Sequence[proto.tracker_pb2.StatusDef], int) ->
+    #     Sequence[api_proto.project_objects_pb2.StatusDef]
+    """Convert protorpc StatusDefs to protoc StatusDefs
+
+    Args:
+      status_defs: Sequence of StatusDefs.
+      project_id: ID of the Project these belong to.
+
+    Returns:
+      Sequence of protoc StatusDefs in the same order they are given in
+      `status_defs`.
+    """
+    resource_names_dict = rnc.ConvertStatusDefNames(
+        self.cnxn, [sd.status for sd in status_defs], project_id, self.services)
+    config = self.services.config.GetProjectConfig(self.cnxn, project_id)
+    mergeable_statuses = set(config.statuses_offer_merge)
+
+    # Rank is only surfaced as positional value in well_known_statuses
+    rank_by_status = {}
+    for rank, sd in enumerate(config.well_known_statuses):
+      rank_by_status[sd.status] = rank
+
+    api_sds = []
+    for sd in status_defs:
+      state = project_objects_pb2.StatusDef.StatusDefState.Value('ACTIVE')
+      if sd.deprecated:
+        state = project_objects_pb2.StatusDef.StatusDefState.Value('DEPRECATED')
+
+      if sd.means_open:
+        status_type = project_objects_pb2.StatusDef.StatusDefType.Value('OPEN')
+      else:
+        if sd.status in mergeable_statuses:
+          status_type = project_objects_pb2.StatusDef.StatusDefType.Value(
+              'MERGED')
+        else:
+          status_type = project_objects_pb2.StatusDef.StatusDefType.Value(
+              'CLOSED')
+
+      api_sd = project_objects_pb2.StatusDef(
+          name=resource_names_dict.get(sd.status),
+          value=sd.status,
+          type=status_type,
+          rank=rank_by_status[sd.status],
+          docstring=sd.status_docstring,
+          state=state,
+      )
+      api_sds.append(api_sd)
+    return api_sds
+
+  def ConvertComponentDef(self, component_def):
+    # type: (proto.tracker_pb2.ComponentDef) ->
+    #     api_proto.project_objects.ComponentDef
+    """Convert a protorpc ComponentDef to a protoc ComponentDef."""
+    return self.ConvertComponentDefs([component_def],
+                                     component_def.project_id)[0]
+
+  def ConvertComponentDefs(self, component_defs, project_id):
+    # type: (Sequence[proto.tracker_pb2.ComponentDef], int) ->
+    #     Sequence[api_proto.project_objects.ComponentDef]
+    """Convert sequence of protorpc ComponentDefs to protoc ComponentDefs
+
+    Args:
+      component_defs: Sequence of protoc ComponentDefs.
+      project_id: ID of the Project these belong to.
+
+    Returns:
+      Sequence of protoc ComponentDefs in the same order they are given in
+      `component_defs`.
+    """
+    resource_names_dict = rnc.ConvertComponentDefNames(
+        self.cnxn, [cd.component_id for cd in component_defs], project_id,
+        self.services)
+    involved_user_ids = tbo.UsersInvolvedInComponents(component_defs)
+    user_resource_names_dict = rnc.ConvertUserNames(involved_user_ids)
+
+    all_label_ids = set()
+    for cd in component_defs:
+      all_label_ids.update(cd.label_ids)
+
+    # If this becomes a performance issue, we should add bulk look up.
+    labels_by_id = {
+        label_id: self.services.config.LookupLabel(
+            self.cnxn, project_id, label_id) for label_id in all_label_ids
+    }
+
+    api_cds = []
+    for cd in component_defs:
+      state = project_objects_pb2.ComponentDef.ComponentDefState.Value('ACTIVE')
+      if cd.deprecated:
+        state = project_objects_pb2.ComponentDef.ComponentDefState.Value(
+            'DEPRECATED')
+
+      api_cd = project_objects_pb2.ComponentDef(
+          name=resource_names_dict.get(cd.component_id),
+          value=cd.path,
+          docstring=cd.docstring,
+          state=state,
+          admins=[
+              user_resource_names_dict.get(admin_id)
+              for admin_id in cd.admin_ids
+          ],
+          ccs=[user_resource_names_dict.get(cc_id) for cc_id in cd.cc_ids],
+          creator=user_resource_names_dict.get(cd.creator_id),
+          modifier=user_resource_names_dict.get(cd.modifier_id),
+          create_time=timestamp_pb2.Timestamp(seconds=cd.created),
+          modify_time=timestamp_pb2.Timestamp(seconds=cd.modified),
+          labels=[labels_by_id[label_id] for label_id in cd.label_ids],
+      )
+      api_cds.append(api_cd)
+    return api_cds
+
+  def ConvertProjectSavedQueries(self, saved_queries, project_id):
+    # type: (Sequence[proto.tracker_pb2.SavedQuery], int) ->
+    #     Sequence(api_proto.project_objects.ProjectSavedQuery)
+    """Convert sequence of protorpc SavedQueries to protoc ProjectSavedQueries
+
+    Args:
+      saved_queries: Sequence of SavedQueries.
+      project_id: ID of the Project these belong to.
+
+    Returns:
+      Sequence of protoc ProjectSavedQueries in the same order they are given in
+      `saved_queries`. In the event any items in `saved_queries` are not found
+      or don't belong to the project, they will be omitted from the result.
+    """
+    resource_names_dict = rnc.ConvertProjectSavedQueryNames(
+        self.cnxn, [sq.query_id for sq in saved_queries], project_id,
+        self.services)
+    api_psqs = []
+    for sq in saved_queries:
+      if sq.query_id not in resource_names_dict:
+        continue
+
+      # TODO(crbug/monorail/7756): Remove base_query_id, avoid confusions.
+      # Until then we have to expand the query by including base_query_id.
+      # base_query_id can only be in the set of DEFAULT_CANNED_QUERIES.
+      if sq.base_query_id:
+        query = '{} {}'.format(tbo.GetBuiltInQuery(sq.base_query_id), sq.query)
+      else:
+        query = sq.query
+
+      api_psqs.append(
+          project_objects_pb2.ProjectSavedQuery(
+              name=resource_names_dict.get(sq.query_id),
+              display_name=sq.name,
+              query=query))
+    return api_psqs