python/google/appengine/ext/deferred/deferred.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 """A module that handles deferred execution of callables via the task queue.
  22
  23 Tasks consist of a callable and arguments to pass to it. The callable and its
  24 arguments are serialized and put on the task queue, which deserializes and
  25 executes them. The following callables can be used as tasks:
  26
  27 1) Functions defined in the top level of a module
  28 2) Classes defined in the top level of a module
  29 3) Instances of classes in (2) that implement __call__
  30 4) Instance methods of objects of classes in (2)
  31 5) Class methods of classes in (2)
  32 6) Built-in functions
  33 7) Built-in methods
  34
  35 The following callables can NOT be used as tasks:
  36 1) Nested functions or closures
  37 2) Nested classes or objects of them
  38 3) Lambda functions
  39 4) Static methods
  40
  41 The arguments to the callable, and the object (in the case of method or object
  42 calls) must all be pickleable.
  43
  44 If you want your tasks to execute reliably, don't use mutable global variables;
  45 they are not serialized with the task and may not be the same when your task
  46 executes as they were when it was enqueued (in fact, they will almost certainly
  47 be different).
  48
  49 If your app relies on manipulating the import path, make sure that the function
  50 you are deferring is defined in a module that can be found without import path
  51 manipulation. Alternately, you can include deferred.TaskHandler in your own
  52 webapp application instead of using the easy-install method detailed below.
  53
  54 When you create a deferred task using deferred.defer, the task is serialized,
  55 and an attempt is made to add it directly to the task queue. If the task is too
  56 big (larger than about 10 kilobytes when serialized), a datastore entry will be
  57 created for the task, and a new task will be enqueued, which will fetch the
  58 original task from the datastore and execute it. This is much less efficient
  59 than the direct execution model, so it's a good idea to minimize the size of
  60 your tasks when possible.
  61
  62 In order for tasks to be processed, you need to set up the handler. Add the
  63 following to your app.yaml handlers section:
  64
  65 handlers:
  66 - url: /_ah/queue/deferred
  67   script: $PYTHON_LIB/google/appengine/ext/deferred/handler.py
  68   login: admin
  69
  70 By default, the deferred module uses the URL above, and the default queue.
  71
  72 Example usage:
  73
  74   def do_something_later(key, amount):
  75     entity = MyModel.get(key)
  76     entity.total += amount
  77     entity.put()
  78
  79   # Use default URL and queue name, no task name, execute ASAP.
  80   deferred.defer(do_something_later, 20)
  81
  82   # Providing non-default task queue arguments
  83   deferred.defer(do_something_later, 20, _queue="foo", countdown=60)
  84 """
  85
  86
  87
  88
  89
  90
  91
  92
  93
  94 import logging
  95 import os
  96 import pickle
  97 import types
  98
  99 from google.appengine.api import taskqueue
 100 from google.appengine.ext import db
 101 from google.appengine.ext import webapp
 102 from google.appengine.ext.webapp.util import run_wsgi_app
 103
 104
 105 _DEFAULT_LOG_LEVEL = logging.INFO
 106 _TASKQUEUE_HEADERS = {"Content-Type": "application/octet-stream"}
 107 _DEFAULT_URL = "/_ah/queue/deferred"
 108 _DEFAULT_QUEUE = "default"
 109
 110
 111 class Error(Exception):
 112   """Base class for exceptions in this module."""
 113
 114
 115 class PermanentTaskFailure(Error):
 116   """Indicates that a task failed, and will never succeed."""
 117
 118
 119 class SingularTaskFailure(Error):
 120   """Indicates that a task failed once."""
 121
 122
 123 def set_log_level(log_level):
 124   """Sets the log level deferred will log to in normal circumstances.
 125
 126   Args:
 127     log_level: one of logging log levels, e.g. logging.DEBUG, logging.INFO, etc.
 128   """
 129   global _DEFAULT_LOG_LEVEL
 130   _DEFAULT_LOG_LEVEL = log_level
 131
 132
 133 def run(data):
 134   """Unpickles and executes a task.
 135
 136   Args:
 137     data: A pickled tuple of (function, args, kwargs) to execute.
 138   Returns:
 139     The return value of the function invocation.
 140   """
 141   try:
 142     func, args, kwds = pickle.loads(data)
 143   except Exception, e:
 144     raise PermanentTaskFailure(e)
 145   else:
 146     return func(*args, **kwds)
 147
 148
 149 class _DeferredTaskEntity(db.Model):
 150   """Datastore representation of a deferred task.
 151
 152   This is used in cases when the deferred task is too big to be included as
 153   payload with the task queue entry.
 154   """
 155   data = db.BlobProperty(required=True)
 156
 157
 158 def run_from_datastore(key):
 159   """Retrieves a task from the datastore and executes it.
 160
 161   Args:
 162     key: The datastore key of a _DeferredTaskEntity storing the task.
 163   Returns:
 164     The return value of the function invocation.
 165   """
 166   entity = _DeferredTaskEntity.get(key)
 167   if not entity:
 168
 169     raise PermanentTaskFailure()
 170   try:
 171     ret = run(entity.data)
 172     entity.delete()
 173   except PermanentTaskFailure:
 174     entity.delete()
 175     raise
 176
 177
 178 def invoke_member(obj, membername, *args, **kwargs):
 179   """Retrieves a member of an object, then calls it with the provided arguments.
 180
 181   Args:
 182     obj: The object to operate on.
 183     membername: The name of the member to retrieve from ojb.
 184     args: Positional arguments to pass to the method.
 185     kwargs: Keyword arguments to pass to the method.
 186   Returns:
 187     The return value of the method invocation.
 188   """
 189   return getattr(obj, membername)(*args, **kwargs)
 190
 191
 192 def _curry_callable(obj, *args, **kwargs):
 193   """Takes a callable and arguments and returns a task queue tuple.
 194
 195   The returned tuple consists of (callable, args, kwargs), and can be pickled
 196   and unpickled safely.
 197
 198   Args:
 199     obj: The callable to curry. See the module docstring for restrictions.
 200     args: Positional arguments to call the callable with.
 201     kwargs: Keyword arguments to call the callable with.
 202   Returns:
 203     A tuple consisting of (callable, args, kwargs) that can be evaluated by
 204     run() with equivalent effect of executing the function directly.
 205   Raises:
 206     ValueError: If the passed in object is not of a valid callable type.
 207   """
 208   if isinstance(obj, types.MethodType):
 209     return (invoke_member, (obj.im_self, obj.im_func.__name__) + args, kwargs)
 210   elif isinstance(obj, types.BuiltinMethodType):
 211     if not obj.__self__:
 212
 213       return (obj, args, kwargs)
 214     else:
 215       return (invoke_member, (obj.__self__, obj.__name__) + args, kwargs)
 216   elif isinstance(obj, types.ObjectType) and hasattr(obj, "__call__"):
 217     return (obj, args, kwargs)
 218   elif isinstance(obj, (types.FunctionType, types.BuiltinFunctionType,
 219                         types.ClassType, types.UnboundMethodType)):
 220     return (obj, args, kwargs)
 221   else:
 222     raise ValueError("obj must be callable")
 223
 224
 225 def serialize(obj, *args, **kwargs):
 226   """Serializes a callable into a format recognized by the deferred executor.
 227
 228   Args:
 229     obj: The callable to serialize. See module docstring for restrictions.
 230     args: Positional arguments to call the callable with.
 231     kwargs: Keyword arguments to call the callable with.
 232   Returns:
 233     A serialized representation of the callable.
 234   """
 235   curried = _curry_callable(obj, *args, **kwargs)
 236   return pickle.dumps(curried, protocol=pickle.HIGHEST_PROTOCOL)
 237
 238
 239 def defer(obj, *args, **kwargs):
 240   """Defers a callable for execution later.
 241
 242   The default deferred URL of /_ah/queue/deferred will be used unless an
 243   alternate URL is explicitly specified. If you want to use the default URL for
 244   a queue, specify _url=None. If you specify a different URL, you will need to
 245   install the handler on that URL (see the module docstring for details).
 246
 247   Args:
 248     obj: The callable to execute. See module docstring for restrictions.
 249         _countdown, _eta, _headers, _name, _target, _transactional, _url,
 250         _retry_options, _queue: Passed through to the task queue - see the
 251         task queue documentation for details.
 252     args: Positional arguments to call the callable with.
 253     kwargs: Any other keyword arguments are passed through to the callable.
 254   Returns:
 255     A taskqueue.Task object which represents an enqueued callable.
 256   """
 257   taskargs = dict((x, kwargs.pop(("_%s" % x), None))
 258                   for x in ("countdown", "eta", "name", "target",
 259                             "retry_options"))
 260   taskargs["url"] = kwargs.pop("_url", _DEFAULT_URL)
 261   transactional = kwargs.pop("_transactional", False)
 262   taskargs["headers"] = dict(_TASKQUEUE_HEADERS)
 263   taskargs["headers"].update(kwargs.pop("_headers", {}))
 264   queue = kwargs.pop("_queue", _DEFAULT_QUEUE)
 265   pickled = serialize(obj, *args, **kwargs)
 266   try:
 267     task = taskqueue.Task(payload=pickled, **taskargs)
 268     return task.add(queue, transactional=transactional)
 269   except taskqueue.TaskTooLargeError:
 270
 271     key = _DeferredTaskEntity(data=pickled).put()
 272     pickled = serialize(run_from_datastore, str(key))
 273     task = taskqueue.Task(payload=pickled, **taskargs)
 274     return task.add(queue)
 275
 276
 277 class TaskHandler(webapp.RequestHandler):
 278   """A webapp handler class that processes deferred invocations."""
 279
 280   def run_from_request(self):
 281     """Default behavior for POST requests to deferred handler."""
 282
 283     if 'X-AppEngine-TaskName' not in self.request.headers:
 284       logging.critical('Detected an attempted XSRF attack. The header '
 285                        '"X-AppEngine-Taskname" was not set.')
 286       self.response.set_status(403)
 287       return
 288
 289
 290
 291     in_prod = (
 292         not self.request.environ.get("SERVER_SOFTWARE").startswith("Devel"))
 293     if in_prod and self.request.environ.get("REMOTE_ADDR") != "0.1.0.2":
 294       logging.critical('Detected an attempted XSRF attack. This request did '
 295                        'not originate from Task Queue.')
 296       self.response.set_status(403)
 297       return
 298
 299
 300     headers = ["%s:%s" % (k, v) for k, v in self.request.headers.items()
 301                if k.lower().startswith("x-appengine-")]
 302     logging.log(_DEFAULT_LOG_LEVEL, ", ".join(headers))
 303
 304     run(self.request.body)
 305
 306   def post(self):
 307
 308     try:
 309       self.run_from_request()
 310     except SingularTaskFailure:
 311
 312
 313       logging.debug("Failure executing task, task retry forced")
 314       self.response.set_status(408)
 315       return
 316     except PermanentTaskFailure, e:
 317
 318       logging.exception("Permanent failure attempting to execute task")
 319
 320
 321 application = webapp.WSGIApplication([(".*", TaskHandler)])
 322
 323
 324 def main():
 325   if os.environ["SERVER_SOFTWARE"].startswith("Devel"):
 326     logging.warn("You are using deferred in a deprecated fashion. Please change"
 327                  " the request handler path for /_ah/queue/deferred in app.yaml"
 328                  " to $PYTHON_LIB/google/appengine/ext/deferred/handler.py to"
 329                  " avoid encountering import errors.")
 330   run_wsgi_app(application)
 331
 332
 333 if __name__ == "__main__":
 334   main()