Derek furst/validate direct ancestors

src/schema/provenance_schema.yaml

@@ -454,10 +454,12 @@ ENTITIES:
         type: list
         before_property_create_validators:
           - validate_no_duplicates_in_list
+          - validate_ancestor_type
         before_property_update_validators:
           - validate_no_duplicates_in_list
           - validate_not_invalid_creation_action
           - validate_id_not_in_direct_ancestor
+          - validate_ancestor_type
         transient: true
         exposed: false
         indexed: false

src/schema/schema_manager.py

@@ -226,6 +226,34 @@ def get_entity_superclass(normalized_entity_class):
     return normalized_superclass
 
 
+"""
+Get the optional subclass (if defined) of the given entity class
+
+Parameters
+----------
+normalized_entity_class : str
+    The normalized target entity class
+
+Returns
+-------
+string or None
+    One of the normalized entity classes if defined. None otherwise
+"""
+def get_entity_subclasses(normalized_entity_class):
+    subclasses = []
+    all_entity_types = get_all_entity_types()
+
+    if normalized_entity_class not in all_entity_types:
+        raise ValueError(f"Unrecognized entity class: {normalized_entity_class}")
+
+    for name, data in _schema["ENTITIES"].items():
+        superclass = data.get("superclass")
+        if superclass and normalize_entity_type(superclass) == normalized_entity_class:
+            subclasses.append(normalize_entity_type(name))
+
+    return subclasses
+
+
 """
 Determine if the Entity type with 'entity_type' is an instance of 'entity_class'.
 Use this function if you already have the Entity type. Use `entity_instanceof(uuid, class)` 

src/schema/schema_neo4j_queries.py

@@ -1798,6 +1798,35 @@ def create_activity_tx(tx, activity_data_dict):
     return node
 
 
+def validate_direct_ancestors(neo4j_driver, entity_uuids, allowed_types, disallowed_property_values=None):
+    disallowed_rules_list = disallowed_property_values
+    query = """
+    UNWIND $uuids AS uid
+    OPTIONAL MATCH (n) WHERE n.uuid = uid
+    WITH uid, n,
+         CASE
+           WHEN n IS NULL THEN false
+           ELSE any(l IN labels(n) WHERE l IN $allowed_labels)
+         END AS label_ok,
+         $disallowed AS rules
+    WITH uid, label_ok, 
+         any(rule IN rules WHERE 
+             n IS NOT NULL 
+             AND n[rule.property] IS NOT NULL 
+             AND n[rule.property] = rule.value
+         ) AS has_forbidden_prop
+    WHERE NOT label_ok OR has_forbidden_prop
+    RETURN DISTINCT uid AS invalid_uuid
+    """
+    with neo4j_driver.session() as session:
+        result = session.run(query, 
+                             uuids=entity_uuids, 
+                             allowed_labels=allowed_types, 
+                             disallowed=disallowed_rules_list)
+
+        return [record["invalid_uuid"] for record in result]
+
+
 """
 Build the property key-value pairs to be used in the Cypher clause for node creation/update
 

src/schema/schema_validators.py

@@ -674,6 +674,34 @@ def validate_sample_category(property_key, normalized_entity_type, request, exis
     if new_data_dict[property_key] != sample_category:
         raise ValueError(f"The case of sample_category '{new_data_dict[property_key]}'"
                          f" must be specified as '{sample_category}'.")
+
+"""
+Validate the provided value of Dataset.direct_ancestor on create via POST and update via PUT
+
+Parameters
+----------
+property_key : str
+    The target property key
+normalized_type : str
+    Submission
+request: Flask request object
+    The instance of Flask request passed in from application request
+existing_data_dict : dict
+    A dictionary that contains all existing entity properties
+new_data_dict : dict
+    The json data in request body, already after the regular validations
+"""
+def validate_ancestor_type(property_key, normalized_entity_type, request, existing_data_dict, new_data_dict):
+    allowed_ancestor_types = ["Dataset", "Sample"]
+    for allowed_ancestor in list(allowed_ancestor_types):
+        subclasses = schema_manager.get_entity_subclasses(schema_manager.normalize_entity_type(allowed_ancestor))
+        allowed_ancestor_types.extend(subclasses)
+    direct_ancestor_uuids = new_data_dict[property_key]
+    disallowed_properties = [{"property": "sample_category", "value": "organ"}]
+    invalid_uuids = schema_neo4j_queries.validate_direct_ancestors(schema_manager.get_neo4j_driver_instance(), direct_ancestor_uuids, allowed_ancestor_types, disallowed_properties)
+    if invalid_uuids:
+        raise ValueError(f"Invalid or not-found direct_ancestor_uuid(s). Allowed entity_types are: {', '.join(allowed_ancestor_types)}. For samples, 'organ' is not allowed. Invalid uuids: {', '.join(invalid_uuids)}")
+
 
 """
 Validate the provided value of Publication.publication_date is in the correct format against ISO 8601 Format: