feat(app): add workflow library get_counts method

Get the counts of workflows for the given tags and/or categories. Made a separate method bc get_many will deserialize all matching workflows, which is unnecessary for this use case.

Author: psychedelicious

invokeai/app/api/routers/workflows.py

@@ -219,3 +219,13 @@ async def get_workflow_thumbnail(
         return response
     except Exception:
         raise HTTPException(status_code=404)
+
+
+@workflows_router.get("/counts", operation_id="get_counts")
+async def get_counts(
+    tags: Optional[list[str]] = Query(default=None, description="The tags to include"),
+    categories: Optional[list[WorkflowCategory]] = Query(default=None, description="The categories to include"),
+) -> int:
+    """Gets a the count of workflows that include the specified tags and categories"""
+
+    return ApiDependencies.invoker.services.workflow_records.get_counts(tags=tags, categories=categories)

invokeai/app/services/workflow_records/workflow_records_base.py

@@ -49,3 +49,12 @@ def get_many(
     ) -> PaginatedResults[WorkflowRecordListItemDTO]:
         """Gets many workflows."""
         pass
+
+    @abstractmethod
+    def get_counts(
+        self,
+        tags: Optional[list[str]],
+        categories: Optional[list[WorkflowCategory]],
+    ) -> int:
+        """Gets the count of workflows for the given tags and categories."""
+        pass

invokeai/app/services/workflow_records/workflow_records_sqlite.py

@@ -237,6 +237,55 @@ def get_many(
             total=total,
         )
 
+    def get_counts(
+        self,
+        tags: Optional[list[str]],
+        categories: Optional[list[WorkflowCategory]],
+    ) -> int:
+        cursor = self._conn.cursor()
+
+        # Start with an empty list of conditions and params
+        conditions: list[str] = []
+        params: list[str | int] = []
+
+        if tags:
+            # Construct a list of conditions for each tag
+            tags_conditions = ["tags LIKE ?" for _ in tags]
+            tags_conditions_joined = " OR ".join(tags_conditions)
+            tags_condition = f"({tags_conditions_joined})"
+
+            # And the params for the tags, case-insensitive
+            tags_params = [f"%{t.strip()}%" for t in tags]
+
+            conditions.append(tags_condition)
+            params.extend(tags_params)
+
+        if categories:
+            # Ensure all categories are valid (is this necessary?)
+            assert all(c in WorkflowCategory for c in categories)
+
+            # Construct a placeholder string for the number of categories
+            placeholders = ", ".join("?" for _ in categories)
+
+            # Construct the condition string & params
+            conditions.append(f"category IN ({placeholders})")
+            params.extend([category.value for category in categories])
+
+        stmt = """--sql
+            SELECT COUNT(*)
+            FROM workflow_library
+            """
+
+        if conditions:
+            # If there are conditions, add a WHERE clause and then join the conditions
+            stmt += " WHERE "
+
+            all_conditions = " AND ".join(conditions)
+            stmt += all_conditions
+
+        cursor.execute(stmt, tuple(params))
+        return cursor.fetchone()[0]
+
     def _sync_default_workflows(self) -> None:
         """Syncs default workflows to the database. Internal use only."""
 
@@ -261,13 +310,13 @@ def _sync_default_workflows(self) -> None:
                 bytes_ = path.read_bytes()
                 workflow_from_file = WorkflowValidator.validate_json(bytes_)
 
-                assert workflow_from_file.id.startswith("default_"), (
-                    f'Invalid default workflow ID (must start with "default_"): {workflow_from_file.id}'
-                )
+                assert workflow_from_file.id.startswith(
+                    "default_"
+                ), f'Invalid default workflow ID (must start with "default_"): {workflow_from_file.id}'
 
-                assert workflow_from_file.meta.category is WorkflowCategory.Default, (
-                    f"Invalid default workflow category: {workflow_from_file.meta.category}"
-                )
+                assert (
+                    workflow_from_file.meta.category is WorkflowCategory.Default
+                ), f"Invalid default workflow category: {workflow_from_file.meta.category}"
 
                 workflows_from_file.append(workflow_from_file)