taskqueue.py

#!/usr/bin/env python
#
# Copyright 2007 Google LLC
#
# 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.
#


"""Task Queue API.

Enables an application to queue background work for itself. Work is done through
webhooks that process tasks pushed from a queue, or workers that manually pull
tasks from a queue. In push queues, tasks will execute in best-effort order of
ETA. Webhooks that fail will cause tasks to be retried at a later time. In pull
queues, workers are responsible of leasing tasks for processing and deleting the
tasks when completed. Multiple queues cab exist with independent throttling
controls.

Webhook URLs can be specified directly for push tasks, or the default URL scheme
can be used, which will translate task names into URLs relative to a queue's
base path. A default queue is also provided for simple usage.
"""

import calendar
import datetime
import logging
import math
import os
import re
import time

from google.appengine.api import apiproxy_stub_map
from google.appengine.api import app_identity
from google.appengine.api import modules
from google.appengine.api import namespace_manager
from google.appengine.api import urlfetch
from google.appengine.api.taskqueue import taskqueue_service_bytes_pb2 as taskqueue_service_pb2
from google.appengine.runtime import apiproxy_errors
from google.appengine.api.taskqueue import cloudtask
from google.appengine.api.taskqueue import cloudtask_transactional
from google.appengine.runtime import context
import six
from six.moves import urllib
import six.moves.urllib.parse














__all__ = [

    'BadTaskStateError',
    'BadTransactionState',
    'BadTransactionStateError',
    'DatastoreError',
    'DuplicateTaskNameError',
    'Error',
    'InternalError',
    'InvalidQueueError',
    'InvalidQueueNameError',
    'InvalidTaskError',
    'InvalidTaskNameError',
    'InvalidUrlError',
    'PermissionDeniedError',
    'TaskAlreadyExistsError',
    'TaskTooLargeError',
    'TombstonedTaskError',
    'TooManyTasksError',
    'TransientError',
    'UnknownQueueError',
    'InvalidTaskRetryOptionsError',
    'InvalidLeaseTimeError',
    'InvalidMaxTasksError',
    'InvalidDeadlineError',
    'InvalidQueueModeError',
    'TransactionalRequestTooLargeError',
    'TaskLeaseExpiredError',
    'QueuePausedError',
    'InvalidEtaError',
    'InvalidTagError',
    'InvalidDispatchDeadlineError',

    'MAX_QUEUE_NAME_LENGTH',
    'MAX_TASK_NAME_LENGTH',
    'MAX_TASK_SIZE_BYTES',
    'MAX_PULL_TASK_SIZE_BYTES',
    'MAX_PUSH_TASK_SIZE_BYTES',
    'MAX_LEASE_SECONDS',
    'MAX_TASKS_PER_ADD',
    'MAX_TASKS_PER_LEASE',
    'MAX_URL_LENGTH',
    'MAX_DISPATCH_DEADLINE',
    'MAX_TAG_LENGTH',
    'MAX_TRANSACTIONAL_REQUEST_SIZE_BYTES',
    'MIN_DISPATCH_DEADLINE',

    'DEFAULT_APP_VERSION',

    'Queue',
    'QueueStatistics',
    'Task',
    'TaskRetryOptions',
    'add',
    'create_rpc'
]


class Error(Exception):
  """Base class for exceptions in this module."""


class UnknownQueueError(Error):
  """The queue specified is unknown."""


class TransientError(Error):
  """A transient error occurred while accessing the queue. Try again later."""


class InternalError(Error):
  """An internal error occurred while accessing this queue.

  If the problem continues, contact the App Engine team through the support
  forum. Be sure to include a description of your problem.
  """


class InvalidTaskError(Error):
  """The parameters, headers, or method of the task is invalid."""


class InvalidTaskNameError(InvalidTaskError):
  """The name of the task is invalid."""


class TaskTooLargeError(InvalidTaskError):
  """The task is too large with its headers and payload."""


class TaskAlreadyExistsError(InvalidTaskError):
  """The task already exists. It has not yet run."""


class TombstonedTaskError(InvalidTaskError):
  """The task has been tombstoned.

  A task with the same name was previously executed in the queue; names should
  be unique within a queue.
  """


class InvalidUrlError(InvalidTaskError):
  """The relative URL used for the task is invalid."""


