Python flask_sqlalchemy 模块，get_debug_queries() 实例源码

def after_request(response):
    for query in get_debug_queries():
        if query.duration >= current_app.config['CIRCULATE_SLOW_DB_QUERY_TIME']:
            current_app.logger.warning(
                'Slow query: %s\nParameters: %s\nDuration: %fs\nContext: %s\n'
                % (query.statement, query.parameters, query.duration,
                   query.context))
    return response

def make_shell_context():
    from pprint import pprint

    from flask_sqlalchemy import get_debug_queries
    from sqlalchemy import desc, func

    from scrobbler.models import Album, Artist, NowPlaying, Scrobble, Session, Token, User

    return dict(
        app=app, db=db, pprint=pprint, gq=get_debug_queries, func=func, desc=desc,
        User=User, Session=Session, Token=Token,
        Scrobble=Scrobble, NowPlaying=NowPlaying,
        Artist=Artist, Album=Album,
    )

def after_request(response):
    for query in get_debug_queries():
        if query.duration >= app.config['DATABASE_QUERY_TIMEOUT']:
            app.logger.warn(
                ('\nContext:{}\nSLOW QUERY: {}\nParameters: {}\n'
                 'Duration: {}\n').format(query.context, query.statement,
                                          query.parameters, query.duration))
    return response

def after_request(response):
    for query in get_debug_queries():
        if query.duration >= current_app.config['FLASKY_SLOW_DB_QUERY_TIME']:
            current_app.logger.warning(
                'Slow query: %s\nParameters: %s\nDuration: %fs\nContext: %s\n'
                % (query.statement, query.parameters, query.duration,
                   query.context))
    return response

def after_request(response):
    for query in get_debug_queries():
        if query.duration >= current_app.config['FLASKY_SLOW_DB_QUERY_TIME']:
            current_app.logger.warning(
                'Slow query: %s\nParameters: %s\nDuration: %fs\nContext: %s\n'
                % (query.statement, query.parameters, query.duration,
                   query.context))
    return response

def after_request(response):
    for query in get_debug_queries():
        if query.duration >= current_app.config['SLOW_DB_QUERY_TIME']:
            current_app.logger.warning(
                'Slow query: %s\nParameters: %s\nDuration: %fs\nContext: %s\n'
                % (query.statement, query.parameters, query.duration,
                   query.context))
    return response

def after_request(response):
    for query in get_debug_queries():
        if query.duration >= current_app.config['FLASKY_SLOW_DB_QUERY_TIME']:
            current_app.logger.warning(
                'Slow query: %s\nParameters: %s\nDuration: %fs\nContext: %s\n'
                % (query.statement, query.parameters, query.duration,
                   query.context))
    return response

def configure_request_filter_handlers(app):
    """
    ???????
    :param app:  app??
    :return:
    """

    @app.before_request
    def before_request():
        print('before request handler')
        update_locale(app)
        login_manager.reload_user()
        # your before request code...

    @app.after_request
    def after_request(response):
        print('after request handler')
        # your after request code...
        # ?????????
        for query in get_debug_queries():
            if query.duration > app.config['DATABASE_QUERY_TIMEOUT']:
                app.logger.warn(
                    ('Context: {}\nSLOW QUERY: {}\nParameters: {}\n'
                     'Duration: {}\n').format(query.context, query.statement, query.parameters, query.duration)
                )

        return response

    @app.teardown_request
    def teardown_request(response):
        db.session.close()  # ??db session
        # your teardown request code...
        return response

def debug_teardown_request(exception):
    """Log slow responses and queries."""

    url = request.url

    response_time = time.time() - g.start

    if response_time >= 0.1:
        msg = '{url} - slow response ({response_time:.2f} seconds)'
        print(colored(msg.format(url=url, response_time=response_time), 'yellow'))

    total_query_time = 0

    queries = defaultdict(lambda: [0, 0])

    for q in get_debug_queries():
        queries[q.statement][0] += 1  # number of time query run
        queries[q.statement][1] += q.duration  # total time for query
        total_query_time += q.duration  # total time for all queries

    # Queries taking longer than 0.1 seconds (slowest first)
    queries = [(k, v[0], v[1]) for k, v in queries.items()]
    queries = [x for x in queries if x[2] >= 0.1]
    queries.sort(key=lambda x: x[1], reverse=True)

    for query, times_run, query_time in queries:
        if times_run > 1:
            message = '{url} - slow query ({query_time:.2f} seconds over {times_run} calls) - {query}'
        else:
            message = '{url} - slow query ({query_time:.2f} seconds) - {query}'

        print(colored(message.format(url=url, query=query, times_run=times_run, query_time=query_time), 'yellow'))

def after_request(response):
    for query in get_debug_queries():
        if query.duration >= current_app.config['FLASKY2_SLOW_DB_QUERY_TIME']:
            current_app.logger.warning(
                'Slow query: %s\nParameters: %s\nDuration: %fs\nContext: %s\n'
                %(query.statement, query.parameters, query.duration, query.context)
                )
    return response

def after_request(response):
    for query in get_debug_queries():
        if query.duration >= current_app.config['FLASKY_SLOW_DB_QUERY_TIME']:
            current_app.logger.warning(
                'Slow query: %s\nParameters: %s\nDuration: %s\nContext: %s\n'
                %(query.statement, query.parameters, query.duration, 
                    query.context))
    return response