使用memory_profiler来分析Flask应用程序中的行

Question

更新：在memory_profiler版本0.53及更高版本中，可以使用@profile装饰任意数量的路径。早期版本只允许装饰一条路线。以下问题仅适用于版本<= 0.52的那些版本的memory_profiler

使用普通的@profile装饰器不适用于两个或更多Flask路线。如何在两个或更多Flask路线中获得逐行内存使用情况分析？

我想简介/ route_one和/ route_two：

from functools import wraps

from memory_profiler import profile

@app.route("/route_one", methods=["GET"])
@profile
def route_one():
    api_live = ping_api()
    if not api_live:
        return make_response('', 503)
    return make_response('', 200)

@app.route("/route_two", methods=["GET"])
@profile
def route_two():
    api_live = ping_api()
    if not api_live:
        return make_response('', 503)
    return make_response('', 200)

当我用上面装饰的路线启动我的烧瓶应用程序时，我收到此错误：

AssertionError: View function mapping is overwriting an existing endpoint function: wrapper

Answer 1

profile装饰器不使用functools.wraps，因此它不保留函数名称。 app.route在执行时将这两个函数视为命名为'profile'。由于Flask不允许多个路由使用相同的名称，因此会引发错误。

在memory_profiler修复此问题之前，您可以通过在要应用@profile装饰器时临时指定端点名称来解决此问题。

@app.route('/one', endpoint='one')
@profile
@login_required
def one():
    return 'one'

@profile应该高于你想要包含在你的分析中的任何装饰器，并且@route应该高于@profile，以便它路由到视图的配置文件版本。

Answer 2

memory_profiler已更新，包含在Flask路线上使用@profile的代码。 memory_profiler版本> = 0.53不会有这个问题。有关更多信息，请参阅this GitHub issue。

---

错误告诉我们在我们尝试映射/装饰的两个路由上使用相同的函数包装器。解决这个问题的方法是使用@wraps。如下所述：What does functools.wraps do? @wraps将内部函数的名称和文档字符串复制到外部包装函数。因此，如果我们使用@wraps，我们可以避免上述错误。

但我们需要在装饰器定义中使用@wraps。我们的配置文件装饰器在memory_profiler库中定义，因此我们需要重新编写该函数以包含@wraps。 memory_profiler profiler函数在这里是https://github.com/pythonprofilers/memory_profiler/blob/master/memory_profiler.py，我们将在其下使用@wraps的修改版本。

在烧瓶应用程序中使用以下代码使用@my_profiler装饰您的路线

from functools import wraps

import memory_profiler
try:
    import tracemalloc
    has_tracemalloc = True
except ImportError:
    has_tracemalloc = False


def my_profiler(func=None, stream=None, precision=1, backend='psutil'):
    """
    Decorator that will run the function and print a line-by-line profile
    """
    backend = memory_profiler.choose_backend(backend)
    if backend == 'tracemalloc' and has_tracemalloc:
        if not tracemalloc.is_tracing():
            tracemalloc.start()
    if func is not None:
        @wraps(func)
        def wrapper(*args, **kwargs):
            prof = memory_profiler.LineProfiler(backend=backend)
            val = prof(func)(*args, **kwargs)
            memory_profiler.show_results(prof, stream=stream,
                                         precision=precision)
            return val

        return wrapper
    else:
        def inner_wrapper(f):
            return profile(f, stream=stream, precision=precision,
                           backend=backend)

        return inner_wrapper

我们现在可以使用我们的固定分析器

@app.route("/route_one", methods=["GET"])
@my_profiler
def route_one():
    api_live = ping_api()
    if not api_live:
        return make_response('', 503)
    return make_response('', 200)

@app.route("/route_two", methods=["GET"])
@my_profiler
def route_two():
    api_live = ping_api()
    if not api_live:
        return make_response('', 503)
    return make_response('', 200)