Source code for flask_mongoengine.panels

"""Debug panel views and logic and related mongoDb event listeners."""
__all__ = ["mongo_command_logger", "MongoDebugPanel"]
import logging
import sys
from dataclasses import dataclass
from typing import List, Union

from flask import current_app
from flask_debugtoolbar.panels import DebugPanel
from jinja2 import ChoiceLoader, PackageLoader
from pymongo import monitoring

logger = logging.getLogger("flask_mongoengine")

[docs]@dataclass class RawQueryEvent: # noinspection PyUnresolvedReferences """Responsible for parsing monitoring events to web panel interface. :param _event: Succeeded or Failed event object from pymongo monitoring. :param _start_event: Started event object from pymongo monitoring. :param _is_query_pass: Boolean status of db query reported by pymongo monitoring. """ _event: Union[monitoring.CommandSucceededEvent, monitoring.CommandFailedEvent] _start_event: monitoring.CommandStartedEvent _is_query_pass: bool @property def time(self): """Query execution time.""" return self._event.duration_micros * 0.001 @property def size(self): """Query object size.""" return sys.getsizeof(self.server_response, 0) / 1024 @property def database(self): """Query database target.""" return self._start_event.database_name @property def collection(self): """Query collection target.""" return self.server_command.get(self.command_name) @property def command_name(self): """Query db level operation/command name.""" return self._event.command_name @property def operation_id(self): """MongoDb operation_id used to match 'start' and 'final' monitoring events.""" return self._start_event.operation_id @property def server_command(self): """Raw MongoDb command send to server.""" return self._start_event.command @property def server_response(self): """Raw MongoDb response received from server.""" return self._event.reply if self._is_query_pass else self._event.failure @property def request_status(self): """Query execution status.""" return "Succeed" if self._is_query_pass else "Failed"
[docs]class MongoCommandLogger(monitoring.CommandListener): """Receive point for :class:`~.pymongo.monitoring.CommandListener` events. Count and parse incoming events for display in debug panel. """ def __init__(self): self.total_time: float = 0 self.started_operations_count: int = 0 self.succeeded_operations_count: int = 0 self.failed_operations_count: int = 0 self.queries: List[RawQueryEvent] = [] self.started_events: dict = {}
[docs] def append_raw_query(self, event, request_status): """Pass 'unknown' events to parser and include final result to final list.""" self.total_time += event.duration_micros start_event = self.started_events.pop(event.operation_id, {}) self.queries.append(RawQueryEvent(event, start_event, request_status)) logger.debug(f"Added record to 'Unknown' section: {self.queries[-1]}")
[docs] def failed(self, event): """Receives 'failed' events. Required to track database answer to request.""" logger.debug(f"Received 'Failed' event from driver: {event}") self.failed_operations_count += 1 self.append_raw_query(event, False)
[docs] def reset_tracker(self): """Resets all counters to default, keeping instance itself the same.""" self.__class__.__init__(self)
[docs] def started(self, event): """Receives 'started' events. Required to track original request context.""" logger.debug(f"Received 'Started' event from driver: {event}") self.started_operations_count += 1 self.started_events[event.operation_id] = event
[docs] def succeeded(self, event): """Receives 'succeeded' events. Required to track database answer to request.""" logger.debug(f"Received 'Succeeded' event from driver: {event}") self.succeeded_operations_count += 1 self.append_raw_query(event, True)
mongo_command_logger = MongoCommandLogger()
[docs]def _maybe_patch_jinja_loader(jinja_env): """Extend jinja_env loader with flask_mongoengine templates folder.""" package_loader = PackageLoader("flask_mongoengine", "templates") if not isinstance(jinja_env.loader, ChoiceLoader): jinja_env.loader = ChoiceLoader([jinja_env.loader, package_loader]) elif package_loader not in jinja_env.loader.loaders: jinja_env.loader.loaders += [package_loader]
[docs]class MongoDebugPanel(DebugPanel): """Panel that shows information about MongoDB operations.""" config_error_message = ( "Pymongo monitoring configuration error. mongo_command_logger should be " "registered before database connection." ) name = "MongoDB" has_content = True def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) _maybe_patch_jinja_loader(self.jinja_env) @property def _context(self) -> dict: """Context for rendering, as property for easy testing.""" return { "queries": mongo_command_logger.queries, "slow_query_limit": current_app.config.get( "MONGO_DEBUG_PANEL_SLOW_QUERY_LIMIT", 100 ), } @property def is_properly_configured(self) -> bool: """Checks that all required watchers registered before Flask application init.""" # noinspection PyProtectedMember if mongo_command_logger not in monitoring._LISTENERS.command_listeners: logger.error(self.config_error_message) return False return True
[docs] def process_request(self, request): """Resets logger stats between each request.""" mongo_command_logger.reset_tracker()
[docs] def nav_title(self) -> str: """Debug toolbar in the bottom right corner.""" return
[docs] def nav_subtitle(self) -> str: """Count operations total time.""" if not self.is_properly_configured: self.has_content = False return self.config_error_message return ( f"{mongo_command_logger.started_operations_count} operations, " f"in {mongo_command_logger.total_time * 0.001: .2f}ms" )
[docs] def title(self) -> str: """Title for 'opened' debug panel window.""" return "MongoDB Operations"
[docs] def url(self) -> str: """Placeholder for internal URLs.""" return ""
[docs] def content(self): """Gathers all template required variables in one dict.""" return self.render("panels/mongo-panel.html", self._context)