Handle empty DatasetResourceId

ingestify/infra/store/dataset/sqlalchemy/repository.py

@@ -268,33 +268,35 @@ def _filter_query(
             if not selectors:
                 raise ValueError("Selectors must contain at least one item")
 
-            attribute_cte = self._build_cte(
-                [selector.filtered_attributes for selector in selectors], "attributes"
-            )
-
-            keys = list(selectors[0].filtered_attributes.keys())
             first_selector = selectors[0].filtered_attributes
+            keys = list(first_selector.keys())
 
-            join_conditions = []
-            for k in keys:
-                if dialect == "postgresql":
-                    column = dataset_table.c.identifier[k]
+            if keys:
+                attribute_cte = self._build_cte(
+                    [selector.filtered_attributes for selector in selectors],
+                    "attributes",
+                )
 
-                    # Take the value from the first selector to determine the type.
-                    # TODO: check all selectors to determine the type
-                    v = first_selector[k]
-                    if isinstance(v, int):
-                        column = column.as_integer()
+                join_conditions = []
+                for k in keys:
+                    if dialect == "postgresql":
+                        column = dataset_table.c.identifier[k]
+
+                        # Take the value from the first selector to determine the type.
+                        # TODO: check all selectors to determine the type
+                        v = first_selector[k]
+                        if isinstance(v, int):
+                            column = column.as_integer()
+                        else:
+                            column = column.as_string()
                     else:
-                        column = column.as_string()
-                else:
-                    column = func.json_extract(dataset_table.c.identifier, f"$.{k}")
+                        column = func.json_extract(dataset_table.c.identifier, f"$.{k}")
 
-                join_conditions.append(attribute_cte.c[k] == column)
+                    join_conditions.append(attribute_cte.c[k] == column)
 
-            query = query.select_from(
-                dataset_table.join(attribute_cte, and_(*join_conditions))
-            )
+                query = query.select_from(
+                    dataset_table.join(attribute_cte, and_(*join_conditions))
+                )
 
         if where:
             query = query.filter(text(where))

ingestify/tests/test_engine.py

@@ -117,6 +117,33 @@ def find_datasets(
         # )
 
 
+class EmptyDatasetResourceIdSource(Source):
+    provider = "fake"
+
+    def find_datasets(
+        self,
+        dataset_type: str,
+        data_spec_versions: DataSpecVersionCollection,
+        dataset_collection_metadata: DatasetCollectionMetadata,
+        **kwargs,
+    ):
+        last_modified = datetime.now(pytz.utc)
+
+        yield (
+            DatasetResource(
+                dataset_resource_id={},
+                provider="fake",
+                dataset_type="match",
+                name="Test Dataset",
+            ).add_file(
+                last_modified=last_modified,
+                data_feed_key="file3",
+                data_spec_version="v1",
+                json_content={"test": "some-content"},
+            )
+        )
+
+
 class BatchSource(Source):
     provider = "batch"
 
@@ -382,3 +409,11 @@ def test_serde(config_file):
         deserialized_event = deserialize(event_dict)
 
         assert event.model_dump_json() == deserialized_event.model_dump_json()
+
+
+def test_empty_dataset_resource_id(config_file):
+    """When a empty DatasetResourceId is passed nothing should break"""
+    engine = get_engine(config_file, "main")
+
+    add_ingestion_plan(engine, EmptyDatasetResourceIdSource("fake-source"))
+    engine.load()