add gene ID detection and probeset conversion utilities for eFP expression endpoint

Author: rmobmina

.flake8

@@ -1,5 +1,8 @@
 [flake8]
 ignore = E501, E203, E121, E123, E126, W503, W504
+per-file-ignores =
+    # DATABASE_SPECIES uses aligned dict values for readability (intentional)
+    api/utils/gene_id_utils.py: E241
 exclude =
     .git,
     .venv,

api/resources/gene_expression.py

@@ -1,9 +1,17 @@
 from flask_restx import Namespace, Resource
 from markupsafe import escape
 
-from api import db
-from api.models.annotations_lookup import AtAgiLookup
-from api.services.efp_data import query_efp_database_dynamic, DYNAMIC_DATABASE_SCHEMAS
+from api.services.efp_data import query_efp_database_dynamic
+from api.utils.bar_utils import BARUtils
+from api.utils.gene_id_utils import (
+    CROSS_SPECIES_DATABASES,
+    DATABASE_SPECIES,
+    PROBESET_DATABASES,
+    convert_gene_to_probeset,
+    is_probeset_id,
+    normalize_gene_id,
+    validate_gene_id,
+)
 
 gene_expression = Namespace(
     'Gene Expression',
@@ -18,47 +26,54 @@
 )
 @gene_expression.param(
     "gene_id",
-    "Gene ID (AGI format like AT1G01010 or probeset like 261585_at)",
+    "Gene ID (e.g. AT1G01010 for Arabidopsis, or a probeset like 261585_at)",
     _in="path",
     default="AT1G01010",
 )
 @gene_expression.param(
     "database",
-    "Database name (e.g., sample_data, klepikova, single_cell)",
+    "Database name (e.g. klepikova, atgenexp, sample_data)",
     _in="path",
     default="klepikova",
 )
 class GeneExpression(Resource):
     def get(self, database, gene_id):
+        database = str(escape(database))
+        gene_id = str(escape(gene_id))
+
+        # 1. Resolve database species and expected input species.
+        #    Cross-species databases (e.g. phelipanche) accept an Arabidopsis AGI
+        #    even though the database itself belongs to a different species.
+        species = DATABASE_SPECIES.get(database)
+        if species is None:
+            return BARUtils.error_exit(f"Unknown database '{database}'"), 400
+        input_species = CROSS_SPECIES_DATABASES.get(database, species)
 
-        database = escape(database)
-        gene_id = escape(gene_id)
+        # 2. If the caller already supplied a probeset ID, use it directly
+        if is_probeset_id(gene_id):
+            query_id = gene_id
+        else:
+            # 3. Validate gene ID format against the expected input species regex
+            if not validate_gene_id(gene_id, input_species):
+                return BARUtils.error_exit(f"Invalid {input_species} gene ID: '{gene_id}'"), 400
 
-        upper_id = gene_id.upper()
-        is_agi = upper_id.startswith("AT") and "G" in upper_id
+            # 4. Normalise (e.g. strip maize transcript suffix _T##)
+            gene_id = normalize_gene_id(gene_id, species)
 
-        # for databases that store probeset IDs, convert AGI to probeset via at_agi_lookup
-        schema = DYNAMIC_DATABASE_SCHEMAS.get(str(database))
-        if schema and is_agi and schema.get("identifier_type") == "probeset":
-            subquery = (
-                db.select(AtAgiLookup.probeset)
-                .where(AtAgiLookup.agi == upper_id)
-                .order_by(AtAgiLookup.date.desc())
-                .limit(1)
-                .subquery()
-            )
-            sq_query = db.session.query(subquery)
-            if sq_query.count() > 0:
-                gene_id = sq_query[0][0]
+            # 5. Microarray / non-direct databases need gene ID → probeset conversion
+            if database in PROBESET_DATABASES:
+                probeset, err = convert_gene_to_probeset(gene_id, species, database)
+                if err:
+                    return BARUtils.error_exit(err), 404
+                query_id = probeset
             else:
-                return {"success": False, "error": f"No probeset found for {gene_id}", "error_code": 404}, 404
+                query_id = gene_id
 
-        result = query_efp_database_dynamic(database, gene_id, sample_ids=None)
+        result = query_efp_database_dynamic(database, query_id)
 
         if result["success"]:
-            return result
-        else:
-            return result, result.get("error_code", 500)
+            return BARUtils.success_exit(result)
+        return BARUtils.error_exit(result["error"]), result.get("error_code", 500)
 
 
 gene_expression.add_resource(GeneExpression, '/expression/<string:database>/<string:gene_id>')