File: //snap/google-cloud-cli/current/lib/googlecloudsdk/api_lib/pubsub/topics.py
# -*- coding: utf-8 -*- #
# Copyright 2017 Google LLC. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Utilities for Cloud Pub/Sub Topics API."""
from __future__ import absolute_import
from __future__ import division
from __future__ import unicode_literals
from apitools.base.py import list_pager
from googlecloudsdk.api_lib.pubsub import utils
from googlecloudsdk.api_lib.util import apis
from googlecloudsdk.command_lib.iam import iam_util
from googlecloudsdk.core import exceptions
CLEAR_MESSAGE_TRANSFORMS_VALUE = []
class PublishOperationException(exceptions.Error):
"""Error when something went wrong with publish."""
class EmptyMessageException(exceptions.Error):
"""Error when no message was specified for a Publish operation."""
class NoFieldsSpecifiedError(exceptions.Error):
"""Error when no fields were specified for a Patch operation."""
class InvalidSchemaSettingsException(exceptions.Error):
"""Error when the schema settings are invalid."""
class ConflictingIngestionSettingsException(exceptions.Error):
"""Error when the ingestion settings are invalid."""
class _TopicUpdateSetting(object):
"""Data container class for updating a topic."""
def __init__(self, field_name, value):
self.field_name = field_name
self.value = value
def GetClientInstance(no_http=False):
return apis.GetClientInstance('pubsub', 'v1', no_http=no_http)
def GetMessagesModule(client=None):
client = client or GetClientInstance()
return client.MESSAGES_MODULE
def ParseMessageEncoding(messages, message_encoding):
enc = message_encoding.lower()
if enc == 'json':
return messages.SchemaSettings.EncodingValueValuesEnum.JSON
elif enc == 'binary':
return messages.SchemaSettings.EncodingValueValuesEnum.BINARY
else:
raise InvalidSchemaSettingsException(
'Unknown message encoding. Options are JSON or BINARY.'
)
class TopicsClient(object):
"""Client for topics service in the Cloud Pub/Sub API."""
def __init__(self, client=None, messages=None):
self.client = client or GetClientInstance()
self.messages = messages or GetMessagesModule(client)
self._service = self.client.projects_topics
self._subscriptions_service = self.client.projects_subscriptions
def _ParseIngestionPlatformLogsSettings(self, ingestion_log_severity):
if ingestion_log_severity:
return self.messages.PlatformLogsSettings(
severity=self.messages.PlatformLogsSettings.SeverityValueValuesEnum(
ingestion_log_severity
)
)
return None
def _ParseIngestionDataSourceSettings(
self,
kinesis_ingestion_stream_arn=None,
kinesis_ingestion_consumer_arn=None,
kinesis_ingestion_role_arn=None,
kinesis_ingestion_service_account=None,
cloud_storage_ingestion_bucket=None,
cloud_storage_ingestion_input_format=None,
cloud_storage_ingestion_text_delimiter=None,
cloud_storage_ingestion_minimum_object_create_time=None,
cloud_storage_ingestion_match_glob=None,
azure_event_hubs_ingestion_resource_group=None,
azure_event_hubs_ingestion_namespace=None,
azure_event_hubs_ingestion_event_hub=None,
azure_event_hubs_ingestion_client_id=None,
azure_event_hubs_ingestion_tenant_id=None,
azure_event_hubs_ingestion_subscription_id=None,
azure_event_hubs_ingestion_service_account=None,
aws_msk_ingestion_cluster_arn=None,
aws_msk_ingestion_topic=None,
aws_msk_ingestion_aws_role_arn=None,
aws_msk_ingestion_service_account=None,
confluent_cloud_ingestion_bootstrap_server=None,
confluent_cloud_ingestion_cluster_id=None,
confluent_cloud_ingestion_topic=None,
confluent_cloud_ingestion_identity_pool_id=None,
confluent_cloud_ingestion_service_account=None,
ingestion_log_severity=None,
):
"""Returns an IngestionDataSourceSettings message from the provided args."""
# For each datasource type, check if all required flags are passed, and
# conditionally construct the source and return the first datasource type
# which is present. We let the argument parser enforce mutual exclusion of
# argument groups.
is_kinesis = (
(kinesis_ingestion_stream_arn is not None)
and (kinesis_ingestion_consumer_arn is not None)
and (kinesis_ingestion_role_arn is not None)
and (kinesis_ingestion_service_account is not None)
)
is_cloud_storage = (cloud_storage_ingestion_bucket is not None) and (
cloud_storage_ingestion_input_format is not None
)
is_azure_event_hubs = (
(azure_event_hubs_ingestion_resource_group is not None)
and (azure_event_hubs_ingestion_namespace is not None)
and (azure_event_hubs_ingestion_event_hub is not None)
and (azure_event_hubs_ingestion_client_id is not None)
and (azure_event_hubs_ingestion_tenant_id is not None)
and (azure_event_hubs_ingestion_subscription_id is not None)
and (azure_event_hubs_ingestion_service_account is not None)
)
is_msk = (
(aws_msk_ingestion_cluster_arn is not None)
and (aws_msk_ingestion_topic is not None)
and (aws_msk_ingestion_aws_role_arn is not None)
and (aws_msk_ingestion_service_account is not None)
)
is_confluent_cloud = (
(confluent_cloud_ingestion_bootstrap_server is not None)
and (confluent_cloud_ingestion_cluster_id is not None)
and (confluent_cloud_ingestion_topic is not None)
and (confluent_cloud_ingestion_identity_pool_id is not None)
and (confluent_cloud_ingestion_service_account is not None)
)
if is_kinesis:
kinesis_source = self.messages.AwsKinesis(
streamArn=kinesis_ingestion_stream_arn,
consumerArn=kinesis_ingestion_consumer_arn,
awsRoleArn=kinesis_ingestion_role_arn,
gcpServiceAccount=kinesis_ingestion_service_account,
)
return self.messages.IngestionDataSourceSettings(
awsKinesis=kinesis_source,
platformLogsSettings=self._ParseIngestionPlatformLogsSettings(
ingestion_log_severity
),
)
elif is_cloud_storage:
cloud_storage_source = self.messages.CloudStorage(
bucket=cloud_storage_ingestion_bucket,
minimumObjectCreateTime=cloud_storage_ingestion_minimum_object_create_time,
matchGlob=cloud_storage_ingestion_match_glob,
)
if cloud_storage_ingestion_input_format == 'text':
cloud_storage_source.textFormat = self.messages.TextFormat(
delimiter=cloud_storage_ingestion_text_delimiter
)
elif cloud_storage_ingestion_input_format == 'avro':
cloud_storage_source.avroFormat = self.messages.AvroFormat()
elif cloud_storage_ingestion_input_format == 'pubsub_avro':
cloud_storage_source.pubsubAvroFormat = self.messages.PubSubAvroFormat()
return self.messages.IngestionDataSourceSettings(
cloudStorage=cloud_storage_source,
platformLogsSettings=self._ParseIngestionPlatformLogsSettings(
ingestion_log_severity
),
)
elif is_azure_event_hubs:
azure_event_hubs_source = self.messages.AzureEventHubs(
resourceGroup=azure_event_hubs_ingestion_resource_group,
namespace=azure_event_hubs_ingestion_namespace,
eventHub=azure_event_hubs_ingestion_event_hub,
clientId=azure_event_hubs_ingestion_client_id,
tenantId=azure_event_hubs_ingestion_tenant_id,
subscriptionId=azure_event_hubs_ingestion_subscription_id,
gcpServiceAccount=azure_event_hubs_ingestion_service_account,
)
return self.messages.IngestionDataSourceSettings(
azureEventHubs=azure_event_hubs_source,
platformLogsSettings=self._ParseIngestionPlatformLogsSettings(
ingestion_log_severity
),
)
elif is_msk:
msk_source = self.messages.AwsMsk(
clusterArn=aws_msk_ingestion_cluster_arn,
topic=aws_msk_ingestion_topic,
awsRoleArn=aws_msk_ingestion_aws_role_arn,
gcpServiceAccount=aws_msk_ingestion_service_account,
)
return self.messages.IngestionDataSourceSettings(
awsMsk=msk_source,
platformLogsSettings=self._ParseIngestionPlatformLogsSettings(
ingestion_log_severity
),
)
elif is_confluent_cloud:
confluent_cloud_source = self.messages.ConfluentCloud(
bootstrapServer=confluent_cloud_ingestion_bootstrap_server,
clusterId=confluent_cloud_ingestion_cluster_id,
topic=confluent_cloud_ingestion_topic,
identityPoolId=confluent_cloud_ingestion_identity_pool_id,
gcpServiceAccount=confluent_cloud_ingestion_service_account,
)
return self.messages.IngestionDataSourceSettings(
confluentCloud=confluent_cloud_source,
platformLogsSettings=self._ParseIngestionPlatformLogsSettings(
ingestion_log_severity
),
)
elif ingestion_log_severity:
raise ConflictingIngestionSettingsException(
'Must set ingestion settings with log severity.'
)
return None
def Create(
self,
topic_ref,
labels=None,
kms_key=None,
message_retention_duration=None,
message_storage_policy_allowed_regions=None,
message_storage_policy_enforce_in_transit=False,
schema=None,
message_encoding=None,
first_revision_id=None,
last_revision_id=None,
kinesis_ingestion_stream_arn=None,
kinesis_ingestion_consumer_arn=None,
kinesis_ingestion_role_arn=None,
kinesis_ingestion_service_account=None,
cloud_storage_ingestion_bucket=None,
cloud_storage_ingestion_input_format=None,
cloud_storage_ingestion_text_delimiter=None,
cloud_storage_ingestion_minimum_object_create_time=None,
cloud_storage_ingestion_match_glob=None,
azure_event_hubs_ingestion_resource_group=None,
azure_event_hubs_ingestion_namespace=None,
azure_event_hubs_ingestion_event_hub=None,
azure_event_hubs_ingestion_client_id=None,
azure_event_hubs_ingestion_tenant_id=None,
azure_event_hubs_ingestion_subscription_id=None,
azure_event_hubs_ingestion_service_account=None,
aws_msk_ingestion_cluster_arn=None,
aws_msk_ingestion_topic=None,
aws_msk_ingestion_aws_role_arn=None,
aws_msk_ingestion_service_account=None,
confluent_cloud_ingestion_bootstrap_server=None,
confluent_cloud_ingestion_cluster_id=None,
confluent_cloud_ingestion_topic=None,
confluent_cloud_ingestion_identity_pool_id=None,
confluent_cloud_ingestion_service_account=None,
ingestion_log_severity=None,
message_transforms_file=None,
tags=None,
enable_vertex_ai_smt=False,
):
"""Creates a Topic.
Args:
topic_ref (Resource): Resource reference to the Topic to create.
labels (LabelsValue): Labels for the topic to create.
kms_key (str): Full resource name of kms_key to set on Topic or None.
message_retention_duration (str): How long to retain messages published to
the Topic.
message_storage_policy_allowed_regions (list[str]): List of Cloud regions
in which messages are allowed to be stored at rest.
message_storage_policy_enforce_in_transit (bool): Whether or not to
enforce in-transit guarantees for this topic using the allowed regions.
schema (Resource): Full resource name of schema used to validate messages
published on Topic.
message_encoding (str): If a schema is set, the message encoding of
incoming messages to be validated against the schema.
first_revision_id (str): If a schema is set, the revision id of the oldest
revision allowed for validation.
last_revision_id (str): If a schema is set, the revision id of the newest
revision allowed for validation.
kinesis_ingestion_stream_arn (str): The Kinesis data stream ARN to ingest
data from.
kinesis_ingestion_consumer_arn (str): The Kinesis data streams consumer
ARN to use for ingestion.
kinesis_ingestion_role_arn (str): AWS role ARN to be used for Federated
Identity authentication with Kinesis.
kinesis_ingestion_service_account (str): The GCP service account to be
used for Federated Identity authentication with Kinesis
cloud_storage_ingestion_bucket (str): The Cloud Storage bucket to ingest
data from.
cloud_storage_ingestion_input_format (str): the format of the data in the
Cloud Storage bucket ('text', 'avro', or 'pubsub_avro').
cloud_storage_ingestion_text_delimiter (optional[str]): delimiter to use
with text format when partioning the object.
cloud_storage_ingestion_minimum_object_create_time (optional[str]): only
Cloud Storage objects with a larger or equal creation timestamp will be
ingested.
cloud_storage_ingestion_match_glob (optional[str]): glob pattern used to
match Cloud Storage objects that will be ingested. If unset, all objects
will be ingested.
azure_event_hubs_ingestion_resource_group (str): The name of the resource
group within an Azure subscription.
azure_event_hubs_ingestion_namespace (str): The name of the Azure Event
Hubs namespace.
azure_event_hubs_ingestion_event_hub (str): The name of the Azure event
hub.
azure_event_hubs_ingestion_client_id (str): The client id of the Azure
Event Hubs application used to authenticate Pub/Sub.
azure_event_hubs_ingestion_tenant_id (str): The tenant id of the Azure
Event Hubs application used to authenticate Pub/Sub.
azure_event_hubs_ingestion_subscription_id (str): The id of the Azure
Event Hubs subscription.
azure_event_hubs_ingestion_service_account (str): The GCP service account
to be used for Federated Identity authentication with Azure Event Hubs.
aws_msk_ingestion_cluster_arn (str): The ARN that uniquely identifies the
MSK cluster.
aws_msk_ingestion_topic (str): The name of the MSK topic that Pub/Sub will
import from.
aws_msk_ingestion_aws_role_arn (str): AWS role ARN to be used for
Federated Identity authentication with MSK.
aws_msk_ingestion_service_account (str): The GCP service account to be
used for Federated Identity authentication with MSK.
confluent_cloud_ingestion_bootstrap_server (str): The address of the
Confluent Cloud bootstrap server. The format is url:port.
confluent_cloud_ingestion_cluster_id (str): The id of the Confluent Cloud
cluster.
confluent_cloud_ingestion_topic (str): The name of the Confluent Cloud
topic that Pub/Sub will import from.
confluent_cloud_ingestion_identity_pool_id (str): The id of the identity
pool to be used for Federated Identity authentication with Confluent
Cloud.
confluent_cloud_ingestion_service_account (str): The GCP service account
to be used for Federated Identity authentication with Confluent Cloud.
ingestion_log_severity (optional[str]): The log severity to use for
ingestion.
message_transforms_file (str): The file path to the JSON or YAML file
containing the message transforms.
tags (TagsValue): The tag Keys/Values to be bound to the topic.
enable_vertex_ai_smt (bool): Whether or not to enable Vertex AI message
transforms.
Returns:
Topic: The created topic.
Raises:
InvalidSchemaSettingsException: If an invalid --schema,
--message-encoding flag comnbination is specified,
or if the --first_revision_id revision is newer than
the --last_revision_id specified.
"""
topic = self.messages.Topic(
name=topic_ref.RelativeName(),
labels=labels,
messageRetentionDuration=message_retention_duration,
)
if kms_key:
topic.kmsKeyName = kms_key
if message_storage_policy_allowed_regions:
message_storage_policy = self.messages.MessageStoragePolicy(
allowedPersistenceRegions=message_storage_policy_allowed_regions
)
if message_storage_policy_enforce_in_transit:
message_storage_policy.enforceInTransit = (
message_storage_policy_enforce_in_transit
)
topic.messageStoragePolicy = message_storage_policy
if schema and message_encoding:
encoding_enum = ParseMessageEncoding(self.messages, message_encoding)
topic.schemaSettings = self.messages.SchemaSettings(
schema=schema,
encoding=encoding_enum,
firstRevisionId=first_revision_id,
lastRevisionId=last_revision_id,
)
topic.ingestionDataSourceSettings = self._ParseIngestionDataSourceSettings(
kinesis_ingestion_stream_arn=kinesis_ingestion_stream_arn,
kinesis_ingestion_consumer_arn=kinesis_ingestion_consumer_arn,
kinesis_ingestion_role_arn=kinesis_ingestion_role_arn,
kinesis_ingestion_service_account=kinesis_ingestion_service_account,
cloud_storage_ingestion_bucket=cloud_storage_ingestion_bucket,
cloud_storage_ingestion_input_format=cloud_storage_ingestion_input_format,
cloud_storage_ingestion_text_delimiter=cloud_storage_ingestion_text_delimiter,
cloud_storage_ingestion_minimum_object_create_time=cloud_storage_ingestion_minimum_object_create_time,
cloud_storage_ingestion_match_glob=cloud_storage_ingestion_match_glob,
azure_event_hubs_ingestion_resource_group=azure_event_hubs_ingestion_resource_group,
azure_event_hubs_ingestion_namespace=azure_event_hubs_ingestion_namespace,
azure_event_hubs_ingestion_event_hub=azure_event_hubs_ingestion_event_hub,
azure_event_hubs_ingestion_client_id=azure_event_hubs_ingestion_client_id,
azure_event_hubs_ingestion_tenant_id=azure_event_hubs_ingestion_tenant_id,
azure_event_hubs_ingestion_subscription_id=azure_event_hubs_ingestion_subscription_id,
azure_event_hubs_ingestion_service_account=azure_event_hubs_ingestion_service_account,
aws_msk_ingestion_cluster_arn=aws_msk_ingestion_cluster_arn,
aws_msk_ingestion_topic=aws_msk_ingestion_topic,
aws_msk_ingestion_aws_role_arn=aws_msk_ingestion_aws_role_arn,
aws_msk_ingestion_service_account=aws_msk_ingestion_service_account,
confluent_cloud_ingestion_bootstrap_server=confluent_cloud_ingestion_bootstrap_server,
confluent_cloud_ingestion_cluster_id=confluent_cloud_ingestion_cluster_id,
confluent_cloud_ingestion_topic=confluent_cloud_ingestion_topic,
confluent_cloud_ingestion_identity_pool_id=confluent_cloud_ingestion_identity_pool_id,
confluent_cloud_ingestion_service_account=confluent_cloud_ingestion_service_account,
ingestion_log_severity=ingestion_log_severity,
)
if message_transforms_file:
try:
topic.messageTransforms = utils.GetMessageTransformsFromFile(
self.messages.MessageTransform,
message_transforms_file,
enable_vertex_ai_smt,
)
except (
utils.MessageTransformsInvalidFormatError,
utils.MessageTransformsEmptyFileError,
utils.MessageTransformsMissingFileError,
) as e:
e.args = (utils.GetErrorMessage(e),)
raise
if tags:
topic.tags = tags
return self._service.Create(topic)
def Get(self, topic_ref):
"""Gets a Topic.
Args:
topic_ref (Resource): Resource reference to the Topic to get.
Returns:
Topic: The topic.
"""
get_req = self.messages.PubsubProjectsTopicsGetRequest(
topic=topic_ref.RelativeName()
)
return self._service.Get(get_req)
def Delete(self, topic_ref):
"""Deletes a Topic.
Args:
topic_ref (Resource): Resource reference to the Topic to delete.
Returns:
Empty: An empty response message.
"""
delete_req = self.messages.PubsubProjectsTopicsDeleteRequest(
topic=topic_ref.RelativeName()
)
return self._service.Delete(delete_req)
def DetachSubscription(self, subscription_ref):
"""Detaches the subscription from its topic.
Args:
subscription_ref (Resource): Resource reference to the Subscription to
detach.
Returns:
Empty: An empty response message.
"""
detach_req = self.messages.PubsubProjectsSubscriptionsDetachRequest(
subscription=subscription_ref.RelativeName()
)
return self._subscriptions_service.Detach(detach_req)
def List(self, project_ref, page_size=100):
"""Lists Topics for a given project.
Args:
project_ref (Resource): Resource reference to Project to list Topics from.
page_size (int): the number of entries in each batch (affects requests
made, but not the yielded results).
Returns:
A generator of Topics in the Project.
"""
list_req = self.messages.PubsubProjectsTopicsListRequest(
project=project_ref.RelativeName(), pageSize=page_size
)
return list_pager.YieldFromList(
self._service,
list_req,
batch_size=page_size,
field='topics',
batch_size_attribute='pageSize',
)
def ListSnapshots(self, topic_ref, page_size=100):
"""Lists Snapshots for a given topic.
Args:
topic_ref (Resource): Resource reference to Topic to list snapshots from.
page_size (int): the number of entries in each batch (affects requests
made, but not the yielded results).
Returns:
A generator of Snapshots for the Topic.
"""
list_req = self.messages.PubsubProjectsTopicsSnapshotsListRequest(
topic=topic_ref.RelativeName(), pageSize=page_size
)
list_snaps_service = self.client.projects_topics_snapshots
return list_pager.YieldFromList(
list_snaps_service,
list_req,
batch_size=page_size,
field='snapshots',
batch_size_attribute='pageSize',
)
def ListSubscriptions(self, topic_ref, page_size=100):
"""Lists Subscriptions for a given topic.
Args:
topic_ref (Resource): Resource reference to Topic to list subscriptions
from.
page_size (int): the number of entries in each batch (affects requests
made, but not the yielded results).
Returns:
A generator of Subscriptions for the Topic..
"""
list_req = self.messages.PubsubProjectsTopicsSubscriptionsListRequest(
topic=topic_ref.RelativeName(), pageSize=page_size
)
list_subs_service = self.client.projects_topics_subscriptions
return list_pager.YieldFromList(
list_subs_service,
list_req,
batch_size=page_size,
field='subscriptions',
batch_size_attribute='pageSize',
)
def Publish(
self, topic_ref, message_body=None, attributes=None, ordering_key=None
):
"""Publishes a message to the given topic.
Args:
topic_ref (Resource): Resource reference to Topic to publish to.
message_body (bytes): Message to send.
attributes (list[AdditionalProperty]): List of attributes to attach to the
message.
ordering_key (string): The ordering key to associate with this message.
Returns:
PublishResponse: Response message with message ids from the API.
Raises:
EmptyMessageException: If neither message nor attributes is
specified.
PublishOperationException: When something went wrong with the publish
operation.
"""
if not message_body and not attributes:
raise EmptyMessageException(
'You cannot send an empty message. You must specify either a '
'MESSAGE, one or more ATTRIBUTE, or both.'
)
message = self.messages.PubsubMessage(
data=message_body,
attributes=self.messages.PubsubMessage.AttributesValue(
additionalProperties=attributes
),
orderingKey=ordering_key,
)
publish_req = self.messages.PubsubProjectsTopicsPublishRequest(
publishRequest=self.messages.PublishRequest(messages=[message]),
topic=topic_ref.RelativeName(),
)
result = self._service.Publish(publish_req)
if not result.messageIds:
# If we got a result with empty messageIds, then we've got a problem.
raise PublishOperationException(
'Publish operation failed with Unknown error.'
)
return result
def SetIamPolicy(self, topic_ref, policy):
"""Sets an IAM policy on a Topic.
Args:
topic_ref (Resource): Resource reference for topic to set IAM policy on.
policy (Policy): The policy to be added to the Topic.
Returns:
Policy: the policy which was set.
"""
request = self.messages.PubsubProjectsTopicsSetIamPolicyRequest(
resource=topic_ref.RelativeName(),
setIamPolicyRequest=self.messages.SetIamPolicyRequest(policy=policy),
)
return self._service.SetIamPolicy(request)
def GetIamPolicy(self, topic_ref):
"""Gets the IAM policy for a Topic.
Args:
topic_ref (Resource): Resource reference for topic to get the IAM policy
of.
Returns:
Policy: the policy for the Topic.
"""
request = self.messages.PubsubProjectsTopicsGetIamPolicyRequest(
resource=topic_ref.RelativeName()
)
return self._service.GetIamPolicy(request)
def AddIamPolicyBinding(self, topic_ref, member, role):
"""Adds an IAM Policy binding to a Topic.
Args:
topic_ref (Resource): Resource reference for subscription to add IAM
policy binding to.
member (str): The member to add.
role (str): The role to assign to the member.
Returns:
Policy: the updated policy.
Raises:
api_exception.HttpException: If either of the requests failed.
"""
policy = self.GetIamPolicy(topic_ref)
iam_util.AddBindingToIamPolicy(self.messages.Binding, policy, member, role)
return self.SetIamPolicy(topic_ref, policy)
def RemoveIamPolicyBinding(self, topic_ref, member, role):
"""Removes an IAM Policy binding from a Topic.
Args:
topic_ref (Resource): Resource reference for subscription to remove IAM
policy binding from.
member (str): The member to remove.
role (str): The role to remove the member from.
Returns:
Policy: the updated policy.
Raises:
api_exception.HttpException: If either of the requests failed.
"""
policy = self.GetIamPolicy(topic_ref)
iam_util.RemoveBindingFromIamPolicy(policy, member, role)
return self.SetIamPolicy(topic_ref, policy)
def Patch(
self,
topic_ref,
labels=None,
kms_key_name=None,
message_retention_duration=None,
clear_message_retention_duration=False,
recompute_message_storage_policy=False,
message_storage_policy_allowed_regions=None,
message_storage_policy_enforce_in_transit=False,
schema=None,
message_encoding=None,
first_revision_id=None,
last_revision_id=None,
clear_schema_settings=None,
clear_ingestion_data_source_settings=False,
kinesis_ingestion_stream_arn=None,
kinesis_ingestion_consumer_arn=None,
kinesis_ingestion_role_arn=None,
kinesis_ingestion_service_account=None,
cloud_storage_ingestion_bucket=None,
cloud_storage_ingestion_input_format=None,
cloud_storage_ingestion_text_delimiter=None,
cloud_storage_ingestion_minimum_object_create_time=None,
cloud_storage_ingestion_match_glob=None,
azure_event_hubs_ingestion_resource_group=None,
azure_event_hubs_ingestion_namespace=None,
azure_event_hubs_ingestion_event_hub=None,
azure_event_hubs_ingestion_client_id=None,
azure_event_hubs_ingestion_tenant_id=None,
azure_event_hubs_ingestion_subscription_id=None,
azure_event_hubs_ingestion_service_account=None,
aws_msk_ingestion_cluster_arn=None,
aws_msk_ingestion_topic=None,
aws_msk_ingestion_aws_role_arn=None,
aws_msk_ingestion_service_account=None,
confluent_cloud_ingestion_bootstrap_server=None,
confluent_cloud_ingestion_cluster_id=None,
confluent_cloud_ingestion_topic=None,
confluent_cloud_ingestion_identity_pool_id=None,
confluent_cloud_ingestion_service_account=None,
ingestion_log_severity=None,
message_transforms_file=None,
clear_message_transforms=False,
enable_vertex_ai_smt=False,
):
"""Updates a Topic.
Args:
topic_ref (Resource): Resource reference for the topic to be updated.
labels (LabelsValue): The Cloud labels for the topic.
kms_key_name (str): The full resource name of the Cloud KMS key to
associate with the topic, or None.
message_retention_duration (str): How long to retain messages.
clear_message_retention_duration (bool): If set, remove retention from the
topic.
recompute_message_storage_policy (bool): True to have the API recalculate
the message storage policy.
message_storage_policy_allowed_regions (list[str]): List of Cloud regions
in which messages are allowed to be stored at rest.
message_storage_policy_enforce_in_transit (bool): Whether or not to
enforce in-transit guarantees for this topic using the allowed regions.
schema (Resource): Full resource name of schema used to validate messages
published on Topic.
message_encoding (str): If a schema is set, the message encoding of
incoming messages to be validated against the schema.
first_revision_id (str): If a schema is set, the revision id of the oldest
revision allowed for validation.
last_revision_id (str): If a schema is set, the revision id of the newest
revision allowed for validation.
clear_schema_settings (bool): If set, clear schema settings from the
topic.
clear_ingestion_data_source_settings (bool): If set, clear
IngestionDataSourceSettings from the topic.
kinesis_ingestion_stream_arn (str): The Kinesis data stream ARN to ingest
data from.
kinesis_ingestion_consumer_arn (str): The Kinesis data streams consumer
ARN to use for ingestion.
kinesis_ingestion_role_arn (str): AWS role ARN to be used for Federated
Identity authentication with Kinesis.
kinesis_ingestion_service_account (str): The GCP service account to be
used for Federated Identity authentication with Kinesis
cloud_storage_ingestion_bucket (str): The Cloud Storage bucket to ingest
data from.
cloud_storage_ingestion_input_format (str): the format of the data in the
Cloud Storage bucket ('text', 'avro', or 'pubsub_avro').
cloud_storage_ingestion_text_delimiter (optional[str]): delimiter to use
with text format when partioning the object.
cloud_storage_ingestion_minimum_object_create_time (optional[str]): only
Cloud Storage objects with a larger or equal creation timestamp will be
ingested.
cloud_storage_ingestion_match_glob (optional[str]): glob pattern used to
match Cloud Storage objects that will be ingested. If unset, all objects
will be ingested.
azure_event_hubs_ingestion_resource_group (str): The name of the resource
group within an Azure subscription.
azure_event_hubs_ingestion_namespace (str): The name of the Azure Event
Hubs namespace.
azure_event_hubs_ingestion_event_hub (str): The name of the Azure event
hub.
azure_event_hubs_ingestion_client_id (str): The client id of the Azure
Event Hubs application used to authenticate Pub/Sub.
azure_event_hubs_ingestion_tenant_id (str): The tenant id of the Azure
Event Hubs application used to authenticate Pub/Sub.
azure_event_hubs_ingestion_subscription_id (str): The id of the Azure
Event Hubs subscription.
azure_event_hubs_ingestion_service_account (str): The GCP service account
to be used for Federated Identity authentication with Azure Event Hubs.
aws_msk_ingestion_cluster_arn (str): The ARN that uniquely identifies the
MSK cluster.
aws_msk_ingestion_topic (str): The name of the MSK topic that Pub/Sub will
import from.
aws_msk_ingestion_aws_role_arn (str): AWS role ARN to be used for
Federated Identity authentication with MSK.
aws_msk_ingestion_service_account (str): The GCP service account to be
used for Federated Identity authentication with MSK.
confluent_cloud_ingestion_bootstrap_server (str): The address of the
Confluent Cloud bootstrap server. The format is url:port.
confluent_cloud_ingestion_cluster_id (str): The id of the Confluent Cloud
cluster.
confluent_cloud_ingestion_topic (str): The name of the Confluent Cloud
topic that Pub/Sub will import from.
confluent_cloud_ingestion_identity_pool_id (str): The id of the identity
pool to be used for Federated Identity authentication with Confluent
Cloud.
confluent_cloud_ingestion_service_account (str): The GCP service account
to be used for Federated Identity authentication with Confluent Cloud.
ingestion_log_severity (optional[str]): The log severity to use for
ingestion.
message_transforms_file (str): The file path to the JSON or YAML file
containing the message transforms.
clear_message_transforms (bool): If set, clears all message transforms
from the topic.
enable_vertex_ai_smt (bool): If set, enables Vertex AI message
transforms.
Returns:
Topic: The updated topic.
Raises:
NoFieldsSpecifiedError: if no fields were specified.
PatchConflictingArgumentsError: if conflicting arguments were provided
InvalidSchemaSettingsException: If an invalid --schema,
--message-encoding flag comnbination is specified,
or if the --first_revision_id revision is newer than
the --last_revision_id specified.
"""
update_settings = []
if labels:
update_settings.append(_TopicUpdateSetting('labels', labels))
if kms_key_name:
update_settings.append(_TopicUpdateSetting('kmsKeyName', kms_key_name))
if message_retention_duration:
update_settings.append(
_TopicUpdateSetting(
'messageRetentionDuration', message_retention_duration
)
)
if clear_message_retention_duration:
update_settings.append(
_TopicUpdateSetting('messageRetentionDuration', None)
)
if recompute_message_storage_policy:
update_settings.append(_TopicUpdateSetting('messageStoragePolicy', None))
elif message_storage_policy_allowed_regions:
message_storage_policy = self.messages.MessageStoragePolicy(
allowedPersistenceRegions=message_storage_policy_allowed_regions
)
if message_storage_policy_enforce_in_transit:
message_storage_policy.enforceInTransit = (
message_storage_policy_enforce_in_transit
)
update_settings.append(
_TopicUpdateSetting('messageStoragePolicy', message_storage_policy)
)
if clear_schema_settings:
update_settings.append(_TopicUpdateSetting('schemaSettings', None))
elif schema and message_encoding:
encoding_enum = ParseMessageEncoding(self.messages, message_encoding)
update_settings.append(
_TopicUpdateSetting(
'schemaSettings',
self.messages.SchemaSettings(
schema=schema,
encoding=encoding_enum,
firstRevisionId=first_revision_id,
lastRevisionId=last_revision_id,
),
)
)
if clear_ingestion_data_source_settings:
update_settings.append(
_TopicUpdateSetting('ingestionDataSourceSettings', None)
)
else:
new_settings = self._ParseIngestionDataSourceSettings(
kinesis_ingestion_stream_arn=kinesis_ingestion_stream_arn,
kinesis_ingestion_consumer_arn=kinesis_ingestion_consumer_arn,
kinesis_ingestion_role_arn=kinesis_ingestion_role_arn,
kinesis_ingestion_service_account=kinesis_ingestion_service_account,
cloud_storage_ingestion_bucket=cloud_storage_ingestion_bucket,
cloud_storage_ingestion_input_format=cloud_storage_ingestion_input_format,
cloud_storage_ingestion_text_delimiter=cloud_storage_ingestion_text_delimiter,
cloud_storage_ingestion_minimum_object_create_time=cloud_storage_ingestion_minimum_object_create_time,
cloud_storage_ingestion_match_glob=cloud_storage_ingestion_match_glob,
azure_event_hubs_ingestion_resource_group=azure_event_hubs_ingestion_resource_group,
azure_event_hubs_ingestion_namespace=azure_event_hubs_ingestion_namespace,
azure_event_hubs_ingestion_event_hub=azure_event_hubs_ingestion_event_hub,
azure_event_hubs_ingestion_client_id=azure_event_hubs_ingestion_client_id,
azure_event_hubs_ingestion_tenant_id=azure_event_hubs_ingestion_tenant_id,
azure_event_hubs_ingestion_subscription_id=azure_event_hubs_ingestion_subscription_id,
azure_event_hubs_ingestion_service_account=azure_event_hubs_ingestion_service_account,
aws_msk_ingestion_cluster_arn=aws_msk_ingestion_cluster_arn,
aws_msk_ingestion_topic=aws_msk_ingestion_topic,
aws_msk_ingestion_aws_role_arn=aws_msk_ingestion_aws_role_arn,
aws_msk_ingestion_service_account=aws_msk_ingestion_service_account,
confluent_cloud_ingestion_bootstrap_server=confluent_cloud_ingestion_bootstrap_server,
confluent_cloud_ingestion_cluster_id=confluent_cloud_ingestion_cluster_id,
confluent_cloud_ingestion_topic=confluent_cloud_ingestion_topic,
confluent_cloud_ingestion_identity_pool_id=confluent_cloud_ingestion_identity_pool_id,
confluent_cloud_ingestion_service_account=confluent_cloud_ingestion_service_account,
ingestion_log_severity=ingestion_log_severity,
)
if new_settings is not None:
update_settings.append(
_TopicUpdateSetting('ingestionDataSourceSettings', new_settings)
)
if message_transforms_file:
try:
update_settings.append(
_TopicUpdateSetting(
'messageTransforms',
utils.GetMessageTransformsFromFile(
self.messages.MessageTransform,
message_transforms_file,
enable_vertex_ai_smt=enable_vertex_ai_smt,
),
)
)
except (
utils.MessageTransformsInvalidFormatError,
utils.MessageTransformsEmptyFileError,
utils.MessageTransformsMissingFileError,
) as e:
e.args = (utils.GetErrorMessage(e),)
raise
if clear_message_transforms:
update_settings.append(
_TopicUpdateSetting(
'messageTransforms', CLEAR_MESSAGE_TRANSFORMS_VALUE
)
)
topic = self.messages.Topic(name=topic_ref.RelativeName())
update_mask = []
for update_setting in update_settings:
setattr(topic, update_setting.field_name, update_setting.value)
update_mask.append(update_setting.field_name)
if not update_mask:
raise NoFieldsSpecifiedError('Must specify at least one field to update.')
patch_req = self.messages.PubsubProjectsTopicsPatchRequest(
updateTopicRequest=self.messages.UpdateTopicRequest(
topic=topic, updateMask=','.join(update_mask)
),
name=topic_ref.RelativeName(),
)
return self._service.Patch(patch_req)