Method search_workflows

workflow_db.py:539–624

Fast search with filters and pagination.

(
        self,
        query: str = "",
        trigger_filter: str = "all",
        complexity_filter: str = "all",
        active_only: bool = False,
        limit: int = 50,
        offset: int = 0,
    )

        return stats

    def search_workflows(
        self,
        query: str = "",
        trigger_filter: str = "all",
        complexity_filter: str = "all",
        active_only: bool = False,
        limit: int = 50,
        offset: int = 0,
    ) -> Tuple[List[Dict], int]:
        """Fast search with filters and pagination."""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row

        # Build WHERE clause
        where_conditions = []
        params = []

        if active_only:
            where_conditions.append("w.active = 1")

        if trigger_filter != "all":
            where_conditions.append("w.trigger_type = ?")
            params.append(trigger_filter)

        if complexity_filter != "all":
            where_conditions.append("w.complexity = ?")
            params.append(complexity_filter)

        # Use FTS search if query provided
        if query.strip():
            # FTS search with ranking
            base_query = """
                SELECT w.*, rank
                FROM workflows_fts fts
                JOIN workflows w ON w.id = fts.rowid
                WHERE workflows_fts MATCH ?
            """
            params.insert(0, query)
        else:
            # Regular query without FTS
            base_query = """
                SELECT w.*, 0 as rank
                FROM workflows w
                WHERE 1=1
            """

        if where_conditions:
            base_query += " AND " + " AND ".join(where_conditions)

        # Count total results
        count_query = f"SELECT COUNT(*) as total FROM ({base_query}) t"
        cursor = conn.execute(count_query, params)
        total = cursor.fetchone()["total"]

        # Get paginated results
        if query.strip():
            base_query += " ORDER BY rank"
        else: