[Data - 3127] Add tag to cohort definitions

charts/d2e-services/templates/atlas-db-init-cm.yaml

@@ -304,3 +304,66 @@ data:
     -- Do not use psql meta-commands (\gset, \if) in this file: trex applies it over
     -- the wire protocol, and the simple query protocol parses the whole file before
     -- executing any of it, so one meta-command stops every statement from running.
+  230_imported_cohort_metadata_tag_group.sql: |
+    -- Tag groups for cohorts imported from external libraries.
+    --
+    -- WebAPI cannot create a root tag through its API (POST /tag requires a parent
+    -- group), so the groups have to be seeded here. Two of them, because their tags
+    -- have opposite cardinality: a cohort's source coexists with other tags, while
+    -- its review status is one-at-a-time. AbstractDaoService.assignTag clears the
+    -- other tags in a single-selection group, which is what retires the previous
+    -- status on re-import. Both groups need allow_custom to accept children, and
+    -- neither is ever assigned to a cohort.
+    DO $$
+    DECLARE
+        source_group_id integer;
+        status_group_id integer;
+    BEGIN
+        -- Group holding the source/provenance tag.
+        SELECT id INTO source_group_id FROM webapi.tag
+        WHERE lower(name) = lower('Imported Cohort Metadata');
+
+        IF source_group_id IS NULL THEN
+            INSERT INTO webapi.tag (name, type, count, show_group, multi_selection,
+                                    permission_protected, mandatory, allow_custom, description)
+            VALUES ('Imported Cohort Metadata', 0, 0, false, true,
+                    false, false, true,
+                    'Container for tags recording where an imported cohort came from')
+            RETURNING id INTO source_group_id;
+            RAISE NOTICE 'Created "Imported Cohort Metadata" tag group (id: %)', source_group_id;
+        ELSE
+            UPDATE webapi.tag
+            SET allow_custom = true, show_group = false, multi_selection = true
+            WHERE id = source_group_id
+              AND (allow_custom IS NOT TRUE OR show_group IS NOT FALSE
+                   OR multi_selection IS NOT TRUE);
+        END IF;
+
+        -- Group holding the mutually exclusive review-status tags.
+        SELECT id INTO status_group_id FROM webapi.tag
+        WHERE lower(name) = lower('Cohort Review Status');
+
+        IF status_group_id IS NULL THEN
+            INSERT INTO webapi.tag (name, type, count, show_group, multi_selection,
+                                    permission_protected, mandatory, allow_custom, description)
+            VALUES ('Cohort Review Status', 0, 0, false, false,
+                    false, false, true,
+                    'Container for the review status of an imported cohort; one applies at a time')
+            RETURNING id INTO status_group_id;
+            RAISE NOTICE 'Created "Cohort Review Status" tag group (id: %)', status_group_id;
+        ELSE
+            UPDATE webapi.tag
+            SET allow_custom = true, show_group = false, multi_selection = false
+            WHERE id = status_group_id
+              AND (allow_custom IS NOT TRUE OR show_group IS NOT FALSE
+                   OR multi_selection IS NOT FALSE);
+        END IF;
+
+    END $$;
+
+    SELECT g.name AS tag_group, g.multi_selection, count(tg.tag_id) AS members
+    FROM webapi.tag g
+    LEFT JOIN webapi.tag_group tg ON tg.group_id = g.id
+    WHERE g.name IN ('Imported Cohort Metadata', 'Cohort Review Status')
+    GROUP BY g.name, g.multi_selection ORDER BY g.name;
+

plugins/flows/_shared_flow_utils/api/PhenotypeTagAPI.py

