Add management command

Author: AlanCoding

ansible_base/lib/management/__init__.py

@@ -0,0 +1 @@
+# Generated by Claude Sonnet 4

ansible_base/lib/management/advisory_locks.py

@@ -0,0 +1,79 @@
+# Generated by Claude Sonnet 4
+import json
+
+from django.core.management.base import BaseCommand
+
+from ansible_base.lib.utils.db import advisory_lock_id_to_debug_info, get_active_advisory_locks
+
+
+class AdvisoryLocksCommand(BaseCommand):
+    """Base command class for advisory lock management."""
+
+    help = "Show currently active PostgreSQL advisory locks"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--format',
+            choices=['table', 'json'],
+            default='table',
+            help='Output format (default: table)',
+        )
+        parser.add_argument(
+            '--show-debug-info',
+            action='store_true',
+            help='Include debug information for lock IDs',
+        )
+
+    def handle(self, *args, **options):
+        locks = get_active_advisory_locks()
+
+        if not locks:
+            self.stdout.write("No active advisory locks found.")
+            return
+
+        if options['format'] == 'json':
+            self._output_json(locks, options['show_debug_info'])
+        else:
+            self._output_table(locks, options['show_debug_info'])
+
+    def _output_json(self, locks, show_debug_info):
+        """Output locks in JSON format."""
+        if show_debug_info:
+            for lock in locks:
+                # Add debug info for the objid (main lock ID)
+                lock['debug_info'] = advisory_lock_id_to_debug_info(lock['objid'])
+
+        self.stdout.write(json.dumps(locks, indent=2, default=str))
+
+    def _output_table(self, locks, show_debug_info):
+        """Output locks in table format."""
+        # Header
+        headers = ['PID', 'Mode', 'Granted', 'Class ID', 'Object ID', 'Object Sub ID']
+        if show_debug_info:
+            headers.extend(['Hex', 'Unsigned CRC32', 'High Bit'])
+
+        self.stdout.write(self.style.SUCCESS(' | '.join(headers)))
+        self.stdout.write('-' * (len(' | '.join(headers)) + 20))
+
+        # Rows
+        for lock in locks:
+            row = [
+                str(lock['pid']),
+                lock['mode'],
+                'Yes' if lock['granted'] else 'No',
+                str(lock['classid']),
+                str(lock['objid']),
+                str(lock['objsubid']),
+            ]
+
+            if show_debug_info:
+                debug_info = advisory_lock_id_to_debug_info(lock['objid'])
+                row.extend(
+                    [
+                        debug_info['hex_representation'],
+                        str(debug_info['unsigned_crc32']),
+                        'Yes' if debug_info['had_high_bit_set'] else 'No',
+                    ]
+                )
+
+            self.stdout.write(' | '.join(row))

test_app/management/commands/advisory_locks.py

@@ -0,0 +1,8 @@
+# Generated by Claude Sonnet 4
+from ansible_base.lib.management.advisory_locks import AdvisoryLocksCommand
+
+
+class Command(AdvisoryLocksCommand):
+    """Management command to show active PostgreSQL advisory locks."""
+
+    pass