Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add opt-in history enforcement for bulk actions #1382

Open
twolfson opened this issue Aug 13, 2024 · 0 comments
Open

Add opt-in history enforcement for bulk actions #1382

twolfson opened this issue Aug 13, 2024 · 0 comments

Comments

@twolfson
Copy link
Contributor

Issue

Problem Statement

As a repo maintainer, I'm always nervous around bulk actions being introduced which don't capture history alongside

Describe the solution you'd like

It'd be nice to have the register function handle enforcement for externally managed models,
as well as a QuerySet and Manager for ones created in the repo.

Describe alternatives you've considered

I've implemented workarounds in my own repos, but this is the second time I'm doing this, so it seems like a good pattern to handle.

Additional context

Here are snippets of code I've used:

For externally defined models (e.g. User):

models.py:

BOUND_EXTERNAL_MODELS = []


def bind_external_model(model):
    # Bind history for model, https://django-simple-history.readthedocs.io/en/latest/quick_start.html#track-history-for-a-third-party-model  # noqa:E501
    # DEV: Due to User coming from Django, `makemigrations` will create its migration inside virtualenv directory (e.g. `/root/.virtualenvs/...` (not git trackable)) without `app=`  # noqa:E501
    register(model, app="MY_APP")

    BOUND_EXTERNAL_MODELS.append(model)
    # DEV: We expect no bulk mechanisms to be called on external models
    #   If we do want this, then please add support to run relevant `full_clean` calls always
    assert model._default_manager == model.objects, f"Expected {model}._default_manager to be {model}.objects"
    assert (
        model._default_manager._queryset_class == QuerySet
    ), f"Expected {model}._default_manager._queryset_class to be QuerySet"


def proxy_queryset_method(klass, method_name):
    original_method = getattr(klass, method_name)

    @wraps(original_method)
    def new_method_to_reject_bound_models(self, *args, **kwargs):
        if self.model in BOUND_EXTERNAL_MODELS:
            raise RuntimeError(
                # Same message as in `signals.py`
                "Attempting to run a bulk action (intentional or not) outside of django-simple-history utilities. "
                "django-simple-history is not attached to bulk actions due to no signals. "
                "Please use `bulk_create_with_history` or `bulk_update_with_history` utilities, "
                f"or the history-less `objects.unsafe_{method_name}` instead."
            )
        return original_method(self, *args, **kwargs)

    setattr(klass, method_name, new_method_to_reject_bound_models)
    setattr(klass, f"unsafe_{method_name}", original_method)


# DEV: We initially bound at `Manager` level but `.filter()` dropped to `QuerySet` level
#   We could've monkey patched at `Manager.get_queryset()` but this feels even stronger (i.e. all scenarios)
proxy_queryset_method(QuerySet, "bulk_create")
proxy_queryset_method(QuerySet, "bulk_update")
proxy_queryset_method(QuerySet, "update")

For repo-defined models

models.py:

class HistoryEnforcingQuerySet(QuerySet):
    def _enforce_bulk_action_with_history(self):
        for frame_info in inspect.stack():
            # Inner frame_info: FrameInfo(
            #   frame=<frame at 0x32bce50, file '***/simple_history/utils.py', line 98, code bulk_create_with_history>,
            #   filename='***/simple_history/utils.py',
            #   lineno=98, function='bulk_create_with_history',
            #   code_context=['        objs_with_id = model_manager.bulk_create(\n'],
            #   index=0
            # )
            # Outer frame_info: FrameInfo(
            #   frame=<frame at 0x7fb177db8420, file 'MY_APP/models.py', ***,  # noqa:E501
            #   filename='MY_APP/models.py',
            #   lineno=374, function='***',
            #   code_context=['            bulk_create_with_history(obj_list, model=Model)\n'],
            #   index=0
            # )
            if (
                frame_info.function in ["bulk_create_with_history", "bulk_update_with_history"]
                and "site-packages/simple_history" in frame_info.filename
            ):
                return

            # WARNING: If we ever need to check against `loaddata`, then this makes relevant tests stop working
            # i.e. Fixtures use `loaddata` so tests will just start passing =/

        raise RuntimeError(
            # Same message as in `models.py`
            "Attempting to run a bulk action (intentional or not) outside of django-simple-history utilities. "
            "django-simple-history is not attached to bulk actions due to no signals. "
            "Please use `bulk_create_with_history` or `bulk_update_with_history` utilities, "
            "or the history-leass `objects.unsafe_*` methods instead."
        )

    # https://docs.djangoproject.com/en/4.2/ref/models/querysets/#bulk-create
    def bulk_create(self, objs, *args, **kwargs):
        self._enforce_bulk_action_with_history()
        return self.unsafe_bulk_create(objs, *args, **kwargs)

    # https://docs.djangoproject.com/en/4.2/ref/models/querysets/#bulk-update
    def bulk_update(self, objs, fields, *args, **kwargs):
        self._enforce_bulk_action_with_history()
        return self.unsafe_bulk_update(objs, fields, *args, **kwargs)

    def update(self, *args, **kwargs):
        raise RuntimeError(
            "QuerySet#update not permitted due to bypassing django-simple-history. "
            "Please use `bulk_create_with_history` or `bulk_update_with_history` utilities, "
            "or the history-leass `objects.unsafe_*` methods instead."
        )

    def unsafe_bulk_create(self, objs, *args, **kwargs):
        return super().bulk_create(objs, *args, **kwargs)

    def unsafe_bulk_update(self, objs, fields, *args, **kwargs):
        return super().bulk_update(objs, fields, *args, **kwargs)

    def unsafe_update(self, objs, fields, *args, **kwargs):
        return super().bulk_update(objs, fields, *args, **kwargs)


class HistoryEnforcingManager(models.Manager.from_queryset(HistoryEnforcingQuerySet)):
    use_in_migrations = False

Alternative: The original was split up with signals.py for enforcement (played nicely to have such signals for other mechanisms)

Note: The original also never used objects directly as its first manager, so the implementation feels a little iffy -- instead it used:

    unsafe_objects = Manager()
    objects = HistoryEnforcingManager()

but I'm confident that what I've written above should work well

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant