redis/redis-py · error · ValueError

Bad query type

Error message

Bad query type {type(query)}

What it means

Raised by _mk_query_args() (redis/commands/search/commands.py:1309) as a ValueError when the query argument to FT.SEARCH is neither a str nor a Query object. The library accepts a raw query string (auto-wrapped in Query) or a Query instance; anything else (dict, int, None, bytes) is rejected before the command is built.

Solutions

  1. Pass a query string or a Query object to search().
  2. Use aggregate() with an AggregateRequest for aggregation queries.
  3. Coerce: query = Query(str(raw)) if raw is not None else Query('*').

Example fix

# before
index.search(user_input)  # user_input is a dict
# after
from redis.commands.search.query import Query
index.search(Query(user_input.get('q', '*')))
Defensive patterns

Strategy: type-guard

Validate before calling

from redis.commands.search.query import Query

def safe_search_query(q):
    if isinstance(q, Query):
        return q
    if isinstance(q, str):
        return Query(q)
    raise TypeError('search query must be str or Query')

Type guard

from typing import Union
from redis.commands.search.query import Query

def is_search_query(q) -> bool:
    return isinstance(q, (str, Query))

Try / catch

try:
    index.search(q)
except ValueError as e:
    if 'Bad query type' in str(e):
        index.search(Query(str(q)))
    else:
        raise

Prevention

When it happens

Trigger: Calling index.search(None), index.search({'q': 'foo'}), index.search(123), or passing a dict/AggregateRequest to search() instead of aggregate().

Common situations: Passing the wrong request type (AggregateRequest belongs to aggregate(), not search()), None from an empty input field, or a parsed JSON/dict that was not converted to a Query.

Related errors


AI-assisted analysis of redis/redis-py@6a6b581b48 (2026-08-10). Data as JSON: /api/errors/128b6c4212cc708d. Report an issue: GitHub.

Appendix: source

Thrown at redis/commands/search/commands.py:1309

        args = []
        if len(query_params) > 0:
            args.append("PARAMS")
            args.append(len(query_params) * 2)
            for key, value in query_params.items():
                args.append(key)
                args.append(value)
        return args

    def _mk_query_args(
        self, query, query_params: Optional[Dict[str, Union[str, int, float, bytes]]]
    ):
        args = [self.index_name]

        if isinstance(query, str):
            # convert the query from a text to a query object
            query = Query(query)
        if not isinstance(query, Query):
            raise ValueError(f"Bad query type {type(query)}")

        args += query.get_args()
        args += self.get_params_args(query_params)

        return args, query

    def search(
        self,
        query: Union[str, Query],
        query_params: Union[Dict[str, Union[str, int, float, bytes]], None] = None,
    ):
        """
        Search the index for a given query, and return a result of documents

        ### Parameters

        - **query**: the search query. Either a text for simple queries with
                     default parameters, or a Query object for complex queries.

View on GitHub (pinned to 6a6b581b48)