Data structure for filters

benchmarks/runner/mocks.py

@@ -1,5 +1,4 @@
 from typing import Any
-from typing import Literal
 from typing import Optional
 
 from pydantic import BaseModel
@@ -13,21 +12,14 @@ class DatasetV2Mock(BaseModel):
     name: str
     zarr_dir: str
     images: list[dict[str, Any]] = Field(default_factory=list)
-    filters: dict[Literal["types", "attributes"], dict[str, Any]] = Field(
-        default_factory=dict
-    )
+    type_filters: dict[str, bool] = Field(default_factory=dict)
+    attribute_filters: dict[str, list[Any]] = Field(default_factory=dict)
     history: list = Field(default_factory=list)
 
     @property
     def image_zarr_urls(self) -> list[str]:
         return [image["zarr_urls"] for image in self.images]
 
-    @validator("filters", always=True)
-    def _default_filters(cls, value):
-        if value == {}:
-            return {"types": {}, "attributes": {}}
-        return value
-
 
 class TaskV2Mock(BaseModel):
     id: int
@@ -77,13 +69,8 @@ class WorkflowTaskV2Mock(BaseModel):
     meta_parallel: Optional[dict[str, Any]] = Field()
     meta_non_parallel: Optional[dict[str, Any]] = Field()
     task: TaskV2Mock = None
-    input_filters: dict[str, Any] = Field(default_factory=dict)
+    type_filters: dict[str, bool] = Field(default_factory=dict)
     order: int
     id: int
     workflow_id: int = 0
     task_id: int
-
-    @validator("input_filters", always=True)
-    def _default_filters(cls, value):
-        if value == {}:
-            return {"types": {}, "attributes": {}}

fractal_server/app/models/v2/dataset.py

@@ -1,6 +1,5 @@
 from datetime import datetime
 from typing import Any
-from typing import Literal
 from typing import Optional
 
 from sqlalchemy import Column
@@ -41,12 +40,11 @@ class Config:
         sa_column=Column(JSON, server_default="[]", nullable=False)
     )
 
-    filters: dict[Literal["attributes", "types"], dict[str, Any]] = Field(
-        sa_column=Column(
-            JSON,
-            nullable=False,
-            server_default='{"attributes": {}, "types": {}}',
-        )
+    type_filters: dict[str, bool] = Field(
+        sa_column=Column(JSON, nullable=False, server_default="{}")
+    )
+    attribute_filters: dict[str, list[Any]] = Field(
+        sa_column=Column(JSON, nullable=False, server_default="{}")
     )
 
     @property

fractal_server/app/models/v2/job.py

@@ -49,3 +49,7 @@ class Config:
     )
     status: str = JobStatusTypeV2.SUBMITTED
     log: Optional[str] = None
+
+    attribute_filters: dict[str, list[Any]] = Field(
+        sa_column=Column(JSON, nullable=False, server_default="{}")
+    )

fractal_server/app/models/v2/workflowtask.py

@@ -1,5 +1,4 @@
 from typing import Any
-from typing import Literal
 from typing import Optional
 
 from sqlalchemy import Column
@@ -25,14 +24,8 @@ class Config:
     args_parallel: Optional[dict[str, Any]] = Field(sa_column=Column(JSON))
     args_non_parallel: Optional[dict[str, Any]] = Field(sa_column=Column(JSON))
 
