python/google/appengine/ext/mapreduce/base_handler.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright 2007 Google Inc.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33 """Base handler class for all mapreduce handlers."""
  34
  35
  36
  37
  38
  39
  40 import httplib
  41 import logging
  42
  43 import google
  44 import simplejson
  45
  46 try:
  47   from google.appengine.ext.mapreduce import pipeline_base
  48 except ImportError:
  49   pipeline_base = None
  50 try:
  51
  52   from google.appengine.ext import cloudstorage
  53   if hasattr(cloudstorage, "_STUB"):
  54     cloudstorage = None
  55 except ImportError:
  56   cloudstorage = None
  57
  58 from google.appengine.ext import webapp
  59 from google.appengine.ext.mapreduce import errors
  60 from google.appengine.ext.mapreduce import json_util
  61 from google.appengine.ext.mapreduce import model
  62 from google.appengine.ext.mapreduce import parameters
  63
  64
  65 class Error(Exception):
  66   """Base-class for exceptions in this module."""
  67
  68
  69 class BadRequestPathError(Error):
  70   """The request path for the handler is invalid."""
  71
  72
  73 class TaskQueueHandler(webapp.RequestHandler):
  74   """Base class for handlers intended to be run only from the task queue.
  75
  76   Sub-classes should implement
  77   1. the 'handle' method for all POST request.
  78   2. '_preprocess' method for decoding or validations before handle.
  79   3. '_drop_gracefully' method if _preprocess fails and the task has to
  80      be dropped.
  81
  82   In Python27 runtime, webapp2 will automatically replace webapp.
  83   """
  84
  85   def __init__(self, *args, **kwargs):
  86
  87
  88
  89
  90
  91     self._preprocess_success = False
  92     super(TaskQueueHandler, self).__init__(*args, **kwargs)
  93     if cloudstorage:
  94       cloudstorage.set_default_retry_params(
  95           cloudstorage.RetryParams(save_access_token=True))
  96
  97   def initialize(self, request, response):
  98     """Initialize.
  99
 100     1. call webapp init.
 101     2. check request is indeed from taskqueue.
 102     3. check the task has not been retried too many times.
 103     4. run handler specific processing logic.
 104     5. run error handling logic if precessing failed.
 105
 106     Args:
 107       request: a webapp.Request instance.
 108       response: a webapp.Response instance.
 109     """
 110     super(TaskQueueHandler, self).initialize(request, response)
 111
 112
 113     if "X-AppEngine-QueueName" not in self.request.headers:
 114       logging.error(self.request.headers)
 115       logging.error("Task queue handler received non-task queue request")
 116       self.response.set_status(
 117           403, message="Task queue handler received non-task queue request")
 118       return
 119
 120
 121     if self.task_retry_count() + 1 > parameters.config.TASK_MAX_ATTEMPTS:
 122       logging.error(
 123           "Task %s has been attempted %s times. Dropping it permanently.",
 124           self.request.headers["X-AppEngine-TaskName"],
 125           self.task_retry_count() + 1)
 126       self._drop_gracefully()
 127       return
 128
 129     try:
 130       self._preprocess()
 131       self._preprocess_success = True
 132
 133     except:
 134       self._preprocess_success = False
 135       logging.error(
 136           "Preprocess task %s failed. Dropping it permanently.",
 137           self.request.headers["X-AppEngine-TaskName"])
 138       self._drop_gracefully()
 139
 140   def post(self):
 141     if self._preprocess_success:
 142       self.handle()
 143
 144   def handle(self):
 145     """To be implemented by subclasses."""
 146     raise NotImplementedError()
 147
 148   def _preprocess(self):
 149     """Preprocess.
 150
 151     This method is called after webapp initialization code has been run
 152     successfully. It can thus access self.request, self.response and so on.
 153     """
 154     pass
 155
 156   def _drop_gracefully(self):
 157     """Drop task gracefully.
 158
 159     When preprocess failed, this method is called before the task is dropped.
 160     """
 161     pass
 162
 163   def task_retry_count(self):
 164     """Number of times this task has been retried."""
 165     return int(self.request.headers.get("X-AppEngine-TaskExecutionCount", 0))
 166
 167   def retry_task(self):
 168     """Ask taskqueue to retry this task.
 169
 170     Even though raising an exception can cause a task retry, it
 171     will flood logs with highly visible ERROR logs. Handlers should uses
 172     this method to perform controlled task retries. Only raise exceptions
 173     for those deserve ERROR log entries.
 174     """
 175     self.response.set_status(httplib.SERVICE_UNAVAILABLE, "Retry task")
 176     self.response.clear()
 177
 178
 179 class JsonHandler(webapp.RequestHandler):
 180   """Base class for JSON handlers for user interface.
 181
 182   Sub-classes should implement the 'handle' method. They should put their
 183   response data in the 'self.json_response' dictionary. Any exceptions raised
 184   by the sub-class implementation will be sent in a JSON response with the
 185   name of the error_class and the error_message.
 186   """
 187
 188   def __init__(self, *args):
 189     """Initializer."""
 190     super(JsonHandler, self).__init__(*args)
 191     self.json_response = {}
 192
 193   def base_path(self):
 194     """Base path for all mapreduce-related urls.
 195
 196     JSON handlers are mapped to /base_path/command/command_name thus they
 197     require special treatment.
 198     """
 199     path = self.request.path
 200     base_path = path[:path.rfind("/")]
 201     if not base_path.endswith("/command"):
 202       raise BadRequestPathError(
 203           "Json handlers should have /command path prefix")
 204     return base_path[:base_path.rfind("/")]
 205
 206   def _handle_wrapper(self):
 207     if self.request.headers.get("X-Requested-With") != "XMLHttpRequest":
 208       logging.error("Got JSON request with no X-Requested-With header")
 209       self.response.set_status(
 210           403, message="Got JSON request with no X-Requested-With header")
 211       return
 212
 213     self.json_response.clear()
 214     try:
 215       self.handle()
 216     except errors.MissingYamlError:
 217       logging.debug("Could not find 'mapreduce.yaml' file.")
 218       self.json_response.clear()
 219       self.json_response["error_class"] = "Notice"
 220       self.json_response["error_message"] = "Could not find 'mapreduce.yaml'"
 221     except Exception, e:
 222       logging.exception("Error in JsonHandler, returning exception.")
 223
 224       self.json_response.clear()
 225       self.json_response["error_class"] = e.__class__.__name__
 226       self.json_response["error_message"] = str(e)
 227
 228     self.response.headers["Content-Type"] = "text/javascript"
 229     try:
 230       output = simplejson.dumps(self.json_response, cls=json_util.JsonEncoder)
 231     except:
 232       logging.exception("Could not serialize to JSON")
 233       self.response.set_status(500, message="Could not serialize to JSON")
 234       return
 235     else:
 236       self.response.out.write(output)
 237
 238   def handle(self):
 239     """To be implemented by sub-classes."""
 240     raise NotImplementedError()
 241
 242
 243 class PostJsonHandler(JsonHandler):
 244   """JSON handler that accepts POST requests."""
 245
 246   def post(self):
 247     self._handle_wrapper()
 248
 249
 250 class GetJsonHandler(JsonHandler):
 251   """JSON handler that accepts GET posts."""
 252
 253   def get(self):
 254     self._handle_wrapper()
 255
 256
 257 class HugeTaskHandler(TaskQueueHandler):
 258   """Base handler for processing HugeTasks."""
 259
 260   class _RequestWrapper(object):
 261     def __init__(self, request):
 262       self._request = request
 263       self._params = model.HugeTask.decode_payload(request)
 264
 265     def get(self, name, default=""):
 266       return self._params.get(name, default)
 267
 268     def set(self, name, value):
 269       self._params[name] = value
 270
 271     def __getattr__(self, name):
 272       return getattr(self._request, name)
 273
 274   def __init__(self, *args, **kwargs):
 275     super(HugeTaskHandler, self).__init__(*args, **kwargs)
 276
 277   def _preprocess(self):
 278     self.request = self._RequestWrapper(self.request)
 279
 280
 281 if pipeline_base:
 282
 283   PipelineBase = pipeline_base.PipelineBase
 284 else:
 285   PipelineBase = None