class InvalidEtaError(InvalidTaskError):
  """The task's ETA is invalid."""


class BadTaskStateError(Error):
  """The task is in the wrong state for the requested operation."""


class InvalidQueueError(Error):
  """The queue's configuration is invalid."""


class InvalidQueueNameError(InvalidQueueError):
  """The name of the queue is invalid."""


class _RelativeUrlError(Error):
  """The relative URL supplied is invalid."""


class PermissionDeniedError(Error):
  """The requested operation is not allowed for this app."""


class DuplicateTaskNameError(Error):
  """Two tasks have the same name.

  When adding multiple tasks to a queue in a batch, more than one task cannot
  have the same name.
  """


class TooManyTasksError(Error):
  """Too many tasks were present in a single function call."""


class DatastoreError(Error):
  """There was a datastore error while accessing the queue."""


class BadTransactionStateError(Error):
  """The state of the current transaction does not permit this operation."""


class InvalidTaskRetryOptionsError(Error):
  """The task retry configuration is invalid."""


class InvalidLeaseTimeError(Error):
  """The lease time period is invalid."""


class InvalidMaxTasksError(Error):
  """The requested maximum number of tasks in `lease_tasks` is invalid."""


class InvalidDeadlineError(Error):
  """The requested deadline in `lease_tasks` is invalid."""


class InvalidQueueModeError(Error):
  """Invokes a pull queue operation on a push queue or vice versa."""


class TransactionalRequestTooLargeError(TaskTooLargeError):
  """The total size of this transaction (including tasks) was too large."""


class TaskLeaseExpiredError(Error):
  """The task lease could not be renewed because it had already expired."""


class QueuePausedError(Error):
  """The queue is paused and cannot process modify task lease requests."""


class InvalidTagError(Error):
  """The specified tag is invalid."""


class InvalidDispatchDeadlineError(Error):
  """The requested dispatch deadline for a `push task` is invalid."""


class _DefaultAppVersionSingleton(object):
  def __repr__(self):
    return '<DefaultApplicationVersion>'


class _UnknownAppVersionSingleton(object):
  def __repr__(self):
    return '<UnknownApplicationVersion>'




BadTransactionState = BadTransactionStateError

MAX_QUEUE_NAME_LENGTH = 100

MAX_PULL_TASK_SIZE_BYTES = 2 ** 20

MAX_PUSH_TASK_SIZE_BYTES = 100 * (2 ** 10)

MAX_TASK_NAME_LENGTH = 500

MAX_TASK_SIZE_BYTES = MAX_PUSH_TASK_SIZE_BYTES

MAX_TASKS_PER_ADD = 100

MAX_TRANSACTIONAL_REQUEST_SIZE_BYTES = 2 ** 20


MAX_URL_LENGTH = 2083

MAX_TASKS_PER_LEASE = 1000

MAX_TAG_LENGTH = 500

MAX_LEASE_SECONDS = 3600 * 24 * 7

MAX_DISPATCH_DEADLINE = datetime.timedelta(hours=24)

MIN_DISPATCH_DEADLINE = datetime.timedelta(seconds=15)


DEFAULT_APP_VERSION = _DefaultAppVersionSingleton()
_UNKNOWN_APP_VERSION = _UnknownAppVersionSingleton()

_DEFAULT_QUEUE = 'default'

_DEFAULT_QUEUE_PATH = '/_ah/queue'

_MAX_COUNTDOWN_SECONDS = 3600 * 24 * 30

_METHOD_MAP = {
    'GET': taskqueue_service_pb2.TaskQueueAddRequest.GET,
    'POST': taskqueue_service_pb2.TaskQueueAddRequest.POST,
    'HEAD': taskqueue_service_pb2.TaskQueueAddRequest.HEAD,
    'PUT': taskqueue_service_pb2.TaskQueueAddRequest.PUT,
    'DELETE': taskqueue_service_pb2.TaskQueueAddRequest.DELETE,
}

_NON_POST_HTTP_METHODS = frozenset(['GET', 'HEAD', 'PUT', 'DELETE'])

_BODY_METHODS = frozenset(['POST', 'PUT', 'PULL'])

_TASK_NAME_PATTERN = r'^[a-zA-Z0-9_-]{1,%s}$' % MAX_TASK_NAME_LENGTH

