Merge pull request #6308 from specify/issue-6266

.env

@@ -47,4 +47,4 @@ LOG_LEVEL=WARNING
 # should only be used during development and troubleshooting and not
 # during general use. Django applications leak memory when operated
 # continuously in debug mode.
-SP7_DEBUG=true
+SP7_DEBUG=true

docker-entrypoint.sh

@@ -12,6 +12,7 @@ if [ "$1" = 've/bin/gunicorn' ] || [ "$1" = 've/bin/python' ]; then
   set +e
   ve/bin/python manage.py base_specify_migration
   ve/bin/python manage.py migrate
+  # ve/bin/python manage.py run_key_migration_functions # Uncomment if you want the key migration functions to run on startup.
   set -e
 fi
 exec "$@"

specifyweb/backend/businessrules/migration_utils.py

@@ -0,0 +1,75 @@
+from typing import List
+
+from specifyweb.backend.businessrules.uniqueness_rules import create_uniqueness_rule
+
+
+def catnum_rule_editable(apps, schema_editor=None):
+    """ Find any CollectionObject catalogNumber must be unique to Collection 
+    rules which are readonly on the frontend (have isDatabaseConstraint=True)
+    and set their isDatabaseConstraint=False.
+
+    Generally should be run only after migration businessrules/0003 has been 
+    applied
+    """
+    UniquenessRule = apps.get_model("businessrules", "UniquenessRule")
+
+    model_rules = UniquenessRule.objects.filter(modelName="Collectionobject", isDatabaseConstraint=True)
+
+    catalog_number_rules: List[int] = []
+    for rule in model_rules: 
+        rule_fields = rule.uniquenessrulefield_set.all()
+
+        fields = rule_fields.filter(isScope=False)
+        scopes = rule_fields.filter(isScope=True)
+
+        # We're only interested in the rule "CollectionObject catalogNumber 
+        # must be unique to Collection"
+        # We check for length of fields and scopes because get() raises an 
+        # exception if more than one result is returned
+        if (len(fields) == 1 and len(scopes) == 1) and (fields.get().fieldPath.lower() == "catalognumber" and scopes.get().fieldPath.lower() == "collection"):
+            catalog_number_rules.append(rule.id)
+
+    rules_to_update = UniquenessRule.objects.filter(id__in=catalog_number_rules)
+    rules_to_update.update(isDatabaseConstraint=False)
+
+
+def catnum_rule_uneditable(apps, schema_editor=None):
+    """ Find any CollectionObject catalogNumber must be unique to Collection 
+    rules which are editable on the frontend (have isDatabaseConstraint=False)
+    and set their isDatabaseConstraint=True.
+
+    Generally should be run when migration businessrules/0003 is being reverted
+    """
+    Discipline = apps.get_model("specify", "Discipline")
+    UniquenessRule = apps.get_model("businessrules", "UniquenessRule")
+
+    for discipline in Discipline.objects.all():
+        model_rules = UniquenessRule.objects.filter(modelName="Collectionobject", discipline_id=discipline.id, isDatabaseConstraint=False)
+
+        has_catalognumber_rule = False
+        matching_rule_ids: List[int] = []
+        for rule in model_rules: 
+            rule_fields = rule.uniquenessrulefield_set.all()
+
+            fields = rule_fields.filter(isScope=False)
+            scopes = rule_fields.filter(isScope=True)
+
+            # We're only interested in the rule "CollectionObject catalogNumber 
+            # must be unique to Collection"
+            # We check for length of fields and scopes because get() raises an 
+            # exception if more than one result is returned
+            if (len(fields) == 1 and len(scopes) == 1) and (fields.get().fieldPath.lower() == "catalognumber" and scopes.get().fieldPath.lower() == "collection"):
+                has_catalognumber_rule = True
+                matching_rule_ids.append(rule.id)
+
+        if has_catalognumber_rule:
+            UniquenessRule.objects.filter(id__in=matching_rule_ids).update(isDatabaseConstraint=True)
+        else:
+            create_uniqueness_rule(
+                "Collectionobject",
+                discipline=discipline,
+                is_database_constraint=True,
+                fields=["catalogNumber"],
+                scopes=["collection"],
+                registry=apps,
+            )

specifyweb/backend/businessrules/migrations/0004_catnum_uniquerule.py

@@ -1,7 +1,6 @@
-from typing import Tuple
-
 from django.db import migrations
 
+from specifyweb.backend.businessrules.migration_utils import catnum_rule_editable, catnum_rule_uneditable
 from specifyweb.backend.businessrules.uniqueness_rules import create_uniqueness_rule
 
 

