随着项目的增长,依赖关系和事件链也在增长,特别是在重写的save()
方法和post_save
和pre_save
信号中。
例:
被重写的A.save
创造了两个与A
相关的对象 - B
和C
。当C
被保存时,post_save
信号被调用,做其他事情,等等......
这些事件下巴怎样才能更明确?有没有办法可视化(自动生成)这样的链/流?我不是在寻找ERD
或Class
图。我需要确保在一个地方做一件事不会影响项目另一端的某些事情,因此简单的可视化将是最好的。
编辑
要清楚,我知道检查动态生成的信号几乎是不可能的。我只是想检查所有(不是动态生成的)post_save
,pre_save
和覆盖的save
方法,并将它们可视化,这样我就可以立即看到发生了什么以及在哪里当我加入某些东西时。
这不是完整的解决方案,但我希望它可以成为一个很好的起点。考虑以下代码:
save
我们可以使用from django.db import models
from django.db.models.signals import pre_save
from django.dispatch import receiver
class A(models.Model):
def save(self, *args, **kwargs):
if not self.pk:
C.objects.create()
class B(models.Model):
pass
class C(models.Model):
b = models.ForeignKey(B, on_delete=models.CASCADE, blank=True)
@receiver(pre_save, sender=C)
def pre_save_c(sender, instance, **kwargs):
if not instance.pk:
b = B.objects.create()
instance.b = b
,django inspect
和get_models()
以这种方式获取应用程序名称列表的依赖项:
signals
这给出了:
import inspect
import re
from collections import defaultdict
from django.apps import apps
from django.db.models import signals
RECEIVER_MODELS = re.compile('sender=(\w+)\W')
SAVE_MODELS = re.compile('(\w+).objects.')
project_signals = defaultdict(list)
for signal in vars(signals).values():
if not isinstance(signal, signals.ModelSignal):
continue
for _, receiver in signal.receivers:
rcode = inspect.getsource(receiver())
rmodel = RECEIVER_MODELS.findall(rcode)
if not rmodel:
continue
auto_by_signals = [
'{} auto create -> {}'.format(rmodel[0], cmodel)
for cmodel in SAVE_MODELS.findall(rcode)
]
project_signals[rmodel[0]].extend(auto_by_signals)
for model in apps.get_models():
is_self_save = 'save' in model().__class__.__dict__.keys()
if is_self_save:
scode = inspect.getsource(model.save)
model_name = model.__name__
for cmodel in SAVE_MODELS.findall(scode):
print('{} auto create -> {}'.format(model_name, cmodel))
for smodels in project_signals.get(cmodel, []):
print(smodels)
更新:通过实例类dict更改方法以找到重写的A auto create -> C
C auto create -> B
。
save
(太长时间不适合评论,缺乏代码作为完整答案)
我现在无法模拟大量代码,但另一个有趣的解决方案,受到Mario Orlandi上述评论的启发,将是某种扫描整个项目并搜索任何被覆盖的保存方法以及前后保存信号的脚本,跟踪创建它们的类/对象。它可以像一系列正则表达式一样简单,它们寻找is_self_save = 'save' in model().__class__.__dict__.keys()
定义,然后是内部任何重写的class
方法。
扫描完所有内容后,可以使用此引用集合根据类名创建依赖关系树(或树集),然后对每个引用进行拓扑排序。任何连接的组件都会说明依赖关系,您可以通过可视化或搜索这些树来以非常简单,自然的方式查看依赖关系。我在django中相对天真,但似乎你可以通过这种方式静态跟踪依赖关系,除非这些方法在不同时间在多个地方被覆盖是很常见的。
Python是一种动态语言,它不适合静态分析。
使用动态语言的推荐方法是使用单元测试来确保满足所有要求。您的项目应该有一个设计文档,从中可以获得所需的功能,然后测试将通过验证您的动态挂钩函数以正确的方式或顺序调用来确保正确的一致性,如文档中所定义。
如果您只想跟踪模型保存,并且对重写的保存方法和信号中发生的其他事情不感兴趣,您可以使用像angio这样的机制。您可以注册一个没有sender参数的全局post_save接收器,一个将为所有模型保存调用的接收器,并在该函数中打印保存的模型名称。然后,编写一个脚本,只为所有现有模型调用save。像下面这样的东西可以工作:
save
具有以下模型结构;
@receiver(models.signals.post_save)
def global_post_save(sender, instance, created, *args, **kwargs):
print(' --> ' + str(sender.__name__))
from django.apps import apps
for model in apps.get_models():
instance = model.objects.first()
if instance:
print('Saving ' + str(model.__name__))
instance.save()
print('\n\n')
该脚本将打印:
class A(models.Model):
...
def save(self, *args, **kwargs):
B.objects.create()
@receiver(post_save, sender=B)
def post_save_b(sender, instance, **kwargs):
C.objects.create()
这只是可以完成的工作的基本草图,可以根据应用程序的结构进行改进。假设您已在每个模型的数据库中都有一个条目。虽然没有改变任何东西,但这种方法也可以节省数据库中的内容,因此可以更好地在测试数据库上运行。
假设您的最终目标是在保存某个模型的实例时跟踪数据库中的更改,一个可能的解决方案是扫描数据库以查找更改而不是源代码。这种方法的优点是它还可以涵盖动态代码。显然,缺点是它只能涵盖数据库更改。
这可以使用简单的测试技术来实现。假设以下型号..
Saving A
--> A
--> B
--> C
Saving B
--> B
--> C
Saving C
--> C
我可以编写一个测试用例来计算所有数据库实例,创建模型实例,再次计算并计算差异。可以使用像from django.db import models
from django.db.models.signals import pre_save, post_save
from django.dispatch import receiver
class B(models.Model):
def save(self, *args, **kwargs):
X.objects.create()
super().save(*args, **kwargs)
class C(models.Model):
y = models.OneToOneField('Y', on_delete=models.CASCADE)
class D(models.Model):
pass
class X(models.Model):
pass
class Y(models.Model):
related = models.ForeignKey('Z', on_delete=models.CASCADE)
class Z(models.Model):
pass
@receiver(pre_save, sender=D)
def pre_save_d(*args, instance, **kwargs):
Z.objects.create()
@receiver(post_save, sender=C)
def pre_save_c(*args, instance, **kwargs):
Y.objects.create(related=Z.objects.create())
这样的工厂创建数据库实例。这是该技术的一个简单但有效的例子。
mommy
我的输出是
class TestModelDependency(TestCase):
def test_dependency(self):
models = apps.get_models()
models = [model for model in models if model._meta.app_label == 'model_effects']
for model in models:
kwargs = self.get_related_attributes(model)
initial_count = self.take_count(models)
mommy.make(model, **kwargs)
final_count = self.take_count(models)
diff = self.diff(initial_count, final_count)
print(f'Creating {model._meta.model_name}')
print(f'Created {" | ".join(f"{v} instance of {k}" for k, v in diff.items())}')
call_command('flush', interactive=False)
@staticmethod
def take_count(models):
return {model._meta.model_name: model.objects.count() for model in models}
@staticmethod
def diff(initial, final):
result = dict()
for k, v in final.items():
i = initial[k]
d = v - i
if d != 0:
result[k] = d
return result
@staticmethod
def get_related_attributes(model):
kwargs = dict()
for field in model._meta.fields:
if any(isinstance(field, r) for r in [ForeignKey, OneToOneField]):
kwargs[field.name] = mommy.make(field.related_model)
return kwargs
对于大型应用程序,它可能很慢,但我在内存中使用sqlite数据库进行测试,并且运行速度非常快。
我在一个类似的Django应用程序中工作,但是当我完成它时,我会评论你在这里展示的用例:
我需要确保在一个地方做一件事不会影响项目另一边的某些事情......
您肯定可以使用一些虚拟信号处理程序编写测试,以便知道某些代码的执行是否会触发不需要的行为,例如:
Creating b
Created 1 instance of b | 1 instance of x
Creating c
Created 1 instance of c | 1 instance of y | 1 instance of z
Creating d
Created 1 instance of d | 1 instance of z
Creating x
Created 1 instance of x
Creating y
Created 1 instance of y
Creating z
Created 1 instance of z
好吧,正如你所说,当出现对这类事物的需求时,是因为项目的规模非常大,如果你使用像你要求的工具,你可以得到这样的结果:
# I use pytest, put this example is suitable also for
# django's TestCase and others
class TestSome:
# For Django TestCase this would be setUp
def setup_method(self, test_method):
self.singals_info = []
def dummy_handler(*args, **kwargs):
# collect_info is a function you must implement, it would
# gather info about signal, sender, instance, etc ... and
# save that info in (for example) self.signals_info.
# You can then use that info for test assertions.
self.collect_info(*args, **kwargs)
# connect your handler to every signal you want to control
post_save.connect(dummy_handler)
def test_foo():
# Your normal test here ...
some_value = some_tested_function()
# Check your signals behave
assert self.signals_behave(self.signals_info)
每次要添加一些代码时,您最终都会解决一个难题,这会保存/修改某些内容。
编写代码会更好,然后,某些测试失败(为您解决难题)并向您展示您的代码错过的行为。
实施这些测试,您的生活将更加轻松。
使用测试的最佳方案:编写代码,如果没有测试失败,您就可以开始处理下一个编程任务了。
使用测试的最糟糕场景:编写代码,一些测试失败,因为您知道代码在哪里破解,只需修复它。
使用该工具的最佳方案:分析工具输出,编写代码,一切正常。
使用该工具的最糟糕情况:分析工具输出,编写代码,失败,重复直到一切正常。
那么,这样的工具会有所帮助吗?当然,但不是确保事情顺利的正确工具,请使用测试。