Pymongo Tab for pyramid debugtoolbar

__init__.py

from pyramid.authentication import AuthTktAuthenticationPolicy
from pyramid.authorization import ACLAuthorizationPolicy
from pyramid.config import Configurator
from pyramid.csrf import CookieCSRFStoragePolicy
from pyramid.security import unauthenticated_userid
from urllib.parse import urlparse
from pymongo import MongoClient
from flex.mongo_toolbar import MongoToolbar, DebugMongo

def main(global_config, **settings):

    db_url = urlparse(settings['mongo_uri'])
    config = Configurator(settings=settings)
    config.add_static_view('static', 'static', cache_max_age=3600)
    config.add_route('home', '/')
   
    debug = 'pyramid_debugtoolbar' in settings.get('pyramid.includes')

    def get_db():
        return MongoClient(db_url.geturl(),
                           serverSelectionTimeoutMS=10000,
                           connectTimeoutMS=3000,
                           socketTimeoutMS=10000,
                           maxPoolSize=200,
                           maxIdleTimeMs=30000,
                           appname='myapp')

    def add_query_log(request):
        return []

    def add_db_conn(request):
        conn = get_db()

        def conn_close(request):
            conn.close()
        db = conn[settings['mongo_db']]
        request.add_finished_callback(conn_close)
        if debug:
            return DebugMongo(db, request)
        return db


    def get_user(request):
        userid = unauthenticated_userid(request)
        if userid is not None:
            return request.db['users'].find_one({'username': request.authenticated_userid})
          
    auth_policy = AuthTktAuthenticationPolicy(settings['auth_ticket_key'], callback=auth_callback, hashalg='sha512')
    config.set_authentication_policy(auth_policy)
    config.set_authorization_policy(ACLAuthorizationPolicy())
    config.set_csrf_storage_policy(CookieCSRFStoragePolicy())
      
    config.add_request_method(get_user, 'user', reify=True)
    config.add_request_method(add_query_log, 'query_log', reify=True)
    config.add_request_method(add_db_conn, 'db', reify=True)
    config.scan()

    return config.make_wsgi_app()

development.ini

[app:main]
use = egg:myapp

mongo_uri = mongodb://user:password@host:port/auth_db
mongo_db = database_name
pyramid.includes =
    pyramid_mako
    pyramid_debugtoolbar
debugtoolbar.includes =
    myapp:MongoToolbar
debugtoolbar.extra_panels =
    myapp:MongoToolbar

mongo.mako

Queries Performed:

<table class="pDebugSortable table table-striped table-condensed tablesorter tablesorter-bootstrap table-bordered">
    <thead>
        <tr>
            % for h in 'Seq','Duration', 'Collection', 'Operation', 'Args', 'Kwargs':
                <th>${h}</th>
            % endfor
        </tr>
    </thead>
    <tbody>
        % for idx, q in enumerate(request.query_log):
            <tr>
                <td>${idx}</td>
                <td>${round(q['duration'].total_seconds()*1000, 2)} ms</td>
                <td>${q['collection']}</td>
                <td>${q['op']}</td>
                <td>${q['args']}</td>
                <td>${q['kwargs']}</td>
            </tr>
        % endfor
    </tbody>
</table>

mongo_toolbar.py

from datetime import datetime

from pyramid_debugtoolbar.panels import DebugPanel


class DebugMongo:
    def __init__(self, conn, request):
        self.conn = conn
        self.request = request

    def _get_conn(self, item):
        coll = self.conn[item]

        def coll_attr(attr):
            target_attr = getattr(coll, attr)
            if hasattr(target_attr, '__call__'):
                # we need to wrap this callable
                def wrapper(*args, **kwargs):
                    start = datetime.now()
                    response = target_attr(*args, **kwargs)
                    duration = datetime.now() - start
                    self.request.query_log.append({
                        'duration': duration,
                        'collection': coll.name,
                        'op': target_attr.__name__,
                        'args': args,
                        'kwargs': kwargs
                    })
                    return response

                return wrapper
            else:
                return target_attr

        class DebugCollection:
            def __init__(self, coll):
                self.coll = coll

            def __getattr__(self, item):
                return coll_attr(item)

        return DebugCollection(coll)

    def __getattr__(self, item):
        return self._get_conn(item)

    def __getitem__(self, item):
        return self._get_conn(item)


class MongoToolbar(DebugPanel):
    """
    MongoDB Debugtoolbar Panel
    """
    name = 'mongodb_panel'
    is_active = True
    has_content = True
    title = 'MongoDB'
    nav_title = 'MongoDB'
    template = 'templates/debugtoolbar/mongo.mako'

    def __init__(self, request):
        super().__init__(request)
        self.data = {'request_path': request}
        self.request = request

    def process_response(self, response):
        self.data['request'] = self.request