python/google/appengine/api/apiproxy_stub_map.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 """Container of APIProxy stubs for more convenient unittesting.
  23
  24 Classes/variables/functions defined here:
  25   APIProxyStubMap: container of APIProxy stubs.
  26   apiproxy: global instance of an APIProxyStubMap.
  27   MakeSyncCall: APIProxy entry point.
  28   UserRPC: User-visible class wrapping asynchronous RPCs.
  29 """
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40 import inspect
  41 import sys
  42 import threading
  43
  44 from google.appengine.api import apiproxy_rpc
  45 from google.appengine.runtime import apiproxy_errors
  46
  47
  48 def CreateRPC(service, stubmap=None):
  49   """Creates a RPC instance for the given service.
  50
  51   The instance is suitable for talking to remote services.
  52   Each RPC instance can be used only once, and should not be reused.
  53
  54   Args:
  55     service: string representing which service to call.
  56     stubmap: optional APIProxyStubMap instance, for dependency injection.
  57
  58   Returns:
  59     the rpc object.
  60
  61   Raises:
  62     AssertionError or RuntimeError if the stub for service doesn't supply a
  63     CreateRPC method.
  64   """
  65   if stubmap is None:
  66     stubmap = apiproxy
  67   stub = stubmap.GetStub(service)
  68   assert stub, 'No api proxy found for service "%s"' % service
  69   assert hasattr(stub, 'CreateRPC'), (('The service "%s" doesn\'t have ' +
  70                                        'a CreateRPC method.') % service)
  71   return stub.CreateRPC()
  72
  73
  74 def MakeSyncCall(service, call, request, response, stubmap=None):
  75   """The APIProxy entry point for a synchronous API call.
  76
  77   Args:
  78     service: string representing which service to call
  79     call: string representing which function to call
  80     request: protocol buffer for the request
  81     response: protocol buffer for the response
  82     stubmap: optional APIProxyStubMap instance, for dependency injection.
  83
  84   Returns:
  85     Response protocol buffer or None. Some implementations may return
  86     a response protocol buffer instead of modifying 'response'.
  87     Caller must use returned value in such cases. If 'response' is modified
  88     then returns None.
  89
  90   Raises:
  91     apiproxy_errors.Error or a subclass.
  92   """
  93   if stubmap is None:
  94     stubmap = apiproxy
  95   return stubmap.MakeSyncCall(service, call, request, response)
  96
  97
  98 class ListOfHooks(object):
  99   """An ordered collection of hooks for a particular API call.
 100
 101   A hook is a function that has exactly the same signature as
 102   a service stub. It will be called before or after an api hook is
 103   executed, depending on whether this list is for precall of postcall hooks.
 104   Hooks can be used for debugging purposes (check certain
 105   pre- or postconditions on api calls) or to apply patches to protocol
 106   buffers before/after a call gets submitted.
 107   """
 108
 109   def __init__(self):
 110     """Constructor."""
 111
 112
 113     self.__content = []
 114
 115
 116     self.__unique_keys = set()
 117
 118   def __len__(self):
 119     """Returns the amount of elements in the collection."""
 120     return self.__content.__len__()
 121
 122   def __Insert(self, index, key, function, service=None):
 123     """Appends a hook at a certain position in the list.
 124
 125     Args:
 126       index: the index of where to insert the function
 127       key: a unique key (within the module) for this particular function.
 128         If something from the same module with the same key is already
 129         registered, nothing will be added.
 130       function: the hook to be added.
 131       service: optional argument that restricts the hook to a particular api
 132
 133     Returns:
 134       True if the collection was modified.
 135     """
 136     unique_key = (key, inspect.getmodule(function))
 137     if unique_key in self.__unique_keys:
 138       return False
 139     num_args = len(inspect.getargspec(function)[0])
 140     if (inspect.ismethod(function)):
 141       num_args -= 1
 142     self.__content.insert(index, (key, function, service, num_args))
 143     self.__unique_keys.add(unique_key)
 144     return True
 145
 146   def Append(self, key, function, service=None):
 147     """Appends a hook at the end of the list.
 148
 149     Args:
 150       key: a unique key (within the module) for this particular function.
 151         If something from the same module with the same key is already
 152         registered, nothing will be added.
 153       function: the hook to be added.
 154       service: optional argument that restricts the hook to a particular api
 155
 156     Returns:
 157       True if the collection was modified.
 158     """
 159     return self.__Insert(len(self), key, function, service)
 160
 161   def Push(self, key, function, service=None):
 162     """Inserts a hook at the beginning of the list.
 163
 164     Args:
 165       key: a unique key (within the module) for this particular function.
 166         If something from the same module with the same key is already
 167         registered, nothing will be added.
 168       function: the hook to be added.
 169       service: optional argument that restricts the hook to a particular api
 170
 171     Returns:
 172       True if the collection was modified.
 173     """
 174     return self.__Insert(0, key, function, service)
 175
 176   def Clear(self):
 177     """Removes all hooks from the list (useful for unit tests)."""
 178     self.__content = []
 179     self.__unique_keys = set()
 180
 181   def Call(self, service, call, request, response, rpc=None, error=None):
 182     """Invokes all hooks in this collection.
 183
 184     NOTE: For backwards compatibility, if error is not None, hooks
 185     with 4 or 5 arguments are *not* called.  This situation
 186     (error=None) only occurs when the RPC request raised an exception;
 187     in the past no hooks would be called at all in that case.
 188
 189     Args:
 190       service: string representing which service to call
 191       call: string representing which function to call
 192       request: protocol buffer for the request
 193       response: protocol buffer for the response
 194       rpc: optional RPC used to make this call
 195       error: optional Exception instance to be passed as 6th argument
 196     """
 197     for key, function, srv, num_args in self.__content:
 198       if srv is None or srv == service:
 199         if num_args == 6:
 200           function(service, call, request, response, rpc, error)
 201         elif error is not None:
 202
 203
 204           pass
 205         elif num_args == 5:
 206           function(service, call, request, response, rpc)
 207         else:
 208           function(service, call, request, response)
 209
 210
 211 class APIProxyStubMap(object):
 212   """Container of APIProxy stubs for more convenient unittesting.
 213
 214   Stubs may be either trivial implementations of APIProxy services (e.g.
 215   DatastoreFileStub, UserServiceStub) or "real" implementations.
 216
 217   For unittests, we may want to mix and match real and trivial implementations
 218   of services in order to better focus testing on individual service
 219   implementations. To achieve this, we allow the client to attach stubs to
 220   service names, as well as define a default stub to be used if no specific
 221   matching stub is identified.
 222   """
 223
 224
 225
 226
 227   def __init__(self, default_stub=None):
 228     """Constructor.
 229
 230     Args:
 231       default_stub: optional stub
 232
 233     'default_stub' will be used whenever no specific matching stub is found.
 234     """
 235     self.__stub_map = {}
 236     self.__default_stub = default_stub
 237     self.__precall_hooks = ListOfHooks()
 238     self.__postcall_hooks = ListOfHooks()
 239
 240   def GetPreCallHooks(self):
 241     """Gets a collection for all precall hooks."""
 242     return self.__precall_hooks
 243
 244   def GetPostCallHooks(self):
 245     """Gets a collection for all precall hooks."""
 246     return self.__postcall_hooks
 247
 248   def ReplaceStub(self, service, stub):
 249     """Replace the existing stub for the specified service with a new one.
 250
 251     NOTE: This is a risky operation; external callers should use this with
 252     caution.
 253
 254     Args:
 255       service: string
 256       stub: stub
 257     """
 258     self.__stub_map[service] = stub
 259
 260
 261
 262
 263
 264
 265
 266     if service == 'datastore':
 267       self.RegisterStub('datastore_v3', stub)
 268
 269   def RegisterStub(self, service, stub):
 270     """Register the provided stub for the specified service.
 271
 272     Args:
 273       service: string
 274       stub: stub
 275     """
 276     assert not self.__stub_map.has_key(service), repr(service)
 277     self.ReplaceStub(service, stub)
 278
 279   def GetStub(self, service):
 280     """Retrieve the stub registered for the specified service.
 281
 282     Args:
 283       service: string
 284
 285     Returns:
 286       stub
 287
 288     Returns the stub registered for 'service', and returns the default stub
 289     if no such stub is found.
 290     """
 291     return self.__stub_map.get(service, self.__default_stub)
 292
 293   def _CopyStubMap(self):
 294     """Get a copy of the stub map. For testing only.
 295
 296     Returns:
 297       Get a shallow copy of the stub map.
 298     """
 299     return dict(self.__stub_map)
 300
 301   def MakeSyncCall(self, service, call, request, response):
 302     """The APIProxy entry point.
 303
 304     Args:
 305       service: string representing which service to call
 306       call: string representing which function to call
 307       request: protocol buffer for the request
 308       response: protocol buffer for the response
 309
 310     Returns:
 311       Response protocol buffer or None. Some implementations may return
 312       a response protocol buffer instead of modifying 'response'.
 313       Caller must use returned value in such cases. If 'response' is modified
 314       then returns None.
 315
 316     Raises:
 317       apiproxy_errors.Error or a subclass.
 318     """
 319
 320
 321     stub = self.GetStub(service)
 322     assert stub, 'No api proxy found for service "%s"' % service
 323     if hasattr(stub, 'CreateRPC'):
 324       rpc = stub.CreateRPC()
 325       self.__precall_hooks.Call(service, call, request, response, rpc)
 326       try:
 327         rpc.MakeCall(service, call, request, response)
 328         rpc.Wait()
 329         rpc.CheckSuccess()
 330       except Exception, err:
 331         self.__postcall_hooks.Call(service, call, request, response, rpc, err)
 332         raise
 333       else:
 334         self.__postcall_hooks.Call(service, call, request, response, rpc)
 335     else:
 336       self.__precall_hooks.Call(service, call, request, response)
 337       try:
 338         returned_response = stub.MakeSyncCall(service, call, request, response)
 339       except Exception, err:
 340         self.__postcall_hooks.Call(service, call, request, response, None, err)
 341         raise
 342       else:
 343         self.__postcall_hooks.Call(service, call, request,
 344                                    returned_response or response)
 345         return returned_response
 346
 347   def CancelApiCalls(self):
 348     if self.__default_stub:
 349       self.__default_stub.CancelApiCalls()
 350
 351
 352 class UserRPC(object):
 353   """Wrapper class for asynchronous RPC.
 354
 355   Simplest low-level usage pattern:
 356
 357     rpc = UserRPC('service', [deadline], [callback])
 358     rpc.make_call('method', request, response)
 359     .
 360     .
 361     .
 362     rpc.wait()
 363     rpc.check_success()
 364
 365   However, a service module normally provides a wrapper so that the
 366   typical usage pattern becomes more like this:
 367
 368     from google.appengine.api import service
 369     rpc = service.create_rpc([deadline], [callback])
 370     service.make_method_call(rpc, [service-specific-args])
 371     .
 372     .
 373     .
 374     rpc.wait()
 375     result = rpc.get_result()
 376
 377   The service.make_method_call() function sets a service- and method-
 378   specific hook function that is called by rpc.get_result() with the
 379   rpc object as its first argument, and service-specific value as its
 380   second argument.  The hook function should call rpc.check_success()
 381   and then extract the user-level result from the rpc.result
 382   protobuffer.  Additional arguments may be passed from
 383   make_method_call() to the get_result hook via the second argument.
 384
 385   Also note wait_any() and wait_all(), which wait for multiple RPCs.
 386   """
 387
 388   __method = None
 389   __get_result_hook = None
 390   __user_data = None
 391   __postcall_hooks_called = False
 392   __must_call_user_callback = False
 393
 394   class MyLocal(threading.local):
 395     """Class to hold per-thread class level attributes."""
 396
 397     may_interrupt_wait = False
 398
 399   __local = MyLocal()
 400
 401   def __init__(self, service, deadline=None, callback=None, stubmap=None):
 402     """Constructor.
 403
 404     Args:
 405       service: The service name.
 406       deadline: Optional deadline.  Default depends on the implementation.
 407       callback: Optional argument-less callback function.
 408       stubmap: optional APIProxyStubMap instance, for dependency injection.
 409     """
 410     if stubmap is None:
 411       stubmap = apiproxy
 412     self.__stubmap = stubmap
 413     self.__service = service
 414     self.__rpc = CreateRPC(service, stubmap)
 415     self.__rpc.deadline = deadline
 416     self.__rpc.callback = self.__internal_callback
 417     self.callback = callback
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431     self.__class__.__local.may_interrupt_wait = False
 432
 433   def __internal_callback(self):
 434     """This is the callback set on the low-level RPC object.
 435
 436     It sets a flag on the current object indicating that the high-level
 437     callback should now be called.  If interrupts are enabled, it also
 438     interrupts the current wait_any() call by raising an exception.
 439     """
 440     self.__must_call_user_callback = True
 441     self.__rpc.callback = None
 442     if self.__class__.__local.may_interrupt_wait and not self.__rpc.exception:
 443
 444
 445
 446
 447
 448
 449
 450
 451       raise apiproxy_errors.InterruptedError(None, self.__rpc)
 452
 453   @property
 454   def service(self):
 455     """Return the service name."""
 456     return self.__service
 457
 458   @property
 459   def method(self):
 460     """Return the method name."""
 461     return self.__method
 462
 463   @property
 464   def deadline(self):
 465     """Return the deadline, if set explicitly (otherwise None)."""
 466     return self.__rpc.deadline
 467
 468   @property
 469   def request(self):
 470     """Return the request protocol buffer object."""
 471     return self.__rpc.request
 472
 473   @property
 474   def response(self):
 475     """Return the response protocol buffer object."""
 476     return self.__rpc.response
 477
 478   @property
 479   def state(self):
 480     """Return the RPC state.
 481
 482     Possible values are attributes of apiproxy_rpc.RPC: IDLE, RUNNING,
 483     FINISHING.
 484     """
 485     return self.__rpc.state
 486
 487   @property
 488   def get_result_hook(self):
 489     """Return the get-result hook function."""
 490     return self.__get_result_hook
 491
 492   @property
 493   def user_data(self):
 494     """Return the user data for the hook function."""
 495     return self.__user_data
 496
 497   def make_call(self, method, request, response,
 498                 get_result_hook=None, user_data=None):
 499     """Initiate a call.
 500
 501     Args:
 502       method: The method name.
 503       request: The request protocol buffer.
 504       response: The response protocol buffer.
 505       get_result_hook: Optional get-result hook function.  If not None,
 506         this must be a function with exactly one argument, the RPC
 507         object (self).  Its return value is returned from get_result().
 508       user_data: Optional additional arbitrary data for the get-result
 509         hook function.  This can be accessed as rpc.user_data.  The
 510         type of this value is up to the service module.
 511
 512     This function may only be called once per RPC object.  It sends
 513     the request to the remote server, but does not wait for a
 514     response.  This allows concurrent execution of the remote call and
 515     further local processing (e.g., making additional remote calls).
 516
 517     Before the call is initiated, the precall hooks are called.
 518     """
 519
 520     assert self.__rpc.state == apiproxy_rpc.RPC.IDLE, repr(self.state)
 521
 522     self.__method = method
 523
 524     self.__get_result_hook = get_result_hook
 525     self.__user_data = user_data
 526
 527     self.__stubmap.GetPreCallHooks().Call(
 528         self.__service, method, request, response, self.__rpc)
 529
 530     self.__rpc.MakeCall(self.__service, method, request, response)
 531
 532   def wait(self):
 533     """Wait for the call to complete, and call callback if needed.
 534
 535     This and wait_any()/wait_all() are the only time callback
 536     functions may be called.  (However, note that check_success() and
 537     get_result() call wait().)  Waiting for one RPC will not cause
 538     callbacks for other RPCs to be called.  Callback functions may
 539     call check_success() and get_result().
 540
 541     Callbacks are called without arguments; if a callback needs access
 542     to the RPC object a Python nested function (a.k.a. closure) or a
 543     bound may be used.  To facilitate this, the callback may be
 544     assigned after the RPC object is created (but before make_call()
 545     is called).
 546
 547     Note: don't confuse callbacks with get-result hooks or precall
 548     and postcall hooks.
 549     """
 550
 551     assert self.__rpc.state != apiproxy_rpc.RPC.IDLE, repr(self.state)
 552
 553     if self.__rpc.state == apiproxy_rpc.RPC.RUNNING:
 554       self.__rpc.Wait()
 555
 556     assert self.__rpc.state == apiproxy_rpc.RPC.FINISHING, repr(self.state)
 557     self.__call_user_callback()
 558
 559   def __call_user_callback(self):
 560     """Call the high-level callback, if requested."""
 561     if self.__must_call_user_callback:
 562       self.__must_call_user_callback = False
 563       if self.callback is not None:
 564         self.callback()
 565
 566   def check_success(self):
 567     """Check for success of the RPC, possibly raising an exception.
 568
 569     This function should be called at least once per RPC.  If wait()
 570     hasn't been called yet, it is called first.  If the RPC caused
 571     an exceptional condition, an exception will be raised here.
 572     The first time check_success() is called, the postcall hooks
 573     are called.
 574     """
 575
 576
 577     self.wait()
 578     try:
 579       self.__rpc.CheckSuccess()
 580     except Exception, err:
 581
 582       if not self.__postcall_hooks_called:
 583         self.__postcall_hooks_called = True
 584         self.__stubmap.GetPostCallHooks().Call(self.__service, self.__method,
 585                                                self.request, self.response,
 586                                                self.__rpc, err)
 587       raise
 588     else:
 589
 590       if not self.__postcall_hooks_called:
 591         self.__postcall_hooks_called = True
 592         self.__stubmap.GetPostCallHooks().Call(self.__service, self.__method,
 593                                                self.request, self.response,
 594                                                self.__rpc)
 595
 596   def get_result(self):
 597     """Get the result of the RPC, or possibly raise an exception.
 598
 599     This implies a call to check_success().  If a get-result hook was
 600     passed to make_call(), that hook is responsible for calling
 601     check_success(), and the return value of the hook is returned.
 602     Otherwise, check_success() is called directly and None is
 603     returned.
 604     """
 605
 606
 607
 608
 609     if self.__get_result_hook is None:
 610       self.check_success()
 611       return None
 612     else:
 613       return self.__get_result_hook(self)
 614
 615   @classmethod
 616   def __check_one(cls, rpcs):
 617     """Check the list of RPCs for one that is finished, or one that is running.
 618
 619     Args:
 620       rpcs: Iterable collection of UserRPC instances.
 621
 622     Returns:
 623       A pair (finished, running), as follows:
 624       (UserRPC, None) indicating the first RPC found that is finished;
 625       (None, UserRPC) indicating the first RPC found that is running;
 626       (None, None) indicating no RPCs are finished or running.
 627     """
 628     rpc = None
 629     for rpc in rpcs:
 630       assert isinstance(rpc, cls), repr(rpc)
 631       state = rpc.__rpc.state
 632       if state == apiproxy_rpc.RPC.FINISHING:
 633         rpc.__call_user_callback()
 634         return rpc, None
 635       assert state != apiproxy_rpc.RPC.IDLE, repr(rpc)
 636     return None, rpc
 637
 638   @classmethod
 639   def wait_any(cls, rpcs):
 640     """Wait until an RPC is finished.
 641
 642     Args:
 643       rpcs: Iterable collection of UserRPC instances.
 644
 645     Returns:
 646       A UserRPC instance, indicating the first RPC among the given
 647       RPCs that finished; or None, indicating that either an RPC not
 648       among the given RPCs finished in the mean time, or the iterable
 649       is empty.
 650
 651     NOTES:
 652
 653     (1) Repeatedly calling wait_any() with the same arguments will not
 654         make progress; it will keep returning the same RPC (the one
 655         that finished first).  The callback, however, will only be
 656         called the first time the RPC finishes (which may be here or
 657         in the wait() method).
 658
 659     (2) It may return before any of the given RPCs finishes, if
 660         another pending RPC exists that is not included in the rpcs
 661         argument.  In this case the other RPC's callback will *not*
 662         be called.  The motivation for this feature is that wait_any()
 663         may be used as a low-level building block for a variety of
 664         high-level constructs, some of which prefer to block for the
 665         minimal amount of time without busy-waiting.
 666     """
 667     assert iter(rpcs) is not rpcs, 'rpcs must be a collection, not an iterator'
 668     finished, running = cls.__check_one(rpcs)
 669     if finished is not None:
 670       return finished
 671     if running is None:
 672       return None
 673     try:
 674       cls.__local.may_interrupt_wait = True
 675       try:
 676         running.__rpc.Wait()
 677       except apiproxy_errors.InterruptedError, err:
 678
 679
 680
 681
 682
 683         err.rpc._exception = None
 684         err.rpc._traceback = None
 685     finally:
 686       cls.__local.may_interrupt_wait = False
 687     finished, runnning = cls.__check_one(rpcs)
 688     return finished
 689
 690   @classmethod
 691   def wait_all(cls, rpcs):
 692     """Wait until all given RPCs are finished.
 693
 694     This is a thin wrapper around wait_any() that loops until all
 695     given RPCs have finished.
 696
 697     Args:
 698       rpcs: Iterable collection of UserRPC instances.
 699
 700     Returns:
 701       None.
 702     """
 703     rpcs = set(rpcs)
 704     while rpcs:
 705       finished = cls.wait_any(rpcs)
 706       if finished is not None:
 707         rpcs.remove(finished)
 708
 709
 710
 711
 712 def GetDefaultAPIProxy():
 713   try:
 714
 715
 716
 717
 718
 719
 720     runtime = __import__('google.appengine.runtime', globals(), locals(),
 721                          ['apiproxy'])
 722     return APIProxyStubMap(runtime.apiproxy)
 723   except (AttributeError, ImportError):
 724     return APIProxyStubMap()
 725
 726
 727
 728
 729 apiproxy = GetDefaultAPIProxy()