[Not intended for use] Historical records

CodeListLibrary_project/clinicalcode/admin.py

@@ -115,19 +115,10 @@ def save_model(self, request, obj, form, change):
         obj.modified = timezone.now()
         obj.save()
 
-#admin.site.register(CodingSystem)
-
-# ############################################
-# # Unregister the original Group admin.
-# admin.site.unregister(Group)
-#
-# # Create a new Group admin.
-# class GroupAdmin(admin.ModelAdmin):
-#     # Use our custom form.
-#     form = GroupAdminForm
-#     #form_class = GroupAdminForm
-#     # Filter permissions horizontal as well.
-#     filter_horizontal = ['permissions']
-#
-# # Register the new Group ModelAdmin.
-# admin.site.register(Group, GroupAdmin)
+# Tests
+from .models.ClinicalConcept import ClinicalConcept
+
+@admin.register(ClinicalConcept)
+class ClinicalConceptAdmin(admin.ModelAdmin):
+    list_display = ['id', 'name']
+    exclude = []

CodeListLibrary_project/clinicalcode/apps.py

@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+class ClinicalCodeAppConfig(AppConfig):
+    name = 'clinicalcode'
+
+    def ready(self):
+        from . import signals

CodeListLibrary_project/clinicalcode/entity_utils/constants.py

@@ -15,6 +15,14 @@ def __contains__(cls, lhs):
         else:
             return True
 
