text: #!/usr/bin/env python
  
  """ A service for handling TaskQueue request from application servers.
  It uses RabbitMQ and Celery to handle tasks. """
  
  import base64
  import datetime
  import hashlib
  import json
  import os
  import socket
  import sys
  import time
  
  from appscale.common import appscale_info
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from .constants import (
    InvalidTarget,
    QueueNotFound,
    TaskNotFound,
    TARGET_REGEX
  )
  from .datastore_client import (
    ApplicationError,
    BadFilterConfiguration,
    DatastoreClient,
    DatastorePermanentError,
    DatastoreTransientError,
    Entity)
  from .protocols import taskqueue_service_pb2
  from .protocols.taskqueue_service_pb2 import (
    TaskQueueServiceError
  )
  from .queue import (
    InvalidLeaseRequest,
    PostgresPullQueue,
    PushQueue
  )
  from .queue_manager import GlobalQueueManager
  from .service_manager import GlobalServiceManager
  from .task import Task
  from .tq_lib import (
    choose_task_name,
    TASK_STATES,
    verify_task_queue_add_request
  )
  from .utils import (
    get_celery_queue_name,
    get_queue_function_name,
    logger
  )
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  
  
  def setup_env():
    """ Sets required environment variables for GAE datastore library. """
    os.environ['AUTH_DOMAIN'] = "appscale.com"
    os.environ['USER_EMAIL'] = ""
    os.environ['USER_NICKNAME'] = ""
    os.environ['APPLICATION_ID'] = ""
  
  
  class DistributedTaskQueue():
    """ AppScale taskqueue layer for the TaskQueue API. """
  
    # The longest a task is allowed to run in days.
    DEFAULT_EXPIRATION = 30
  
    # The default maximum number to retry a task, where 0 or None is unlimited.
    DEFAULT_MAX_RETRIES = 0
  
    # The default amount of min/max time we wait before retrying a task
    # in seconds.
    DEFAULT_MIN_BACKOFF = 1
    DEFAULT_MAX_BACKOFF = 3600.0
  
    # Default number of times we double the backoff value.
    DEFAULT_MAX_DOUBLINGS = 1000
  
    # Kind used for storing task names.
    TASK_NAME_KIND = "__task_name__"
  
    def __init__(self, zk_client):
      """ DistributedTaskQueue Constructor.
  
      Args:
        zk_client: A KazooClient.
      """
      setup_env()
  
      self.load_balancers = list(appscale_info.get_load_balancer_ips())
      self.queue_manager = GlobalQueueManager(zk_client)
      self.service_manager = GlobalServiceManager(zk_client)
      self.datastore_client = DatastoreClient()
  
    def get_queue(self, app, queue):
      """ Fetches a Queue object.
  
      Args:
        app: A string containing the application ID.
        queue: A string specifying the name of the queue.
      Returns:
        A Queue object or None.
      """
      try:
        return self.queue_manager[app][queue]
      except KeyError:
        raise QueueNotFound(
          'The queue {} is not defined for {}'.format(queue, app))
  
    def __parse_json_and_validate_tags(self, json_request, tags):
      """ Parses JSON and validates that it contains the proper tags.
  
      Args:
        json_request: A JSON string.
        tags: The tags to validate if they are in the JSON string.
      Returns:
        A dictionary dumped from the JSON string.
      """
      try:
        json_response = json.loads(json_request)
      except ValueError:
        json_response = {"error": True,
                         "reason": "Badly formed JSON"}
        return json_response
  
      for tag in tags:
        if tag  not in json_response:
          json_response = {'error': True,
                           'reason': 'Missing ' + tag + ' tag'}
          break
      return json_response
  
    def fetch_queue_stats(self, app_id, http_data):
      """ Gets statistics about tasks in queues.
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      epoch = datetime.datetime.utcfromtimestamp(0)
      request = taskqueue_service_pb2.TaskQueueFetchQueueStatsRequest()
      request.ParseFromString(http_data)
      response = taskqueue_service_pb2.TaskQueueFetchQueueStatsResponse()
  
      for queue_name in request.queue_name:
        try:
          queue = self.get_queue(app_id, queue_name.decode('utf-8'))
        except QueueNotFound as error:
          return b'', TaskQueueServiceError.UNKNOWN_QUEUE, str(error)
  
        stats_response = response.queuestats.add()
  
        if isinstance(queue, PostgresPullQueue):
          num_tasks = queue.total_tasks()
          oldest_eta = queue.oldest_eta()
        else:
          num_tasks = self.datastore_client.query_count(app_id,
            [("state =", TASK_STATES.QUEUED),
             ("queue =", queue_name),
             ("app_id =", app_id)])
          # This is not supported for push queues yet.
          oldest_eta = None
  
        # -1 is used to indicate an absence of a value.
        oldest_eta_usec = (int((oldest_eta - epoch).total_seconds() * 1000000)
                           if oldest_eta else -1)
  
        stats_response.num_tasks = num_tasks
        stats_response.oldest_eta_usec = oldest_eta_usec
  
      return response.SerializeToString(), 0, ""
  
    def purge_queue(self, app_id, http_data):
      """
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      request = taskqueue_service_pb2.TaskQueuePurgeQueueRequest()
      request.ParseFromString(http_data)
      response = taskqueue_service_pb2.TaskQueuePurgeQueueResponse()
  
      try:
        queue = self.get_queue(app_id, request.queue_name.decode('utf-8'))
      except QueueNotFound as error:
        return b'', TaskQueueServiceError.UNKNOWN_QUEUE, str(error)
  
      queue.purge()
      return (response.SerializeToString(), 0, "")
  
    def delete(self, app_id, http_data):
      """ Delete a task.
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      request = taskqueue_service_pb2.TaskQueueDeleteRequest()
      request.ParseFromString(http_data)
      response = taskqueue_service_pb2.TaskQueueDeleteResponse()
  
      try:
        queue = self.get_queue(app_id, request.queue_name.decode('utf-8'))
      except QueueNotFound as error:
        return b'', TaskQueueServiceError.UNKNOWN_QUEUE, str(error)
  
      for task_name in request.task_name:
        queue.delete_task(Task({'id': task_name.decode('utf-8')}))
  
        response.result.append(TaskQueueServiceError.OK)
  
      return response.SerializeToString(), 0, ""
  
    def query_and_own_tasks(self, app_id, http_data):
      """ Lease pull queue tasks.
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      request = taskqueue_service_pb2.TaskQueueQueryAndOwnTasksRequest()
      request.ParseFromString(http_data)
      response = taskqueue_service_pb2.TaskQueueQueryAndOwnTasksResponse()
  
      try:
        queue = self.get_queue(app_id, request.queue_name.decode('utf-8'))
      except QueueNotFound as error:
        return b'', TaskQueueServiceError.UNKNOWN_QUEUE, str(error)
  
      tag = None
      if request.HasField("tag"):
        tag = request.tag.decode('utf-8')
  
      tasks = queue.lease_tasks(request.max_tasks, request.lease_seconds,
                                group_by_tag=request.group_by_tag, tag=tag)
  
      for task in tasks:
        task_pb = response.task.add()
        task_pb.CopyFrom(task.encode_lease_pb())
  
      return response.SerializeToString(), 0, ""
  
    def add(self, source_info, http_data):
      """ Adds a single task to the task queue.
  
      Args:
        source_info: A dictionary containing the application, module, and version
         ID that is sending this request.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      # Just call bulk add with one task.
      request = taskqueue_service_pb2.TaskQueueAddRequest()
      request.ParseFromString(http_data)
      request.app_id = source_info['app_id'].encode('utf-8')
      response = taskqueue_service_pb2.TaskQueueAddResponse()
      bulk_request = taskqueue_service_pb2.TaskQueueBulkAddRequest()
      bulk_response = taskqueue_service_pb2.TaskQueueBulkAddResponse()
      bulk_request.add_request.add().CopyFrom(request)
  
      try:
        self.__bulk_add(source_info, bulk_request, bulk_response)
      except QueueNotFound as error:
        return b'', TaskQueueServiceError.UNKNOWN_QUEUE, str(error)
      except DatastorePermanentError as error:
        return b'', TaskQueueServiceError.INTERNAL_ERROR, str(error)
      except BadFilterConfiguration as error:
        return b'', TaskQueueServiceError.INTERNAL_ERROR, str(error)
  
      if len(bulk_response.taskresult) == 1:
        result = bulk_response.taskresult[0].result
      else:
        return (response.SerializeToString(), TaskQueueServiceError.INTERNAL_ERROR,
                "Task did not receive a task response.")
  
      if result != TaskQueueServiceError.OK:
        return (response.SerializeToString(), result, "Task did not get an OK status.")
      elif bulk_response.taskresult[0].HasField("chosen_task_name"):
        response.chosen_task_name = bulk_response.taskresult[0].chosen_task_name
  
      return (response.SerializeToString(), 0, "")
  
    def bulk_add(self, source_info, http_data):
      """ Adds multiple tasks to the task queue.
  
      Args:
        source_info: A dictionary containing the application, module, and version
         ID that is sending this request.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      request = taskqueue_service_pb2.TaskQueueBulkAddRequest()
      request.ParseFromString(http_data)
      response = taskqueue_service_pb2.TaskQueueBulkAddResponse()
  
      try:
        self.__bulk_add(source_info, request, response)
      except QueueNotFound as error:
        return b'', TaskQueueServiceError.UNKNOWN_QUEUE, str(error)
      except DatastorePermanentError as error:
        return b'', TaskQueueServiceError.INTERNAL_ERROR, str(error)
      except BadFilterConfiguration as error:
        return b'', TaskQueueServiceError.INTERNAL_ERROR, str(error)
  
      return (response.SerializeToString(), 0, "")
  
    def __bulk_add(self, source_info, request, response):
      """ Function for bulk adding tasks.
  
      Args:
        source_info: A dictionary containing the application, module, and version
         ID that is sending this request.
        request: taskqueue_service_pb2.TaskQueueBulkAddRequest.
        response: taskqueue_service_pb2.TaskQueueBulkAddResponse.
      Raises:
        apiproxy_error.ApplicationError.
      """
      if len(request.add_request) == 0:
        return
  
      now = datetime.datetime.utcfromtimestamp(time.time())
  
      # Assign names if needed and validate tasks.
      error_found = False
      namespaced_names = [None for _ in request.add_request]
      for index, add_request in enumerate(request.add_request):
        task_result = response.taskresult.add()
  
        if (add_request.HasField("mode") and
            add_request.mode == taskqueue_service_pb2.TaskQueueMode.PULL):
          queue = self.get_queue(add_request.app_id.decode('utf-8'),
                                 add_request.queue_name.decode('utf-8'))
          if not isinstance(queue, PostgresPullQueue):
            task_result.result = TaskQueueServiceError.INVALID_QUEUE_MODE
            error_found = True
            continue
  
          encoded_payload = base64.urlsafe_b64encode(add_request.body)
          task_info = {'payloadBase64': encoded_payload.decode('utf-8'),
                       'leaseTimestamp': add_request.eta_usec}
          if add_request.HasField("task_name"):
            task_info['id'] = add_request.task_name.decode('utf-8')
          if add_request.HasField("tag"):
            task_info['tag'] = add_request.tag.decode('utf-8')
  
          new_task = Task(task_info)
          queue.add_task(new_task)
          task_result.result = TaskQueueServiceError.OK
          task_result.chosen_task_name = new_task.id.encode('utf-8')
          continue
  
        result = verify_task_queue_add_request(add_request.app_id,
                                                      add_request, now)
        # Tasks go from SKIPPED to OK once they're run. If there are
        # any failures from other tasks then we pass this request
        # back as skipped.
        if result == TaskQueueServiceError.SKIPPED:
          task_name = None
          if add_request.HasField("task_name"):
            task_name = add_request.task_name
  
          namespaced_name = choose_task_name(
            add_request.app_id.decode('utf-8'),
            add_request.queue_name.decode('utf-8'),
            user_chosen=task_name.decode('utf-8'))
  
          add_request.task_name = namespaced_name.encode('utf-8')
  
          # Wait until the result is known before setting the chosen task name on
          # the response.
          namespaced_names[index] = namespaced_name
        else:
          error_found = True
          task_result.result = result
      if error_found:
        return
  
      for index, add_request in enumerate(request.add_request):
        task_result = response.taskresult[index]
        if (add_request.HasField("mode") and
            add_request.mode == taskqueue_service_pb2.TaskQueueMode.PULL):
          continue
  
        try:
          self.__enqueue_push_task(source_info, add_request)
        except ApplicationError as error:
          task_result.result = error.application_error
        except InvalidTarget as e:
          logger.error(e.args[0])
          task_result.result = TaskQueueServiceError.INVALID_REQUEST
        else:
          task_result.result = TaskQueueServiceError.OK
          if namespaced_names[index] is not None:
            task_result.chosen_task_name = \
              namespaced_names[index].encode('utf-8')
  
    def __method_mapping(self, method):
      """ Maps an int index to a string.
  
      Args:
        method: int representing a http method.
      Returns:
        A string version of the method.
     """
      if method == taskqueue_service_pb2.TaskQueueQueryTasksResponse.Task.GET:
        return 'GET'
      elif method == taskqueue_service_pb2.TaskQueueQueryTasksResponse.Task.POST:
        return  'POST'
      elif method == taskqueue_service_pb2.TaskQueueQueryTasksResponse.Task.HEAD:
        return  'HEAD'
      elif method == taskqueue_service_pb2.TaskQueueQueryTasksResponse.Task.PUT:
        return 'PUT'
      elif method == taskqueue_service_pb2.TaskQueueQueryTasksResponse.Task.DELETE:
        return 'DELETE'
  
    def __get_task_name(self, project_id, task_name, retries=3):
      """ Checks if a given task name entity exists.
  
      Args:
        project_id: A string specifying a project ID.
        task_name: A string specifying the task name key.
        retries: An integer specifying how many times to retry the get.
      """
      try:
        return self.datastore_client.get(project_id, task_name)
      except DatastoreTransientError as error:
        retries -= 1
        if retries >= 0:
          logger.warning('Error while checking task name: {}. '
                         'Retrying'.format(error))
          return self.__get_task_name(task_name, retries)
  
        raise
  
    def __create_task_name(self, project_id, queue_name, task_name, retries=3):
      """ Creates a new datastore_client.Entity object.
  
      Args:
        project_id: A string specifying a project ID.
        queue_name: A string specifying a queue name.
        task_name: A string specifying the task name key.
        retries: An integer specifying how many times to retry the create.
      """
      entity = Entity(key_name=task_name, state=TASK_STATES.QUEUED,
                      queue=queue_name, app_id=project_id)
      try:
        self.datastore_client.put(project_id, entity)
        return
      except DatastoreTransientError as error:
        retries -= 1
        if retries >= 0:
          logger.warning('Error creating task name: {}. Retrying'.format(error))
          return self.__create_task_name(project_id, queue_name, task_name,
                                         retries)
  
        raise
  
    def __check_and_store_task_names(self, request):
      """ Tries to fetch the taskqueue name, if it exists it will raise an
      exception.
  
      We store a receipt of each enqueued task in the datastore. If we find that
      task in the datastore, we will raise an exception. If the task is not
      in the datastore, then it is assumed this is the first time seeing the
      tasks and we create a receipt of the task in the datastore to prevent
      a duplicate task from being enqueued.
  
      Args:
        request: A taskqueue_service_pb2.TaskQueueAddRequest.
      Raises:
        An ApplicationError of TASK_ALREADY_EXISTS.
      """
      task_name = request.task_name.decode('utf-8')
  
      try:
        item = self.__get_task_name(request.app_id, task_name)
      except DatastoreTransientError:
        logger.exception('Unable to check task name')
        raise ApplicationError(
          taskqueue_service_pb2.TaskQueueServiceError.INTERNAL_ERROR)
  
      logger.debug("Task name {0}".format(task_name))
      if item:
        if item.state == TASK_STATES.QUEUED:
          logger.warning("Task already exists")
          raise ApplicationError(
            TaskQueueServiceError.TASK_ALREADY_EXISTS)
        else:
          # If a task with the same name has already been processed, it should
          # be tombstoned for some time to prevent a duplicate task.
          raise ApplicationError(
            TaskQueueServiceError.TOMBSTONED_TASK)
  
      logger.debug('Creating task name {}'.format(task_name))
      try:
        self.__create_task_name(request.app_id, request.queue_name,
                                task_name)
      except DatastoreTransientError:
        logger.exception('Unable to create task name')
        raise ApplicationError(
          TaskQueueServiceError.INTERNAL_ERROR)
  
    def __enqueue_push_task(self, source_info, request):
      """ Enqueues a batch of push tasks.
  
      Args:
        source_info: A dictionary containing the application, module, and version
         ID that is sending this request.
        request: A taskqueue_service_pb2.TaskQueueAddRequest.
      """
      self.__validate_push_task(request)
      self.__check_and_store_task_names(request)
      headers = self.get_task_headers(request)
      args = self.get_task_args(source_info, headers, request)
      countdown = int(headers['X-AppEngine-TaskETA']) - \
                  int(datetime.datetime.now().strftime("%s"))
  
      push_queue = self.get_queue(request.app_id.decode('utf-8'),
                                  request.queue_name.decode('utf-8'))
      task_func = get_queue_function_name(push_queue.name)
      celery_queue = get_celery_queue_name(
        request.app_id.decode('utf-8'), push_queue.name)
  
      push_queue.celery.send_task(
        task_func,
        kwargs={'headers': headers, 'args': args},
        expires=args['expires'],
        acks_late=True,
        countdown=countdown,
        queue=celery_queue,
        routing_key=celery_queue,
      )
  
    def get_task_args(self, source_info, headers, request):
      """ Gets the task args used when making a task web request.
  
      Args:
        source_info: A dictionary containing the application, module, and version
         ID that is sending this request.
        headers: The request headers, used to determine target.
        request: A taskqueue_service_pb2.TaskQueueAddRequest.
      Returns:
        A dictionary used by a task worker.
      """
      args = {}
      args['task_name'] = request.task_name
      args['app_id'] = request.app_id
      args['queue_name'] = request.queue_name
      args['method'] = self.__method_mapping(request.method)
      args['body'] = request.body
      args['description'] = request.description
  
      # Set defaults.
      args['max_retries'] = self.DEFAULT_MAX_RETRIES
      args['expires'] = self.__when_to_expire(request)
      args['max_retries'] = self.DEFAULT_MAX_RETRIES
      args['max_backoff_sec'] = self.DEFAULT_MAX_BACKOFF
      args['min_backoff_sec'] = self.DEFAULT_MIN_BACKOFF
      args['max_doublings'] = self.DEFAULT_MAX_DOUBLINGS
  
      # Load queue info into cache.
      app_id = self.__cleanse(request.app_id.decode('utf-8'))
      queue_name = request.queue_name
  
      # Use queue defaults.
      queue = self.get_queue(app_id, queue_name.decode('utf-8'))
      if queue is not None:
        if not isinstance(queue, PushQueue):
          raise Exception('Only push queues are implemented')
  
        args['max_retries'] = queue.task_retry_limit
        args['min_backoff_sec'] = queue.min_backoff_seconds
        args['max_backoff_sec'] = queue.max_backoff_seconds
        args['max_doublings'] = queue.max_doublings
  
      # Override defaults.
      if request.HasField("retry_parameters"):
        retry_params = request.retry_parameters
        if retry_params.HasField("retry_limit"):
          args['max_retries'] = retry_params.retry_limit
        if retry_params.HasField("min_backoff_sec"):
          args['min_backoff_sec'] = retry_params.min_backoff_sec
        if retry_params.HasField("max_backoff_sec"):
          args['max_backoff_sec'] = retry_params.max_backoff_sec
        if retry_params.HasField("max_doublings"):
          args['max_doublings'] = retry_params.max_doublings
  
      target_url = "http://{ip}:{port}".format(
        ip=self.load_balancers[0],
        port=self.get_module_port(app_id, source_info, target_info=[]))
  
      try:
        host = headers[b'Host'].decode('utf-8')
      except KeyError:
        host = None
      else:
        host = host if TARGET_REGEX.match(host) else None
  
      # Try to set target based on queue config.
      if queue.target:
        target_url = self.get_target_url(app_id, source_info, queue.target)
      # If we cannot get anything from the queue config, we try the target from
      # the request (python sdk will set the target via the Host header). Java
      # sdk does not include Host header, so we catch the KeyError.
      elif host:
        target_url = self.get_target_url(app_id, source_info, host)
  
      args['url'] = "{target}{url}".format(target=target_url,
                                           url=request.url.decode('utf-8'))
      return args
  
    def get_target_url(self, app_id, source_info, target):
      """ Gets the url for the target using the queue's target defined in the
      configuration file or the request's host header.
  
      Args:
        app_id: The application id, used to lookup module port.
        source_info: A dictionary containing the source version and module ids.
        target: A string containing the value of queue.target or the host header.
      Returns:
         A url as a string for the given target.
      """
      target_info = target.split('.')
      return "http://{ip}:{port}".format(
        ip=self.load_balancers[0],
        port=self.get_module_port(app_id, source_info, target_info))
  
    def get_module_port(self, app_id, source_info, target_info):
      """ Gets the port for the desired version and module or uses the current
      running version and module.
  
      Args:
       app_id: The application id, used to lookup port.
       source_info: A dictionary containing the source version and module ids.
       target_info: A list containing [version, module]
      Returns:
        An int containing the port for the target.
      Raises:
        InvalidTarget if the app_id, module, and version cannot be found in
          self.service_manager which maintains a dict of zookeeper info.
      """
      try:
        target_module = target_info.pop()
      except IndexError:
        target_module = source_info['module_id']
      try:
        target_version = target_info.pop()
      except IndexError:
        target_version = source_info['version_id']
      try:
        port = self.service_manager[app_id][target_module][target_version]
      except KeyError:
        err_msg = "target '{version}.{module}' does not exist".format(
          version=target_version, module=target_module)
        raise InvalidTarget(err_msg)
      return port
  
    def get_task_headers(self, request):
      """ Gets the task headers used for a task web request.
  
      Args:
        request: A taskqueue_service_pb2.TaskQueueAddRequest
      Returns:
        A dictionary of key/values for a web request.
      """
      headers = {}
      for header in request.header:
        headers[header.key] = header.value
  
      eta = self.__when_to_run(request)
  
      # This header is how we authenticate that it's an internal request
      secret = appscale_info.get_secret()
      secret_hash = hashlib.sha1(request.app_id + '/'.encode('utf-8') + \
                        secret.encode('utf-8')).hexdigest()
      headers['X-AppEngine-Fake-Is-Admin'] = secret_hash
      headers['X-AppEngine-QueueName'] = request.queue_name
      headers['X-AppEngine-TaskName'] = request.task_name
      headers['X-AppEngine-TaskRetryCount'] = '0'
      headers['X-AppEngine-TaskExecutionCount'] = '0'
      headers['X-AppEngine-TaskETA'] = str(int(eta.strftime("%s")))
      return headers
  
    def __when_to_run(self, request):
      """ Returns a datetime object of when a task should execute.
  
      Args:
        request: A taskqueue_service_pb2.TaskQueueAddRequest.
      Returns:
        A datetime object for when the nearest time to run the
       task is.
      """
      if request.HasField("eta_usec"):
        eta = request.eta_usec
        return datetime.datetime.fromtimestamp(eta/1000000)
      else:
        return datetime.datetime.now()
  
    def __when_to_expire(self, request):
      """ Returns a datetime object of when a task should expire.
  
      Args:
        request: A taskqueue_service_pb2.TaskQueueAddRequest.
      Returns:
        A datetime object of when the task should expire.
      """
      if request.HasField("retry_parameters") and \
             request.retry_parameters.HasField("age_limit_sec"):
        limit = request.retry_parameters.age_limit_sec
        return datetime.datetime.now() + datetime.timedelta(seconds=limit)
      else:
        return datetime.datetime.now() + \
                     datetime.timedelta(days=self.DEFAULT_EXPIRATION)
  
    def __validate_push_task(self, request):
      """ Checks to make sure the task request is valid.
  
      Args:
        request: A taskqueue_service_pb2.TaskQueueAddRequest.
      Raises:
        ApplicationError upon invalid tasks.
      """
      if not request.HasField("queue_name"):
        raise ApplicationError(
          TaskQueueServiceError.INVALID_QUEUE_NAME)
      if not request.HasField("task_name"):
        raise ApplicationError(
          TaskQueueServiceError.INVALID_TASK_NAME)
      if not request.HasField("app_id"):
        raise ApplicationError(
          TaskQueueServiceError.UNKNOWN_QUEUE)
      if not request.HasField("url"):
        raise ApplicationError(TaskQueueServiceError.INVALID_URL)
      if (request.HasField("mode") and
          request.mode == taskqueue_service_pb2.TaskQueueMode.PULL):
        raise ApplicationError(
          TaskQueueServiceError.INVALID_QUEUE_MODE)
  
    def modify_task_lease(self, app_id, http_data):
      """
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      request = taskqueue_service_pb2.TaskQueueModifyTaskLeaseRequest()
      request.ParseFromString(http_data)
  
      try:
        queue = self.get_queue(app_id, request.queue_name.decode('utf-8'))
      except QueueNotFound as error:
        return b'', TaskQueueServiceError.UNKNOWN_QUEUE, str(error)
  
      task_info = {'id': request.task_name.decode('utf-8'),
                   'leaseTimestamp': request.eta_usec}
      try:
        # The Python AppServer sets eta_usec with a resolution of 1 second,
        # so update_lease can't be used. It checks with millisecond precision.
        task = queue.update_task(Task(task_info), request.lease_seconds)
      except InvalidLeaseRequest as lease_error:
        return b'', TaskQueueServiceError.TASK_LEASE_EXPIRED, str(lease_error)
      except TaskNotFound as error:
        return b'', TaskQueueServiceError.TASK_LEASE_EXPIRED, str(error)
  
      epoch = datetime.datetime.utcfromtimestamp(0)
      updated_usec = int((task.leaseTimestamp - epoch).total_seconds() * 1000000)
      response = taskqueue_service_pb2.TaskQueueModifyTaskLeaseResponse()
      response.updated_eta_usec = updated_usec
      return response.SerializeToString(), 0, ""
  
    def fetch_queue(self, app_id, http_data):
      """
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      # TODO implement.
      request = taskqueue_service_pb2.TaskQueueFetchQueuesRequest(http_data)
      response = taskqueue_service_pb2.TaskQueueFetchQueuesResponse()
      return (response.SerializeToString(), 0, "")
  
    def query_tasks(self, app_id, http_data):
      """
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      # TODO implement.
      request = taskqueue_service_pb2.TaskQueueQueryTasksRequest(http_data)
      response = taskqueue_service_pb2.TaskQueueQueryTasksResponse()
      return (response.SerializeToString(), 0, "")
  
    def fetch_task(self, app_id, http_data):
      """
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      # TODO implement.
      request = taskqueue_service_pb2.TaskQueueFetchTaskRequest(http_data)
      response = taskqueue_service_pb2.TaskQueueFetchTaskResponse()
      return (response.SerializeToString(), 0, "")
  
    def force_run(self, app_id, http_data):
      """
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      # TODO implement.
      request = taskqueue_service_pb2.TaskQueueForceRunRequest(http_data)
      response = taskqueue_service_pb2.TaskQueueForceRunResponse()
      return (response.SerializeToString(), 0, "")
  
    def pause_queue(self, app_id, http_data):
      """
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      # TODO implement.
      request = taskqueue_service_pb2.TaskQueuePauseQueueRequest(http_data)
      response = taskqueue_service_pb2.TaskQueuePauseQueueResponse()
      return (response.SerializeToString(), 0, "")
  
    def delete_group(self, app_id, http_data):
      """
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      # TODO implement.
      request = taskqueue_service_pb2.TaskQueueDeleteGroupRequest(http_data)
      response = taskqueue_service_pb2.TaskQueueDeleteGroupResponse()
      return (response.SerializeToString(), 0, "")
  
    def update_storage_limit(self, app_id, http_data):
      """
  
      Args:
        app_id: The application ID.
        http_data: The payload containing the protocol buffer request.
      Returns:
        A tuple of a encoded response, error code, and error detail.
      """
      # TODO implement.
      request = taskqueue_service_pb2.TaskQueueUpdateStorageLimitRequest(http_data)
      response = taskqueue_service_pb2.TaskQueueUpdateStorageLimitResponse()
      return (response.SerializeToString(), 0, "")
  
    def __cleanse(self, str_input):
      """ Removes any questionable characters which might be apart of a remote
      attack.
  
      Args:
        str_input: The string to cleanse.
      Returns:
        A string which has questionable characters replaced.
      """
      for char in "~./\\!@#$%&*()]\+=|":
        str_input = str_input.replace(char, "_")
      return str_input
  
    def __is_localhost(self, hostname):
      """ Determines if the hostname is that of the current host.
  
      Args:
        hostname: A string representing the hostname.
      Returns:
        True if its the localhost, false otherwise.
      """
      if socket.gethostname() == hostname:
        return True
      elif socket.gethostbyname(socket.gethostname()) == hostname:
        return True
      else:
        return False
  