Reusable commands

codeforlife/commands/__init__.py

@@ -0,0 +1,7 @@
+"""
+© Ocado Group
+Created on 15/11/2024 at 12:18:24(+00:00).
+"""
+
+from .load_fixtures import LoadFixtures
+from .summarize_fixtures import SummarizeFixtures

codeforlife/commands/load_fixtures.py

@@ -0,0 +1,40 @@
+"""
+© Ocado Group
+Created on 10/06/2024 at 10:44:45(+01:00).
+"""
+
+import os
+import typing as t
+
+from django.apps import apps
+from django.core.management import call_command
+from django.core.management.base import BaseCommand
+
+
+# pylint: disable-next=missing-class-docstring
+class LoadFixtures(BaseCommand):
+    help = "Loads all the fixtures of the specified apps."
+
+    required_app_labels: t.Set[str] = set()
+
+    def add_arguments(self, parser):
+        parser.add_argument("app_labels", nargs="*", type=str)
+
+    def handle(self, *args, **options):
+        fixture_labels: t.List[str] = []
+        for app_label in {*options["app_labels"], *self.required_app_labels}:
+            app_config = apps.app_configs[app_label]
+            fixtures_path = os.path.join(app_config.path, "fixtures")
+
+            self.stdout.write(f"{app_label} fixtures ({fixtures_path}):")
+            for fixture_label in os.listdir(fixtures_path):
+                if fixture_label in fixture_labels:
+                    self.stderr.write(f"Duplicate fixture: {fixture_label}")
+                    return
+
+                self.stdout.write(f"    - {fixture_label}")
+                fixture_labels.append(fixture_label)
+
+            self.stdout.write()
+
+        call_command("loaddata", *fixture_labels)

codeforlife/commands/summarize_fixtures.py

@@ -0,0 +1,86 @@
+"""
+© Ocado Group
+Created on 22/02/2024 at 09:24:27(+00:00).
+"""
+
+import json
+import os
+import typing as t
+from dataclasses import dataclass
+from itertools import groupby
+
+from django.apps import apps
+from django.core.management.base import BaseCommand
+
+from ..types import JsonDict
+
+
+@dataclass(frozen=True)
+class Fixture:
+    """A data model fixture."""
+
+    model: str
+    pk: t.Any
+    fields: JsonDict
+
+
+FixtureDict = t.Dict[str, t.List[Fixture]]
+
+
+# pylint: disable-next=missing-class-docstring
+class SummarizeFixtures(BaseCommand):
+    help = "Summarizes all the listed fixtures."
+
+    required_app_labels: t.Set[str] = set()
+
+    def add_arguments(self, parser):
+        parser.add_argument("app_labels", nargs="*", type=str)
+
+    def _write_pks_per_model(self, fixtures: t.List[Fixture], indents: int = 0):
+        def get_model(fixture: Fixture):
+            return fixture.model.lower()
+
+        fixtures.sort(key=get_model)
+
+        self.stdout.write(f'{"    " * indents}Primary keys per model:')
+
+        for model, group in groupby(fixtures, key=get_model):
+            pks = [fixture.pk for fixture in group]
+            pks.sort()
+
+            self.stdout.write(f'{"    " * (indents + 1)}- {model}: {pks}')
+
+    def write_pks_per_model(self, fixtures: FixtureDict):
+        """Write all the sorted primary keys per model."""
+        self._write_pks_per_model(
+            [
+                fixture
+                for file_fixtures in fixtures.values()
+                for fixture in file_fixtures
+            ]
+        )
+
+    def write_pks_per_file(self, fixtures: FixtureDict):
+        """Write all the sorted primary keys per file, per model."""
+        self.stdout.write("Primary keys per file:")
+
+        for file, file_fixtures in fixtures.items():
+            self.stdout.write(f"    - {file}")
+            self._write_pks_per_model(file_fixtures, indents=2)
+
+    def handle(self, *args, **options):
+        fixtures: FixtureDict = {}
+        for app_label in {*options["app_labels"], *self.required_app_labels}:
+            app_config = apps.app_configs[app_label]
+            fixtures_path = os.path.join(app_config.path, "fixtures")
+
+            for fixture_name in os.listdir(fixtures_path):
+                fixture_path = os.path.join(fixtures_path, fixture_name)
+                with open(fixture_path, "r", encoding="utf-8") as fixture:
+                    fixtures[fixture_path] = [
+                        Fixture(**fixture) for fixture in json.load(fixture)
+                    ]
+
+        self.write_pks_per_model(fixtures)
+        self.stdout.write()
+        self.write_pks_per_file(fixtures)

