S.O.L.I.D

编写功能性、可重用性和清洁代码的简单原则

代码就像幽默。当你不得不解释它时，它就很糟糕了：）

作为软件工程师，我们必须牢记一个重要的事实，那就是我们写的代码会被其他人看到，他们可能想在我们创造的基础上进行开发。请不要让他们的生活变得更加艰难！

如果你希望你的代码在未来被尽可能多的使用和重复使用，并且需要最小的改动，你需要写出整洁的代码。

易于理解

容易发现错误

易于编辑

当设计一个系统时，这是由很多模块组合在一起完成的。

把一个设计复杂的系统想象成一个拼图。模块是拼图的碎片，而拼图中的每一块都有确切的、唯一的位置，与特定形状的拼图相连，不能在任何其他位置。

另一方面，一个精心设计的系统应该更像乐高砖。任何两块乐高砖都可以相互连接，不会有太大的麻烦--所有的乐高砖都是相互兼容的。

当设计一个软件系统时，你想让它像乐高一样。你希望能够很容易地把模块连接在一起，重新使用它们，并且在将来仍然能够理解每个模块的用途是什么。

这就是为什么找到一种评估设计复杂性的方法真的很重要--它将帮助你更好地应用面向对象的设计原则，并建立起受控的系统。

有两个与任何软件的开发有关的评估概念：耦合和内聚。

耦合

关注一个模块和其他模块之间的复杂性。一个模块的耦合度可以捕捉到模块之间相互联系的复杂性。

如果模块之间有很强的依赖性，那么这个模块就被称为紧密耦合的模块--这就像拼图。如果一个模块不强烈依赖其他模块，并且发现很容易与其他模块连接，那么这个模块与其他模块是松散耦合的--这就像乐高。

耦合度是由模块之间的关系数量来衡量的。也就是说，随着模块之间的调用次数增加或共享数据量大，耦合度也会增加。所以，从逻辑上讲，一个高耦合度的设计会有更多的错误。

内聚

着重于模块内部的复杂性。它衡量的是一个给定模块内各功能片段之间的关系强度。

如果一个模块有明确的目的，并且只执行一项任务，这意味着该模块具有高内聚性。另一方面，如果该模块试图执行一个以上的任务，并封装了一个以上的目的，那么该模块的内聚力就很低。

最佳做法是，如果模块有一个以上的责任，那么可能是时候分离模块的责任了。你希望你的模块具有低耦合或松耦合，而不是紧耦合，高内聚。

所以，面向对象的编程包含了一些原则，帮助你理解对特定设计模式和软件架构的需求。

在面向对象编程的世界里，有许多设计规则、模式和原则。其中有五条原则通常被归为一组，称为S.O.L.I.D原则--由Robert C. Martin（Bob叔叔）定义的规则。

SOLID原则告诉我们如何将我们的函数和数据结构安排成类，以及这些类应该如何协同工作。

什么是S.O.L.I.D

S：单一责任原则(SRP)

O: 开放-封闭原则(CR)

L: 利斯科夫替代原则(LSP)

I: 界面隔离原则(ISP)

D: 依赖性反转原则(DIP)

image.png

让我们更多地讨论每个原则。

S: 单一责任原则(SRP)

一个类应该只有一个责任。

这个原则说，类应该只有一个明确的工作和一个责任。

但什么是责任呢？

应该只有一个原因让一个类必须被改变。

你的类的责任越多，你需要改变它的频率就越高。

在你做任何改变之前，你需要问一个简单的问题。

class Person(object):
    def __init__(self):
        self.__method()

    def __method(self):
        print('I am student')

class Moreperson(Person):
    def __method(self):
        print('I am teacher')

Moreperson()

输出：
I am student

"你的class的责任是什么？"

如果你的答案中包括 "和 "这个词，你就很可能违背了单一责任原则。

通过一个例子可以很好地理解这个原则。想象一下，有一个类的雇员，执行以下操作。

计算付款。

交付报告。

在数据库中保存数据。

在这里，这个类有多种责任，而且有太多不同的原因需要改变。因此，它不符合单一责任原则。

为了应用单一责任原则，你需要将每个不同的责任分离到不同的类中。

这个原则使软件更容易实现，防止未来改变后出现意外的副作用，并减少耦合。

注意：这个原则不仅适用于类，也适用于软件组件和微服务。

O: 开放-封闭原则（CR）

软件实体（类、模块功能）应该对扩展开放，但对修改封闭。

Robert C. Martin认为这个原则是 "面向对象设计的最重要原则"。

需求随着时间的推移而变化。开放-封闭原则的目标是写出的代码不必在每次需求改变时都要进行修改。

我们怎样才能在不改变模块本身的情况下，改变模块的作用呢？

开放-封闭原则指出，如果需求发生变化，这些变化应该通过添加新的代码来实现，而不是通过改变已经工作的旧代码。

抽象是关键

一个模块有可能操纵一个抽象的东西。模块可以被关闭修改，因为它依赖于一个固定的抽象概念。而模块的行为可以通过创建新的抽象衍生物来扩展。

这一原则使你能够写出健壮的、可维护的和可重用的软件组件。

开放-封闭原则说：永远不要重写代码。

L：利斯科夫替代原则。(LSP)

BaseType必须可以被它的subTypes所替代

这个原则定义了子类的对象必须可以替代其超类的对象而不破坏应用。

这要求子类的对象具有与它们的超类相同的行为，如果一个超类可以做某项工作，那么子类有必要可以做与它的超类相同的工作。

利斯科夫替代原则与继承密切相关。它通过关注超类及其子类型的行为，扩展了开放-封闭原则。

通过一个例子可以很好地理解这个原则。想象一下，有一个执行以下操作的类Animal：

说话

吃东西

走路

我们有一个子类，猫，它是一种动物，继承了超类动物，并且有它的行为，说话、吃东西和走路。因此，这适用于利斯科夫替代原则。

class Animal:  #(object)
    def __init__(self,talk,eat,walk):
        self.talk = talk
        self.eat = eat
        self.walk = walk

cat = Animal('喵喵','鱼','猫步')
print(cat.talk)
print(cat.eat)
print(cat.walk)

喵喵
鱼
猫步

但是我们假设有一个子类，蜗牛snail，它仍然是一个动物，但是不能说话，所以它不能继承超类的动物。这就破坏了Liskov替代原则。

snail = Animal('不能说话','叶子','爬行')
print(snail.talk)

总而言之，如果你对父类型说的是真的，那么它在整个链条中都必须是真的。如果你说父类型可以做某事，那么所有的子类型也需要能够做这件事。

如何修改上面的动物的类？

先抽象出猫和蜗牛共同的属性：

• 走
• 吃
  Animal中只保留以上两个属性。
  分别继承Animal类的属性后，再增加各自独特属性。

如何修改上面的动物的类？
猫和蜗牛共同的属性：eat,walk
猫和蜗牛不同的属性: call
共同的部分：

#  python3中所有类都可以继承于object基类
class Animal(object):  
   def __init__(self, eat, walk):
       self.eat = eat
       self.walk = walk

不要忘记从Animal类引入属性

class Cat(Animal):
   def __init__(self,eat,walk,call):
       super(Cat, self).__init__(eat,walk)  
       self.call = call

if __name__ == '__main__':  # 单模块被引用时下面代码不会受影响，用于调试
     c = Cat( '鱼', '猫步', '喵喵',)  #  Cat继承了父类Animal的吃和走的属性
     c.call()  # 输出 喵喵 会叫 ，Cat继承了父类Animal的call方法 

至于蜗牛没有叫的属性，就不必增加call的属性即可！
抽象出两个属性可以继承给Cat子类；

当然，我们也可以给蜗牛定义一个子类，不增加任何属性：

class Animal(object):
   def __init__(self, eat, walk):
       self.eat = eat
       self.walk = walk

snail = Animal('叶子','爬行')
print(snail.eat)

总结更多类的实例化调用写法：

#  python3中所有类都可以继承于object基类
class Animal(object):
   def __init__(self, name, age):
       self.name = name
       self.age = age

   def call(self):
       print(self.name, '会叫')

cat = Animal('kitty',5)
print('cat.name:',cat.name)
# 同时output-> kitty 会叫

print('cat.call:',cat.call)
print('cat.call():',cat.call())  #None
# TypeError: call() missing 1 required positional argument: 'self'

######
# 现在我们需要定义一个Cat 猫类继承于Animal，猫类比动物类多一个sex属性。
######
class Cat(Animal):
   def __init__(self,name,age,sex):
       super(Cat, self).__init__(name,age)  # 不要忘记从Animal类引入属性
       self.sex=sex

if __name__ == '__main__':  # 单模块被引用时下面代码不会受影响，用于调试
   c = Cat('ketty', 2, '男')  #  Cat继承了父类Animal的属性
   c.call()  # 输出 喵喵 会叫 ，Cat继承了父类Animal的方法

'''
注意：一定要用 super(Cat, self).__init__(name,age) 去初始化父类，
否则，继承自 Animal的 Cat子类将没有 name和age两个属性。
函数super(Cat, self)将返回当前类继承的父类，即 Animal，然后调用__init__()方法，
注意self参数已在super()中传入，在__init__()中将隐式传递，不能再写出self。

Python 对子类方法的重构
上面例子中 Animal 的子类 Cat 继承了父类的属性和方法，但是我们猫类 Cat 有自己的
叫声 '喵喵' ，这时我们可以对父类的 Call() 方法进行重构。如下：
'''

class Cat(Animal):
   def __init__(self, name, age, sex):
       super(Cat, self).__init__(name,age)
       self.sex = sex

   def call(self):
       return print(self.name,'会喵喵叫')

   def genderJudge(self):
        if self.sex == '男':
            return print(f"{self.name}不是母猫！")

if __name__ == '__main__':
   c = Cat('丁大猫', 2, '男')
   c.call()  # 输出-> 丁大猫 会“喵喵”叫
   c.genderJudge()

#  python3中所有类都可以继承于object基类

class Animal(object):
   def __init__(self,name, eat, walk):
       self.name = name
       self.eat = eat
       self.walk = walk

   def call(self):
       print(self.name, '会叫')

Cat = Animal
c = Cat('kitty','鱼', '猫步') #,喵喵'
print(Cat.call)
print(c.name,c.call()) #kitty None
print('c.call-1:',c.call)
print('c.call-2:',c.call())

#不要忘记从Animal类引入属性

class Cat(Animal):
   def __init__(self,name,eat,walk,call):
       super(Cat, self).__init__(name,eat,walk)
       self.call = call

cat1 = Cat('kitty','鱼', '猫步','喵喵') #
print('cat-1.walk:',cat1.walk)
print('cat-1.call:',cat1.call)

# 单模块被引用时下面代码不会受影响，用于调试
if __name__ == '__main__':
     cat2 = Cat('ada','鱼', '猫步','miao')
     # Cat继承了父类Animal的吃和走的属性
     print('cat2.eat',cat2.eat)
     print('cat2.call:',cat2.call)
     cat2.call
     # 输出 喵喵 会叫 ，Cat继承了父类Animal的方法


#
class Cat(Animal):
   def __init__(self,name,eat,walk,gender):
       super(Cat, self).__init__(name,eat,walk)
       self.gender = gender

if __name__ == '__main__':
     cat3 = Cat('brown','鱼', '猫步','male cat')
     # Cat继承了父类Animal的吃和走的属性
     print('cat2.eat',cat3.eat)
     print('cat2.call:',cat3.call)
     cat3.call()
     # 输出 brown会叫 ，Cat继承了父类Animal的方法