specifyweb/backend/businessrules/migrations/0005_cojo.py

@@ -6,7 +6,7 @@
     Applies the COJO uniqueness rule to the database.
 """
 def apply_migration(apps, schema_editor):
-    cojo_rules = DEFAULT_UNIQUENESS_RULES["CollectionObjectGroupJoin"]
+    cojo_rules = DEFAULT_UNIQUENESS_RULES["Collectionobjectgroupjoin"]
     Discipline = apps.get_model('specify', 'Discipline')
 
     for rule in cojo_rules:

specifyweb/backend/businessrules/migrations/0008_fix_global_default_rules.py

@@ -0,0 +1,18 @@
+import logging
+from django.db import migrations
+from specifyweb.backend.businessrules.uniqueness_rules import fix_global_default_rules
+
+logger = logging.getLogger(__name__)
+
+def apply_migration(apps, schema_editor):
+    fix_global_default_rules(apps)
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('businessrules', '0007_more_uniqueness_rules'),
+    ]
+
+    operations = [
+        migrations.RunPython(apply_migration, migrations.RunPython.noop, atomic=True)
+    ]

specifyweb/backend/businessrules/uniqueness_rules.json

@@ -78,7 +78,7 @@
   "Collectionobject": [
     {
       "rule": [["catalogNumber"], ["collection"]],
-      "isDatabaseConstraint": true
+      "isDatabaseConstraint": false
     },
     {
       "rule": [["uniqueIdentifier"], []],
@@ -89,7 +89,7 @@
       "isDatabaseConstraint": false
     }
   ],
-  "CollectionObjectGroupJoin": [
+  "Collectionobjectgroupjoin": [
     {
       "rule": [["childCo"], []],
       "isDatabaseConstraint": true

specifyweb/backend/businessrules/uniqueness_rules.py

@@ -6,13 +6,14 @@
 from collections.abc import Iterable
 
 from django.apps import apps
-from django.db import connections
+from django.db import connections, transaction
 from django.db.models import Q, Count
 from django.db.migrations.recorder import MigrationRecorder
 from django.core.exceptions import ObjectDoesNotExist
 from specifyweb.specify.api.crud import get_model
 from specifyweb.specify.datamodel import datamodel
 from specifyweb.middleware.general import serialize_django_obj
+from specifyweb.specify.models import Discipline
 from specifyweb.specify.utils.scoping import in_same_scope
 from .orm_signal_handler import orm_signal_handler
 from .exceptions import BusinessRuleException
@@ -73,7 +74,8 @@ def validate_unique(model, instance):
     rules = UniquenessRule.objects.filter(modelName=model_name)
     for rule in rules:
         rule_fields = UniquenessRuleField.objects.filter(uniquenessrule=rule)
-        if not rule_is_global(tuple(field.fieldPath for field in rule_fields.filter(isScope=True))) and not in_same_scope(rule, instance):
+        if not rule_is_global(tuple(field.fieldPath for field in rule_fields.filter(isScope=True))) \
+            and not in_same_scope(rule, instance):
             continue
 
         field_names = [
@@ -152,7 +154,8 @@ class UniquenessCheck(TypedDict):
     fields: list[ViolatedUniquenessCheck]
 
 
-def check_uniqueness(model_name: str, raw_fields: list[str], raw_scopes: list[str], registry=None) -> UniquenessCheck | None:
+def check_uniqueness(model_name: str, raw_fields: list[str], raw_scopes: list[str], registry=None) \
+    -> UniquenessCheck | None:
     """
     Given a model, a list of fields, and a list of scopes, check whether there
     are models of model_name which have duplicate values of fields in scopes. 