codeforlife/user/management/commands/load_fixtures.py

@@ -3,36 +3,9 @@
 Created on 10/06/2024 at 10:44:45(+01:00).
 """
 
-import os
-import typing as t
-
-from django.apps import apps
-from django.core.management import call_command
-from django.core.management.base import BaseCommand
+from ....commands import LoadFixtures
 
 
 # pylint: disable-next=missing-class-docstring
-class Command(BaseCommand):
-    help = "Loads all the fixtures of the specified apps."
-
-    def add_arguments(self, parser):
-        parser.add_argument("app_labels", nargs="*", type=str)
-
-    def handle(self, *args, **options):
-        fixture_labels: t.List[str] = []
-        for app_label in {*options["app_labels"], "user"}:
-            app_config = apps.app_configs[app_label]
-            fixtures_path = os.path.join(app_config.path, "fixtures")
-
-            self.stdout.write(f"{app_label} fixtures ({fixtures_path}):")
-            for fixture_label in os.listdir(fixtures_path):
-                if fixture_label in fixture_labels:
-                    self.stderr.write(f"Duplicate fixture: {fixture_label}")
-                    return
-
-                self.stdout.write(f"    - {fixture_label}")
-                fixture_labels.append(fixture_label)
-
-            self.stdout.write()
-
-        call_command("loaddata", *fixture_labels)
+class Command(LoadFixtures):
+    required_app_labels = {"user"}

codeforlife/user/management/commands/summarize_fixtures.py

@@ -3,82 +3,9 @@
 Created on 22/02/2024 at 09:24:27(+00:00).
 """
 
-import json
-import os
-import typing as t
-from dataclasses import dataclass
-from itertools import groupby
-
-from django.apps import apps
-from django.core.management.base import BaseCommand
-
-from ....types import JsonDict
-
-
-@dataclass(frozen=True)
-class Fixture:
-    """A data model fixture."""
-
-    model: str
-    pk: t.Any
-    fields: JsonDict
-
-
-FixtureDict = t.Dict[str, t.List[Fixture]]
+from ....commands import SummarizeFixtures
 
 
 # pylint: disable-next=missing-class-docstring
-class Command(BaseCommand):
-    help = "Summarizes all the listed fixtures."
-
-    def add_arguments(self, parser):
-        parser.add_argument("app_labels", nargs="*", type=str)
-
-    def _write_pks_per_model(self, fixtures: t.List[Fixture], indents: int = 0):
-        def get_model(fixture: Fixture):
-            return fixture.model.lower()
-
-        fixtures.sort(key=get_model)
-
-        self.stdout.write(f'{"    " * indents}Primary keys per model:')
-
-        for model, group in groupby(fixtures, key=get_model):
-            pks = [fixture.pk for fixture in group]
-            pks.sort()
-
-            self.stdout.write(f'{"    " * (indents + 1)}- {model}: {pks}')
-
-    def write_pks_per_model(self, fixtures: FixtureDict):
-        """Write all the sorted primary keys per model."""
-        self._write_pks_per_model(
-            [
-                fixture
-                for file_fixtures in fixtures.values()
-                for fixture in file_fixtures
-            ]
-        )
-
-    def write_pks_per_file(self, fixtures: FixtureDict):
-        """Write all the sorted primary keys per file, per model."""
-        self.stdout.write("Primary keys per file:")
-
-        for file, file_fixtures in fixtures.items():
-            self.stdout.write(f"    - {file}")
-            self._write_pks_per_model(file_fixtures, indents=2)
-
-    def handle(self, *args, **options):
-        fixtures: FixtureDict = {}
-        for app_label in {*options["app_labels"], "user"}:
-            app_config = apps.app_configs[app_label]
-            fixtures_path = os.path.join(app_config.path, "fixtures")
-
-            for fixture_name in os.listdir(fixtures_path):
-                fixture_path = os.path.join(fixtures_path, fixture_name)
-                with open(fixture_path, "r", encoding="utf-8") as fixture:
-                    fixtures[fixture_path] = [
-                        Fixture(**fixture) for fixture in json.load(fixture)
-                    ]
-
-        self.write_pks_per_model(fixtures)
-        self.stdout.write()
-        self.write_pks_per_file(fixtures)
+class Command(SummarizeFixtures):
+    required_app_labels = {"user"}