blob: 60aebd7b53e9089f4842c3fe1240407a6f4a258d [file] [log] [blame]
# 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