@@ -0,0 +1,149 @@
+import requests
+from prefect.logging import get_run_logger
+
+from _shared_flow_utils.api.BaseAPI import BaseAPI
+
+
+class PhenotypeTagAPI(BaseAPI):
+    """Resolve the WebAPI tags applied to Phenotype Library cohort imports.
+
+    Each cohort carries two: its source, and its review status. The groups holding
+    them are seeded in SQL and never assigned to a cohort themselves.
+    """
+
+    # Seeded by services/atlas-db-init/230_imported_cohort_metadata_tag_group.sql.
+    # Two groups because multi_selection is a property of the group: statuses are mutually exclusive, the source tag is not.
+    REQUEST_TIMEOUT = (10, 30)
+
+    SOURCE_GROUP = "Imported Cohort Metadata"
+    STATUS_GROUP = "Cohort Review Status"
+    PHENOTYPE_LIBRARY_TAG = "Phenotype Library"
+
+    def __init__(self):
+        super().__init__()
+        # The d2e-webapi plugin exposes no /tag routes, so tag work goes straight
+        # to WebAPI through the d2e-compat shim, which performs the same token
+        # exchange the plugin's own calls trigger.
+        self.tag_url = self.get_service_route("webapi") + "tag"
+        self.headers = self.get_options()
+
+    def _get_tags(self, dataset_id: str) -> list:
+        headers = self.headers.copy()
+        headers["datasetId"] = dataset_id
+        response = requests.get(
+            self.tag_url, headers=headers, verify=self.get_verify_value(),
+            timeout=self.REQUEST_TIMEOUT
+        )
+        if response.status_code != 200:
+            raise Exception(
+                f"Failed to list tags: {response.status_code} - {response.text}"
+            )
+        return response.json()
+
+    def _create_tag(self, dataset_id: str, name: str, group_id: int) -> dict:
+        """Create a tag inside one of the seeded groups.
+
+        groups must be non-empty: WebAPI 400s on an empty list and 500s if the
+        field is absent, which is why groups cannot be created through the API.
+        """
+        headers = self.headers.copy()
+        headers["datasetId"] = dataset_id
+        payload = {
+            "name": name,
+            # The nested "groups": [] is required, not noise. TagDTOToTagConverter
+            # converts each group reference recursively through itself, and that
+            # second pass dereferences source.getGroups() -- a group reference
+            # without the field NPEs into a 500 ConversionFailedException.
+            "groups": [{"id": group_id, "groups": []}],
+            "allowCustom": False,
+            "showGroup": False,
+            "multiSelection": False,
+            "permissionProtected": False,
+            "mandatory": False,
+        }
+        response = requests.post(
+            self.tag_url,
+            headers=headers,
+            json=payload,
+            verify=self.get_verify_value(),
+            timeout=self.REQUEST_TIMEOUT,
+        )
+        if response.status_code not in [200, 201]:
+            raise Exception(
+                f"Failed to create tag '{name}': "
+                f"{response.status_code} - {response.text}"
+            )
+        return response.json()
+
+    def _require_group(self, by_name: dict, group_name: str) -> dict:
+        """Look up a seeded tag group, failing with something actionable."""
+        group = by_name.get(group_name.lower())
+        if group is None:
+            # Not created here: WebAPI refuses to create a tag with no parent
+            # group, so a root group cannot come from the API at all.
+            raise Exception(
+                f"Tag group '{group_name}' does not exist in WebAPI. It cannot be "
+                f"created through the API; apply "
+                f"services/atlas-db-init/230_imported_cohort_metadata_tag_group.sql "
+                f"(restarting the webapi-init container does this) and retry."
+            )
+        if not group.get("allowCustom", False):
+            raise Exception(
+                f"Tag group '{group_name}' (id {group['id']}) has allowCustom "
+                f"disabled, so tags cannot be attached to it."
+            )
+        return group
+
+    def resolve_import_tags(self, dataset_id: str, statuses) -> tuple[dict, dict]:
+        """Return the provenance tag and a {status: tag} map, creating what is missing.
+
+        Tag names are unique case-insensitively (tags_name_idx on lower(name)), so
+        an existing tag is reused rather than recreated.
+        """
+        logger = get_run_logger()
+        by_name = {tag["name"].lower(): tag for tag in self._get_tags(dataset_id)}
+
+        source_group = self._require_group(by_name, self.SOURCE_GROUP)
+        status_group = self._require_group(by_name, self.STATUS_GROUP)
+
+        wanted = [(self.PHENOTYPE_LIBRARY_TAG, source_group)]
+        wanted += [(status, status_group) for status in sorted(set(statuses))]
+
+        for name, group in wanted:
+            if name.lower() not in by_name:
+                by_name[name.lower()] = self._create_tag(dataset_id, name, group["id"])
+                logger.info(f"Created cohort tag '{name}' under '{group['name']}'")
+
+        status_tags = {
+            status: by_name[status.lower()] for status in sorted(set(statuses))
+        }
+        return by_name[self.PHENOTYPE_LIBRARY_TAG.lower()], status_tags
+
+    def assign_tags_to_cohorts(self, dataset_id: str, tag_ids: list, cohort_ids: list) -> None:
+        """Attach a set of tags to a set of cohort definitions in one request.
+
+        Not the per-cohort POST /cohortdefinition/{id}/tag: that takes one tag id,
+        so a full import costs ~2200 requests and exceeds Trex's rate limit of
+        5000 per 15 minutes. WebAPI routes each pair through the same assignTag,
+        so re-assigning stays a no-op and the single-selection swap still applies.
+        """
+        if not tag_ids or not cohort_ids:
+            return
+        headers = self.headers.copy()
+        headers["datasetId"] = dataset_id
+        payload = {
+            "tags": list(tag_ids),
+            "assets": {"cohorts": list(cohort_ids)},
+        }
+        response = requests.post(
+            f"{self.tag_url}/multiAssign",
+            headers=headers,
+            json=payload,
+            verify=self.get_verify_value(),
+            timeout=self.REQUEST_TIMEOUT,
+        )
+        if response.status_code not in [200, 201, 204]:
+            raise Exception(
+                f"Failed to assign tags {list(tag_ids)} to {len(cohort_ids)} cohort "
+                f"definitions: {response.status_code} - {response.text}"
+            )

