python - 使用 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)

当我使用上述装饰路线启动我的 flask 应用程序时，出现此错误:

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

Answer 1

memory_profiler 已更新为包含在 Flask 路由上使用 @profile 的代码。 memory_profiler version >= 0.53 不会有这个问题。参见 this GitHub issue获取更多信息。

---

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

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

在您的 Flask 应用程序中使用以下代码用@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)