def __init__(self):
"""
Middleware init is called once per server on startup - do the heavy
lifting here.
"""
# If disabled or not enabled raise MiddleWareNotUsed so django
# processes next middleware.
self.ENABLED = getattr(settings, 'BANS_ENABLED', False)
self.DEBUG = getattr(settings, 'BANS_DEBUG', False)
if not self.ENABLED:
raise MiddlewareNotUsed("bans are not enabled via settings.py")
if self.DEBUG:
print "Bans status = enabled"
# Populate various 'banish' buckets
for ban in Ban.objects.all():
if self.DEBUG:
print ban
cache.add('BAN:'+ban.address, '1', None)
python类MiddlewareNotUsed()的实例源码
def __init__(self):
self.ENABLED = getattr(settings,'SCOOKIE_ENABLED', False)
if not self.ENABLED:
raise MiddlewareNotUsed("Security Cookie not enabled in settings.py")
def load_middleware(self):
"""
Populate middleware lists from settings.MIDDLEWARE_CLASSES.
Must be called after the environment is fixed (see __call__ in subclasses).
"""
self._view_middleware = []
self._template_response_middleware = []
self._response_middleware = []
self._exception_middleware = []
request_middleware = []
for middleware_path in settings.MIDDLEWARE_CLASSES:
mw_class = import_by_path(middleware_path)
try:
mw_instance = mw_class()
except MiddlewareNotUsed:
continue
if hasattr(mw_instance, 'process_request'):
request_middleware.append(mw_instance.process_request)
if hasattr(mw_instance, 'process_view'):
self._view_middleware.append(mw_instance.process_view)
if hasattr(mw_instance, 'process_template_response'):
self._template_response_middleware.insert(0, mw_instance.process_template_response)
if hasattr(mw_instance, 'process_response'):
self._response_middleware.insert(0, mw_instance.process_response)
if hasattr(mw_instance, 'process_exception'):
self._exception_middleware.insert(0, mw_instance.process_exception)
# We only assign to this when initialization is complete as it is used
# as a flag for initialization being complete.
self._request_middleware = request_middleware
def __init__(self, get_response):
self.get_response = get_response
if not redisAvailable:
raise MiddlewareNotUsed()
self.r = redis.Redis(
host='127.0.0.1',
port=6379)
try:
self.r.ping()
except redis.ConnectionError:
raise MiddlewareNotUsed()
def test_disabled(self):
mock_get_response = object()
with self.assertRaises(MiddlewareNotUsed):
Middleware(mock_get_response)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if not RequestQueryCountConfig.enabled():
raise MiddlewareNotUsed()
self.force_debug_cursor = False
self.initial_queries = 0
self.final_queries = None
self.connection = connections[DEFAULT_DB_ALIAS]