L1L`XPL0IT

✏️ Edit: pull.py

📁 //snap/google-cloud-cli/current/lib/surface/pubsub/subscriptions/pull.py

# -*- coding: utf-8 -*- #
# Copyright 2015 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.
"""Cloud Pub/Sub subscription pull command."""

from apitools.base.py import exceptions as api_ex

from googlecloudsdk.api_lib.pubsub import subscriptions
from googlecloudsdk.api_lib.util import exceptions as util_ex
from googlecloudsdk.calliope import base
from googlecloudsdk.calliope import exceptions
from googlecloudsdk.command_lib.pubsub import flags
from googlecloudsdk.command_lib.pubsub import resource_args
from googlecloudsdk.command_lib.pubsub import util

MESSAGE_FORMAT = """\
table[box](
  message.data.decode(base64).decode(utf-8),
  message.messageId,
  message.orderingKey,
  message.attributes.list(separator='\n'),
  deliveryAttempt,
  ackId.if(NOT auto_ack)
)
"""

MESSAGE_FORMAT_WITH_ACK_STATUS = """\
table[box](
  message.data.decode(base64).decode(utf-8),
  message.messageId,
  message.orderingKey,
  message.attributes.list(separator='\n'),
  deliveryAttempt,
  ackId.if(NOT auto_ack),
  ackStatus.if(auto_ack)
)
"""

def _Run(args, max_messages, return_immediately=False):
  """Pulls messages from a subscription."""
  client = subscriptions.SubscriptionsClient()

subscription_ref = args.CONCEPTS.subscription.Parse()
  pull_response = client.Pull(subscription_ref, max_messages,
                              return_immediately)

failed_ack_ids = {}
  ack_ids_and_failure_reasons = []
  if args.auto_ack and pull_response.receivedMessages:
    ack_ids = [message.ackId for message in pull_response.receivedMessages]
    try:
      client.Ack(ack_ids, subscription_ref)
    except api_ex.HttpError as error:
      exc = util_ex.HttpException(error)
      ack_ids_and_failure_reasons = util.ParseExactlyOnceErrorInfo(
          exc.payload.details)
      # If the failure doesn't have more information (specifically for exactly
      # once related failures), assume all the ack ids have failed with the
      # same status.
      if not ack_ids_and_failure_reasons:
        for ack_id in ack_ids:
          failed_ack_ids[ack_id] = 'FAILURE_' + str(error.status_code)

if not failed_ack_ids:
      for ack_ids_and_failure_reason in ack_ids_and_failure_reasons:
        failed_ack_ids[ack_ids_and_failure_reason[
            'AckId']] = ack_ids_and_failure_reason['FailureReason']

if not args.auto_ack:
    return pull_response.receivedMessages

# We attempted to auto-ack this message. Augment the response with ackStatus.
  return_val = []
  for message in pull_response.receivedMessages:
    # Copy the message into a separate object so we can mutate it.
    message_copy = {}
    for field in message.all_fields():
      value = getattr(message, field.name)
      if value:
        message_copy[field.name] = value
    if message.ackId in failed_ack_ids:
      message_copy['ackStatus'] = failed_ack_ids[message.ackId]
    else:
      message_copy['ackStatus'] = 'SUCCESS'
    return_val.append(message_copy)
  return return_val

@base.ReleaseTracks(base.ReleaseTrack.GA)
class Pull(base.ListCommand):
  """Pulls one or more Cloud Pub/Sub messages from a subscription."""

detailed_help = {
      'DESCRIPTION':
          """\
          Returns one or more messages from the specified Cloud Pub/Sub
          subscription, if there are any messages enqueued.

By default, this command returns only one message from the
          subscription. Use the `--limit` flag to specify the max messages to
          return.

Please note that this command is not guaranteed to return all the
          messages in your backlog or the maximum specified in the --limit
          argument.  Receiving fewer messages than available occasionally
          is normal."""
  }

@staticmethod
  def Args(parser):
    parser.display_info.AddFormat(MESSAGE_FORMAT_WITH_ACK_STATUS)
    resource_args.AddSubscriptionResourceArg(parser, 'to pull messages from.')
    flags.AddPullFlags(parser)

base.LIMIT_FLAG.SetDefault(parser, 1)

def Run(self, args):
    return _Run(args, args.limit)

@base.ReleaseTracks(base.ReleaseTrack.BETA, base.ReleaseTrack.ALPHA)
class PullBeta(Pull):
  """Pulls one or more Cloud Pub/Sub messages from a subscription."""

def Run(self, args):
    if args.IsSpecified('limit'):
      if args.IsSpecified('max_messages'):
        raise exceptions.ConflictingArgumentsException('--max-messages',
                                                       '--limit')
      max_messages = args.limit
    else:
      max_messages = args.max_messages

return_immediately = False
    if args.IsSpecified('return_immediately'):
      return_immediately = args.return_immediately
    elif args.IsSpecified('wait'):
      return_immediately = not args.wait

return _Run(args, max_messages, return_immediately)

Cancel

💻 Terminal

📤 Upload

📁 //snap/google-cloud-cli/current/lib/surface/pubsub/subscriptions

///snap/google-cloud-cli/current/lib/surface/pubsub/subscriptions

Name	Size	Perm	Modified	Actions
📁 __pycache__/	-	0775	2026-09-29 12:17	🗑 Delete
📁 config/	-	0775	2026-09-29 12:17	🗑 Delete
__init__.py	837B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
ack.py	4.1K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
ack.yaml	92B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
add_iam_policy_binding.yaml	2.5K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
create.py	10.4K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
create.yaml	92B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
delete.py	2.5K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
delete.yaml	92B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
describe.py	1.2K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
describe.yaml	91B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
get_iam_policy.py	1.6K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
get_iam_policy.yaml	913B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
list.py	1.8K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
list.yaml	91B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
modify_ack_deadline.py	4K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
modify_ack_deadline.yaml	88B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
modify_message_ack_deadline.py	4.8K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
modify_message_ack_deadline.yaml	92B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
modify_push_config.py	2.4K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
pull.py	5.5K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
pull.yaml	92B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
remove_iam_policy_binding.yaml	2.4K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
seek.py	2.1K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
seek.yaml	92B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
set_iam_policy.py	1.8K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
set_iam_policy.yaml	1.3K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
update.py	11K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
update.yaml	92B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del