@@ -180,16 +183,32 @@ def check_uniqueness(model_name: str, raw_fields: list[str], raw_scopes: list[st
 
     duplicates_field = '__duplicates'
 
-    duplicates = django_model.objects.values(
-        *all_fields).annotate(**{duplicates_field: Count('id')}).filter(strict_filters).filter(**{f"{duplicates_field}__gt": 1}).order_by(f'-{duplicates_field}')
+    duplicates = (
+        django_model.objects
+        .values(*all_fields)
+        .annotate(**{duplicates_field: Count('id')})
+        .filter(strict_filters)
+        .filter(**{f"{duplicates_field}__gt": 1})
+        .order_by(f'-{duplicates_field}')
+    )
 
     total_duplicates = sum(duplicate[duplicates_field]
                            for duplicate in duplicates)
 
     final = {
         "totalDuplicates": total_duplicates,
-        "fields": [{"duplicates": duplicate[duplicates_field], "fields": {field: value for field, value in duplicate.items() if field != duplicates_field}}
-                   for duplicate in duplicates]}
+        "fields": [
+            {
+                "duplicates": duplicate[duplicates_field],
+                "fields": {
+                    field: value
+                    for field, value in duplicate.items()
+                    if field != duplicates_field
+                },
+            }
+            for duplicate in duplicates
+        ],
+    }
     return final
 
 
@@ -224,7 +243,6 @@ def serialize_multiple_django(matchable, field_map, fields):
 def join_with_and(fields):
     return ' and '.join(fields)
 
-
 def apply_default_uniqueness_rules(discipline, registry=None):
     for table, rules in DEFAULT_UNIQUENESS_RULES.items():
         model_name = getattr(datamodel.get_table(table), "django_name", None)
@@ -234,8 +252,7 @@ def apply_default_uniqueness_rules(discipline, registry=None):
             fields, scopes = rule["rule"]
             isDatabaseConstraint = rule["isDatabaseConstraint"]
 
-            create_uniqueness_rule(
-                model_name, discipline, isDatabaseConstraint, fields, scopes, registry)
+            create_uniqueness_rule(model_name, discipline, isDatabaseConstraint, fields, scopes, registry)
 
 
 def create_uniqueness_rule(model_name, raw_discipline, is_database_constraint, fields, scopes, registry=None):
@@ -246,26 +263,32 @@ def create_uniqueness_rule(model_name, raw_discipline, is_database_constraint, f
 
     discipline = None if rule_is_global(scopes) else raw_discipline
 
-    candidate_rules = UniquenessRule.objects.filter(modelName=model_name, isDatabaseConstraint=is_database_constraint, discipline=discipline)
+    candidate_rules = UniquenessRule.objects.filter(modelName=model_name,
+                                                    isDatabaseConstraint=is_database_constraint,
+                                                    discipline=discipline)
+    expected_fields = set(fields)
+    expected_scopes = set(scopes)
 
-    for rule in candidate_rules: 
+    for rule in candidate_rules:
         all_fields = rule.uniquenessrulefield_set.all()
-        matching_fields = all_fields.filter(fieldPath__in=fields, isScope=False)
-        matching_scopes = all_fields.filter(fieldPath__in=scopes, isScope=True)
+        existing_fields = set(
+            all_fields.filter(isScope=False).values_list("fieldPath", flat=True)
+        )
+        existing_scopes = set(
+            all_fields.filter(isScope=True).values_list("fieldPath", flat=True)
+        )
         # If the rule already exists, skip creating the rule
-        if len(matching_fields) == len(fields) and len(matching_scopes) == len(scopes): 
+        if existing_fields == expected_fields and existing_scopes == expected_scopes:
             return
 
+    logger.info(f"Creating uniqueness rule on {model_name} with fields {fields} and scopes {scopes} for the discipline {discipline.name if discipline else 'Global'}")
     rule = UniquenessRule.objects.create(
         discipline=discipline, modelName=model_name, isDatabaseConstraint=is_database_constraint)
 
     for field in fields:
-        UniquenessRuleField.objects.create(
-            uniquenessrule=rule, fieldPath=field, isScope=False)
+        UniquenessRuleField.objects.create(uniquenessrule=rule, fieldPath=field, isScope=False)
     for scope in scopes:
-        UniquenessRuleField.objects.create(
-            uniquenessrule=rule, fieldPath=scope, isScope=True)
-
+        UniquenessRuleField.objects.create(uniquenessrule=rule, fieldPath=scope, isScope=True)
 
 def remove_uniqueness_rule(model_name, raw_discipline, is_database_constraint, fields, scopes, registry=None):
     UniquenessRule = registry.get_model(
@@ -275,15 +298,22 @@ def remove_uniqueness_rule(model_name, raw_discipline, is_database_constraint, f
 
     discipline = None if rule_is_global(scopes) else raw_discipline
 
-    candidate_rules = UniquenessRule.objects.filter(modelName=model_name, isDatabaseConstraint=is_database_constraint, discipline=discipline)
+    candidate_rules = UniquenessRule.objects.filter(
+        modelName=model_name, isDatabaseConstraint=is_database_constraint, discipline=discipline)
 
+    expected_fields = set(fields)
+    expected_scopes = set(scopes)
     rule_ids = []
     for rule in candidate_rules: 
         all_fields = rule.uniquenessrulefield_set.all()
-        matching_fields = all_fields.filter(fieldPath__in=fields, isScope=False)
-        matching_scopes = all_fields.filter(fieldPath__in=scopes, isScope=True)
+        existing_fields = set(
+            all_fields.filter(isScope=False).values_list("fieldPath", flat=True)
+        )
+        existing_scopes = set(
+            all_fields.filter(isScope=True).values_list("fieldPath", flat=True)
+        )
         # If the rule exists, add it to the list of rules to be deleted
-        if len(matching_fields) == len(fields) and len(matching_scopes) == len(scopes): 
+        if existing_fields == expected_fields and existing_scopes == expected_scopes:
             rule_ids.append(rule.id)
 
     UniquenessRuleField.objects.filter(
@@ -297,6 +327,46 @@ def remove_uniqueness_rule(model_name, raw_discipline, is_database_constraint, f
 """
 GLOBAL_RULE_FIELDS = ["division", 'institution']
 
-
 def rule_is_global(scopes: Iterable[str]) -> bool:
-    return len(scopes) == 0 or any(any(scope_field.lower() in GLOBAL_RULE_FIELDS for scope_field in scope.split('__')) for scope in scopes)
+    return len(scopes) == 0 \
+        or any(any(scope_field.lower() in GLOBAL_RULE_FIELDS for scope_field in scope.split('__')) for scope in scopes)
+
+def fix_global_default_rules(registry=None):
+    UniquenessRule = registry.get_model('businessrules', 'UniquenessRule') \
+        if registry \
+        else models.UniquenessRule
+
+    global_rule_signatures = {
+        (
+            rule.modelName,
+            rule.isDatabaseConstraint,
+            frozenset(
+                rule.uniquenessrulefield_set.values_list("fieldPath", "isScope")
+            ),
+        )
+        for rule in UniquenessRule.objects.filter(
+            discipline__isnull=True
+        ).prefetch_related("uniquenessrulefield_set")
+    }
+
+    discipline_ids = (
+        UniquenessRule.objects.exclude(discipline__isnull=True)
+        .values_list("discipline_id", flat=True)
+        .distinct()
+    )
+
+    for discipline_id in discipline_ids:
+        with transaction.atomic():
+            for rule in UniquenessRule.objects.filter(
+                discipline_id=discipline_id
+            ).prefetch_related("uniquenessrulefield_set"):
+                signature = (
+                    rule.modelName,
+                    rule.isDatabaseConstraint,
+                    frozenset(
+                        rule.uniquenessrulefield_set.values_list("fieldPath", "isScope")
+                    ),
+                )
+                if signature in global_rule_signatures:
+                    rule.uniquenessrulefield_set.all().delete()
+                    rule.delete()

specifyweb/backend/patches/migration_utils.py

@@ -0,0 +1,38 @@
+from django.db.models import F
+
+# REFACTOR: Use ALL_TRESS in specify/tree_views.py?
+SPECIFY_TREES = ["Taxon", "Geography", "Storage",
+                 "Geologictimeperiod", "Lithostrat"]
+
+
+def apply_migrations(app_registry, schema_editor=None):
+    update_is_accepted(app_registry, schema_editor)
+    update_coordinates(app_registry, schema_editor)
+
+def update_is_accepted(app_registry, schema_editor=None):
+    db_alias = schema_editor.connection.alias if schema_editor is not None else "default"
+    for tree in SPECIFY_TREES:
+        tree_filters = {
+            "isaccepted": False,
+            "accepted" + tree.lower() + "__isnull": True
+        }
+
+        tree_model = app_registry.get_model("specify", tree)
+        tree_model._base_manager.using(db_alias).filter(**tree_filters).update(isaccepted=True)
+
+
+def update_coordinates(app_registry, schema_editor=None):
+    db_alias = schema_editor.connection.alias if schema_editor is not None else "default"
+    Locality = app_registry.get_model("specify", "Locality")
+
+    Locality._base_manager.using(db_alias).filter(lat1text__isnull=True, latitude1__isnull=False) \
+        .update(lat1text=F("latitude1"))
+
+    Locality._base_manager.using(db_alias).filter(long1text__isnull=True, longitude1__isnull=False) \
+        .update(long1text=F("longitude1"))
+
+    Locality._base_manager.using(db_alias).filter(lat2text__isnull=True, latitude2__isnull=False) \
+        .update(lat2text=F("latitude2"))
+
+    Locality._base_manager.using(db_alias).filter(long2text__isnull=True, longitude2__isnull=False) \
+        .update(long2text=F("longitude2"))