[ENH] query feature (#875)

* wip: query features

* wip: search based on metadata

* wip: separate return and filters of features

* wip: restructure from feedback

* run black

* wip: refactor query feature

* fix query building and testing

* test the or functionality

* fix jsonpath string generation

* fix tests for most the schemas

* fix tests

* formatting

* formatting

Author: jdkent

store/neurostore/ingest/extracted_features.py

@@ -10,8 +10,7 @@
 from neurostore.models import (
     Pipeline,
     PipelineConfig,
-    PipelineRun,
-    PipelineRunResult,
+    PipelineStudyResult,
 )
 
 
@@ -26,15 +25,13 @@ def ingest_feature(feature_directory):
         db.session.query(Pipeline)
         .filter(
             Pipeline.name == pipeline_info["name"],
-            Pipeline.version == pipeline_info["version"],
         )
         .first()
     )
     # create a pipeline if it does not exist
     if not pipeline:
         pipeline = Pipeline(
             name=pipeline_info["name"],
-            version=pipeline_info["version"],
             description=pipeline_info.get("description"),
             study_dependent=(
                 True if pipeline_info.get("type", False) == "dependent" else False
@@ -58,29 +55,25 @@ def ingest_feature(feature_directory):
         .filter(
             PipelineConfig.pipeline_id == pipeline.id,
             PipelineConfig.config_hash == config_hash,
+            PipelineConfig.version == pipeline_info["version"],
         )
         .first()
     )
     # create a pipeline config if it does not exist
     if not pipeline_config:
         pipeline_config = PipelineConfig(
             pipeline_id=pipeline.id,
+            version=pipeline_info["version"],
             config=pipeline_info["arguments"],
             config_hash=config_hash,
         )
         db.session.add(pipeline_config)
 
-    # create a new pipeline run
-    pipeline_run = PipelineRun(
-        pipeline_id=pipeline.id,
-        config_id=pipeline_config.id,
-    )
-
     # get a list of all the paper directories in the feature directory
     paper_dirs = [d for d in Path(feature_directory).iterdir() if d.is_dir()]
 
     # for each subject directory, read the results.json file and the info.json file
-    pipeline_run_results = []
+    pipeline_study_results = []
     for paper_dir in paper_dirs:
         with open(op.join(paper_dir, "results.json")) as f:
             results = json.load(f)
@@ -91,17 +84,16 @@ def ingest_feature(feature_directory):
         # use the directory name as the base_study_id
         base_study_id = paper_dir.name
         # create a new result record
-        pipeline_run_results.append(
-            PipelineRunResult(
+        pipeline_study_results.append(
+            PipelineStudyResult(
                 base_study_id=base_study_id,
-                data=results,
+                result_data=results,
                 date_executed=parse_date(info["date"]),
                 file_inputs=info["inputs"],
-                run=pipeline_run,
+                config=pipeline_config,
             )
         )
 
-    db.session.add(pipeline_run)
-    db.session.add_all(pipeline_run_results)
+    db.session.add_all(pipeline_study_results)
 
     db.session.commit()

store/neurostore/models/__init__.py

@@ -14,9 +14,7 @@
     AnalysisConditions,
     Pipeline,
     PipelineConfig,
-    PipelineRun,
-    PipelineRunResult,
-    PipelineRunResultVote,
+    PipelineStudyResult,
 )
 from .auth import User, Role
 
@@ -38,7 +36,5 @@
     "Role",
     "Pipeline",
     "PipelineConfig",
-    "PipelineRun",
-    "PipelineRunResult",
-    "PipelineRunResultVote",
+    "PipelineStudyResult",
 ]

store/neurostore/models/data.py

@@ -1,18 +1,20 @@
+import re
+
 import sqlalchemy as sa
 from sqlalchemy import exists
 from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy import ForeignKeyConstraint
+from sqlalchemy import ForeignKeyConstraint, func
 from sqlalchemy.ext.associationproxy import association_proxy
-
 from sqlalchemy.ext.mutable import MutableDict
-from sqlalchemy.orm import relationship, backref
-from sqlalchemy.sql import func
+from sqlalchemy.orm import relationship, backref, validates, aliased
 import shortuuid
 
 from .migration_types import TSVector
 from ..database import db
 
+SEMVER_REGEX = r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$"  # noqa E501
+
 
 def _check_type(x):
     """check annotation key type"""
@@ -193,6 +195,9 @@ class BaseStudy(BaseMixin, db.Model):
     versions = relationship(
         "Study", backref=backref("base_study"), passive_deletes=True
     )
+    pipeline_study_results = relationship(
+        "PipelineStudyResult", backref=backref("base_study"), passive_deletes=True
+    )
 
     __table_args__ = (
         db.CheckConstraint(level.in_(["group", "meta"])),
@@ -263,6 +268,71 @@ def update_has_images_and_points(self):
         self.has_images = self.images_exist
         self.has_coordinates = self.points_exist
 
+    def display_features(self, pipelines=None):
+        """
+        Display pipeline features for the base study.
+        Only loads and returns features if pipelines are explicitly specified.
+
+        Args:
+            pipelines (list, optional): List of pipeline names to display features from.
+                                      If None or empty, returns empty dict.
+        """
+        if not pipelines:
+            return {}
+
+        # Create aliases for the tables
+        PipelineAlias = aliased(Pipeline)
+        PipelineConfigAlias = aliased(PipelineConfig)
+        PipelineStudyResultAlias = aliased(PipelineStudyResult)
+
+        # Get latest results subquery
+        latest_results = (
+            db.session.query(
+                PipelineStudyResultAlias.base_study_id,
+                PipelineAlias.name.label("pipeline_name"),
+                func.max(PipelineStudyResultAlias.date_executed).label("max_date"),
+            )
+            .join(
+                PipelineConfigAlias,
+                PipelineStudyResultAlias.config_id == PipelineConfigAlias.id,
+            )
+            .join(PipelineAlias, PipelineConfigAlias.pipeline_id == PipelineAlias.id)
+            .filter(PipelineStudyResultAlias.base_study_id == self.id)
+            .filter(PipelineAlias.name.in_(pipelines))
+            .group_by(PipelineStudyResultAlias.base_study_id, PipelineAlias.name)
+            .subquery()
+        )
+
+        # Main query joining with latest results
+        query = (
+            db.session.query(
+                PipelineStudyResultAlias.result_data,
+                PipelineAlias.name.label("pipeline_name"),
+            )
+            .join(
+                PipelineConfigAlias,
+                PipelineStudyResultAlias.config_id == PipelineConfigAlias.id,
+            )
+            .join(PipelineAlias, PipelineConfigAlias.pipeline_id == PipelineAlias.id)
+            .join(
+                latest_results,
+                (
+                    PipelineStudyResultAlias.base_study_id
+                    == latest_results.c.base_study_id
+                )
+                & (PipelineAlias.name == latest_results.c.pipeline_name)
+                & (PipelineStudyResultAlias.date_executed == latest_results.c.max_date),
+            )
+        )
+
+        # Execute query and build response
+        results = query.all()
+        features = {}
+        for result in results:
+            features[result.pipeline_name] = result.result_data
+
+        return features
+
 
 class Study(BaseMixin, db.Model):
     __tablename__ = "studies"
@@ -544,7 +614,6 @@ class Pipeline(BaseMixin, db.Model):
 
     name = db.Column(db.String)
     description = db.Column(db.String)
-    version = db.Column(db.String)
     study_dependent = db.Column(db.Boolean, default=False)
     ace_compatible = db.Column(db.Boolean, default=False)
     pubget_compatible = db.Column(db.Boolean, default=False)
@@ -557,55 +626,39 @@ class PipelineConfig(BaseMixin, db.Model):
     pipeline_id = db.Column(
         db.Text, db.ForeignKey("pipelines.id", ondelete="CASCADE"), index=True
     )
+    version = db.Column(db.String)
     config = db.Column(JSONB)
+    executed_at = db.Column(
+        db.DateTime(timezone=True)
+    )  # when the pipeline was executed on the filesystem (not when it was ingested)
     config_hash = db.Column(db.String, index=True)
     pipeline = relationship(
         "Pipeline", backref=backref("configs", passive_deletes=True)
     )
 
+    @validates("version")
+    def validate_version(self, key, value):
+        if not re.match(SEMVER_REGEX, value):
+            raise ValueError(f"Invalid version format: {value}")
+        return value
 
-class PipelineRun(BaseMixin, db.Model):
-    __tablename__ = "pipeline_runs"
 
-    pipeline_id = db.Column(
-        db.Text, db.ForeignKey("pipelines.id", ondelete="CASCADE"), index=True
-    )
+class PipelineStudyResult(BaseMixin, db.Model):
+    __tablename__ = "pipeline_study_results"
+
     config_id = db.Column(
         db.Text, db.ForeignKey("pipeline_configs.id", ondelete="CASCADE"), index=True
     )
-    config = relationship(
-        "PipelineConfig", backref=backref("runs", passive_deletes=True)
-    )
-    run_index = db.Column(db.Integer())
-
-
-class PipelineRunResult(BaseMixin, db.Model):
-    __tablename__ = "pipeline_run_results"
-
-    run_id = db.Column(
-        db.Text, db.ForeignKey("pipeline_runs.id", ondelete="CASCADE"), index=True
-    )
     base_study_id = db.Column(db.Text, db.ForeignKey("base_studies.id"), index=True)
     date_executed = db.Column(db.DateTime(timezone=True))
-    data = db.Column(JSONB)
+    result_data = db.Column(JSONB)
     file_inputs = db.Column(JSONB)
-    run = relationship("PipelineRun", backref=backref("results", passive_deletes=True))
-
-
-class PipelineRunResultVote(BaseMixin, db.Model):
-    __tablename__ = "pipeline_run_result_votes"
-
-    run_result_id = db.Column(
-        db.Text,
-        db.ForeignKey("pipeline_run_results.id", ondelete="CASCADE"),
-        index=True,
+    status = db.Column(
+        db.Enum("SUCCESS", "FAILURE", "ERROR", "UNKNOWN", name="status_enum")
     )
-    user_id = db.Column(db.Text, db.ForeignKey("users.external_id"), index=True)
-    accurate = db.Column(db.Boolean)
-    run_result = relationship(
-        "PipelineRunResult", backref=backref("votes", passive_deletes=True)
+    config = relationship(
+        "PipelineConfig", backref=backref("results", passive_deletes=True)
     )
-    user = relationship("User", backref=backref("votes", passive_deletes=True))
 
 
 # from . import event_listeners  # noqa E402

store/neurostore/openapi

@@ -18,9 +18,7 @@
 from .pipeline import (
     PipelinesView,
     PipelineConfigsView,
-    PipelineRunsView,
-    PipelineRunResultsView,
-    PipelineRunResultVotesView,
+    PipelineStudyResultsView,
 )
 
 __all__ = [
@@ -37,7 +35,5 @@
     "UsersView",
     "PipelinesView",
     "PipelineConfigsView",
-    "PipelineRunsView",
-    "PipelineRunResultsView",
-    "PipelineRunResultVotesView",
+    "PipelineStudyResultsView",
 ]