I：接口隔离原则。(ISP)

客户端不应该被迫依赖他们不使用的接口。

根据接口隔离原则，不应该强迫客户实现他们不需要或对他们不重要的方法。

这意味着一个小的特定接口要比一个大的接口好。

隔离 = 分割

所以，界面隔离原则意味着将你的界面分割成更小的界面。

通过一个例子可以很好地理解这个原则。

假设你有一个类，餐厅接口，执行以下操作。

在线支付

在线订餐

当面支付

当面下单

在Restaurant接口中定义了四个方法：在线支付、当面支付、在线订购和当面订购。我们有两个子类。

在线客户

当面交易客户

由于这四个方法是餐厅接口的一部分，任何子类都必须实现这四个方法。但是由于子类Online Client不需要实现Pay in Person或Order in person，所以它将被迫使用这些方法，即使它不需要这些方法。

这里的餐厅接口打破了接口分离原则以及单一责任原则，因为不同的支付方法被归入了一个接口。

为了克服这个问题，我们需要将支付和订单的功能分成两个独立的类。支付接口和订单接口。每个客户都使用支付接口和订单接口的一个实现。例如，在线客户端将使用在线支付和在线订单。

接口分离原则有助于保持类和接口的清晰、紧凑和重点。

为了应用单一责任原则，你需要将每个不同的责任分离到不同的类中。这一原则使软件更容易实现，防止未来改变后出现意外的副作用，并减少耦合。

注意：这个原则不仅适用于类，也适用于软件组件和微服务。

D: 依赖性反转原则。(DIP)

抽象不应该依赖于细节。细节应该依赖于抽象。

这个原则指出，高层模块不应该依赖低层模块，但两者都应该依赖抽象。

抽象不应该依赖于细节，但细节应该依赖于抽象。

这与开放-封闭原则和Liskov替代原则密切相关，因为它们都依赖于接口。

这个原则可以通过一个例子很好地理解。

电脑鼠标有很多类型。这与开放-封闭原则和利斯科夫替代原则密切相关，因为它们都依赖于界面。通过一个例子可以很好地理解这个原理。

有许多类型的电脑鼠标。现在，想象一下，每台电脑只能使用一种特定类型的鼠标。你必须购买你的电脑系统支持的确切类型的鼠标。在这种情况下，如果你的电脑只适用于无线鼠标，那么有线或轨迹球鼠标或任何其他类型的鼠标将无法与你的电脑一起使用。

幸运的是，并不是这样的，因为电脑并不依赖特定的鼠标端口或类型。无论你的鼠标是什么类型，它都有一个共同的界面（至少有两个按钮和移动屏幕指针的能力）。

关键点

为了拯救你的团队，最好是写出可读、可重用、直奔主题的代码。

SOLID原则是用来编写更好的代码的OOP规则。

SOLID的五项原则也是为同一要点服务的。

对数据科学家有用的Python装饰器

Marton Trencseni - 2022年5月22日 星期日 - Python

简介
在这篇文章中，我将展示一些对数据科学家有用的@decorators。重温一下Bytepawn以前关于装饰器的文章也是很有用的。
构建一个玩具Python @dataclass装饰器
Python装饰器模式
所有以Python为标签的Bytepawn帖子
ipython笔记本已经在Github上了。
@parallel
让我们假设我写了一个非常低效的方法来寻找素数。

from sympy import isprime

def generate_primes(domain: int=1000*1000, num_attempts: int=1000) -> list[int]:
    primes: set[int] = set()
    seed(time())
    for _ in range(num_attempts):
        candidate: int = randint(4, domain)
        if isprime(candidate):
            primes.add(candidate)
    return sorted(primes)

print(len(generate_primes()))

输出类似的结果。
88

