以编程方式为类生成方法

11

我有大约20个方法需要重定向到一个包装方法,该方法接受原始方法和其他参数:

class my_socket(parent):

    def _in(self, method, *args, **kwargs):
        # do funky stuff

    def recv(self, *args, **kwargs):
        return self._in(super().recv, *args, **kwargs)

    def recv_into(self, *args, **kwargs):
        return self._in(super().recv_into, *args, **kwargs)

    # and so on...

我如何通过编程方式添加更多这样的方法?在一切开始看起来不对之前,这是我所能做到的:

for method in 'recv', 'recvfrom', 'recvfrom_into', 'recv_into', ...:
    setattr(my_socket, method, ???)

我可以通过在类定义中进行赋值或其他更自然的方式来实现这个吗?

class my_socket(parent):

    def makes_recv_methods(name):
        # wraps call to name

    def recv_meh = makes_recv_methods('recv_meh')

在可能的情况下,我更倾向于使用 __get__ 等函数而不是来自 types 的魔法功能。


你不能把代码重构一下,然后用@decorator执行“funky stuff”吗?或者我有什么地方理解错了吗? - Kimvais
@Kimvais:是的,但我该如何将装饰函数与类名绑定起来呢?最后会变成@decorator('recv_into'): def recv_into(self): pass,对吧? - Matt Joiner
当没有超类时,使用super有什么意义? - Jochen Ritzel
4个回答

9

我会通过运行一些代码从列表中生成方法来完成它,可以将此放入装饰器中。这与类定义分开,使得代码更加简洁易懂。

import functools

def wrap_method(cls, name):
    # This unbound method will be pulled from the superclass.
    wrapped = getattr(cls, name)
    @functools.wraps(wrapped)
    def wrapper(self, *args, **kwargs):
        return self._in(wrapped.__get__(self, cls), *args, **kwargs)
    return wrapper

def wrap_methods(cls):
    for name in cls.WRAP_ATTRS:
        setattr(cls, name, wrap_method(cls, name))
    return cls

@wrap_methods
class my_socket(parent_class):
    WRAP_ATTRS = ['recv', 'recvfrom'] # ... + more method names

    def _in(self, method, *args, **kwargs):
        # do funky stuff

我认为Matt想在_in中调用该方法,也许是在修改参数后。相反,你可以将绑定方法作为第一个参数传递:self._in(wrapped.__get__(self, cls), *args, **kwargs) - Eryk Sun
啊,是的 - 改变以包含那个。 - babbageclunk
1
读完那些废话后,我决定反对它。有太多的间接引用,让我感到恶心。我希望我们最终能够得到匿名函数。 - Matt Joiner
Python有匿名函数。除了好的品味外,没有什么能阻止wrapper(此代码中生成的唯一闭包)成为lambda。;) - babbageclunk

0

我想进一步阐述已接受的答案。我希望可能有一个非常长的装饰器方法列表应用于一个非常长的方法列表。

import functools


def wrap_method(cls, name, wrapper_method_name):
    # This unbound method will be pulled from the superclass.
    wrapped = getattr(cls, name, wrapper_method_name)

    @functools.wraps(wrapped)
    def wrapper(self, *args, **kwargs):
        wrapper_method = getattr(self, wrapper_method_name)
        return wrapper_method(wrapped.__get__(self, cls), *args, **kwargs)

    return wrapper


def wrap_methods(cls):
    for wrapper_method_name in cls.WRAPPER_METHOD_NAMES:
        for name in cls.WRAPPED_METHODS:
            setattr(cls, name, wrap_method(cls, name, wrapper_method_name))
    return cls

