def _ReloadQueuesFromYaml()

in src/google/appengine/api/taskqueue/taskqueue_stub.py [0:0]
65 lines of code
18 McCabe index (conditional complexity)

  def _ReloadQueuesFromYaml(self):
    """Update the queue map with the contents of the queue.yaml file.

    This function will remove queues that no longer exist in the queue.yaml
    file.

    If no queue yaml parser has been defined, this function is a no-op.
    """
    if not self._queue_yaml_parser:
      return

    queue_info = self._queue_yaml_parser()

    if queue_info and queue_info.queue:
      queues = queue_info.queue
    else:
      queues = []

    old_queues = set(self._queues)
    new_queues = set()

    for entry in queues:
      queue_name = entry.name
      new_queues.add(queue_name)

      retry_parameters = None


      if entry.bucket_size:
        bucket_size = entry.bucket_size
      else:
        bucket_size = DEFAULT_BUCKET_SIZE
      if entry.retry_parameters:
        retry_parameters = queueinfo.TranslateRetryParameters(
            entry.retry_parameters)

      if entry.mode == 'pull':
        mode = QUEUE_MODE.PULL
        if entry.rate is not None:
          logging.warning(
              'Refill rate must not be specified for pull-based queue. '
              'Please check queue.yaml file.')
      else:
        mode = QUEUE_MODE.PUSH
        if entry.rate is None:
          logging.warning('Refill rate must be specified for push-based queue. '
                          'Please check queue.yaml file.')
      max_rate = entry.rate

      if entry.acl is not None:
        acl = taskqueue_service_pb2.TaskQueueAcl()
        for acl_entry in entry.acl:
          acl.user_email.append(six.ensure_binary(acl_entry.user_email))
      else:
        acl = None

      if self._queues.get(queue_name) is None:

        self._ConstructQueue(
            queue_name,
            bucket_capacity=bucket_size,
            user_specified_rate=max_rate,
            queue_mode=mode,
            acl=acl,
            retry_parameters=retry_parameters,
            target=entry.target)
      else:


        queue = self._queues[queue_name]
        queue.bucket_size = bucket_size
        queue.user_specified_rate = max_rate
        queue.acl = acl
        queue.queue_mode = mode
        queue.retry_parameters = retry_parameters
        if mode == QUEUE_MODE.PUSH:
          eta = queue.Oldest()
          if eta:
            self._update_newest_eta(_UsecToSec(eta))

    if DEFAULT_QUEUE_NAME not in self._queues:
      self._ConstructAutomaticQueue(DEFAULT_QUEUE_NAME)


    new_queues.add(DEFAULT_QUEUE_NAME)
    if not self._all_queues_valid:

      for queue_name in old_queues - new_queues:



        del self._queues[queue_name]