Allow queries with same slug; "enabled" flag added

Author: morlandi

CHANGELOG.rst

@@ -3,6 +3,13 @@
 History
 =======
 
+v1.2.5
+------
+- "enabled" boolean field added
+- "slug" is no longer unique, and a Query.objects.get_query_for_slug(query_name)
+  is available for retriving the active query associated with a certain slug
+  and detect any duplication
+
 v1.2.4
 ------
 * QUERY_INSPECTOR_QUERY_STOCK_QUERIES can be either a list, or a callable which returns a list

query_inspector/admin.py

@@ -23,8 +23,10 @@
 @admin.register(Query)
 class QueryAdmin(admin.ModelAdmin):
 
-    list_display = ("slug", 'stock', "title", "list_parameters", )
-    list_filter = ('stock', 'from_view', 'from_materialized_view', )
+    list_display = ("slug", 'is_duplicated', 'enabled', 'stock', "title", "list_parameters", )
+    list_editable = ('enabled', )
+    list_filter = ('enabled', 'stock', 'from_view', 'from_materialized_view', )
+    search_fields = ['slug', ]
     save_on_top = True
 
     fieldsets = (

query_inspector/migrations/0007_query_enabled_alter_query_slug.py

@@ -0,0 +1,22 @@
+# Generated by Django 4.1.11 on 2024-02-22 09:25
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("query_inspector", "0006_query_from_materialized_view_query_from_view"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="query",
+            name="enabled",
+            field=models.BooleanField(default=True),
+        ),
+        migrations.AlterField(
+            model_name="query",
+            name="slug",
+            field=models.SlugField(),
+        ),
+    ]

query_inspector/models.py

@@ -10,16 +10,45 @@
 _named_parameters_sqlite_re = re.compile(r"\$([^ )]+)")
 
 
+class QueryManager(models.Manager):
+
+    def get_active_query_from_slug(self, slug):
+        """
+        Since slug is no longer unique, you can use this helper to identify the active query
+        associated with a certain slug (thus excluding all disabled queries);
+        in case of ambiguity it returns None
+
+        Example:
+
+            my_query = Query.objects.get_query_for_slug(query_name)
+
+        """
+        queryset = (
+            self.get_queryset()
+            .filter(
+                enabled=True,
+                #slug=slug,
+            )
+        )
+        return queryset.get(slug=slug)
+        # if queryset.count() == 1:
+        #     return queryset.first()
+        # return None
+
+
 class Query(models.Model):
     title = models.CharField(blank=True, max_length=128)
-    slug = models.SlugField(unique=True)
+    slug = models.SlugField(unique=False)
+    enabled = models.BooleanField(null=False, default=True)
     sql = models.TextField(null=False, blank=True)
     default_parameters = models.JSONField(null=False, default=dict, blank=True)
     notes = models.TextField(null=False, blank=True)
     stock = models.BooleanField(null=False, default=False, editable=False)
     from_view = models.BooleanField(null=False, default=False, editable=False)
     from_materialized_view = models.BooleanField(null=False, default=False, editable=False)
 
+    objects = QueryManager()
+
     class Meta:
         abstract = False
         verbose_name = _("Query")
@@ -40,6 +69,22 @@ def can_execute(self, request):
             return True
         return False
 
+    @property
+    def is_duplicated(self):
+        """
+        Returns True iif this query is enabled, and another enabled query
+        having the same slug has been detected
+        """
+        duplicated = False
+        if self.enabled:
+            try:
+                the_query = Query.objects.get_active_query_from_slug(self.slug)
+            except Query.MultipleObjectsReturned:
+                duplicated = True
+            except Exception as e:
+                pass
+        return duplicated
+
     def extract_named_parameters(self):
 
         def is_sqlite_engine():

query_inspector/templates/admin/query_inspector/query/change_list_results.html

@@ -0,0 +1,49 @@
+{% load i18n query_inspector_tags %}
+
+<style>
+  .duplicated, .duplicated a { color: #DD4646 !important; }
+  .dimmed, .dimmed a { color: #ccc !important; }
+</style>
+
+{% if result_hidden_fields %}
+<div class="hiddenfields">{# DIV for HTML validation #}
+{% for item in result_hidden_fields %}{{ item }}{% endfor %}
+</div>
+{% endif %}
+{% if results %}
+<div class="results">
+<table id="result_list">
+<thead>
+<tr>
+{% for header in result_headers %}
+<th scope="col"{{ header.class_attrib }}>
+   {% if header.sortable %}
+     {% if header.sort_priority > 0 %}
+       <div class="sortoptions">
+         <a class="sortremove" href="{{ header.url_remove }}" title="{% translate "Remove from sorting" %}"></a>
+         {% if num_sorted_fields > 1 %}<span class="sortpriority" title="{% blocktranslate with priority_number=header.sort_priority %}Sorting priority: {{ priority_number }}{% endblocktranslate %}">{{ header.sort_priority }}</span>{% endif %}
+         <a href="{{ header.url_toggle }}" class="toggle {% if header.ascending %}ascending{% else %}descending{% endif %}" title="{% translate "Toggle sorting" %}"></a>
+       </div>
+     {% endif %}
+   {% endif %}
+   <div class="text">{% if header.sortable %}<a href="{{ header.url_primary }}">{{ header.text|capfirst }}</a>{% else %}<span>{{ header.text|capfirst }}</span>{% endif %}</div>
+   <div class="clear"></div>
+</th>{% endfor %}
+</tr>
+</thead>
+<tbody>
+{% for result in results %}
+    {% if result.form and result.form.non_field_errors %}
+        <tr><td colspan="{{ result|length }}">{{ result.form.non_field_errors }}</td></tr>
+    {% endif %}
+
+    {% obj_from_result_list cl.result_list forloop.counter0 as query_obj %}
+    <tr class="{% if not query_obj.enabled %}dimmed {% endif %}{% if query_obj.is_duplicated %}duplicated {% endif %}">
+        {% for item in result %}{{ item }}{% endfor %}
+    </tr>
+
+{% endfor %}
+</tbody>
+</table>
+</div>
+{% endif %}

query_inspector/templatetags/query_inspector_tags.py

@@ -43,6 +43,10 @@ def ipdb(element):
     ipdb.set_trace()
     return element
 
+@register.simple_tag
+def obj_from_result_list(result_list, index):
+    instance = result_list[index]
+    return instance
 
 @register.filter
 def format_datetime(dt, include_time=True, include_seconds=False, exclude_date=False):