然后我意识到，如果我在所有的CPU线程上并行运行原始的generate_primes()，我可以得到一个 "免费 "的速度提升。这很常见，定义一个@parallel是有意义的。

def parallel(func=None, args=(), merge_func=lambda x:x, parallelism = cpu_count()):
    def decorator(func: Callable):
        def inner(*args, **kwargs):
            results = Parallel(n_jobs=parallelism)(delayed(func)(*args, **kwargs) for i in range(parallelism))
            return merge_func(results)
        return inner
    if func is None:
        # decorator was used like @parallel(...)
        return decorator
    else:
        # decorator was used like @parallel, without parens
        return decorator(func)

有了这个，只需一行，我们就可以将我们的函数并行化。

@parallel(merge_func=lambda li: sorted(set(chain(*li)))))
def generate_primes(...): # 同样的签名，没有变化
    ... # 同样的代码，没有任何变化

print(len(generate_primes()))

输出类似的东西。
1281

在我的例子中，我的Macbook有8个核心，16个线程（cpu_count()是16），所以我生成了16倍的素数。注意。

唯一的开销是必须定义一个merge_func，它将函数的不同运行结果合并成一个结果，以便向装饰函数（本例中为generate_primes()）的外部调用者隐藏并行性。在这个玩具例子中，我只是合并了列表，并通过使用 set() 确保素数是唯一的。

有许多Python库和方法（例如线程与进程）可以实现并行。这个例子使用了joblib.Parallel()的进程并行，它在Darwin + python3 + ipython上运行良好，并且避免了对Python全局解释器锁（GIL）的锁定。
@production
有时我们会写一个复杂的流水线，有一些额外的步骤，我们只想在某些环境下运行。例如，在我们的本地开发环境中做一些事情，但在生产环境中不做，反之亦然。如果能够对函数进行装饰，让它们只在某些环境下运行，而在其他地方不做任何事情，那就更好了。

实现这一目标的方法之一是使用一些简单的装饰器。 @production表示我们只想在prod上运行的东西，@development表示我们只想在dev中运行的东西，我们甚至可以引入一个@inactive，将函数完全关闭。这种方法的好处是，这种方式可以在代码/Github中跟踪部署历史和当前状态。另外，我们可以在一行中做出这些改变，从而使提交更简洁；例如 例如，@inactive比整个代码块被注释掉的大提交要干净。

@parallel(merge_func=lambda li: sorted(set(chain(*li))))
def generate_primes(...): # same signature, nothing changes
    ... # same code, nothing changes

print(len(generate_primes()))

1281

在我的例子中，我的Macbook有8个核心，16个线程（cpu_count()是16），所以我产生了16倍的素数。注意。

唯一的开销是必须定义一个merge_func，它将函数的不同运行结果合并为一个结果，以便向装饰函数（本例中为 generate_primes()）的外部调用者隐藏并行性。在这个玩具例子中，我只是合并了列表，并通过使用 set() 确保素数是唯一的。

有许多Python库和方法（例如线程与进程）可以实现并行。这个例子使用了joblib.Parallel()的进程并行，它在Darwin + python3 + ipython上运行良好，并且避免了对Python全局解释器锁（GIL）的锁定。
@production

有时我们会写一个复杂的流水线，有一些额外的步骤，我们只想在某些环境下运行。例如，在我们的本地开发环境中做一些事情，但在生产环境中不做，反之亦然。如果能够对函数进行装饰，让它们只在某些环境下运行，而在其他地方不做任何事情，那就更好了。

production_servers = [...]

def production(func: Callable):
    def inner(*args, **kwargs):
        if gethostname() in production_servers:
            return func(*args, **kwargs)
        else:
            print('This host is not a production server, skipping function decorated with @production...')
    return inner

def development(func: Callable):
    def inner(*args, **kwargs):
        if gethostname() not in production_servers:
            return func(*args, **kwargs)
        else:
            print('This host is a production server, skipping function decorated with @development...')
    return inner

def inactive(func: Callable):
    def inner(*args, **kwargs):
        print('Skipping function decorated with @inactive...')
    return inner

@production
def foo():
    print('Running in production, touching databases!')

foo()

@development
def foo():
    print('Running in production, touching databases!')

foo()

@inactive
def foo():
    print('Running in production, touching databases!')

foo()

Output

Running in production, touching databases!
This host is a production server, skipping function decorated with @development...
Skipping function decorated with @inactive...

这个想法可以适用于其他框架/环境。
@deployable
在我目前的工作中，我们使用Airflow进行ETL/数据管道。我们有一个丰富的辅助函数库，可以在内部构建适当的DAG，所以用户（数据科学家）不必担心这个问题。
最常用的是dag_vertica_create_table_as()，它在我们的Vertica DWH上运行一个SELECT，每晚将结果转储到一个表中。

dag = dag_vertica_create_table_as(
    table='my_aggregate_table',
    owner='Marton Trencseni (marton.trencseni@maf.ae)',
    schedule_interval='@daily',
    ...
    select="""
    SELECT
        ...
    FROM
        ...
    """
)


CREATE TABLE my_aggregate_table AS
SELECT ...

实现这一目标的方法之一是使用一些简单的装饰器。 @production表示我们只想在prod上运行的东西，@development表示我们只想在dev中运行的东西，我们甚至可以引入一个@inactive，将函数完全关闭。这种方法的好处是，这种方式可以在代码/Github中跟踪部署历史和当前状态。另外，我们可以在一行中做出这些改变，从而使提交更简洁；例如 例如，@inactive比整个代码块被注释掉的大提交要干净。

def deployable(func):
    def inner(*args, **kwargs):
        if 'deploy' in kwargs:
            if kwargs['deploy'].lower() in ['production', 'prod'] and gethostname() not in production_servers:
                print('This host is not a production server, skipping...')
                return
            if kwargs['deploy'].lower() in ['development', 'dev'] and gethostname() not in development_servers:
                print('This host is not a development server, skipping...')
                return
            if kwargs['deploy'].lower() in ['skip', 'none']:
                print('Skipping...')
                return
            del kwargs['deploy'] # to avoid func() throwing an unexpected keyword exception
        return func(*args, **kwargs)
    return inner

@deployable
def dag_vertica_create_table_as(...): # same signature, nothing changes
    ... # code signature, nothing changes

@deployable
def dag_vertica_create_or_replace_view_as(...): # same signature, nothing changes
    ... # code signature, nothing changes

@deployable
def dag_vertica_train_predict_model(...): # same signature, nothing changes
    ... # code signature, nothing changes

在生产中运行，正在接触数据库
这个主机是生产服务器，跳过用@development装饰的函数...
跳过用@inactive装饰的函数...

这个想法可以适用于其他框架/环境。
@deployable
在我目前的工作中，我们使用Airflow进行ETL/数据管道。我们有一个丰富的辅助函数库，可以在内部构建适当的DAG，所以用户（数据科学家）不必担心这个问题。

最常用的是dag_vertica_create_table_as()，它在我们的Vertica DWH上运行一个SELECT，每晚将结果转储到一个表中。

如果我们在这里停止，什么也不会发生，我们不会破坏任何东西。然而，现在我们可以到我们使用这些函数的DAG文件中，增加1行。

dag = dag_vertica_create_table_as(
    deploy='development', # the function will return None on production
    ...
)


dag = dag_vertica_create_table_as(
    table='my_aggregate_table',
    owner='Marton Trencseni (marton.trencseni@maf.ae)',
    schedule_interval='@daily',
    ...
    select="""
    选择
        ...
    从
        ...
    """
)

然后这就变成了对DWH的查询，大致是这样。
CREATE TABLE my_aggregate_table AS
选择...

