Merge pull request #407 from VariantEffect/estelle/geneOntologyControlledKeywords

Controlled keywords - gene ontology

Author: jstone-dev

alembic/versions/0b29eefbe1dd_controlled_keyword_modification.py

@@ -0,0 +1,46 @@
+"""controlled keyword modification
+
+Revision ID: 0b29eefbe1dd
+Revises: 2b7a977e7e98
+Create Date: 2025-08-07 15:45:43.241567
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '0b29eefbe1dd'
+down_revision = '2b7a977e7e98'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    # Rename existing columns
+    op.alter_column('controlled_keywords', 'value', new_column_name='label', existing_type=sa.String(), existing_nullable=False)
+    op.alter_column('controlled_keywords', 'vocabulary', new_column_name='system', existing_type=sa.String(), existing_nullable=True)
+    # Add new columns
+    op.add_column('controlled_keywords', sa.Column('code', sa.String(), nullable=True))
+    op.add_column('controlled_keywords', sa.Column('version', sa.String(), nullable=True))
+    # Drop and recreate the correct unique constraint
+    op.drop_constraint('ix_controlled_keywords_key_value', 'controlled_keywords', type_='unique')
+    op.create_unique_constraint('ix_controlled_keywords_key_label', 'controlled_keywords', ['key', 'label'])
+    # ### end Alembic commands ###
+
+    # TODO: Will modify this part when we get the final GO terms.
+    op.execute(
+        """INSERT INTO controlled_keywords (key, label, system, code, version, special, description, creation_date, modification_date) VALUES ('Phenotypic Assay Mechanism', 'Other', NULL, NULL, NULL, False, 'The Gene Ontology (GO) is a structured, standardized representation of biological knowledge.', NOW(), NOW())"""
+    )
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('controlled_keywords', 'label', new_column_name='value', existing_type=sa.String(), existing_nullable=False)
+    op.alter_column('controlled_keywords', 'system', new_column_name='vocabulary', existing_type=sa.String(), existing_nullable=True)
+    op.drop_constraint('ix_controlled_keywords_key_label', 'controlled_keywords', type_='unique')
+    op.create_unique_constraint('ix_controlled_keywords_key_value', 'controlled_keywords', ['key', 'value'])
+    op.drop_column('controlled_keywords', 'version')
+    op.drop_column('controlled_keywords', 'code')
+    # ### end Alembic commands ###

src/mavedb/lib/experiments.py

@@ -100,7 +100,7 @@ def search_experiments(
         query = query.filter(
             Experiment.keyword_objs.any(
                 ExperimentControlledKeywordAssociation.controlled_keyword.has(
-                    ControlledKeyword.value.in_(search.keywords)
+                    ControlledKeyword.label.in_(search.keywords)
                 )
             )
         )

src/mavedb/lib/keywords.py

@@ -6,16 +6,16 @@
 from mavedb.models.controlled_keyword import ControlledKeyword
 
 
-def search_keyword(db: Session, key: str, value: Optional[str]):
+def search_keyword(db: Session, key: str, label: Optional[str]):
     lower_key = key.lower().strip()
-    lower_value = value.lower().strip() if value is not None else None
+    lower_label = label.lower().strip() if label is not None else None
     query = db.query(ControlledKeyword)
     if lower_key:
         query = query.filter(func.lower(ControlledKeyword.key) == lower_key)
-    if lower_value:
-        query = query.filter(func.lower(ControlledKeyword.value) == lower_value)
+    if lower_label:
+        query = query.filter(func.lower(ControlledKeyword.label) == lower_label)
 
     controlled_keyword = query.one_or_none()
     if controlled_keyword is None:
-        raise ValueError(f"Invalid keyword {key} or {value}")
+        raise ValueError(f"Invalid keyword {key} or {label}")
     return controlled_keyword

src/mavedb/lib/score_sets.py

@@ -208,7 +208,7 @@ def search_score_sets(db: Session, owner_or_contributor: Optional[User], search:
             ScoreSet.experiment.has(
                 Experiment.keyword_objs.any(
                     ExperimentControlledKeywordAssociation.controlled_keyword.has(
-                        ControlledKeyword.value.in_(search.keywords)
+                        ControlledKeyword.label.in_(search.keywords)
                     )
                 )
             )

src/mavedb/lib/validation/keywords.py

@@ -1,32 +1,41 @@
+import re
 from typing import Optional
 
 from mavedb.lib.validation.exceptions import ValidationError
 from mavedb.lib.validation.utilities import is_null
 
 
+def validate_code(key: str, label: str, code: Optional[str]):
+    if key.lower() == "phenotypic assay mechanism" and label.lower() != "other":
+        # The Gene Ontology accession is a unique seven digit identifier prefixed by GO:.
+        # e.g. GO:0005739, GO:1904659, or GO:0016597.
+        if code is None or not re.match(r"^GO:\d{7}$", code):
+            raise ValidationError("Invalid Gene Ontology accession.")
+
+
 # TODO: value will not be Optional when we confirm the final controlled keyword list.
-def validate_description(value: str, key: str, description: Optional[str]):
-    if value.lower() == "other" and (description is None or description.strip() == ""):
+def validate_description(label: str, key: str, description: Optional[str]):
+    if label.lower() == "other" and (description is None or description.strip() == ""):
         raise ValidationError(
             "Other option does not allow empty description.", custom_loc=["body", "keywordDescriptions", key]
         )
 
 
 def validate_duplicates(keywords: list):
     keys = []
-    values = []
+    labels = []
     for k in keywords:
         keys.append(k.keyword.key.lower())  # k: ExperimentControlledKeywordCreate object
-        if k.keyword.value.lower() != "other":
-            values.append(k.keyword.value.lower())
+        if k.keyword.label.lower() != "other":
+            labels.append(k.keyword.label.lower())
 
     keys_set = set(keys)
-    values_set = set(values)
+    labels_set = set(labels)
 
     if len(keys) != len(keys_set):
         raise ValidationError("Duplicate keys found in keywords.")
-    if len(values) != len(values_set):
-        raise ValidationError("Duplicate values found in keywords.")
+    if len(labels) != len(labels_set):
+        raise ValidationError("Duplicate labels found in keywords.")
 
 
 def validate_keyword(keyword: str):
@@ -47,7 +56,7 @@ def validate_keyword(keyword: str):
 
 
 def validate_keyword_keys(keywords: list):
-    keyword_dict = {k.keyword.key.lower(): k.keyword.value.lower() for k in keywords}
+    keyword_dict = {k.keyword.key.lower(): k.keyword.label.lower() for k in keywords}
     variant_library_method = keyword_dict.get("variant library creation method", "")
 
     if variant_library_method == "endogenous locus library method":

src/mavedb/models/controlled_keyword.py

@@ -10,10 +10,12 @@ class ControlledKeyword(Base):
 
     id = Column(Integer, primary_key=True, index=True)
     key = Column(String, nullable=False)
-    value = Column(String, nullable=False)
-    vocabulary = Column(String, nullable=True)
+    label = Column(String, nullable=False)
+    system = Column(String, nullable=True)
+    code = Column(String, nullable=True)
+    version = Column(String, nullable=True)
     special = Column(Boolean, nullable=True)
     description = Column(String, nullable=True)
     creation_date = Column(Date, nullable=False, default=date.today)
     modification_date = Column(Date, nullable=False, default=date.today, onupdate=date.today)
-    __table_args__ = (UniqueConstraint("key", "value", name="ix_controlled_keywords_key_value"),)
+    __table_args__ = (UniqueConstraint("key", "label", name="ix_controlled_keywords_key_label"),)

src/mavedb/models/experiment.py

@@ -161,8 +161,10 @@ def keywords(self) -> list[dict]:
                 {
                     "keyword": {
                         "key": controlled_keyword.key,
-                        "value": controlled_keyword.value,
-                        "vocabulary": controlled_keyword.vocabulary,
+                        "label": controlled_keyword.label,
+                        "system": controlled_keyword.system,
+                        "code": controlled_keyword.code,
+                        "version": controlled_keyword.version,
                         "special": controlled_keyword.special,
                         "description": controlled_keyword.description,
                     },
@@ -190,8 +192,8 @@ async def set_keywords(self, db, keywords: list):
                     controlled_keyword=await self._find_keyword(
                         db,
                         keyword_obj.keyword.key,
-                        keyword_obj.keyword.value,
-                        keyword_obj.keyword.vocabulary,
+                        keyword_obj.keyword.label,
+                        keyword_obj.keyword.code,
                     ),
                     description=keyword_obj.description,
                 )
@@ -211,15 +213,15 @@ async def _find_or_create_legacy_keyword(self, db, keyword_text):
             keyword_obj = LegacyKeyword(text=keyword_text)
         return keyword_obj
 
-    async def _find_keyword(self, db, key: str, value: str, vocabulary: Optional[str]):
+    async def _find_keyword(self, db, key: str, label: str, code: Optional[str]):
         query = (
-            db.query(ControlledKeyword).filter(ControlledKeyword.key == key).filter(ControlledKeyword.value == value)
+            db.query(ControlledKeyword).filter(ControlledKeyword.key == key).filter(ControlledKeyword.label == label)
         )
-        if vocabulary:
-            query = query.filter(ControlledKeyword.vocabulary == vocabulary)
+        if code:
+            query = query.filter(ControlledKeyword.code == code)
         controlled_keyword_obj = query.one_or_none()
         if controlled_keyword_obj is None:
-            raise ValueError(f"Unknown keyword {key}:{value}")
+            raise ValueError(f"Unknown keyword {key}:{label}")
         return controlled_keyword_obj
 
 

src/mavedb/routers/controlled_keywords.py

@@ -32,7 +32,7 @@ def fetch_keywords_by_key(
     items = (
         db.query(ControlledKeyword)
         .filter(func.lower(ControlledKeyword.key) == lower_key)
-        .order_by(ControlledKeyword.value)
+        .order_by(ControlledKeyword.label)
         .all()
     )
     if not items:
@@ -41,8 +41,8 @@ def fetch_keywords_by_key(
 
 
 @router.post("/search/{key}/{value}", status_code=200, response_model=keyword.Keyword)
-def search_keyword_by_key_and_value(key: str, value: str, db: Session = Depends(deps.get_db)) -> ControlledKeyword:
+def search_keyword_by_key_and_value(key: str, label: str, db: Session = Depends(deps.get_db)) -> ControlledKeyword:
     """
     Search keywords.
     """
-    return _search_keyword(db, key, value)
+    return _search_keyword(db, key, label)

src/mavedb/routers/experiments.py

@@ -279,22 +279,22 @@ async def create_experiment(
     for publication in publication_identifiers:
         setattr(publication, "primary", publication.identifier in primary_identifiers)
 
-    # TODO: Controlled keywords currently is allowed none value.
+    # TODO: Controlled keywords currently is allowed none label.
     #  Will be changed in the future when we get the final list.
     keywords: list[ExperimentControlledKeywordAssociation] = []
     if item_create.keywords:
-        all_values_none = all(k.keyword.value is None for k in item_create.keywords)
-        if all_values_none is False:
+        all_labels_none = all(k.keyword.label is None for k in item_create.keywords)
+        if all_labels_none is False:
             # Users may choose part of keywords from dropdown menu. Remove not chosen keywords from the list.
-            filtered_keywords = list(filter(lambda k: k.keyword.value is not None, item_create.keywords))
+            filtered_keywords = list(filter(lambda k: k.keyword.label is not None, item_create.keywords))
             try:
                 validate_keyword_list(filtered_keywords)
             except ValidationError as e:
                 raise HTTPException(status_code=422, detail=str(e))
             for upload_keyword in filtered_keywords:
                 try:
                     description = upload_keyword.description
-                    controlled_keyword = search_keyword(db, upload_keyword.keyword.key, upload_keyword.keyword.value)
+                    controlled_keyword = search_keyword(db, upload_keyword.keyword.key, upload_keyword.keyword.label)
                     experiment_controlled_keyword = ExperimentControlledKeywordAssociation(
                         controlled_keyword=controlled_keyword,
                         description=description,
@@ -417,10 +417,10 @@ async def update_experiment(
     item.raw_read_identifiers = raw_read_identifiers
 
     if item_update.keywords:
-        all_values_none = all(k.keyword.value is None for k in item_update.keywords)
-        if all_values_none is False:
+        all_labels_none = all(k.keyword.label is None for k in item_update.keywords)
+        if all_labels_none is False:
             # Users may choose part of keywords from dropdown menu. Remove not chosen keywords from the list.
-            filtered_keywords = list(filter(lambda k: k.keyword.value is not None, item_update.keywords))
+            filtered_keywords = list(filter(lambda k: k.keyword.label is not None, item_update.keywords))
             try:
                 validate_keyword_list(filtered_keywords)
             except ValidationError as e:

src/mavedb/routers/statistics.py

@@ -150,8 +150,8 @@ def experiment_keyword_statistics(
         raise HTTPException(500, "No association table associated with the keywords field when one was expected.")
 
     query = _join_model_and_filter_unpublished(
-        select(ControlledKeyword.value, func.count(ControlledKeyword.value)).join(queried_assc), queried_model
-    ).group_by(ControlledKeyword.value)
+        select(ControlledKeyword.label, func.count(ControlledKeyword.label)).join(queried_assc), queried_model
+    ).group_by(ControlledKeyword.label)
 
     return _count_for_identifier_in_query(db, query)