_TASK_NAME_RE = re.compile(_TASK_NAME_PATTERN)

_QUEUE_NAME_PATTERN = r'^[a-zA-Z0-9-]{1,%s}$' % MAX_QUEUE_NAME_LENGTH

_QUEUE_NAME_RE = re.compile(_QUEUE_NAME_PATTERN)

_ERROR_MAPPING = {
    taskqueue_service_pb2.TaskQueueServiceError.UNKNOWN_QUEUE:
        UnknownQueueError,
    taskqueue_service_pb2.TaskQueueServiceError.TRANSIENT_ERROR:
        TransientError,
    taskqueue_service_pb2.TaskQueueServiceError.INTERNAL_ERROR:
        InternalError,
    taskqueue_service_pb2.TaskQueueServiceError.TASK_TOO_LARGE:
        TaskTooLargeError,
    taskqueue_service_pb2.TaskQueueServiceError.INVALID_TASK_NAME:
        InvalidTaskNameError,
    taskqueue_service_pb2.TaskQueueServiceError.INVALID_QUEUE_NAME:
        InvalidQueueNameError,
    taskqueue_service_pb2.TaskQueueServiceError.INVALID_URL:
        InvalidUrlError,
    taskqueue_service_pb2.TaskQueueServiceError.INVALID_QUEUE_RATE:
        InvalidQueueError,
    taskqueue_service_pb2.TaskQueueServiceError.PERMISSION_DENIED:
        PermissionDeniedError,
    taskqueue_service_pb2.TaskQueueServiceError.TASK_ALREADY_EXISTS:
        TaskAlreadyExistsError,
    taskqueue_service_pb2.TaskQueueServiceError.TOMBSTONED_TASK:
        TombstonedTaskError,
    taskqueue_service_pb2.TaskQueueServiceError.INVALID_ETA:
        InvalidEtaError,
    taskqueue_service_pb2.TaskQueueServiceError.INVALID_REQUEST:
        Error,
    taskqueue_service_pb2.TaskQueueServiceError.UNKNOWN_TASK:
        Error,
    taskqueue_service_pb2.TaskQueueServiceError.TOMBSTONED_QUEUE:
        Error,
    taskqueue_service_pb2.TaskQueueServiceError.DUPLICATE_TASK_NAME:
        DuplicateTaskNameError,
    taskqueue_service_pb2.TaskQueueServiceError.INVALID_QUEUE_MODE:
        InvalidQueueModeError,

    taskqueue_service_pb2.TaskQueueServiceError.TOO_MANY_TASKS:
        TooManyTasksError,
    taskqueue_service_pb2.TaskQueueServiceError.TRANSACTIONAL_REQUEST_TOO_LARGE:
        TransactionalRequestTooLargeError,
    taskqueue_service_pb2.TaskQueueServiceError.TASK_LEASE_EXPIRED:
        TaskLeaseExpiredError,
    taskqueue_service_pb2.TaskQueueServiceError.QUEUE_PAUSED:
        QueuePausedError,
    taskqueue_service_pb2.TaskQueueServiceError.INVALID_TAG:
        InvalidTagError,

}



class _UTCTimeZone(datetime.tzinfo):
  """UTC time zone."""

  ZERO = datetime.timedelta(0)

  def utcoffset(self, dt):
    return self.ZERO

  def dst(self, dt):
    return self.ZERO

  def tzname(self, dt):
    return 'UTC'

  def __repr__(self):
    return '_UTCTimeZone()'


_UTC = _UTCTimeZone()


def _parse_relative_url(relative_url):
  """Parses a relative URL and splits it into its path and query string.

  Args:
      relative_url: The relative URL, starting with a '/'.

  Returns:
      Tuple (path, query) where:
        path: The path in the relative URL.
        query: The query string in the URL without the '?'' character.

  Raises:
     _RelativeUrlError: If the `relative_url` is invalid for any reason.
  """
  if not relative_url:
    raise _RelativeUrlError('The relative URL is empty')
  (scheme, netloc, path, query,
   fragment) = six.moves.urllib.parse.urlsplit(relative_url)
  if scheme or netloc:
    raise _RelativeUrlError('Relative URL cannot have a scheme or location')
  if fragment:
    raise _RelativeUrlError('Relative URL cannot specify a fragment')
  if not path or path[0] != '/':
    raise _RelativeUrlError('The relative URL path must start with "/"')
  return path, query