plugins/flows/base/phenotype_plugin/flow.py

@@ -9,6 +9,7 @@
 from _shared_flow_utils.types import UserType
 from _shared_flow_utils.dao.DBDao import DBDao
 from _shared_flow_utils.api.PhenotypeAPI import PhenotypeAPI
+from _shared_flow_utils.api.PhenotypeTagAPI import PhenotypeTagAPI
 
 os.environ["plugin_name"] = "phenotype_plugin"
 
@@ -30,7 +31,6 @@ def validate_integer_string(input_string: str) -> bool:
         logger.info(
             "Cohorts ID is set to 'default', retrieving all cohorts from the Phenotype."
         )
-        logger.warning("Cohort 921 is not supported currently, it will be skipped.")
         return True
     else:
         input_string = input_string.strip()
@@ -42,10 +42,6 @@ def validate_integer_string(input_string: str) -> bool:
                 error_message = f"""Input CohortsId: {input_string} is not supported, use ',' as seperator, e.g.: '3,4,25' """
                 logger.error(error_message)
                 raise ValueError(error_message)
-            if num.strip() == "921":
-                logger.warning(
-                    "Cohort 921 is not supported currently, it will be skipped."
-                )
         return True
 
 
@@ -83,6 +79,7 @@ def get_cohort_definitions(cohorts_id: str, vocabschema_name: str, materialize:
                     "cohortName": str(result[i].rx2("cohortName")[0]),
                     "json": str(result[i].rx2("json")[0]),
                     "sql": str(result[i].rx2("sql")[0]),
+                    "status": str(result[i].rx2("status")[0]),
                 }
                 cohort_definitions.append(cohort_def)
             return cohort_definitions
@@ -106,25 +103,68 @@ def atlas_cohort_definitions(
     """
     logger = get_run_logger()
     phenotype_api = PhenotypeAPI()
+    phenotype_tag_api = PhenotypeTagAPI()
     created_cohorts = []
     name_index = phenotype_api.get_cohort_name_index(dataset_id)
     logger.info(f"Indexed {len(name_index)} existing WebAPI cohort definitions")
 
+    # get_cohort_definitions.R already maps blank/NA to "Unspecified".
+    statuses = {cohort_def["status"] for cohort_def in cohort_definitions}
+    phenotype_library_tag, status_tags = phenotype_tag_api.resolve_import_tags(
+        dataset_id, statuses
+    )
+
+    # Collected while writing, applied in bulk afterwards -- see tag_cohort_definitions.
+    cohort_ids_by_status = {}
+
     for cohort_def in cohort_definitions:
         try:
+            status = cohort_def["status"]
             result = phenotype_api.create_single_cohort_definition(
                 cohort_def, dataset_id, user_name, name_index
             )
             created_cohorts.append(result)
+            cohort_ids_by_status.setdefault(status, []).append(result["id"])
         except Exception as e:
             error_message = (
-                f"Failed to save cohort {cohort_def['cohortId']}: {str(e)}"
+                f"Failed to save cohort {cohort_def['cohortId']}: {str(e)}. "
+                f"{len(created_cohorts)} cohorts were written before this and are "
+                f"still untagged; re-run the flow to finish them."
             )
             logger.error(error_message)
             raise Exception(error_message) from e
+
+    tag_cohort_definitions(
+        phenotype_tag_api, dataset_id, phenotype_library_tag, status_tags,
+        cohort_ids_by_status,
+    )
     return created_cohorts
 
 
+def tag_cohort_definitions(phenotype_tag_api, dataset_id: str, phenotype_library_tag: dict,
+                           status_tags: dict, cohort_ids_by_status: dict) -> None:
+    """Record where the cohorts came from, and what their review status is.
+    """
+    logger = get_run_logger()
+    all_cohort_ids = [i for ids in cohort_ids_by_status.values() for i in ids]
+    if not all_cohort_ids:
+        return
+
+    phenotype_tag_api.assign_tags_to_cohorts(
+        dataset_id, [phenotype_library_tag["id"]], all_cohort_ids
+    )
+    logger.info(
+        f"Tagged {len(all_cohort_ids)} cohort definitions as "
+        f"'{phenotype_library_tag['name']}'"
+    )
+
+    for status, cohort_ids in sorted(cohort_ids_by_status.items()):
+        phenotype_tag_api.assign_tags_to_cohorts(
+            dataset_id, [status_tags[status]["id"]], cohort_ids
+        )
+        logger.info(f"Tagged {len(cohort_ids)} cohort definitions as '{status}'")
+
+
 @task(log_prints=True)
 def materialize_cohort_definitions(
     dbdao: DBDao,

plugins/flows/base/phenotype_plugin/get_cohort_definitions.R

@@ -18,24 +18,26 @@ get_cohort_definitions <- function(cohortsID, vocabschemaName, materialize = FAL
     vocabschemaName <- toString(vocabschemaName)
     library('PhenotypeLibrary')
     library('CirceR')
+    phenotypeLog <- PhenotypeLibrary::getPhenotypeLog(showHidden = FALSE)
+
     create_cohort_definitionsets <- function(cohortsID, vocabschemaName) {
-        # CirceR version 1.1.1 does not support cohort 344, and CirceR version 1.3.3 (currently used) does not support cohort 921
         if (is.character(cohortsID) && cohortsID == 'default') {
-            cohorts <- PhenotypeLibrary::getPhenotypeLog()
-            cohortDefinitionSets <- PhenotypeLibrary::getPlCohortDefinitionSet(cohorts$cohortId[1:nrow(cohorts)])
-            cohortDefinitionSets <- cohortDefinitionSets[cohortDefinitionSets$cohortId!=921,]
+            cohortDefinitionSets <- PhenotypeLibrary::getPlCohortDefinitionSet(phenotypeLog$cohortId[1:nrow(phenotypeLog)])
             for (i in 1:nrow(cohortDefinitionSets)) {
                 cohortDefinitionSets$sql[i] <- CirceR::buildCohortQuery(cohortDefinitionSets$json[i], options = CirceR::createGenerateOptions(generateStats = TRUE, vocabularySchema = vocabschemaName))
             }
         } else if (class(cohortsID) == "integer") {
-            if (921 %in% cohortsID) {
-                cohortsID <- cohortsID[cohortsID!=921]
-            }
             cohortDefinitionSets <- PhenotypeLibrary::getPlCohortDefinitionSet(cohortsID)
             for (i in 1:nrow(cohortDefinitionSets)) {
                 cohortDefinitionSets$sql[i] <- CirceR::buildCohortQuery(cohortDefinitionSets$json[i], options = CirceR::createGenerateOptions(generateStats = TRUE, vocabularySchema = vocabschemaName))
             }
         }
+
+        # getPlCohortDefinitionSet returns only cohortId/cohortName/json/sql; the
+        # status lives in the phenotype log, so carry it across for tagging.
+        cohortDefinitionSets$status <- phenotypeLog$status[match(cohortDefinitionSets$cohortId, phenotypeLog$cohortId)]
+        cohortDefinitionSets$status[is.na(cohortDefinitionSets$status) | cohortDefinitionSets$status == ""] <- "Unspecified"
+
         return(cohortDefinitionSets)
     }
 
@@ -51,7 +53,8 @@ get_cohort_definitions <- function(cohortsID, vocabschemaName, materialize = FAL
                 cohortId = cohortDefinitionSets$cohortId[i],
                 cohortName = cohortDefinitionSets$cohortName[i],
                 json = cohortDefinitionSets$json[i],
-                sql = cohortDefinitionSets$sql[i]
+                sql = cohortDefinitionSets$sql[i],
+                status = cohortDefinitionSets$status[i]
             )
         }
         return(result_list)

plugins/flows/base/renv.lock

@@ -400,11 +400,11 @@
     },
     "PhenotypeLibrary": {
       "Package": "PhenotypeLibrary",
-      "Version": "3.36.0",
+      "Version": "3.37.0",
       "Source": "GitHub",
       "Type": "Package",
       "Title": "The OHDSI Phenotype Library",
-      "Date": "2025-04-02",
+      "Date": "2026-04-09",
       "Author": "Gowtham Rao [aut, cre]",
       "Maintainer": "Gowtham Rao <rao@ohdsi.org>",
       "Description": "A repository to store the content of the OHDSI Phenotype library.",
@@ -435,8 +435,8 @@
       "RemoteHost": "api.github.com",
       "RemoteUsername": "OHDSI",
       "RemoteRepo": "PhenotypeLibrary",
-      "RemoteRef": "v3.36.0",
-      "RemoteSha": "bf0180d522382c733e50b14a0ba55434508cc90b"
+      "RemoteRef": "v3.37.0",
+      "RemoteSha": "40106e4e2892d883c208b9df10ad56711da3e21d"
     },
     "R6": {
       "Package": "R6",