这里是包装原始类的类。
@wrap_methods
class WrappedConnection(BaseConnection):
    """
    This class adds some quality-of-life improvements to the BaseConnection class.
    -WRAPPED_METHODS are wrapped by WRAPPER_METHOD_NAMES
    -wrappers can be toggled on and off.

    example:
    connection = WrappedConnection(show_messages=True, log_errors=False, keep_authenticated=False)

    default:
    connection = WrappedConnection(show_messages=False, log_errors=True, keep_authenticated=True)
    """
    WRAPPER_METHOD_NAMES = ['log_errors', 'keep_authenticated', 'show_messages']
    WRAPPED_METHODS = ['a_method', 'b_method', 'c_method', 'd_method']
    MESSAGE_OVERRIDE_MAP = {"a_method": "a_method_message_override_attribute", 
                            "b_method": "b_method_message_override_attribute"}

    def keep_authenticated(self, method, *args, **kwargs):
        """
        If the session has expired, the session is re-authenticated. The incident is logged by the default logger.
        This option can be turned off by setting keep_authenticated during initialization of a WrappedConnection object.
        - connection = WrappedConnection(keep_authenticated=False) # why would you ever do this


        :param method: (method) method to be wrapped
        :param args: (args) passed args
        :param kwargs: (kwargs) passed kwargs
        :return: (method) method wrapped by @keep_authenticated
        """
        response, expired_session = method(*args, **kwargs), None
        if response["errors"] and self._keep_authenticated:
            expired_session = list(filter(lambda x: 'expired session' in x, response["errors"]))
        if expired_session:
            self.__init__()
            logging.info('Session has been re-authenticated.')
            response = method(*args, **kwargs)
        return response

    def log_errors(self, method, *args, **kwargs):
        """
        If there is an error the incident is logged. This option can be turned off by setting log_errors
        during initialization of a WrappedConnection object.
        - connection = WrappedConnection(log_errors=False)

        :param method: (method) method to be wrapped
        :param args: (args) passed args
        :param kwargs: (kwargs) passed kwargs
        :return: (method) method wrapped by @log_errors
        """
        response = method(*args, **kwargs)
        if response["errors"] and self._log_errors:
            errors = response["errors"]
            logging.error(errors)
        return response

    def show_messages(self, method, *args, **kwargs):
        """
        Shows the xml that is sent during the request. This option can be turned on by setting show_messages during
        initialization of a WrappedConnection object.
        - connection = WrappedConnection(show_messages=True)

        :param method: (method) method to be wrapped
        :param args: (args) passed args
        :param kwargs: (kwargs) passed kwargs
        :return: (method) method wrapped by @show_messages
        """
        response = method(*args, **kwargs)
        if self._show_messages:
            message_override_attr = WrappedConnection.MESSAGE_OVERRIDE_MAP.get(method.__name__)
            if message_override_attr:
                message_override = getattr(self, message_override_attr)
                print(BeautifulSoup(message_override, "xml").prettify())
            else:
                self._show_message(method.__name__, *args, **kwargs)
        return response

    def __init__(self, *args, keep_authenticated=True, log_errors=True, show_messages=False, **kwargs):
        super(WrappedConnection, self).__init__(*args, **kwargs)
        self._keep_authenticated = keep_authenticated
        self._log_errors = log_errors
        self._show_messages = show_messages

0

威尔伯福斯提出的方案可行,但仅使用面向对象编程有一种更简单的方法:

def wrap_method(wrapped):
    @functools.wraps(wrapped)
    def wrapper(self, *args, **kwargs):
        return self._in(wrapped.__get__(self, cls), *args, **kwargs)
    return wrapper

class Parent:

    def _in(self, method, *args, **kwargs):
        return method(*args, **kwargs)


    @wrap_method
    def recv(self, *args, **kwargs):
        return # whatever

    @wrap_method
    def recv_into(self, *args, **kwargs):
        return # whatever

class MySocket(Parent):

    def _in(self, method, *args, **kwargs):
        # do funky stuff

很遗憾,我不能对父类进行更改。 - Matt Joiner
好吧,提供一种不需要元编程的替代方案也无妨。我赞成Wilberforce的回答,但我倾向于同意@Matt的观点。楼主试图实现的内容是危险的。他应该尝试设计他的代码,以便他可以不使用类似这样的技巧而得到想要的结果。 - Simon Bergot

-3
你可以使用cog
class MySocket(Parent):
"""[[[cog
import cog
l = ['in','out']
for item in l:
    cog.outl("def _{0}(self, method, *args, **kwargs):".format(item))

]]]"""
#[[[end]]]

这种方法的优点是易于更新,可以在结束注释之外不触及代码,并且如果需要的话,您可以微调生成的代码。

我曾成功地在另一个项目中使用cog来生成样板文件,并与非生成代码混合在一起。它从输入文件中读取指令并将其存储在字典中。然后对于每个样板部分,它使用该字典的相应部分来确定要编写的内容。

我在一个地方编辑指令文件,而不是在样板中的二十个不同位置进行编辑。


2
在我看来,在Python中永远不需要代码生成,因为你总可以找到一种重构样板的方法。 - Simon Bergot
@Simon 我使用COG的主要语言不是Python :) 根据他想做什么,代码生成是完全有意义的。这就是编译器或解释器所做的,接收您的指令并输出代码。 - Spencer Rathbun

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接