def _flatten_params(params):
  """Converts a dictionary of parameters to a list of parameters.

  Any unicode strings in keys or values will be encoded as UTF-8.

  Args:
      params: Dictionary mapping parameter keys to values. Values will be
          converted to a string and added to the list as tuple (key, value). If
          a values is iterable and not a string, each contained value will be
          added as a separate (key, value) tuple.

  Returns:
      List of (key, value) tuples.
  """

  def get_string(value):
    if isinstance(value, six.text_type):
      return value.encode('utf8')
    elif isinstance(value, six.binary_type):
      return value
    else:




      return six.ensure_binary(str(value))

  param_list = []
  for key, value in six.iteritems(params):
    key = get_string(key)
    if isinstance(value, (six.text_type, six.binary_type)):
      param_list.append((key, get_string(value)))
    else:
      try:
        iterator = iter(value)
      except TypeError:
        param_list.append((key, get_string(value)))
      else:
        param_list.extend((key, get_string(v)) for v in iterator)

  return param_list


def _MakeAsyncCall(method, request, response, get_result_hook=None, rpc=None):
  """Internal helper to schedule an asynchronous RPC.

  Args:
      method: The name of the taskqueue_service method that should be called,
          for example: `BulkAdd`.
      request: The protocol buffer that contains the request argument.
      response: The protocol buffer to be populated with the response.
      get_result_hook: An optional hook function used to process results. See
          `UserRPC.make_call()` for more information.
      rpc: An optional UserRPC object that will be used to make the call.

  Returns:
      A UserRPC object; either the object that was passed in as the RPC
      argument, or a new object if no RPC was passed in.
  """
  if rpc is None:
    rpc = create_rpc()
  assert rpc.service == 'taskqueue', repr(rpc.service)
  rpc.make_call(method, request, response, get_result_hook, None)
  return rpc


def _TranslateError(error, detail=''):
  """Translates a `TaskQueueServiceError` into an exception.

  Args:
      error: Value from TaskQueueServiceError enum.
      detail: A human-readable description of the error.

  Returns:
      The corresponding Exception sub-class for that error code.
  """
  if (isinstance(error, int) and
      error >= taskqueue_service_pb2.TaskQueueServiceError.DATASTORE_ERROR):
    from google.appengine.api import datastore
    datastore_exception = datastore._DatastoreExceptionFromErrorCodeAndDetail(
        error - taskqueue_service_pb2.TaskQueueServiceError.DATASTORE_ERROR,
        detail)

    class JointException(datastore_exception.__class__, DatastoreError):
      """There was a datastore error while accessing the queue."""
      __msg = (u'taskqueue.DatastoreError caused by: %s %s' %
               (datastore_exception.__class__, detail))

      def __str__(self):
        return JointException.__msg

    return JointException()
  else:
    exception_class = _ERROR_MAPPING.get(error, None)
    if exception_class:
      return exception_class(detail)
    else:
      return Error('Application error %s: %s' % (error, detail))


def _ValidateDeadline(deadline):
  if not isinstance(deadline, (int, int, float)):
    raise TypeError(
        'deadline must be numeric')

  if deadline <= 0:
    raise InvalidDeadlineError(
        'Negative or zero deadline requested')


def _ValidateDispatchDeadline(dispatch_deadline_usec):
  """Validates the dispatch_deadline_usec's type and value."""
  if not isinstance(dispatch_deadline_usec, int):
    raise TypeError('deadline must be an integer')

  deadline_in_time_delta = datetime.timedelta(seconds=dispatch_deadline_usec /
                                              1000000)
  if deadline_in_time_delta > MAX_DISPATCH_DEADLINE:
    raise InvalidDispatchDeadlineError(
        'Deadline cannot be more than {duration}'.format(
            duration=MAX_DISPATCH_DEADLINE))

  if deadline_in_time_delta < MIN_DISPATCH_DEADLINE:
    raise InvalidDispatchDeadlineError(
        'Deadline cannot be less than {duration}'.format(
            duration=MIN_DISPATCH_DEADLINE))


