Блог

По мере роста проекта растут зависимости и цепочки событий, особенно в переопределенных 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
  

Мы можем получить зависимости для списка имен приложений, используя inspect , django 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)
  

Это дает:

 A auto create -> C
C auto create -> B
  

Обновлено: измените метод на найденный, переопределенный save классом экземпляра dict .

 is_self_save = 'save' in model().__class__.__dict__.keys()
  

(Слишком длинный, чтобы вписаться в комментарий, не хватает кода для полного ответа)

Я не могу сейчас смоделировать тонну кода, но другим интересным решением, вдохновленным комментарием Марио Орланди выше, будет какой-то скрипт, который сканирует весь проект и ищет любые переопределенные методы сохранения и сигналы сохранения до и после, отслеживая класс / объект, который их создает. Это может быть так же просто, как серия выражений регулярных выражений, которые ищут class определения, за которыми следуют любые переопределенные save методы внутри.

После того, как вы все отсканировали, вы можете использовать эту коллекцию ссылок для создания дерева зависимостей (или набора деревьев) на основе имени класса, а затем топологически отсортировать каждое из них. Любые подключенные компоненты будут иллюстрировать зависимости, и вы можете визуализировать или искать эти деревья, чтобы увидеть зависимости очень простым и естественным способом. Я относительно наивен в django, но, похоже, вы могли бы статически отслеживать зависимости таким образом, если только эти методы не переопределяются в нескольких местах в разное время.

Если вы хотите отслеживать только сохранения моделей и не заинтересованы в других вещах, происходящих внутри переопределенных методов сохранения и сигналов, вы можете использовать механизм, подобный angio. Вы можете зарегистрировать глобальный приемник post_save без аргумента sender, который будет вызываться для всех сохранений модели, и распечатать сохраненное имя модели в этой функции. Затем напишите сценарий, чтобы просто вызвать 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('nn')
  

Со следующей структурой модели;

 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
  

И мой вывод

 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
  

Для больших приложений это может быть медленно, но я использую базу данных sqlite в памяти для тестирования, и она работает довольно быстро.

Я работаю в приложении Django, которое делает нечто подобное, но пока я это делаю, я прокомментирую вариант использования, который вы представили здесь:

Мне нужно быть уверенным, что выполнение одной вещи в одном месте не повлияет на что-то на другой стороне проекта…

Вы, конечно, могли бы писать тесты с некоторыми фиктивными обработчиками сигналов, чтобы узнать, вызывает ли выполнение определенного кода нежелательное поведение, например:

 # 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)
  

Почему это лучше, чем иметь скрипт, который показывает цепочку событий?

Ну, как вы говорите, когда возникает необходимость в подобных вещах, это потому, что размер проекта очень большой, и если вы используете инструмент, подобный тому, который вы запрашиваете, вы можете получить такой результат:

 Save A -> Creates B -> Creates C
Save B -> Creates D
Save B -> Creates C
.
.
.
# Imagine here 3 or 4 more lines.
  

В конечном итоге вы будете решать головоломку каждый раз, когда захотите добавить какой-то код, который что-то сохраняет / изменяет.

Однако …

Было бы лучше написать свой код, а затем выполнить какой-то тест с ошибкой (разрешающий головоломку для вас) и показывающий вам, где именно ваш код будет вести себя неправильно.

Заключение:

Реализует эти тесты, и ваша жизнь станет проще.

Наилучший сценарий с использованием тестов: напишите свой код, и если ни один тест не провалился, вы готовы к решению следующей задачи программирования.

Худший сценарий с использованием тестов: напишите свой код, какой-то тест завершается неудачей, поскольку вы знаете, где именно ваш код сломался, просто исправьте это.

Лучший сценарий с использованием инструмента: проанализируйте вывод инструмента, напишите свой код, все в порядке.

Наихудший сценарий с использованием инструмента: проанализируйте вывод инструмента, напишите свой код, что-то не удается, повторяйте, пока все не ок.

Итак, такой инструмент был бы полезен? Конечно, но это не правильный инструмент для обеспечения того, чтобы все было хорошо, используйте для этого тесты.