diff --git a/backend/core/migrations/0070_complianceassessmentsnapshot.py b/backend/core/migrations/0070_complianceassessmentsnapshot.py new file mode 100644 index 000000000..5478f9d97 --- /dev/null +++ b/backend/core/migrations/0070_complianceassessmentsnapshot.py @@ -0,0 +1,54 @@ +# Generated by Django 5.1.8 on 2025-04-25 19:06 + +import django.db.models.deletion +import uuid +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0069_auto_20250414_2023"), + ] + + operations = [ + migrations.CreateModel( + name="ComplianceAssessmentSnapshot", + fields=[ + ( + "id", + models.UUIDField( + default=uuid.uuid4, + editable=False, + primary_key=True, + serialize=False, + ), + ), + ( + "created_at", + models.DateTimeField(auto_now_add=True, verbose_name="Created at"), + ), + ( + "updated_at", + models.DateTimeField(auto_now=True, verbose_name="Updated at"), + ), + ( + "is_published", + models.BooleanField(default=False, verbose_name="published"), + ), + ("format", models.IntegerField(default=1)), + ("revision", models.IntegerField(default=1)), + ("key_indicators", models.JSONField()), + ("data", models.JSONField()), + ( + "compliance_assessment", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + to="core.complianceassessment", + ), + ), + ], + options={ + "abstract": False, + }, + ), + ] diff --git a/backend/core/models.py b/backend/core/models.py index e130ac602..5775c0b81 100644 --- a/backend/core/models.py +++ b/backend/core/models.py @@ -3059,7 +3059,7 @@ class Meta: verbose_name = _("Compliance assessment") verbose_name_plural = _("Compliance assessments") - def upsert_daily_metrics(self): + def key_indicators(self): per_status = dict() per_result = dict() for item in self.get_requirements_status_count(): @@ -3067,7 +3067,11 @@ def upsert_daily_metrics(self): for item in self.get_requirements_result_count(): per_result[item[1]] = item[0] - total = RequirementAssessment.objects.filter(compliance_assessment=self).count() + total = ( + RequirementAssessment.objects.filter(compliance_assessment=self) + .filter(requirement__assessable=True) + .count() + ) data = { "reqs": { "total": total, @@ -3077,7 +3081,40 @@ def upsert_daily_metrics(self): "score": self.get_global_score(), }, } + return data + + def create_snapshot(self): + logger.info(f"creating snapshot for {self.name}") + key_indicators = self.key_indicators() + requirements_assessments = [] + for ra in RequirementAssessment.objects.filter(compliance_assessment=self): + # ).exclude(result=RequirementAssessment.Result.NOT_ASSESSED):#probably a risky optimization + entry = { + "id": str(ra.id), + "result": ra.result, + "status": ra.status, + "score": ra.score, + "doc_score": ra.documentation_score, + } + requirements_assessments.append(entry) + data = {"requirements_assessments": requirements_assessments} + ac = [ + {"id": str(ac.id), "status": ac.status} + for ac in AppliedControl.objects.filter( + requirement_assessments__compliance_assessment=self + ) + ] + if len(ac) > 0: + data.update({"applied_controls": ac}) + snapshot = ComplianceAssessmentSnapshot.objects.create_snapshot( + compliance_assessment=self, + key_indicators=key_indicators, + data=data, + ) + logger.info(f"created snapshot {snapshot.id} for {self.name}") + def upsert_daily_metrics(self): + data = self.key_indicators() HistoricalMetric.update_daily_metric( model=self.__class__.__name__, object_id=self.id, data=data ) @@ -4338,6 +4375,39 @@ class Meta: verbose_name_plural = "Task nodes" +class SnapshotManager(models.Manager): + def create_snapshot(self, compliance_assessment, key_indicators, data, format=1): + latest = ( + self.filter(compliance_assessment=compliance_assessment) + .order_by("-revision") + .first() + ) + + revision = 1 + if latest: + revision = latest.revision + 1 + + return self.create( + compliance_assessment=compliance_assessment, + revision=revision, + key_indicators=key_indicators, + data=data, + format=format, + ) + + +class ComplianceAssessmentSnapshot(AbstractBaseModel): + format = models.IntegerField(default=1) + compliance_assessment = models.ForeignKey( + "ComplianceAssessment", on_delete=models.CASCADE + ) + revision = models.IntegerField(default=1) + key_indicators = models.JSONField() + data = models.JSONField() + + objects = SnapshotManager() + + common_exclude = ["created_at", "updated_at"] auditlog.register(