def create_rpc(deadline=None, callback=None):
  """Creates an RPC object for use with the Task Queue API.

  Args:
    deadline: Optional deadline in seconds for the operation; the default
        value is a system-specific deadline, which is typically 5 seconds. After
        the deadline, a `DeadlineExceededError` error will be returned.
    callback: Optional function to be called with the Task Queue service
        returns results successfully when `get_result()`, `check_success()`,
        or `wait()` is invoked on the RPC object. The function is
        called without arguments. The function is not called in a background
        process or thread; the function is only called when one of the above
        methods is called by the application. The function is called even if
        the request fails or the RPC deadline elapses.

  Returns:
    An `apiproxy_stub_map.UserRPC` object specialized for this service.
  """
  if deadline is not None:
    _ValidateDeadline(deadline)
  return apiproxy_stub_map.UserRPC('taskqueue', deadline, callback)


class TaskRetryOptions(object):
  """The options used to decide when a failed task will be retried.

  Tasks executing in the task queue can fail for many reasons. If a task fails
  to execute, which is indicated by returning any HTTP status code outside of
  the range 200-299, App Engine retries the task until it succeeds. By default,
  the system gradually reduces the retry rate to avoid flooding your application
  with too many requests, but schedules retry attempts to recur at a maximum of
  once per hour until the task succeeds. 503 errors, however, are treated as
  special cases and should not be returned by user code.

  The `TaskRetryOptions` class provides the properties that you can use to
  decide when to retry a failed task at runtime.
  """

  __CONSTRUCTOR_KWARGS = frozenset(
      ['min_backoff_seconds', 'max_backoff_seconds',
       'task_age_limit', 'max_doublings', 'task_retry_limit'])

  def __init__(self, **kwargs):
    """Initializer.

    Args:
      min_backoff_seconds: Optional; the minimum number of seconds to wait
          before retrying a task after it fails.
      max_backoff_seconds: Optional; the maximum number of seconds to wait
          before retrying a task after it fails.
      task_age_limit: Optional; the number of seconds after creation that a
          failed task will no longer be retried. The given value is rounded up
          to the nearest integer. If `task_retry_limit` is also specified, the
          task will be retried until both limits are reached.
      max_doublings: Optional; the maximum number of times that the interval
          between failed task retries will be doubled before the increase
          becomes constant. The constant is:
          `2**(max_doublings - 1) * min_backoff_seconds`.
      task_retry_limit: Optional; the maximum number of times to retry a
          failed task before giving up. In push queues, the counter is
          incremented each time App Engine tries the tasks, up to the
          specified limit. If `task_age_limit` is also specified, the task
          will be retried until both limits are reached.

    Raises:
        InvalidTaskRetryOptionsError: If any of the parameters are invalid.
    """
    args_diff = set(six.iterkeys(kwargs)) - self.__CONSTRUCTOR_KWARGS
    if args_diff:
      raise TypeError('Invalid arguments: %s' % ', '.join(args_diff))

    self.__min_backoff_seconds = kwargs.get('min_backoff_seconds')
    if (self.__min_backoff_seconds is not None and
        self.__min_backoff_seconds < 0):
      raise InvalidTaskRetryOptionsError(
          'The minimum retry interval cannot be negative')

    self.__max_backoff_seconds = kwargs.get('max_backoff_seconds')
    if (self.__max_backoff_seconds is not None and
        self.__max_backoff_seconds < 0):
      raise InvalidTaskRetryOptionsError(
          'The maximum retry interval cannot be negative')

    if (self.__min_backoff_seconds is not None and
        self.__max_backoff_seconds is not None and
        self.__max_backoff_seconds < self.__min_backoff_seconds):
      raise InvalidTaskRetryOptionsError(
          'The maximum retry interval cannot be less than the '
          'minimum retry interval')

    self.__max_doublings = kwargs.get('max_doublings')
    if self.__max_doublings is not None and self.__max_doublings < 0:
      raise InvalidTaskRetryOptionsError(
          'The maximum number of retry interval doublings cannot be negative')

    self.__task_retry_limit = kwargs.get('task_retry_limit')
    if self.__task_retry_limit is not None and self.__task_retry_limit < 0:
      raise InvalidTaskRetryOptionsError(
          'The maximum number of retries cannot be negative')

    self.__task_age_limit = kwargs.get('task_age_limit')
    if self.__task_age_limit is not None:
      if self.__task_age_limit < 0:
        raise InvalidTaskRetryOptionsError(
            'The expiry countdown cannot be negative')
      self.__task_age_limit = int(math.ceil(self.__task_age_limit))

  @property
  def min_backoff_seconds(self):
    """The minimum number of seconds to wait before retrying a task."""
    return self.__min_backoff_seconds

  @property
  def max_backoff_seconds(self):
    """The maximum number of seconds to wait before retrying a task."""
    return self.__max_backoff_seconds

  @property
  def task_age_limit(self):
    """The number of seconds after which a failed task will not be retried."""
    return self.__task_age_limit

  @property
  def max_doublings(self):
    """The number of times that the retry interval will be doubled."""
    return self.__max_doublings

  @property
  def task_retry_limit(self):
    """The number of times that a failed task will be retried."""
    return self.__task_retry_limit

  def __repr__(self):
    properties = ['%s=%r' % (attr, getattr(self, attr)) for attr in
                  self.__CONSTRUCTOR_KWARGS]
    return 'TaskRetryOptions(%s)' % ', '.join(properties)


