python - 有没有一种 Pythonic 方法可以将可选功能与函数的主要目的分离？

Question

语境

假设我有以下 Python 代码:

def example_function(numbers, n_iters):
    sum_all = 0
    for number in numbers:
        for _ in range(n_iters):
            number = halve(number)
        sum_all += number
    return sum_all


ns = [1, 3, 12]
print(example_function(ns, 3))

example_function这里只是简单地遍历 ns 中的每个元素列出并将它们减半 3 次，同时累积结果。运行此脚本的输出很简单:

2.0

因为 1/(2^3)*(1+3+12) = 2。

现在，假设(出于任何原因，可能是调试或日志记录)，我想显示一些关于 example_function 的中间步骤的信息。正在服用。也许我会把这个函数重写成这样的:

def example_function(numbers, n_iters):
    sum_all = 0
    for number in numbers:
        print('Processing number', number)
        for i_iter in range(n_iters):
            number = number/2
            print(number)
        sum_all += number
        print('sum_all:', sum_all)
    return sum_all

现在，当使用与以前相同的参数调用时，会输出以下内容:

Processing number 1
0.5
0.25
0.125
sum_all: 0.125
Processing number 3
1.5
0.75
0.375
sum_all: 0.5
Processing number 12
6.0
3.0
1.5
sum_all: 2.0

这正是我想要的。然而，这有点违背函数应该只做一件事的原则，现在是 example_function 的代码。略长且复杂。对于这样一个简单的函数，这不是问题，但是在我的上下文中，我有相当复杂的函数相互调用，并且打印语句通常涉及比这里显示的更复杂的步骤，导致我的代码的复杂性大大增加(对于一个在我的函数中，与日志记录相关的代码行多于与其实际目的相关的代码行!)。

此外，如果我后来决定不再需要在我的函数中使用任何打印语句，我将不得不通过 example_function并删除所有 print手动声明以及与此功能相关的任何变量，这是一个既乏味又容易出错的过程。

如果我想在函数执行期间总是有打印或不打印的可能性，情况会变得更糟，导致我要么声明两个极其相似的函数(一个带有 print 语句，一个没有)，这对于维护，或定义类似:

def example_function(numbers, n_iters, debug_mode=False):
    sum_all = 0
    for number in numbers:
        if debug_mode:
            print('Processing number', number)
        for i_iter in range(n_iters):
            number = number/2
            if debug_mode:
                print(number)
        sum_all += number
        if debug_mode:
            print('sum_all:', sum_all)
    return sum_all

这会导致函数臃肿且(希望如此)不必要地复杂，即使在我们的 example_function 的简单情况下也是如此。 .

问题

有没有一种 Python 方法可以将打印功能与 example_function 的原始功能“分离”？ ?

更一般地说，是否有一种 Pythonic 方法可以将可选功能与函数的主要目的分离？

到目前为止我已经尝试过:

我目前找到的解决方案是使用回调进行解耦。例如，可以重写 example_function像这样:

def example_function(numbers, n_iters, callback=None):
    sum_all = 0
    for number in numbers:
        for i_iter in range(n_iters):
            number = number/2

            if callback is not None:
                callback(locals())
        sum_all += number
    return sum_all

然后定义一个回调函数来执行我想要的任何打印功能:

def print_callback(locals):
    print(locals['number'])

并调用 example_function像这样:

ns = [1, 3, 12]
example_function(ns, 3, callback=print_callback)

然后输出:

0.5
0.25
0.125
1.5
0.75
0.375
6.0
3.0
1.5
2.0

这成功地将打印功能与 example_function 的基本功能分离。 .但是，这种方法的主要问题是回调函数只能在 example_function 的特定部分运行。 (在这种情况下，在将当前数字减半之后)，并且所有的打印都必须在那里发生。这有时会迫使回调函数的设计非常复杂(并使某些行为无法实现)。

例如，如果想要实现与我在问题的前一部分中所做的完全相同的打印类型(显示正在处理的数字及其相应的减半)，则结果回调将是:

def complicated_callback(locals):
    i_iter = locals['i_iter']
    number = locals['number']
    if i_iter == 0:
        print('Processing number', number*2)
    print(number)
    if i_iter == locals['n_iters']-1:
        print('sum_all:', locals['sum_all']+number)

这导致与以前完全相同的输出:

