move the rule configuration to the dataspace

Signed-off-by: tdruez <tdruez@aboutcode.org>

Author: tdruez

dje/admin.py

@@ -1141,7 +1141,11 @@ class DataspaceConfigurationInline(DataspacedFKMixin, admin.StackedInline):
         ),
     ]
     # Do not include the Dataspace related FKs on addition as the Dataspace does not exist yet
-    fieldsets = [("", {"fields": ("homepage_layout",)})] + add_fieldsets
+    policy_rules_fieldset = (
+        "Policy Rules",
+        {"fields": ("policy_rules_config",)},
+    )
+    fieldsets = [("", {"fields": ("homepage_layout",)})] + add_fieldsets + [policy_rules_fieldset]
     can_delete = False
 
     def get_fieldsets(self, request, obj=None):

dje/migrations/0016_dataspaceconfiguration_policy_rules_config.py

@@ -0,0 +1,18 @@
+# Generated by Django 6.0.6 on 2026-07-15 08:25
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dje', '0015_alter_dataspaceconfiguration_purldb_api_key_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='dataspaceconfiguration',
+            name='policy_rules_config',
+            field=models.JSONField(blank=True, default=dict, help_text='Override default policy rule settings for this dataspace. '),
+        ),
+    ]

dje/models.py

@@ -614,6 +614,14 @@ class DataspaceConfiguration(DataspaceForeignKeyValidationMixin, models.Model):
         ),
     )
 
+    policy_rules_config = models.JSONField(
+        blank=True,
+        default=dict,
+        help_text=_(
+            "Override default policy rule settings for this dataspace. "
+        ),
+    )
+
     def __str__(self):
         return f"{self.dataspace}"
 

policy/admin.py

@@ -23,12 +23,9 @@
 from dje.admin import dejacode_site
 from dje.list_display import AsColored
 from policy.forms import AssociatedPolicyForm
-from policy.forms import PolicyRuleForm
 from policy.forms import UsagePolicyForm
 from policy.models import AssociatedPolicy
-from policy.models import PolicyRule
 from policy.models import UsagePolicy
-from policy.rules import RULE_REGISTRY
 
 License = apps.get_model("license_library", "license")
 
@@ -162,37 +159,3 @@ def download_license_dump_view(self, request):
         response["Content-Disposition"] = 'attachment; filename="license_policies.yml"'
 
         return response
-
-
-@admin.register(PolicyRule, site=dejacode_site)
-class PolicyRuleAdmin(DataspacedAdmin):
-    form = PolicyRuleForm
-    list_display = ("name", "rule_type", "threshold", "is_active", "get_dataspace")
-    list_filter = DataspacedAdmin.list_filter + ("rule_type", "is_active")
-    readonly_fields = DataspacedAdmin.readonly_fields + ("parameters_schema_hint",)
-    activity_log = False
-    actions = []
-    actions_to_remove = ["copy_to", "compare_with"]
-    email_notification_on = ()
-
-    short_description = (
-        "You can define Policy Rules that automatically detect compliance violations "
-        "across your products and trigger notifications."
-    )
-
-    long_description = linebreaksbr(
-        "A Policy Rule defines a type of automated check to run against your products. "
-        "When the number of detected issues exceeds the configured threshold, a "
-        "ProductPolicyViolation is recorded.\n"
-        "Set the rule type to match a registered evaluation handler and configure the "
-        "threshold (0 means any violation triggers the rule). "
-    )
-
-    def parameters_schema_hint(self, obj):
-        handler = RULE_REGISTRY.get(obj.rule_type)
-        if not handler or not handler.parameters_schema:
-            return "No parameters supported for this rule type."
-        lines = [f"{key}: {desc}" for key, desc in handler.parameters_schema.items()]
-        return mark_safe("<br>".join(lines))
-
-    parameters_schema_hint.short_description = "Supported parameters"

policy/engine.py

@@ -8,34 +8,47 @@
 
 from django.utils import timezone
 
-from policy.models import PolicyRule
 from policy.rules import RULE_REGISTRY
 from product_portfolio.models import ProductPolicyViolation
 
 
-def evaluate_rule(policy_rule, product):
+def get_effective_config(rule_type, dataspace):
     """
-    Evaluate a single PolicyRule against a product, create or update the
-    ProductPolicyViolation record.
-    Returns the ProductPolicyViolation instance, or None if no violation exists.
+    Resolve threshold, parameters, and is_active for a rule type in a given dataspace.
+
+    Reads the dataspace-level override from DataspaceConfiguration.policy_rules_config,
+    falling back to the code defaults defined on the rule handler.
     """
-    rule_handler = RULE_REGISTRY.get(policy_rule.rule_type)
-    if not rule_handler:
-        return
+    handler = RULE_REGISTRY[rule_type]
+    try:
+        rule_config = dataspace.configuration.policy_rules_config.get(rule_type, {})
+    except AttributeError:
+        rule_config = {}
+
+    return {
+        "is_active": rule_config.get("is_active", True),
+        "threshold": rule_config.get("threshold", handler.default_threshold),
+        "parameters": rule_config.get("parameters", {}),
+    }
+
 
-    violation_count = rule_handler.count_violations(policy_rule, product)
+def evaluate_rule(rule_type, product, threshold, parameters):
+    """Evaluate a single rule against a product and record the violation if triggered."""
+    handler = RULE_REGISTRY[rule_type]
+    violation_count = handler.count_violations(product, threshold, parameters)
 
-    lookup = {"policy_rule": policy_rule, "product": product, "resolved": False}
+    lookup = {"rule_type": rule_type, "product": product, "resolved": False}
 
     if violation_count > 0:
         violation, created = ProductPolicyViolation.objects.get_or_create(
             **lookup,
-            defaults={"dataspace": policy_rule.dataspace, "violation_count": violation_count},
+            defaults={"dataspace": product.dataspace, "violation_count": violation_count},
         )
         if not created:
             violation.violation_count = violation_count
             violation.save()
         return violation
+
     else:
         ProductPolicyViolation.objects.filter(**lookup).update(
             resolved=True,
@@ -46,12 +59,17 @@ def evaluate_rule(policy_rule, product):
 
 def evaluate_rules(product):
     """
-    Evaluate all active PolicyRules for the given product.
+    Evaluate all rules in RULE_REGISTRY for the given product.
+
     Returns the list of active ProductPolicyViolation instances.
     """
     violations = []
-    for policy_rule in PolicyRule.objects.scope(product.dataspace).active():
-        violation = evaluate_rule(policy_rule, product)
+    for rule_type in RULE_REGISTRY:
+        config = get_effective_config(rule_type, product.dataspace)
+        if not config["is_active"]:
+            continue
+
+        violation = evaluate_rule(rule_type, product, config["threshold"], config["parameters"])
         if violation:
             violations.append(violation)
 

policy/forms.py

@@ -11,8 +11,6 @@
 
 from dje.forms import ColorCodeFormMixin
 from dje.forms import DataspacedAdminForm
-from policy.models import PolicyRule
-from policy.rules import RULE_REGISTRY
 
 
 class UsagePolicyForm(ColorCodeFormMixin, DataspacedAdminForm):
@@ -94,21 +92,3 @@ def get_ct(app_label, model):
             self.add_error("to_policy", msg)
 
         return cleaned_data
-
-
-class PolicyRuleForm(DataspacedAdminForm):
-    class Meta:
-        model = PolicyRule
-        fields = "__all__"
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.fields["rule_type"].widget = forms.Select(
-            choices=[(key, handler.label) for key, handler in RULE_REGISTRY.items()]
-        )
-
-    def clean_rule_type(self):
-        value = self.cleaned_data["rule_type"]
-        if value not in RULE_REGISTRY:
-            raise forms.ValidationError(f"Unknown rule type: {value}")
-        return value

policy/migrations/0003_policyrule.py

@@ -20,7 +20,6 @@
 from dje.models import DataspacedModel
 from dje.models import DataspacedQuerySet
 from dje.models import colored_icon_mixin_factory
-from policy.rules import RULE_REGISTRY
 
 ColoredIconMixin = colored_icon_mixin_factory(
     verbose_name="usage policy",
@@ -247,57 +246,6 @@ def save(self, *args, **kwargs):
         super().save(*args, **kwargs)
 
 
-class PolicyRuleQuerySet(DataspacedQuerySet):
-    def active(self):
-        return self.filter(is_active=True)
-
-
-class PolicyRule(DataspacedModel):
-    name = models.CharField(
-        max_length=100,
-        help_text=_("Descriptive name for this policy rule."),
-    )
-    rule_type = models.CharField(
-        max_length=50,
-        help_text=_("The type of evaluation performed by this rule."),
-    )
-    threshold = models.PositiveIntegerField(
-        default=0,
-        help_text=_("Minimum number of violations required to trigger this rule (0 means any)."),
-    )
-    is_active = models.BooleanField(
-        default=True,
-        help_text=_("Only active rules are evaluated."),
-    )
-    parameters = models.JSONField(
-        blank=True,
-        default=dict,
-        help_text=_(
-            "Optional rule-specific parameters as a JSON object. "
-            "Supported keys depend on the chosen rule type."
-        ),
-    )
-
-    objects = PolicyRuleQuerySet.as_manager()
-
-    class Meta:
-        unique_together = (("dataspace", "uuid"), ("dataspace", "name"))
-        ordering = ["name"]
-
-    def __str__(self):
-        return self.name
-
-    @property
-    def rule_label(self):
-        handler = RULE_REGISTRY.get(self.rule_type)
-        return handler.label if handler else self.rule_type
-
-    @property
-    def rule_description(self):
-        handler = RULE_REGISTRY.get(self.rule_type)
-        return handler.description if handler else ""
-
-
 class AbstractPolicyViolation(models.Model):
     """Shared fields for all concrete policy violation models. No DB table."""
 

policy/models.py

@@ -17,7 +17,7 @@ class BaseRule:
     description = None
     parameters_schema = {}
 
-    def count_violations(self, policy_rule, product):
+    def count_violations(self, product, threshold, parameters):
         """Count objects violating the rule for the given product."""
         raise NotImplementedError
 
@@ -27,15 +27,15 @@ class PackageBaseRule(BaseRule):
 
     package_filter = {}
 
-    def count_violations(self, policy_rule, product):
+    def count_violations(self, product, threshold, parameters):
         Package = apps.get_model("component_catalog", "package")
 
         count = Package.objects.filter(
             productpackages__product=product,
             **self.package_filter,
         ).count()
 
-        return count if count > policy_rule.threshold else 0
+        return count if count > threshold else 0
 
 
 class LicensePolicyErrorRule(PackageBaseRule):
@@ -73,20 +73,20 @@ class VulnerabilityDetectedRule(BaseRule):
         "min_risk_score": "Minimum risk score (0.0-10.0). Default: any vulnerability.",
     }
 
-    def count_violations(self, policy_rule, product):
+    def count_violations(self, product, threshold, parameters):
         Package = apps.get_model("component_catalog", "package")
 
         packages = Package.objects.filter(
             productpackages__product=product,
             risk_score__isnull=False,
         )
 
-        min_risk_score = policy_rule.parameters.get("min_risk_score")
+        min_risk_score = parameters.get("min_risk_score")
         if min_risk_score is not None:
             packages = packages.filter(risk_score__gte=min_risk_score)
 
         count = packages.count()
-        return count if count > policy_rule.threshold else 0
+        return count if count > threshold else 0
 
 
 RULE_REGISTRY = {

policy/rules.py

@@ -1,4 +1,4 @@
-# Generated by Django 6.0.6 on 2026-07-08 08:39
+# Generated by Django 6.0.6 on 2026-07-15 08:25
 
 import django.db.models.deletion
 import dje.models
@@ -9,8 +9,7 @@
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('dje', '0015_alter_dataspaceconfiguration_purldb_api_key_and_more'),
-        ('policy', '0003_policyrule'),
+        ('dje', '0016_dataspaceconfiguration_policy_rules_config'),
         ('product_portfolio', '0017_scancodeproject_import_options'),
     ]
 
@@ -25,13 +24,13 @@ class Migration(migrations.Migration):
                 ('last_checked', models.DateTimeField(auto_now=True, help_text='The date and time of the last evaluation.')),
                 ('resolved', models.BooleanField(default=False, help_text='Indicates whether this violation has been resolved.')),
                 ('resolved_date', models.DateTimeField(blank=True, help_text='The date and time when this violation was resolved.', null=True)),
+                ('rule_type', models.CharField(help_text='The rule type from the rule registry that triggered this violation.', max_length=50)),
                 ('dataspace', models.ForeignKey(editable=False, help_text='A Dataspace is an independent, exclusive set of DejaCode data, which can be either nexB master reference data or installation-specific data.', on_delete=django.db.models.deletion.PROTECT, to='dje.dataspace')),
-                ('policy_rule', models.ForeignKey(help_text='The policy rule that triggered this violation.', on_delete=django.db.models.deletion.CASCADE, related_name='product_violations', to='policy.policyrule')),
                 ('product', models.ForeignKey(help_text='The product in the context of which this violation was detected.', on_delete=django.db.models.deletion.CASCADE, related_name='policy_violations', to='product_portfolio.product')),
             ],
             options={
                 'ordering': ['-detected_date'],
-                'unique_together': {('dataspace', 'uuid'), ('policy_rule', 'product')},
+                'unique_together': {('dataspace', 'uuid'), ('rule_type', 'product')},
             },
             bases=(dje.models.DataspaceForeignKeyValidationMixin, models.Model),
         ),