class Task(object):
  """Represents a single task on a queue.

  The `Task` class enables an application to queue background work. Work is done
  through webhooks that process tasks pushed from a push queue, or workers that
  manually pull tasks from a pull queue.

  In push queues, most tasks are delivered in best-effort order of ETA.

  Note:
      Occasionally, tasks might be delivered out of order of ETA. However, for
      corner cases, tasks are delivered out of order for an extended period of
      time. You should not rely on tasks being delivered in order, as the
      results aren't always consistent.

  Webhooks that fail cause tasks to be retried at a later time. You can
  configure the rate and number of retries for failed tasks. You can specify
  webhook URLs directly for push tasks. You can also use the default URL scheme,
  which translates task names into URLs that are relative to a queue's base
  path. A default queue is also provided for simple usage.

  In pull queues, workers are responsible for leasing tasks, processing them,
  and deleting them after processing. You can configure the number of task
  retries, which is based on how many times the task has been leased. You can
  define multiple queues with independent throttling controls.

  You set the various properties for a task in the constructor. Once the `Task`
  object is instantiated, you insert that task into a queue. A task instance can
  be inserted into one queue only.
  """


  __CONSTRUCTOR_KWARGS = frozenset([
      'countdown', 'eta', 'headers', 'method', 'name', 'params',
      'retry_options', 'tag', 'target', 'url', '_size_check',
      'dispatch_deadline_usec'
  ])


  __eta_posix = None
  __target = None


  def __init__(self, payload=None, **kwargs):
    """Initializer.

    Args:
      payload: Optional; the payload data for this task. This argument is only
          allowed for `POST` and `PUT` methods and pull tasks. In push queues,
          the payload is delivered to the webhook or backend in the body of an
          HTTP request. In pull queues, the payload is fetched by workers as
          part of the response from `lease_tasks()`.
      name: Optional; the name to give the task. If you do not specify a name,
          a name is auto-generated when added to a queue and assigned to this
          object. The name must match the `_TASK_NAME_PATTERN` regular
          expression. Avoid sequential names, such as counters or timestamps, as
          these names can lead to decreased availability and performance.
      method: Optional; the HTTP method to use when accessing the webhook. The
          default value is `POST`. This argument is not used for pull queues.
      url: Optional; the relative URL where the webhook that should handle
          this task is located for this application. You can include a query
          string in this value unless it is being used in a `POST` method. You
          cannot specify a URL for pull tasks.
      headers: Optional; a dictionary of headers to pass to the webhook. The
          values in the dictionary can be iterable to indicate repeated header
          fields. You cannot specify headers for pull tasks. In a push task,
          if you do not specify a `Content-Type` header, the default value of
          `text/plain` will be used. In a push task, if you specify a `Host`
          header, you cannot use the `target` keyword argument. Any headers that
          use the `X-AppEngine` prefix will also be dropped.
      params: Optional; a dictionary of parameters to use for the task. For
          `POST` requests and PULL tasks, these parameters are encoded as
          `application/x-www-form-urlencoded` and set to the payload. For both
          `POST` and pull requests, you cannot specify parameters if you
          already specified a `payload`. In `PUT` requests, parameters are
          converted to a query string if the URL contains a query string, or if
          the task already has a `payload`. Do not specify parameters if the URL
          contains a query string and the method is `GET`.
      countdown: Optional; time in seconds into the future that this task
          should run or be leased. The default value is zero. Do not specify a
          countdown if you also specified an `eta`, as it sets the ETA to a
          value of now + `countdown`.
      eta: Optional; a `datetime.datetime` specifying the absolute time at
          which the task should be run or leased. The `eta` argument must not
          be specified if `countdown` is specified. This value can be time
          zone-aware or time zone-naive. If the value is set to None, the
          default value is now. For pull tasks, no worker will be able to
          lease this task before the time indicated by the `eta` argument.
      retry_options: Optional; a `TaskRetryOptions` object used to control
          how the task will be retried if it fails. For pull tasks, only the
          `task_retry_limit` option is allowed. For push tasks, you can use
          the `min_backoff_seconds`, `max_backoff_seconds`, `task_age_limit`,
          `max_doublings`, and `task_retry_limit` options.
      target: Optional; a string that names a module or version, a frontend
          version, or a backend on which to run the task. The string is
          prepended to the domain name of your app. If you set the `target`,
          do not specify a `Host` header in the dictionary for the `headers`
          argument. For pull tasks, do not specify a target.
      tag: Optional; the tag to be used when grouping by tag (pull tasks only).
      dispatch_deadline_usec: A duration of time that serves as a limit on the
          dispatch duration; if the task's end point does not respond to the
          request within this deadline, the task is canceled (and retried
          depending on the config).

    Raises:
      InvalidDispatchDeadlineError: If the `dispatch_deadline_usec` is not
      within valid limits.
      InvalidEtaError: If the `eta` is too far into the future.
      InvalidTagError: If the tag is too long.
      InvalidTaskError: If any of the parameters are invalid.
      InvalidTaskNameError: If the task name is invalid.
      InvalidUrlError: If the task URL is invalid or too long.
      TaskTooLargeError: If the task with its associated payload is too large.
    """
    args_diff = set(six.iterkeys(kwargs)) - self.__CONSTRUCTOR_KWARGS
    if args_diff:
      raise TypeError('Invalid arguments: %s' % ', '.join(args_diff))

    self.__name = kwargs.get('name')
    if self.__name and not _TASK_NAME_RE.match(self.__name):
      raise InvalidTaskNameError(
          'The task name does not match expression "%s"; found %s' %
          (_TASK_NAME_PATTERN, self.__name))

    self.__default_url, self.__relative_url, query = Task.__determine_url(
        kwargs.get('url', ''))
    self.__headers = urlfetch._CaselessDict()
    self.__headers.update(kwargs.get('headers', {}))
    self.__method = kwargs.get('method', 'POST').upper()
    self.__tag = kwargs.get('tag')
    self.__payload = None
    self.__retry_count = 0
    self.__queue_name = None
    self.__dispatch_deadline_usec = kwargs.get('dispatch_deadline_usec')


    size_check = kwargs.get('_size_check', True)
    params = kwargs.get('params', {})


    apps_namespace = namespace_manager.google_apps_namespace()
    if apps_namespace is not None:
      self.__headers.setdefault('X-AppEngine-Default-Namespace', apps_namespace)

    self.__headers.setdefault('X-AppEngine-Current-Namespace',
                              namespace_manager.get_namespace())
    if query and params:
      raise InvalidTaskError('Query string and parameters both present; '
                             'only one of these can be supplied')

    if self.__method != 'PULL' and self.__tag is not None:
      raise InvalidTaskError('tag can only be specified for pull tasks')

    if self.__method == 'PULL':
      if not self.__default_url:
        raise InvalidTaskError('url must not be specified for pull tasks')
      if kwargs.get('headers'):
        raise InvalidTaskError('headers must not be specified for pull tasks')
      if kwargs.get('target'):
        raise InvalidTaskError('target must not be specified for pull tasks')
      if params:
        if payload:
          raise InvalidTaskError(
              'Message body and parameters both present for '
              'PULL method; only one of these can be supplied')
        payload = Task.__encode_params(params)
      if payload is None:
        raise InvalidTaskError('payload must be specified for pull task')
      self.__payload = Task.__convert_payload(payload, self.__headers)
    elif self.__method == 'POST':
      if payload and params:
        raise InvalidTaskError('Message body and parameters both present for '
                               'POST method; only one of these can be '
                               'supplied')
      elif query:
        raise InvalidTaskError('POST method cannot have a query string; '
                               'use the "params" keyword argument instead')
      elif params:
        self.__payload = Task.__encode_params(params)
        self.__headers.setdefault(
            'content-type', 'application/x-www-form-urlencoded')
      elif payload is not None:
        self.__payload = Task.__convert_payload(payload, self.__headers)
    elif self.__method in _NON_POST_HTTP_METHODS:
      if payload and self.__method not in _BODY_METHODS:
        raise InvalidTaskError(
            'Payload can only be specified for methods %s' %
            ', '.join(_BODY_METHODS))
      if payload:
        self.__payload = Task.__convert_payload(payload, self.__headers)
      if params:
        query = Task.__encode_params(params)
      if query:
        self.__relative_url = '%s?%s' % (self.__relative_url, query)
    else:
      raise InvalidTaskError('Invalid method: %s' % self.__method)

    self.__target = kwargs.get('target')
    self.__resolve_hostname_and_target()

    self.__headers_list = _flatten_params(self.__headers)
    self.__eta_posix = Task.__determine_eta_posix(
        kwargs.get('eta'), kwargs.get('countdown'))
    self.__eta = None
    self.__retry_options = kwargs.get('retry_options')
    self.__enqueued = False
    self.__deleted = False

    if self.__eta_posix - time.time() > _MAX_COUNTDOWN_SECONDS:
      raise InvalidEtaError('ETA too far in the future')

    if size_check:
      if self.__method == 'PULL':
        max_task_size_bytes = MAX_PULL_TASK_SIZE_BYTES
      else:
        max_task_size_bytes = MAX_PUSH_TASK_SIZE_BYTES
      if self.size > max_task_size_bytes:
        raise TaskTooLargeError('Task size must be less than %d; found %d' %
                                (max_task_size_bytes, self.size))
      if self.__tag and len(self.__tag) > MAX_TAG_LENGTH:
        raise InvalidTagError(
            'Tag must be <= %d bytes. Got a %d byte tag.' % (
                MAX_TAG_LENGTH, len(self.__tag)))

    if self.__dispatch_deadline_usec is not None:
      _ValidateDispatchDeadline(self.__dispatch_deadline_usec)

  def __resolve_hostname_and_target(self):
    """Resolve the values of the target parameter and the `Host' header.

    Requires that the attributes __target and __headers exist before this method
    is called.

    This function should only be called once from the __init__ function of the
    Task class.

    Raises:
      InvalidTaskError: If the task is invalid.
    """





    if context.get('HTTP_HOST', None) is None:
      logging.warning(
          'The HTTP_HOST environment variable was not set, but is required '
          'to determine the correct value for the `Task.target\' property. '
          'Please update your unit tests to specify a correct value for this '
          'environment variable.')

    if self.__target is not None and 'Host' in self.__headers:
      raise InvalidTaskError(
          'A host header cannot be set when a target is specified.')
    elif self.__target is not None:
      host = self.__host_from_target(self.__target)
      if host:


        self.__headers['Host'] = host
    elif 'Host' in self.__headers:
      self.__target = self.__target_from_host(self.__headers['Host'])
    else:
      if context.get('HTTP_HOST', None):
        self.__headers['Host'] = context.get('HTTP_HOST')
        self.__target = self.__target_from_host(self.__headers['Host'])
      else:


        self.__target = _UNKNOWN_APP_VERSION

  @staticmethod
  def __target_from_host(host):
    """Calculate the value of the target parameter from a host header.

    Args:
      host: A string representing the hostname for this task.

    Returns:
      A string containing the target of this task, or the constant
      `DEFAULT_APP_VERSION` if it is the default version.

      If this code is running in a unit-test where the environment variable
      `DEFAULT_VERSION_HOSTNAME` is not set then the constant
      `_UNKNOWN_APP_VERSION` is returned.
    """
    default_hostname = app_identity.get_default_version_hostname()
    if default_hostname is None:



      return _UNKNOWN_APP_VERSION

    if host.endswith(default_hostname):

      version_name = host[:-(len(default_hostname) + 1)]
      if version_name:
        return version_name





    return DEFAULT_APP_VERSION

  @staticmethod