在现实中，它更复杂：我们首先运行今天的查询，如果今天的查询成功创建，则有条件地删除昨天的查询。这个条件逻辑（以及其他一些针对我们环境的意外的复杂性，例如必须发布GRANTs）导致DAG有9个步骤，但这不是这里的重点，也超出了本文的范围。

在过去的两年里，我们已经创建了近500个DAG，所以我们扩大了Airflow EC2实例的规模，并引入了独立的开发和生产环境。如果能有一种方法来标记DAG是应该在开发环境还是生产环境中运行，在代码/Github中跟踪这一点，并使用相同的机制来确保DAG不会意外地运行在错误的环境中，那就更好了。

大约有10个类似的便利函数，如

dag_vertica_create_or_replace_view_as()和
dag_vertica_train_predict_model()

等，我们希望这些dag_xxx()函数的所有调用都可以在生产和开发之间切换（或者到处跳过）。
然而，上一节中的@production和@development装饰器在这里不起作用，因为我们不想将

代码块

dag_vertica_create_table_as()

切换为永远不在其中一个环境中运行。我们希望能够在每次调用时进行设置，并且在我们所有的dag_xxxx()函数中都有这个功能，而不需要复制/粘贴代码。我们想要的是在我们所有的dag_xxxx()函数中添加一个部署参数（有一个好的默认值），这样我们就可以在我们的DAG中添加这个参数，以增加安全性。我们可以通过@deployable装饰器来实现这一点。

然后我们可以将装饰器添加到我们的函数定义中（每个函数添加1行）

@deployable
def dag_vertica_create_table_as(...)。# 相同的签名，没有变化
    ... #代码签名，没有任何变化

@deployable
def dag_vertica_create_or_replace_view_as(...): # 同样的签名，没有任何变化
    ... # 代码签名，没有任何变化

@deployable
def dag_vertica_train_predict_model(...): # 同样的签名，没有任何变化
    ... # 代码签名，没有任何变化

如果我们在这里停止，什么也不会发生，我们没有破坏任何东西。然而，现在我们可以去使用这些函数的DAG文件，并添加1行。

dag = dag_vertica_create_table_as(
    deploy='development', # 该函数在生产中会返回 None
    ...
)

@redirect (stdout)

有时我们写一个大的函数，也会调用其他代码，各种信息都被打印()出来。或者，我们可能有一个bug，有一堆print()，想在打印出来的内容上加上行号，这样就更容易参考了。

在这些情况下，@redirect可能是有用的。这个装饰器将print()的标准输出重定向到我们自己的逐行打印机，我们可以对它做任何我们想做的事情（包括扔掉它）。

def redirect(func=None, line_print: Callable = None):
    def decorator(func: Callable):
        def inner(*args, **kwargs):
            with StringIO() as buf, redirect_stdout(buf):
                func(*args, **kwargs)
                output = buf.getvalue()
            lines = output.splitlines()
            if line_print is not None:
                for line in lines:
                    line_print(line)
            else:
                width = floor(log(len(lines), 10)) + 1
                for i, line in enumerate(lines):
                    i += 1
                    print(f'{i:0{width}}: {line}')
        return inner
    if func is None:
        # decorator was used like @redirect(...)
        return decorator
    else:
        # decorator was used like @redirect, without parens
        return decorator(func)

如果我们使用redirect()而不指定明确的line_print()函数，它就会打印行数，但要加上行号。

@redirect
def print_lines(num_lines):
    for i in range(num_lines):
        print(f'Line #{i+1}')

print_lines(10)

Output:
01: Line #1
02: Line #2
03: Line #3
04: Line #4
05: Line #5
06: Line #6
07: Line #7
08: Line #8
09: Line #9
10: Line #10

如果我们想把所有的打印文本保存到一个变量中，我们也可以实现这一点。

lines = []
def save_lines(line):
    lines.append(line)

