Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Django database query alerting middleware
class DatabaseQueryAlertingMiddleware:
"""
Middleware that alerts when a threshold for maximum number of database queries or total query time
is exceeded when loading views in development mode.
"""
max_queries_threshold = 30
max_time_threshold = 5 # seconds
def __init__(self, get_response):
if not settings.DEBUG:
raise MiddlewareNotUsed()
self.get_response = get_response
def __call__(self, request):
if not self._should_process_request(request):
return self.get_response(request)
response = self.get_response(request)
total_queries, total_time = self._get_total_queries_and_total_query_time()
if total_queries > self.max_queries_threshold:
msg = 'Number of queries for this view ({0} queries) exceeded max threshold of {1} queries'.format(
total_queries, self.max_queries_threshold,
)
messages.warning(request, msg)
logger.debug(msg)
if total_time > self.max_time_threshold:
msg = 'Total query time for this view ({0:.2f} seconds) exceeded max threshold of {1} seconds'.format(
total_time, self.max_time_threshold
)
messages.warning(request, msg)
logger.debug(msg)
return response
def _should_process_request(self, request) -> bool:
return not (
request.is_ajax() or
request.path_info.startswith('/favicon.ico') or
request.path_info.startswith(settings.STATIC_URL) or
request.path_info.startswith(settings.MEDIA_URL)
)
def _get_total_queries_and_total_query_time(self) -> Tuple[int, float]:
total_queries = len(connection.queries)
total_time = sum(float(query['time']) for query in connection.queries)
return total_queries, total_time
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.