-    input_filters: dict[
-        Literal["attributes", "types"], dict[str, Any]
-    ] = Field(
-        sa_column=Column(
-            JSON,
-            nullable=False,
-            server_default='{"attributes": {}, "types": {}}',
-        )
+    type_filters: dict[str, bool] = Field(
+        sa_column=Column(JSON, nullable=False, server_default="{}")
     )
 
     # Task

fractal_server/app/routes/api/v2/_aux_functions.py

@@ -21,7 +21,6 @@
 from ....models.v2 import WorkflowTaskV2
 from ....models.v2 import WorkflowV2
 from ....schemas.v2 import JobStatusTypeV2
-from fractal_server.images import Filters
 
 
 async def _get_project_check_owner(
@@ -336,7 +335,7 @@ async def _workflow_insert_task(
     meta_non_parallel: Optional[dict[str, Any]] = None,
     args_non_parallel: Optional[dict[str, Any]] = None,
     args_parallel: Optional[dict[str, Any]] = None,
-    input_filters: Optional[Filters] = None,
+    type_filters: Optional[dict[str, bool]] = None,
     db: AsyncSession,
 ) -> WorkflowTaskV2:
     """
@@ -350,7 +349,7 @@ async def _workflow_insert_task(
         meta_non_parallel:
         args_non_parallel:
         args_parallel:
-        input_filters:
+        type_filters:
         db:
     """
     db_workflow = await db.get(WorkflowV2, workflow_id)
@@ -376,12 +375,6 @@ async def _workflow_insert_task(
     if final_meta_non_parallel == {}:
         final_meta_non_parallel = None
 
-    # Prepare input_filters attribute
-    if input_filters is None:
-        input_filters_kwarg = {}
-    else:
-        input_filters_kwarg = dict(input_filters=input_filters)
-
     # Create DB entry
     wf_task = WorkflowTaskV2(
         task_type=task_type,
@@ -390,7 +383,7 @@ async def _workflow_insert_task(
         args_parallel=args_parallel,
         meta_parallel=final_meta_parallel,
         meta_non_parallel=final_meta_non_parallel,
-        **input_filters_kwarg,
+        type_filters=(type_filters or dict()),
     )
     db_workflow.task_list.append(wf_task)
     flag_modified(db_workflow, "task_list")

fractal_server/app/routes/api/v2/images.py

@@ -15,7 +15,6 @@
 from fractal_server.app.db import get_async_db
 from fractal_server.app.models import UserOAuth
 from fractal_server.app.routes.auth import current_active_user
-from fractal_server.images import Filters
 from fractal_server.images import SingleImage
 from fractal_server.images import SingleImageUpdate
 from fractal_server.images.tools import find_image_by_zarr_url
@@ -38,7 +37,8 @@ class ImagePage(BaseModel):
 
 class ImageQuery(BaseModel):
     zarr_url: Optional[str]
-    filters: Filters = Field(default_factory=Filters)
+    type_filters: dict[str, bool] = Field(default_factory=dict)
+    attribute_filters: dict[str, list[Any]] = Field(default_factory=dict)
 
 
 @router.post(
@@ -124,7 +124,11 @@ async def query_dataset_images(
         images = [
             image
             for image in images
-            if match_filter(image, Filters(**dataset.filters))
+            if match_filter(
+                image,
+                type_filters=dataset.type_filters,
+                attribute_filters=dataset.attribute_filters,
+            )
         ]
 
     attributes = {}
@@ -160,7 +164,8 @@ async def query_dataset_images(
                 for image in images
                 if match_filter(
                     image,
-                    Filters(**query.filters.dict()),
+                    type_filters=query.type_filters,
+                    attribute_filters=query.attribute_filters,
                 )
             ]
 

fractal_server/app/routes/api/v2/workflowtask.py

@@ -109,7 +109,7 @@ async def replace_workflowtask(
         task_type=task.type,
         task=task,
         # old-task values
-        input_filters=old_workflow_task.input_filters,
+        type_filters=old_workflow_task.type_filters,
         # possibly new values
         args_non_parallel=_args_non_parallel,
         args_parallel=_args_parallel,
@@ -183,7 +183,7 @@ async def create_workflowtask(
         meta_parallel=new_task.meta_parallel,
         args_non_parallel=new_task.args_non_parallel,
         args_parallel=new_task.args_parallel,
-        input_filters=new_task.input_filters,
+        type_filters=new_task.type_filters,
         db=db,
     )
 
@@ -274,7 +274,7 @@ async def update_workflowtask(
             if not actual_args:
                 actual_args = None
             setattr(db_wf_task, key, actual_args)
-        elif key in ["meta_parallel", "meta_non_parallel", "input_filters"]:
+        elif key in ["meta_parallel", "meta_non_parallel", "type_filters"]:
             setattr(db_wf_task, key, value)
         else:
             raise HTTPException(

fractal_server/app/runner/filenames.py

@@ -1,5 +1,6 @@
 HISTORY_FILENAME = "history.json"
-FILTERS_FILENAME = "filters.json"
+TYPE_FILTERS_FILENAME = "type_filters.json"
+ATTRIBUTE_FILTERS_FILENAME = "attribute_filters.json"
 IMAGES_FILENAME = "images.json"
 METADATA_FILENAME = "metadata.json"
 SHUTDOWN_FILENAME = "shutdown"

fractal_server/app/runner/v2/runner.py

@@ -7,15 +7,14 @@
 from typing import Callable
 from typing import Optional
 
-from ....images import Filters
 from ....images import SingleImage
 from ....images.tools import filter_image_list
 from ....images.tools import find_image_by_zarr_url
 from ....images.tools import match_filter
 from ..exceptions import JobExecutionError
-from ..filenames import FILTERS_FILENAME
 from ..filenames import HISTORY_FILENAME
 from ..filenames import IMAGES_FILENAME
+from ..filenames import TYPE_FILTERS_FILENAME
 from .runner_functions import no_op_submit_setup_call
 from .runner_functions import run_v2_task_compound
 from .runner_functions import run_v2_task_non_parallel
@@ -47,7 +46,7 @@ def execute_tasks_v2(
     # Initialize local dataset attributes
     zarr_dir = dataset.zarr_dir
     tmp_images = deepcopy(dataset.images)
-    tmp_filters = deepcopy(dataset.filters)
+    tmp_type_filters = deepcopy(dataset.type_filters)
     tmp_history = []
 
     for wftask in wf_task_list:
@@ -58,19 +57,14 @@ def execute_tasks_v2(
         # PRE TASK EXECUTION
 
         # Get filtered images
-        pre_filters = dict(
-            types=copy(tmp_filters["types"]),
-            attributes=copy(tmp_filters["attributes"]),
-        )
-        pre_filters["types"].update(wftask.input_filters["types"])
-        pre_filters["attributes"].update(wftask.input_filters["attributes"])
+        pre_type_filters = copy(tmp_type_filters)
+        pre_type_filters.update(wftask.type_filters)
         filtered_images = filter_image_list(
-            images=tmp_images,
-            filters=Filters(**pre_filters),
+            images=tmp_images, type_filters=pre_type_filters
         )
         # Verify that filtered images comply with task input_types
         for image in filtered_images:
-            if not match_filter(image, Filters(types=task.input_types)):
+            if not match_filter(image, type_filters=task.input_types):
                 raise JobExecutionError(
                     "Invalid filtered image list\n"
                     f"Task input types: {task.input_types=}\n"
@@ -249,19 +243,12 @@ def execute_tasks_v2(
             else:
                 tmp_images.pop(img_search["index"])
 
-        # Update filters.attributes:
-        # current + (task_output: not really, in current examples..)
-        if current_task_output.filters is not None:
-            tmp_filters["attributes"].update(
-                current_task_output.filters.attributes
-            )
-
         # Find manifest ouptut types
         types_from_manifest = task.output_types
 
         # Find task-output types
-        if current_task_output.filters is not None:
-            types_from_task = current_task_output.filters.types
+        if current_task_output.type_filters is not None:
+            types_from_task = current_task_output.type_filters
         else:
             types_from_task = {}
 
@@ -279,8 +266,8 @@ def execute_tasks_v2(
             )
 
         # Update filters.types
-        tmp_filters["types"].update(types_from_manifest)
-        tmp_filters["types"].update(types_from_task)
+        tmp_type_filters.update(types_from_manifest)
+        tmp_type_filters.update(types_from_task)
 
         # Update history (based on _DatasetHistoryItemV2)
         history_item = _DatasetHistoryItemV2(
@@ -299,8 +286,8 @@ def execute_tasks_v2(
         # information
         with open(workflow_dir_local / HISTORY_FILENAME, "w") as f:
             json.dump(tmp_history, f, indent=2)
-        with open(workflow_dir_local / FILTERS_FILENAME, "w") as f:
-            json.dump(tmp_filters, f, indent=2)
+        with open(workflow_dir_local / TYPE_FILTERS_FILENAME, "w") as f:
+            json.dump(tmp_type_filters, f, indent=2)
         with open(workflow_dir_local / IMAGES_FILENAME, "w") as f:
             json.dump(tmp_images, f, indent=2)
 
@@ -311,7 +298,7 @@ def execute_tasks_v2(
     # represent the new attributes (to replace the original ones)
     result = dict(
         history=tmp_history,
-        filters=tmp_filters,
+        type_filters=tmp_type_filters,
         images=tmp_images,
     )
     return result

fractal_server/app/runner/v2/task_interface.py

@@ -6,7 +6,6 @@
 from pydantic import validator
 
 from ....images import SingleImageTaskOutput
-from fractal_server.images import Filters
 from fractal_server.urls import normalize_url
 
 
@@ -16,7 +15,8 @@ class TaskOutput(BaseModel, extra=Extra.forbid):
         default_factory=list
     )
     image_list_removals: list[str] = Field(default_factory=list)
-    filters: Filters = Field(default_factory=Filters)
+    type_filters: dict[str, bool] = Field(default_factory=dict)
+    attribute_filters: dict[str, list[Any]] = Field(default_factory=dict)
 
     def check_zarr_urls_are_unique(self) -> None:
         zarr_urls = [img.zarr_url for img in self.image_list_updates]