Added a Clean old history command to clean older history entries by cron/task (#675)

* Tests and Documentation Completed

* Running Make Format

* Updating Authors and Changes rst

* Update docs/utils.rst

fixing misspeling

Co-authored-by: Ross Mechanic <[email protected]>

* Update docs/utils.rst

Accepting @rossmechanic changes

Co-authored-by: Ross Mechanic <[email protected]>

* Fixing Code Review from @rossmechanic

Co-authored-by: Ross Mechanic <[email protected]>

Author: PauloPeres

.gitignore

@@ -17,3 +17,5 @@ MANIFEST
 test_files/
 venv/
 .DS_Store
+env
+.vscode

AUTHORS.rst

@@ -99,6 +99,7 @@ Authors
 - Ulysses Vilela
 - `vnagendra <https://github.com/vnagendra>`_
 - `yakimka <https://github.com/yakimka>`_
+- `Paulo Peres <https://github.com/PauloPeres>`_
 
 Background
 ==========

CHANGES.rst

@@ -1,5 +1,7 @@
 Changes
 =======
+2.10.1 (2020-06-16)
+- added ``clean_old_history`` management command (gh-675)
 
 2.10.0 (2020-04-27)
 -------------------

docs/utils.rst

@@ -23,3 +23,26 @@ so you can schedule, for instance, an hourly cronjob such as
 .. code-block:: bash
 
     $ python manage.py clean_duplicate_history -m 60 --auto
+
+
+clean_old_history
+-----------------------
+
+You may want to remove historical records that have existed for a certain amount of time. 
+
+If you find yourself with a lot of old history you can schedule the
+``clean_old_history`` command
+
+.. code-block:: bash
+
+    $ python manage.py clean_old_history --auto
+
+You can use ``--auto`` to remove old historial entries 
+with ``HistoricalRecords`` or enumerate specific models as args.
+You may also specify a  ``--days`` parameter, which indicates how many 
+days of records you want to keep. The default it 30 days, meaning that
+all records older than 30 days would be removed.
+
+.. code-block:: bash
+
+    $ python manage.py clean_old_history --days 60 --auto

simple_history/management/commands/clean_old_history.py

@@ -0,0 +1,74 @@
+from django.utils import timezone
+from django.db import transaction
+
+from . import populate_history
+from ... import models, utils
+from ...exceptions import NotHistoricalModelError
+
+
+class Command(populate_history.Command):
+    args = "<app.model app.model ...>"
+    help = "Scans HistoricalRecords for old entries " "and deletes them."
+
+    DONE_CLEANING_FOR_MODEL = "Removed {count} historical records for {model}\n"
+
+    def add_arguments(self, parser):
+        parser.add_argument("models", nargs="*", type=str)
+        parser.add_argument(
+            "--auto",
+            action="store_true",
+            dest="auto",
+            default=False,
+            help="Automatically search for models with the HistoricalRecords field "
+            "type",
+        )
+        parser.add_argument(
+            "--days",
+            help="Only Keep the last X Days of history, default is 30",
+            action="store_true",
+            dest="days",
+            default=30,
+        )
+
+        parser.add_argument(
+            "-d", "--dry", action="store_true", help="Dry (test) run only, no changes"
+        )
+
+    def handle(self, *args, **options):
+        self.verbosity = options["verbosity"]
+
+        to_process = set()
+        model_strings = options.get("models", []) or args
+
+        if model_strings:
+            for model_pair in self._handle_model_list(*model_strings):
+                to_process.add(model_pair)
+
+        elif options["auto"]:
+            to_process = self._auto_models()
+
+        else:
+            self.log(self.COMMAND_HINT)
+
+        self._process(to_process, days_back=options["days"], dry_run=options["dry"])
+
+    def _process(self, to_process, days_back=None, dry_run=True):
+
+        start_date = timezone.now() - timezone.timedelta(days=days_back)
+        for model, history_model in to_process:
+            history_model_manager = history_model.objects
+            history_model_manager = history_model_manager.filter(
+                history_date__lt=start_date
+            )
+            found = len(history_model_manager)
+            self.log("{0} has {1} old historical entries".format(model, found), 2)
+            if not found:
+                continue
+            if not dry_run:
+                history_model_manager.delete()
+
+            self.log(self.DONE_CLEANING_FOR_MODEL.format(model=model, count=found))
+
+    def log(self, message, verbosity_level=1):
+        if self.verbosity >= verbosity_level:
+            self.stdout.write(message)

simple_history/tests/tests/test_commands.py

@@ -6,7 +6,11 @@
 from six.moves import cStringIO as StringIO
 
 from simple_history import models as sh_models
-from simple_history.management.commands import populate_history, clean_duplicate_history
+from simple_history.management.commands import (
+    populate_history,
+    clean_duplicate_history,
+    clean_old_history,
+)
 from ..models import (
     Book,
     CustomManagerNameModel,
@@ -381,3 +385,196 @@ def test_auto_cleanup_custom_history_field(self):
             "<class 'simple_history.tests.models.CustomManagerNameModel'>\n",
         )
         self.assertEqual(CustomManagerNameModel.log.all().count(), 2)
+
+
+class TestCleanOldHistory(TestCase):
+    command_name = "clean_old_history"
+    command_error = (management.CommandError, SystemExit)
+
+    def test_no_args(self):
+        out = StringIO()
+        management.call_command(self.command_name, stdout=out, stderr=StringIO())
+        self.assertIn(clean_old_history.Command.COMMAND_HINT, out.getvalue())
+
+    def test_bad_args(self):
+        test_data = (
+            (clean_old_history.Command.MODEL_NOT_HISTORICAL, ("tests.place",)),
+            (clean_old_history.Command.MODEL_NOT_FOUND, ("invalid.model",)),
+            (clean_old_history.Command.MODEL_NOT_FOUND, ("bad_key",)),
+        )
+        for msg, args in test_data:
+            out = StringIO()
+            self.assertRaises(
+                self.command_error,
+                management.call_command,
+                self.command_name,
+                *args,
+                stdout=StringIO(),
+                stderr=out
+            )
+            self.assertIn(msg, out.getvalue())
+
+    def test_no_historical(self):
+        out = StringIO()
+        with replace_registry({"test_place": Place}):
+            management.call_command(self.command_name, auto=True, stdout=out)
+        self.assertIn(clean_old_history.Command.NO_REGISTERED_MODELS, out.getvalue())
+
+    def test_auto_dry_run(self):
+        p = Poll.objects.create(
+            question="Will this be deleted?", pub_date=datetime.now()
+        )
+        p.save()
+
+        # not related to dry_run test, just for increasing coverage :)
+        # create instance with single-entry history older than "minutes"
+        # so it is skipped
+        p = Poll.objects.create(
+            question="Will this be deleted?", pub_date=datetime.now()
+        )
+        h = p.history.first()
+        h.history_date -= timedelta(days=31)
+        h.save()
+
+        self.assertEqual(Poll.history.all().count(), 3)
+        out = StringIO()
+        management.call_command(
+            self.command_name,
+            auto=True,
+            days=20,
+            dry=True,
+            stdout=out,
+            stderr=StringIO(),
+        )
+        self.assertEqual(
+            out.getvalue(),
+            "Removed 1 historical records for "
+            "<class 'simple_history.tests.models.Poll'>\n",
+        )
+        self.assertEqual(Poll.history.all().count(), 3)
+
+    def test_auto_cleanup(self):
+        p = Poll.objects.create(
+            question="Will this be deleted?", pub_date=datetime.now()
+        )
+        self.assertEqual(Poll.history.all().count(), 1)
+        p.save()
+        self.assertEqual(Poll.history.all().count(), 2)
+        p.question = "Maybe this one won't...?"
+        p.save()
+        self.assertEqual(Poll.history.all().count(), 3)
+        out = StringIO()
+        h = p.history.first()
+        h.history_date -= timedelta(days=40)
+        h.save()
+        management.call_command(
+            self.command_name, auto=True, stdout=out, stderr=StringIO()
+        )
+        self.assertEqual(
+            out.getvalue(),
+            "Removed 1 historical records for "
+            "<class 'simple_history.tests.models.Poll'>\n",
+        )
+        self.assertEqual(Poll.history.all().count(), 2)
+
+    def test_auto_cleanup_verbose(self):
+        p = Poll.objects.create(
+            question="Will this be deleted?", pub_date=datetime.now()
+        )
+        self.assertEqual(Poll.history.all().count(), 1)
+        p.save()
+        p.question = "Maybe this one won't...?"
+        p.save()
+        h = p.history.first()
+        h.history_date -= timedelta(days=40)
+        h.save()
+        self.assertEqual(Poll.history.all().count(), 3)
+        out = StringIO()
+        management.call_command(
+            self.command_name,
+            "tests.poll",
+            auto=True,
+            verbosity=2,
+            stdout=out,
+            stderr=StringIO(),
+        )
+
+        self.assertEqual(
+            out.getvalue(),
+            "<class 'simple_history.tests.models.Poll'> has 1 old historical entries\n"
+            "Removed 1 historical records for "
+            "<class 'simple_history.tests.models.Poll'>\n",
+        )
+        self.assertEqual(Poll.history.all().count(), 2)
+
+    def test_auto_cleanup_dated(self):
+        the_time_is_now = datetime.now()
+        p = Poll.objects.create(
+            question="Will this be deleted?", pub_date=the_time_is_now
+        )
+        self.assertEqual(Poll.history.all().count(), 1)
+        p.save()
+        p.save()
+        self.assertEqual(Poll.history.all().count(), 3)
+        p.question = "Or this one...?"
+        p.save()
+        p.save()
+        self.assertEqual(Poll.history.all().count(), 5)
+
+        for h in Poll.history.all()[2:]:
+            h.history_date -= timedelta(days=30)
+            h.save()
+
+        management.call_command(
+            self.command_name, auto=True, days=20, stdout=StringIO(), stderr=StringIO(),
+        )
+        self.assertEqual(Poll.history.all().count(), 2)
+
+    def test_auto_cleanup_dated_extra_one(self):
+        the_time_is_now = datetime.now()
+        p = Poll.objects.create(
+            question="Will this be deleted?", pub_date=the_time_is_now
+        )
+        self.assertEqual(Poll.history.all().count(), 1)
+        p.save()
+        p.save()
+        self.assertEqual(Poll.history.all().count(), 3)
+        p.question = "Or this one...?"
+        p.save()
+        p.save()
+        p.save()
+        p.save()
+        self.assertEqual(Poll.history.all().count(), 7)
+
+        for h in Poll.history.all()[2:]:
+            h.history_date -= timedelta(days=30)
+            h.save()
+
+        management.call_command(
+            self.command_name, auto=True, days=20, stdout=StringIO(), stderr=StringIO(),
+        )
+        # We will remove the 3 ones that we are marking as old
+        self.assertEqual(Poll.history.all().count(), 2)
+
+    def test_auto_cleanup_custom_history_field(self):
+        m = CustomManagerNameModel.objects.create(name="John")
+        self.assertEqual(CustomManagerNameModel.log.all().count(), 1)
+        m.save()
+        self.assertEqual(CustomManagerNameModel.log.all().count(), 2)
+        m.name = "Ivan"
+        m.save()
+        h = m.log.first()
+        h.history_date -= timedelta(days=40)
+        h.save()
+        self.assertEqual(CustomManagerNameModel.log.all().count(), 3)
+        out = StringIO()
+        management.call_command(
+            self.command_name, auto=True, stdout=out, stderr=StringIO()
+        )
+
+        self.assertEqual(
+            out.getvalue(),
+            "Removed 1 historical records for "
+            "<class 'simple_history.tests.models.CustomManagerNameModel'>\n",
+        )
+        self.assertEqual(CustomManagerNameModel.log.all().count(), 2)