+class HISTORICAL_CHANGE_TYPE(int, enum.Enum):
+    '''
+        Historical change type for History Mixin
+    '''
+    CREATED = 1
+    EDITED = 2
+    DELETED = 3
+
 class TAG_TYPE(int, enum.Enum):
     '''
         Tag types used for differentiate Collections & Tags

CodeListLibrary_project/clinicalcode/management/commands/concept_migrate.py

@@ -0,0 +1,89 @@
+from django.core.management.base import BaseCommand
+
+import os
+import glob
+import json
+
+from ...models.ClinicalConcept import ClinicalConcept
+from ...models.ClinicalConcept import ClinicalRuleset
+
+from ...models.CodingSystem import CodingSystem
+from ...models.Concept import Concept
+from ...models.Component import Component
+from ...models.CodeList import CodeList
+from ...models.Code import Code
+
+class Command(BaseCommand):
+    IS_DEBUG = True
+
+    help = 'Example usage of historical models'
+
+    def __get_log_style(self, style):
+        if not isinstance(style, str):
+            return style
+
+        style = style.upper()
+        if style == 'SUCCESS':
+            return self.style.SUCCESS
+        elif style == 'NOTICE':
+            return self.style.NOTICE
+        elif style == 'WARNING':
+            return self.style.WARNING
+        elif style == 'ERROR':
+            return self.style.ERROR
+        return self.style.SUCCESS
+
+    def __log(self, message, style='SUCCESS'):
+        style = self.__get_log_style(style)
+        self.stdout.write(style(message))
+
+    def __migrate_concept(self, *args, **kwargs):
+        ids = kwargs.get('ids') or 'ALL'
+        ids = ids.upper().split(',')
+        if ids[0] == 'ALL':
+            ids = list(Concept.objects.all().values_list('id', flat=True))
+
+        self.__log(f'Migrating Concepts #{len(ids)}')
+
+    def __test_historical_records(self, *args, **kwargs):
+        # Create Concept
+        self.__log('Create Concept')
+        clin_concept = ClinicalConcept(
+            name='COVID-19 with rules',
+            coding_system=CodingSystem.objects.get(id=4)
+        )
+        clin_concept.save()
+        self.__log(f'Created initial: ClinicalConcept<id={clin_concept.entity_id}, version={clin_concept.version_id}>', style=self.style.MIGRATE_LABEL)
+
+        # Add a rule
+        clin_ruleset = ClinicalRuleset(name='Some ruleset')
+        clin_ruleset.save()
+        clin_concept.rulesets.add(clin_ruleset)
+
+        # Update the concept for some reason
+        self.__log('Update Concept')
+        clin_concept.name = 'COVID-19 with change'
+        clin_concept.save()
+        self.__log(f'Diff: {clin_concept.history.get(version_id=1).get_delta(clin_concept)}', style=self.style.MIGRATE_HEADING)
+
+        # Remove the ruleset(s) and update the Concept
+        self.__log('Remove Rules')
+        clin_concept.name = 'COVID-19 with removed rules'
+        clin_concept.coding_system = CodingSystem.objects.get(id=7)
+        clin_concept.save()
+        clin_concept.rulesets.clear()
+        self.__log(f'Diff: {clin_concept.history.get(version_id=2).get_delta(clin_concept)}', style=self.style.MIGRATE_HEADING)
+
+        self.__log('Result:')
+        self.__log(f'1. ClinicalConcept<id={clin_concept.entity_id}, version={clin_concept.version_id}>', style=self.style.MIGRATE_LABEL)
+        self.__log(f'2. Historical records: {list(clin_concept.history.all())}', style=self.style.MIGRATE_LABEL)
+
+    def add_arguments(self, parser):
+        if not self.IS_DEBUG:
+            parser.add_argument('-i', '--ids', type=str, help='Only migrate a specific instance, or instances (using a comma delimiter)')
+
+    def handle(self, *args, **kwargs):
+        if not self.IS_DEBUG:
+            self.__migrate_concept(*args, **kwargs)
+            return
+        self.__test_historical_records(*args, **kwargs)

CodeListLibrary_project/clinicalcode/managers/HistoricalManager.py

@@ -0,0 +1,63 @@
+from django.db import models
+from django.db.models import QuerySet
+
+from ..entity_utils import constants
+
+class HistoricalQuerySet(QuerySet):
+    '''
+        Replicate QuerySet behaviour of historical records
+
+        e.g.
+            ClinicalConcept.history.get(entity_id__in=[1, 2]).latest_of_each()
+
+    '''
+    def filter(self, *args, **kwargs):
+        return super().filter(*args, **kwargs)
+
+    def latest_of_each(self):
+        latest_versions = self.order_by('-entity_id', '-version_date') \
+                                .distinct('entity_id') \
+                                .values_list('pk', flat=True)
+        return self.filter(pk__in=latest_versions)
+
+class CurrentRecordManager(models.Manager):    
+    def get_super_queryset(self):
+        queryset = super().get_queryset()
+        latest_live_versions = queryset.exclude(change_reason=constants.HISTORICAL_CHANGE_TYPE.DELETED.value) \
+                                .order_by('-entity_id', '-version_date') \
+                                .distinct('entity_id') \
+                                .values_list('pk', flat=True)
+        return queryset.filter(pk__in=latest_live_versions)
+
+    def get_queryset(self):
+        return self.get_super_queryset()
+
+class HistoricalRecordManager(models.Manager):
+    '''
+        Replicate class level behaviour of historical records
+
+        e.g.
+            ClinicalConcept.history.all()
+    '''
+    def __init__(self, model, instance=None):
+        super().__init__()
+        self.model = model
+        self.instance = instance
+
+    def get_super_queryset(self):
+        return super().get_queryset()
+
+    def get_queryset(self):
+        if self.instance is None:
+            return self.get_super_queryset()
+        return self.get_super_queryset().filter(entity_id=self.instance.entity_id)
+
+    def most_recent(self):
+        if not self.instance:
+            return
+        return self.get_queryset().order_by('-version_date').first()
+
+    def earliest(self):
+        if not self.instance:
+            return
+        return self.get_queryset().order_by('version_date').first()

CodeListLibrary_project/clinicalcode/migrations/0099_clinicalcodeitem_clinicalconcept_legacyclinicaldata_and_more.py

@@ -0,0 +1,115 @@
+# Generated by Django 4.0.10 on 2023-04-26 20:01
+
+import clinicalcode.entity_utils.constants
+import clinicalcode.mixins.DeltaMixin
+from django.conf import settings
+import django.contrib.postgres.fields
+from django.db import migrations, models
+import django.db.models.deletion
+import django.utils.timezone
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('auth', '0012_alter_user_first_name_max_length'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('clinicalcode', '0098_brand_footer_images_historicalbrand_footer_images'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ClinicalCodeItem',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('attributes', django.contrib.postgres.fields.ArrayField(base_field=models.CharField(max_length=250), blank=True, null=True, size=None)),
+                ('code', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='clinicalcode.code')),
+            ],
+        ),
+        migrations.CreateModel(
+            name='ClinicalConcept',
+            fields=[
+                ('id', models.AutoField(primary_key=True, serialize=False)),
+                ('entity_id', models.BigIntegerField(db_index=True, editable=False)),
+                ('version_id', models.BigIntegerField(db_index=True, default=1, editable=False)),
+                ('version_date', models.DateTimeField(blank=True, db_index=True, default=django.utils.timezone.now, editable=False)),
+                ('created_date', models.DateTimeField(blank=True, db_index=True, default=django.utils.timezone.now, editable=False)),
+                ('change_reason', models.TextField(blank=True, null=True)),
+                ('change_type', models.IntegerField(choices=[('CREATED', 1), ('EDITED', 2), ('DELETED', 3)], default=clinicalcode.entity_utils.constants.HISTORICAL_CHANGE_TYPE['CREATED'])),
+                ('name', models.CharField(max_length=250)),
+                ('code_attribute_header', django.contrib.postgres.fields.ArrayField(base_field=models.CharField(max_length=100), blank=True, null=True, size=None)),
+                ('coding_system', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='coded_concepts', to='clinicalcode.codingsystem')),
+                ('parent', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='child_concepts', to='clinicalcode.genericentity')),
+                ('root_concept', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='forked_concepts', to='clinicalcode.clinicalconcept')),
+            ],
+            options={
+                'ordering': ('name',),
+            },
+            bases=(models.Model, clinicalcode.mixins.DeltaMixin.DeltaModelMixin),
+        ),
+        migrations.CreateModel(
+            name='LegacyClinicalData',
+            fields=[
+                ('id', models.AutoField(primary_key=True, serialize=False)),
+                ('entry_date', models.DateField()),
+                ('author', models.CharField(max_length=1000)),
+                ('description', models.TextField(blank=True, null=True)),
+                ('tags', django.contrib.postgres.fields.ArrayField(base_field=models.IntegerField(), blank=True, null=True, size=None)),
+                ('collections', django.contrib.postgres.fields.ArrayField(base_field=models.IntegerField(), blank=True, null=True, size=None)),
+                ('validation_performed', models.BooleanField(default=False, null=True)),
+                ('validation_description', models.TextField(blank=True, null=True)),
+                ('publication_doi', models.CharField(max_length=100)),
+                ('publication_link', models.URLField(max_length=1000)),
+                ('secondary_publication_links', models.TextField(blank=True, null=True)),
+                ('paper_published', models.BooleanField(default=False, null=True)),
+                ('source_reference', models.CharField(max_length=250)),
+                ('citation_requirements', models.TextField(blank=True, null=True)),
+                ('parent', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='legacy_details', to='clinicalcode.clinicalconcept')),
+            ],
+        ),
+        migrations.CreateModel(
+            name='ContentPermission',
+            fields=[
+                ('id', models.AutoField(primary_key=True, serialize=False)),
+                ('world_access', models.IntegerField(choices=[('NONE', 1), ('VIEW', 2), ('EDIT', 3)], default=clinicalcode.entity_utils.constants.GROUP_PERMISSIONS['NONE'])),
+                ('group', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, to='auth.group')),
+                ('owner', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'ordering': ('id',),
+            },
+        ),
+        migrations.CreateModel(
+            name='ClinicalRuleset',
+            fields=[
+                ('id', models.AutoField(primary_key=True, serialize=False)),
+                ('name', models.CharField(max_length=250)),
+                ('source_type', models.IntegerField(choices=[('CONCEPT', 1), ('QUERY_BUILDER', 2), ('EXPRESSION', 3), ('SELECT_IMPORT', 4), ('FILE_IMPORT', 5), ('SEARCH_TERM', 6)], default=clinicalcode.entity_utils.constants.CLINICAL_CODE_SOURCE['SEARCH_TERM'])),
+                ('logical_type', models.IntegerField(choices=[('INCLUDE', 1), ('EXCLUDE', 2)], default=clinicalcode.entity_utils.constants.CLINICAL_RULE_TYPE['INCLUDE'])),
+                ('codes', models.ManyToManyField(blank=True, related_name='rulesets', through='clinicalcode.ClinicalCodeItem', to='clinicalcode.code')),
+            ],
+            options={
+                'ordering': ('name',),
+            },
+        ),
+        migrations.AddField(
+            model_name='clinicalconcept',
+            name='rulesets',
+            field=models.ManyToManyField(blank=True, related_name='parent_concepts', to='clinicalcode.clinicalruleset'),
+        ),
+        migrations.AddField(
+            model_name='clinicalcodeitem',
+            name='ruleset',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='clinicalcode.clinicalruleset'),
+        ),
+        migrations.AddField(
+            model_name='genericentity',
+            name='permissions',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='permitted_entities', to='clinicalcode.contentpermission'),
+        ),
+        migrations.AddField(
+            model_name='historicalgenericentity',
+            name='permissions',
+            field=models.ForeignKey(blank=True, db_constraint=False, null=True, on_delete=django.db.models.deletion.DO_NOTHING, related_name='+', to='clinicalcode.contentpermission'),
+        ),
+    ]