diff --git a/charts/d2e-services/templates/atlas-db-init-cm.yaml b/charts/d2e-services/templates/atlas-db-init-cm.yaml index 769b1acce2..cd3bac53ed 100644 --- a/charts/d2e-services/templates/atlas-db-init-cm.yaml +++ b/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; + diff --git a/plugins/flows/_shared_flow_utils/api/PhenotypeTagAPI.py b/plugins/flows/_shared_flow_utils/api/PhenotypeTagAPI.py new file mode 100644 index 0000000000..0d6203932b --- /dev/null +++ b/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}" + ) diff --git a/plugins/flows/base/phenotype_plugin/flow.py b/plugins/flows/base/phenotype_plugin/flow.py index df0e4a303d..c8a0cfe05b 100644 --- a/plugins/flows/base/phenotype_plugin/flow.py +++ b/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, diff --git a/plugins/flows/base/phenotype_plugin/get_cohort_definitions.R b/plugins/flows/base/phenotype_plugin/get_cohort_definitions.R index 32359c7b63..d71213c598 100644 --- a/plugins/flows/base/phenotype_plugin/get_cohort_definitions.R +++ b/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) diff --git a/plugins/flows/base/renv.lock b/plugins/flows/base/renv.lock index 3ac56a1838..f45fc2d1f8 100644 --- a/plugins/flows/base/renv.lock +++ b/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 ", "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", diff --git a/plugins/functions/d2e-webapi/src/dto/cohortdefinition.ts b/plugins/functions/d2e-webapi/src/dto/cohortdefinition.ts index 915707ccbf..256221143d 100644 --- a/plugins/functions/d2e-webapi/src/dto/cohortdefinition.ts +++ b/plugins/functions/d2e-webapi/src/dto/cohortdefinition.ts @@ -48,20 +48,20 @@ const WebAPICohortUserDto = z.object({ const WebAPICohortTagDto = z.object({ name: z.string(), id: z.number(), - hasWriteAccess: z.boolean(), - modifiedBy: WebAPICohortUserDto, - createdBy: WebAPICohortUserDto, - createdDate: z.string(), - modifiedDate: z.string(), - icon: z.string(), + hasWriteAccess: z.boolean().optional(), + modifiedBy: WebAPICohortUserDto.nullish(), + createdBy: WebAPICohortUserDto.nullish(), + createdDate: z.union([z.number(), z.string()]), + modifiedDate: z.union([z.number(), z.string()]).nullish(), + icon: z.string().nullish(), permissionProtected: z.boolean(), multiSelection: z.boolean(), mandatory: z.boolean(), type: z.enum(["SYSTEM", "CUSTOM", "PRIZM"]), - description: z.string(), + description: z.string().nullish(), count: z.number(), groups: z.array(z.unknown()), - color: z.string(), + color: z.string().nullish(), showGroup: z.boolean(), allowCustom: z.boolean(), }); diff --git a/services/atlas-db-init/230_imported_cohort_metadata_tag_group.sql b/services/atlas-db-init/230_imported_cohort_metadata_tag_group.sql new file mode 100644 index 0000000000..da6073900b --- /dev/null +++ b/services/atlas-db-init/230_imported_cohort_metadata_tag_group.sql @@ -0,0 +1,61 @@ +-- 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;