@redirect(line_print=save_lines)
def print_lines(num_lines):
    for i in range(num_lines):
        print(f'Line #{i+1}')

print_lines(3)
print(lines)

Output:
['Line #1', 'Line #2', 'Line #3']

重定向stdout的实际工作是由contextlib.redirect_stdout完成的，如StackOverflow线程中所示。
@stacktrace
下一个装饰器模式是 @stacktrace，当函数被调用和从函数返回值时，它会发出有用的信息。

def stacktrace(func=None, exclude_files=['anaconda']):
    def tracer_func(frame, event, arg):
        co = frame.f_code
        func_name = co.co_name
        caller_filename = frame.f_back.f_code.co_filename
        if func_name == 'write':
            return # ignore write() calls from print statements
        for file in exclude_files:
            if file in caller_filename:
                return # ignore in ipython notebooks
        args = str(tuple([frame.f_locals[arg] for arg in frame.f_code.co_varnames]))
        if args.endswith(',)'):
            args = args[:-2] + ')'
        if event == 'call':
            print(f'--> Executing: {func_name}{args}')
            return tracer_func
        elif event == 'return':
            print(f'--> Returning: {func_name}{args} -> {repr(arg)}')
        return

    def decorator(func: Callable):
        def inner(*args, **kwargs):
            settrace(tracer_func)
            func(*args, **kwargs)
            settrace(None)
        return inner
    if func is None:
        # decorator was used like @stacktrace(...)
        return decorator
    else:
        # decorator was used like @stacktrace, without parens
        return decorator(func)

有了这个，我们就可以装饰我们希望追踪开始的最上面的函数，并且我们将得到关于分支的有用输出。

def b():
    print('...')

@stacktrace
def a(arg):
    print(arg)
    b()
    return 'world'

a('foo')

Output:
--> Executing: a('foo')
foo
--> Executing: b()
...
--> Returning: b() -> None
--> Returning: a('foo') -> 'world'

这里唯一的诀窍是隐藏调用栈中不感兴趣的部分。在我的例子中，我是在Anaconda上的ipython中运行这段代码的，所以我隐藏了代码在路径中有Anaconda的文件中的部分调用栈（否则我在上面的片段中会得到大约50-100个无用的调用栈条目）。这是通过装饰器的exclude_files参数完成的。
@traceclass

与上述类似，我们可以定义一个装饰器@traceclass，用于类，以获得其成员的执行痕迹。这包括在之前的装饰器帖子中，在那里它只是被称为@trace，并且有一个bug（在原来的帖子中已经修复）。这个装饰器

def traceclass(cls: type):
    def make_traced(cls: type, method_name: str, method: Callable):
        def traced_method(*args, **kwargs):
            print(f'--> Executing: {cls.__name__}::{method_name}()')
            return method(*args, **kwargs)
        return traced_method
    for name in cls.__dict__.keys():
        if callable(getattr(cls, name)) and name != '__class__':
            setattr(cls, name, make_traced(cls, name, getattr(cls, name)))
    return cls

We can use it like:
@traceclass
class Foo:
    i: int = 0
    def __init__(self, i: int = 0):
        self.i = i
    def increment(self):
        self.i += 1
    def __str__(self):
        return f'This is a {self.__class__.__name__} object with i = {self.i}'

f1 = Foo()
f2 = Foo(4)
f1.increment()
print(f1)
print(f2)

Output:
--> Executing: Foo::__init__()
--> Executing: Foo::__init__()
--> Executing: Foo::increment()
--> Executing: Foo::__str__()
This is a Foo object with i = 1
--> Executing: Foo::__str__()
This is a Foo object with i = 4

结论
在Python中，函数是一等公民，而装饰器是强大的语法糖，利用这一功能为程序员提供了一种看似 "神奇 "的方式来构建函数和类的有用组合。这些是5个装饰器，可能对在ipython笔记本中工作的数据科学家特别有用。
感谢Zsolt的错误修正和改进建议。