Method calls are imperative: tell an object to do something
window.resize(x=640, y=480)
Pub/Sub events are generally reactive: something happened or will happen
resized(window, x=640, y=480)
>>> from blinker import signal
>>> window_resized = signal('window_resized')
>>> window_resized.send('some window', x=640, y=480)
[]
>>>
>>> def receiver(sender, **kw):
... print "received %r %r" % (sender, kw)
...
>>> window_resized.connect(receiver)
>>> window_resized.send('some window', x=640, y=480)
received 'some window' {'x': 640, 'y': 480}
[(<function receiver at 0x3243b0>, None)]
from blinker import signal
started = signal('request-started')
ended = signal('request-ended')
def my_wsgi_app(environ, start_response):
started.send(app, environ=environ)
response = do_something(...)
ended.send(app, environ=environ, response=response)
return response
from blinker import signal
started = signal('request-started')
ended = signal('request-ended')
from blinker import signal
from sqlalchemy import ConnectionProxy
sql_executed = signal('sql-executed')
class Watcher(ConnectionProxy):
def cursor_execute(self, execute, cursor, statement, parameters,
context, executemany):
if not sql_executed.receivers:
return execute(cursor, statement, parameters, context)
start = time.time()
r = execute(cursor, statement, parameters, context)
sql_executed.send(cursor, statement=statement, parameters=parameters,
elapsed=(time.time() - start))
return r
Extend the gunicorn!
def my_app(environ, start_response):
...
def pre_request(worker, req):
"""Called just before a worker processes the request."""
def post_request(worker, req):
"""Called after a worker processes the request."""
from blinker import signal
request_started = signal('request_started')
current_environ = None
@request_started.connect
def stash_environ(app, environ):
global current_environ
current_environ = environ
def pre_request(worker, req):
signal('pre-request').send(worker, req=req)
def post_request(worker, req):
signal('post-request').send(worker, req=req, environ=current_environ)
[auto-import]
sql_engine = idealist.model.utilities.engine
mail = idealist.mail
traps = idealist.webapp.traps
@config.post_config_load.connect
def autoinstall(config):
requested = config.slice('webapp.traps')
for trap, value in requested.items():
try:
enabler = traps[trap]
except KeyError:
raise LookupError("Unknown trap webapp.traps.%s" % trap)
else:
enabler(value)