Processing number 1.0
0.5
0.25
0.125
sum_all: 0.125
Processing number 3.0
1.5
0.75
0.375
sum_all: 0.5
Processing number 12.0
6.0
3.0
1.5
sum_all: 2.0

但是编写、阅读和调试都很痛苦。

Answer 1

如果您需要函数外部的功能来使用函数内部的数据，那么函数内部需要一些消息传递系统来支持这一点。
没有办法解决这个问题。函数中的局部变量与外部完全隔离。

日志模块非常擅长设置消息系统。
它不仅限于打印日志消息 - 使用自定义处理程序，您可以做任何事情。

添加消息系统类似于您的回调示例，除了可以在 example_function 内的任何位置指定处理“回调”(日志处理程序)的位置。
(通过将消息发送到记录器)。发送消息时可以指定日志处理程序所需的任何变量(您仍然可以使用 locals() ，但最好明确
声明你需要的变量)。

一个新的example_function可能看起来像:

import logging

# Helper function
def send_message(logger, level=logging.DEBUG, **kwargs):
  logger.log(level, "", extra=kwargs)

# Your example function with logging information
def example_function(numbers, n_iters):
    logger = logging.getLogger("example_function")
    # If you have a logging system set up, then we don't want the messages sent here to propagate to the root logger
    logger.propagate = False
    sum_all = 0
    for number in numbers:
        send_message(logger, action="processing", number=number)
        for i_iter in range(n_iters):
            number = number/2
            send_message(logger, action="division", i_iter=i_iter, number=number)
        sum_all += number
        send_message(logger, action="sum", sum=sum_all)
    return sum_all

这指定了可以处理消息的三个位置。
就其本身而言，这个 example_function除了 example_function 的功能之外不会做任何事情本身。
它不会打印出任何东西，也不会执行任何其他功能。

为 example_function 添加额外的功能，那么您将需要向记录器添加处理程序。

例如，如果您想对发送的变量进行一些打印(类似于您的 debugging 示例)，那么您定义
自定义处理程序，并将其添加到 example_function记录器:

class ExampleFunctionPrinter(logging.Handler):
    def emit(self, record):
        if record.action == "processing":
          print("Processing number {}".format(record.number))
        elif record.action == "division":
          print(record.number)
        elif record.action == "sum":
          print("sum_all: {}".format(record.sum))

example_function_logger = logging.getLogger("example_function")
example_function_logger.setLevel(logging.DEBUG)
example_function_logger.addHandler(ExampleFunctionPrinter())

如果您想在图表上绘制结果，则只需定义另一个处理程序:

class ExampleFunctionDivisionGrapher(logging.Handler):
    def __init__(self, grapher):
      self.grapher = grapher

    def emit(self, record):
      if record.action == "division":
        self.grapher.plot_point(x=record.i_iter, y=record.number)

example_function_logger = logging.getLogger("example_function")
example_function_logger.setLevel(logging.DEBUG)
example_function_logger.addHandler(
    ExampleFunctionDivisionGrapher(MyFancyGrapherClass())
)

您可以定义和添加您想要的任何处理程序。它们将完全独立于 example_function 的功能。 ,
并且只能使用 example_function 的变量给他们。

虽然日志记录可以用作消息传递系统，但迁移到成熟的消息传递系统可能会更好，例如 PyPubSub ,
这样它就不会干扰您可能正在执行的任何实际日志记录:

from pubsub import pub

# Your example function
def example_function(numbers, n_iters):
    sum_all = 0
    for number in numbers:
        pub.sendMessage("example_function.processing", number=number)
        for i_iter in range(n_iters):
            number = number/2
            pub.sendMessage("example_function.division", i_iter=i_iter, number=number)
        sum_all += number
        pub.sendMessage("example_function.sum", sum=sum_all)
    return sum_all

# If you need extra functionality added in, then subscribe to the messages.
# Otherwise nothing will happen, other than the normal example_function functionality.
def handle_example_function_processing(number):
    print("Processing number {}".format(number))

def handle_example_function_division(i_iter, number):
    print(number)

def handle_example_function_sum(sum):
    print("sum_all: {}".format(sum))

pub.subscribe(
    "example_function.processing",
    handle_example_function_processing
)
pub.subscribe(
    "example_function.division",
    handle_example_function_division
)
pub.subscribe(
    "example_function.sum",
    handle_example_function_sum
)