salmon/routing.py

   1 """
   2 The meat of Salmon, doing all the work that actually takes an email and makes
   3 sure that your code gets it.
   4
   5 The three most important parts for a programmer are the Router variable, the
   6 StateStorage base class, and the @route, @route_like, and @stateless decorators.
   7
   8 The salmon.routing.Router variable (it's not a class, just named like one) is
   9 how the whole system gets to the Router.  It is an instance of RoutingBase and
  10 there's usually only one.
  11
  12 The salmon.routing.StateStorage is what you need to implement if you want
  13 Salmon to store the state in a different way.  By default the
  14 salmon.routing.Router object just uses a default MemoryStorage to do its job.
  15 If you want to use a custom storage, then in your boot modiule you would set
  16 salmon.routing.Router.STATE_STORE to what you want to use.
  17
  18 Finally, when you write a state handler, it has functions that act as state
  19 functions for dealing with each state.  To tell the Router what function should
  20 handle what email you use a @route decorator.  To tell the Route that one
  21 function routes the same as another use @route_like.  In the case where a state
  22 function should run on every matching email, just use the @stateless decorator
  23 after a @route or @route_like.
  24
  25 If at any time you need to debug your routing setup just use the salmon routes
  26 command.
  27
  28 Routing Control
  29 ===============
  30
  31 To control routing there are a set of decorators that you apply to your
  32 functions.
  33
  34 * @route -- The main routing function that determines what addresses you are
  35   interested in.
  36 * @route_like -- Says that this function routes like another one.
  37 * @stateless -- Indicates this function always runs on each route encountered, and
  38   no state is maintained.
  39 * @locking -- Use this if you want this handler to be run one call at a time.
  40 * @state_key_generator -- Used on a function that knows how to make your state
  41   keys for the module, for example if module_name + message.To is needed to maintain
  42   state.
  43
  44 It's best to put @route or @route_like as the first decorator, then the others
  45 after that.
  46
  47 The @state_key_generator is different since it's not intended to go on a handler
  48 but instead on a simple function, so it shouldn't be combined with the others.
  49 """
  50 from functools import wraps
  51 from importlib import reload
  52 import logging
  53 import re
  54 import shelve
  55 import sys
  56 import threading
  57 import warnings
  58
  59 ROUTE_FIRST_STATE = 'START'
  60 LOG = logging.getLogger("routing")
  61 SALMON_SETTINGS_VARIABLE_NAME = "_salmon_settings"
  62 _DEFAULT_VALUE = object()
  63
  64
  65 def DEFAULT_STATE_KEY(mod, msg):
  66     return mod
  67
  68
  69 class StateStorage:
  70     """
  71     The base storage class you need to implement for a custom storage
  72     system.
  73     """
  74     def get(self, key, sender):
  75         """
  76         You must implement this so that it returns a single string
  77         of either the state for this combination of arguments, OR
  78         the ROUTE_FIRST_STATE setting.
  79         """
  80         raise NotImplementedError("You have to implement a StateStorage.get.")
  81
  82     def set(self, key, sender, state):
  83         """
  84         Set should take the given parameters and consistently set the state for
  85         that combination such that when StateStorage.get is called it gives back
  86         the same setting.
  87         """
  88         raise NotImplementedError("You have to implement a StateStorage.set.")
  89
  90     def clear(self):
  91         """
  92         This should clear ALL states, it is only used in unit testing, so you
  93         can have it raise an exception if you want to make this safer.
  94         """
  95         raise NotImplementedError("You have to implement a StateStorage.clear for unit testing to work.")
  96
  97
  98 class MemoryStorage(StateStorage):
  99     """
 100     The default simplified storage for the Router to hold the states.  This
 101     should only be used in testing, as you'll lose all your contacts and their
 102     states if your server shuts down.
 103     """
 104     def __init__(self):
 105         self.states = {}
 106         self.lock = threading.RLock()
 107
 108     def get(self, key, sender):
 109         key = self.key(key, sender)
 110         try:
 111             return self.states[key]
 112         except KeyError:
 113             return ROUTE_FIRST_STATE
 114
 115     def set(self, key, sender, state):
 116         with self.lock:
 117             key = self.key(key, sender)
 118             if state == ROUTE_FIRST_STATE:
 119                 try:
 120                     del self.states[key]
 121                 except KeyError:
 122                     pass
 123             else:
 124                 self.states[key] = state
 125
 126     def key(self, key, sender):
 127         return repr([key, sender])
 128
 129     def clear(self):
 130         self.states.clear()
 131
 132
 133 class ShelveStorage(MemoryStorage):
 134     """
 135     Uses Python's shelve to store the state of the Routers to disk rather than
 136     in memory like with MemoryStorage.  This will get you going on a small
 137     install if you need to persist your states (most likely), but if you
 138     have a database, you'll need to write your own StateStorage that
 139     uses your ORM or database to store.  Consider this an example.
 140
 141     NOTE: Because of shelve limitations you can only use ASCII encoded keys.
 142     """
 143     def __init__(self, database_path):
 144         """Database path depends on the backing library use by Python's shelve."""
 145         super().__init__()
 146         self.database_path = database_path
 147
 148     def get(self, key, sender):
 149         """
 150         This will lock the internal thread lock, and then retrieve from the
 151         shelf whatever key you request.  If the key is not found then it
 152         will set (atomically) to ROUTE_FIRST_STATE.
 153         """
 154         with self.lock:
 155             self.states = shelve.open(self.database_path)
 156             value = super().get(key.encode('ascii'), sender)
 157             self.states.close()
 158             return value
 159
 160     def set(self, key, sender, state):
 161         """
 162         Acquires the self.lock and then sets the requested state in the shelf.
 163         """
 164         with self.lock:
 165             self.states = shelve.open(self.database_path)
 166             super().set(key.encode('ascii'), sender, state)
 167             self.states.close()
 168
 169     def clear(self):
 170         """
 171         Primarily used in the debugging/unit testing process to make sure the
 172         states are clear.  In production this could be a bad thing.
 173         """
 174         with self.lock:
 175             self.states = shelve.open(self.database_path)
 176             super().clear()
 177             self.states.close()
 178
 179
 180 class RoutingBase:
 181     """
 182     The self is a globally accessible class that is actually more like a
 183     glorified module.  It is used mostly internally by the salmon.routing
 184     decorators (route, route_like, stateless) to control the routing
 185     mechanism.
 186
 187     It keeps track of the registered routes, their attached functions, the
 188     order that these routes should be evaluated, any default routing captures,
 189     and uses the MemoryStorage by default to keep track of the states.
 190
 191     You can change the storage to another implementation by simple setting:
 192
 193         Router.STATE_STORE = OtherStorage()
 194
 195     in your settings module.
 196
 197     RoutingBase assumes that both your STATE_STORE and handlers are
 198     thread-safe. For handlers that cannot be made thread-safe, use @locking and
 199     RoutingBase will use locks to make sure that handler is only called one
 200     call at a time. Please note that this will have a negative impact on
 201     performance.
 202
 203     NOTE: See @state_key_generator for a way to change what the key is to
 204     STATE_STORE for different state control options.
 205     """
 206
 207     def __init__(self):
 208         self.REGISTERED = {}
 209         self.ORDER = []
 210         self.DEFAULT_CAPTURES = {}
 211         self.STATE_STORE = MemoryStorage()
 212         self.HANDLERS = {}
 213         self.RELOAD = False
 214         self.LOG_EXCEPTIONS = True
 215         self.UNDELIVERABLE_QUEUE = None
 216         self.lock = threading.RLock()
 217         self.call_lock = threading.RLock()
 218
 219     def register_route(self, format, func):
 220         """
 221         Registers this function func into the routes mapping based on the
 222         format given.  Format should be a regex string ready to be handed to
 223         re.compile.
 224         """
 225         with self.lock:
 226             if format in self.REGISTERED:
 227                 self.REGISTERED[format][1].append(func)
 228             else:
 229                 self.ORDER.append(format)
 230                 self.REGISTERED[format] = (re.compile(format, re.IGNORECASE), [func])
 231
 232     def match(self, address):
 233         """
 234         This is a generator that goes through all the routes and
 235         yields each match it finds.  It expects you to give it a
 236         blah@blah.com address, NOT "Joe Blow" <blah@blah.com>.
 237         """
 238         for format in self.ORDER:
 239             regex, functions = self.REGISTERED[format]
 240             match = regex.match(address)
 241             if match:
 242                 yield functions, match.groupdict()
 243
 244     def defaults(self, **captures):
 245         """
 246         Updates the defaults for routing captures with the given settings.
 247
 248         You use this in your handlers or your settings module to set
 249         common regular expressions you'll have in your @route decorators.
 250         This saves you typing, but also makes it easy to reconfigure later.
 251
 252         For example, many times you'll have a single host="..." regex
 253         for all your application's routes.  Put this in your settings.py
 254         file using route_defaults={'host': '...'} and you're done.
 255         """
 256         with self.lock:
 257             self.DEFAULT_CAPTURES.update(captures)
 258
 259     def get_state(self, module_name, message):
 260         """Returns the state that this module is in for the given message (using its from)."""
 261         key = self.state_key(module_name, message)
 262         return self.STATE_STORE.get(key, message.From)
 263
 264     def in_state(self, func, message):
 265         """
 266         Determines if this function is in the state for the to/from in the
 267         message.  Doesn't apply to @stateless state handlers.
 268         """
 269         state = self.get_state(func.__module__, message)
 270         return state and state == func.__name__
 271
 272     def in_error(self, func, message):
 273         """
 274         Determines if the this function is in the 'ERROR' state,
 275         which is a special state that self puts handlers in that throw
 276         an exception.
 277         """
 278         state = self.get_state(func.__module__, message)
 279         return state and state == 'ERROR'
 280
 281     def state_key(self, module_name, message):
 282         """
 283         Given a module_name we need to get a state key for, and a
 284         message that has information to make the key, this function
 285         calls any registered @state_key_generator and returns that
 286         as the key.  If none is given then it just returns module_name
 287         as the key.
 288         """
 289         key_func = self.HANDLERS.get(module_name, DEFAULT_STATE_KEY)
 290         return key_func(module_name, message)
 291
 292     def set_state(self, module_name, message, state):
 293         """
 294         Sets the state of the given module (a string) according to the message to the requested
 295         state (a string).  This is also how you can force another FSM to a required state.
 296         """
 297         key = self.state_key(module_name, message)
 298         self.STATE_STORE.set(key, message.From, state)
 299
 300     def _collect_matches(self, message):
 301         in_state_found = False
 302
 303         for functions, matchkw in self.match(message.To):
 304             for func in functions:
 305                 if salmon_setting(func, 'stateless'):
 306                     yield func, matchkw
 307                 elif not in_state_found and self.in_state(func, message):
 308                     in_state_found = True
 309                     yield func, matchkw
 310
 311     def _enqueue_undeliverable(self, message):
 312         if self.UNDELIVERABLE_QUEUE is not None:
 313             LOG.debug("Message to %r from %r undeliverable, putting in undeliverable queue (# of recipients: %d).",
 314                       message.To, message.From, len(message.To))
 315             self.UNDELIVERABLE_QUEUE.push(message)
 316         else:
 317             LOG.debug("Message to %r from %r didn't match any handlers. (# recipients: %d)",
 318                       message.To, message.From, len(message.To))
 319
 320     def deliver(self, message):
 321         """
 322         The meat of the whole Salmon operation, this method takes all the
 323         arguments given, and then goes through the routing listing to figure out
 324         which state handlers should get the gear.  The routing operates on a
 325         simple set of rules:
 326
 327             1) Match on all functions that match the given To in their
 328             registered format pattern.
 329             2) Call all @stateless state handlers functions.
 330             3) Call the first method that's in the right state for the From/To.
 331
 332         It will log which handlers are being run, and you can use the 'salmon route'
 333         command to inspect and debug routing problems.
 334
 335         If you have an ERROR state function, then when your state blows up, it will
 336         transition to ERROR state and call your function right away.  It will then
 337         stay in the ERROR state unless you return a different one.
 338         """
 339         if self.RELOAD:
 340             self.reload()
 341
 342         called_count = 0
 343
 344         for func, matchkw in self._collect_matches(message):
 345             LOG.debug("Matched %r against %s.", message.To, func.__name__)
 346
 347             if salmon_setting(func, 'locking'):
 348                 with self.call_lock:
 349                     self.call_safely(func, message, matchkw)
 350             else:
 351                 self.call_safely(func, message,  matchkw)
 352
 353             called_count += 1
 354
 355         if called_count == 0:
 356             self._enqueue_undeliverable(message)
 357
 358     def call_safely(self, func, message, kwargs):
 359         """
 360         Used by self to call a function and log exceptions rather than
 361         explode and crash.
 362         """
 363         from salmon.server import SMTPError
 364
 365         try:
 366             func(message, **kwargs)
 367             LOG.debug("Message to %s was handled by %s.%s",
 368                       message.To, func.__module__, func.__name__)
 369         except SMTPError:
 370             raise
 371         except Exception:
 372             self.set_state(func.__module__, message, 'ERROR')
 373
 374             if self.UNDELIVERABLE_QUEUE is not None:
 375                 self.UNDELIVERABLE_QUEUE.push(message)
 376
 377             if self.LOG_EXCEPTIONS:
 378                 LOG.exception("!!! ERROR handling %s.%s", func.__module__, func.__name__)
 379             else:
 380                 raise
 381
 382     def clear_states(self):
 383         """Clears out the states for unit testing."""
 384         with self.lock:
 385             self.STATE_STORE.clear()
 386
 387     def clear_routes(self):
 388         """Clears out the routes for unit testing and reloading."""
 389         with self.lock:
 390             self.REGISTERED.clear()
 391             del self.ORDER[:]
 392
 393     def load(self, handlers):
 394         """
 395         Loads the listed handlers making them available for processing.
 396         This is safe to call multiple times and to duplicate handlers
 397         listed.
 398         """
 399         with self.lock:
 400             for module in handlers:
 401                 try:
 402                     __import__(module, globals(), locals())
 403
 404                     if module not in self.HANDLERS:
 405                         # they didn't specify a key generator, so use the
 406                         # default one for now
 407                         self.HANDLERS[module] = DEFAULT_STATE_KEY
 408                 except ImportError:
 409                     if self.LOG_EXCEPTIONS:
 410                         LOG.exception("ERROR IMPORTING %r MODULE:" % module)
 411                     else:
 412                         raise
 413
 414     def reload(self):
 415         """
 416         Performs a reload of all the handlers and clears out all routes,
 417         but doesn't touch the internal state.
 418         """
 419         with self.lock:
 420             self.clear_routes()
 421             for module in list(sys.modules.keys()):
 422                 if module in self.HANDLERS:
 423                     try:
 424                         reload(sys.modules[module])
 425                     except (TypeError, NameError, ImportError):
 426                         if self.LOG_EXCEPTIONS:
 427                             LOG.exception("ERROR RELOADING %r MODULE:" % module)
 428                         else:
 429                             raise
 430
 431
 432 Router = RoutingBase()
 433
 434
 435 class route:
 436     """
 437     The @route decorator is attached to state handlers to configure them in the
 438     Router so they handle messages for them.  The way this works is, rather than
 439     just routing working on only messages being sent to a state handler, it also uses
 440     the state of the sender.  It's like having routing in a web application use
 441     both the URL and an internal state setting to determine which method to run.
 442
 443     However, if you'd rather than this state handler process all messages
 444     matching the @route then tag it @stateless.  This will run the handler
 445     no matter what and not change the user's state.
 446     """
 447
 448     def __init__(self, format, **captures):
 449         r"""
 450         Sets up the pattern used for the Router configuration.  The format
 451         parameter is a simple pattern of words, captures, and anything you
 452         want to ignore.  The captures parameter is a mapping of the words in
 453         the format to regex that get put into the format.  When the pattern is
 454         matched, the captures are handed to your state handler as keyword
 455         arguments.
 456
 457         For example, if you have::
 458
 459             @route("(list_name)-(action)@(host)",
 460                 list_name='[a-z]+',
 461                 action='[a-z]+', host='test\.com')
 462             def STATE(message, list_name=None, action=None, host=None):
 463                 pass
 464
 465         Then this will be translated so that list_name is replaced with [a-z]+,
 466         action with [a-z]+, and host with 'test.com' to produce a regex with the
 467         right format and named captures to that your state handler is called
 468         with the proper keyword parameters.
 469
 470         You should also use the Router.defaults() to set default things like the
 471         host so that you are not putting it into your code.
 472         """
 473         self.captures = Router.DEFAULT_CAPTURES.copy()
 474         self.captures.update(captures)
 475         self.format = self.parse_format(format, self.captures)
 476
 477     def __call__(self, func):
 478         """Returns either a decorator that does a stateless routing or
 479         a normal routing."""
 480         self.setup_accounting(func)
 481
 482         if salmon_setting(func, 'stateless'):
 483             @wraps(func)
 484             def routing_wrapper(message, *args, **kw):
 485                 func(message, *args, **kw)
 486         else:
 487             @wraps(func)
 488             def routing_wrapper(message, *args, **kw):
 489                 next_state = func(message, *args, **kw)
 490
 491                 if next_state:
 492                     Router.set_state(next_state.__module__, message, next_state.__name__)
 493
 494         Router.register_route(self.format, routing_wrapper)
 495         return routing_wrapper
 496
 497     def __get__(self, obj, of_type=None):
 498         """
 499         This is NOT SUPPORTED.  It is here just so that if you try to apply
 500         this decorator to a class's method it will barf on you.
 501         """
 502         raise NotImplementedError("Not supported on methods yet, only module functions.")
 503
 504     def parse_format(self, format, captures):
 505         """Does the grunt work of conversion format+captures into the regex."""
 506         for key in captures:
 507             format = format.replace("(" + key + ")", "(?P<%s>%s)" % (key, captures[key]))
 508         return "^" + format + "$"
 509
 510     def setup_accounting(self, func):
 511         """Sets up an accounting map attached to the func for routing decorators."""
 512         salmon_setting(func, 'format', self.format)
 513         salmon_setting(func, 'captures', self.captures)
 514
 515
 516 def salmon_setting(func, key, value=_DEFAULT_VALUE):
 517     """Get or set a salmon setting on a handler function"""
 518     try:
 519         salmon_settings = getattr(func, SALMON_SETTINGS_VARIABLE_NAME)
 520     except AttributeError:
 521         salmon_settings = {}
 522         setattr(func, SALMON_SETTINGS_VARIABLE_NAME, salmon_settings)
 523     if value is not _DEFAULT_VALUE:
 524         salmon_settings[key] = value
 525     else:
 526         return salmon_settings.get(key)
 527
 528
 529 def has_salmon_settings(func):
 530     return hasattr(func, SALMON_SETTINGS_VARIABLE_NAME)
 531
 532
 533 class route_like(route):
 534     """
 535     Many times you want your state handler to just accept mail like another
 536     handler.  Use this, passing in the other function.  It even works across
 537     modules.
 538     """
 539     def __init__(self, func):
 540         self.format = salmon_setting(func, 'format')
 541         self.captures = salmon_setting(func, 'captures')
 542         if self.format is None or self.captures is None:
 543             raise TypeError("{} is missing a @route".format(func))
 544
 545
 546 def stateless(func):
 547     """
 548     This simple decorator is attached to a handler to indicate to the
 549     Router.deliver() method that it does NOT maintain state or care about it.
 550     This is how you create a handler that processes all messages matching the
 551     given format+captures in a @route.
 552
 553     Another way to think about a @stateless handler is that it is a pass-through
 554     handler that does its processing and then passes the results on to others.
 555
 556     Stateless handlers are NOT guaranteed to run before the handler with state.
 557     """
 558     salmon_setting(func, 'stateless', True)
 559     return func
 560
 561
 562 def nolocking(func):
 563     """
 564     Does nothing, as no locking is the default now
 565     """
 566     warnings.warn("@nolocking is redundant and can be safely removed from your handler %s" % func,
 567                   category=DeprecationWarning, stacklevel=2)
 568     return func
 569
 570
 571 def locking(func):
 572     """
 573     Salmon assumes your handlers are thread-safe, but is not always the case.
 574     Put this decorator on any state functions that are not thread-safe for
 575     whatever reason.
 576     """
 577     salmon_setting(func, 'locking', True)
 578     return func
 579
 580
 581 def state_key_generator(func):
 582     """
 583     Used to indicate that a function in your handlers should be used
 584     to determine what they key is for state storage.  It should be a
 585     function that takes the module_name and message being worked on
 586     and returns a string.
 587     """
 588     Router.HANDLERS[func.__module__] = func
 589     return func