Translator

The translator turns a SQL string into a Redis FT.SEARCH or FT.AGGREGATE command. It does not execute anything; use Executor for that.

Translator

Translator

Translator(
    schema_registry: SchemaRegistry | AsyncSchemaRegistry,
)

Translates SQL queries to Redis FT.SEARCH/FT.AGGREGATE commands.

Initialize translator with schema registry.

Parameters:

Name	Type	Description	Default
schema_registry	SchemaRegistry | AsyncSchemaRegistry	Registry containing index schemas. Can be either sync (SchemaRegistry) or async (AsyncSchemaRegistry) - only the sync get_schema() method is used.	required

Source code in sql_redis/translator.py

def __init__(self, schema_registry: SchemaRegistry | AsyncSchemaRegistry) -> None:
    """Initialize translator with schema registry.

    Args:
        schema_registry: Registry containing index schemas. Can be either
            sync (SchemaRegistry) or async (AsyncSchemaRegistry) - only
            the sync get_schema() method is used.
    """
    self._schema_registry = schema_registry
    self._parser = SQLParser()
    self._query_builder = QueryBuilder()

parse

parse(sql: str) -> ParsedQuery

Parse a SQL SELECT into a ParsedQuery AST.

Useful when callers need the parsed result before translation (e.g., to extract the index name for async schema loading).

Parameters:

Name	Type	Description	Default
sql	str	SQL SELECT statement.	required

Returns:

Type	Description
ParsedQuery	ParsedQuery with extracted index, fields, conditions, etc.

Source code in sql_redis/translator.py

def parse(self, sql: str) -> ParsedQuery:
    """Parse a SQL SELECT into a ParsedQuery AST.

    Useful when callers need the parsed result before translation
    (e.g., to extract the index name for async schema loading).

    Args:
        sql: SQL SELECT statement.

    Returns:
        ParsedQuery with extracted index, fields, conditions, etc.
    """
    return self._parser.parse(sql)

translate

translate(sql: str) -> TranslatedQuery

Translate a SQL SELECT into a Redis search command.

Parameters:

Name	Type	Description	Default
sql	str	SQL SELECT statement.	required

Returns:

Type	Description
TranslatedQuery	TranslatedQuery with command details.

Raises:

Type	Description
ValueError	If SQL is invalid or references unknown index/fields.

Source code in sql_redis/translator.py

def translate(self, sql: str) -> TranslatedQuery:
    """Translate a SQL SELECT into a Redis search command.

    Args:
        sql: SQL SELECT statement.

    Returns:
        TranslatedQuery with command details.

    Raises:
        ValueError: If SQL is invalid or references unknown index/fields.
    """
    parsed = self._parser.parse(sql)
    return self.translate_parsed(parsed)

translate_parsed

translate_parsed(parsed: ParsedQuery) -> TranslatedQuery

Translate a pre-parsed query into a Redis search command.

This avoids re-parsing SQL when the caller has already parsed it (e.g., AsyncExecutor extracts the index name before translation).

Parameters:

Name	Type	Description	Default
parsed	ParsedQuery	A ParsedQuery from SQLParser.parse().	required

Returns:

Type	Description
TranslatedQuery	TranslatedQuery with command details.

Raises:

Type	Description
ValueError	If the index or a field is unknown.

Source code in sql_redis/translator.py

def translate_parsed(self, parsed: ParsedQuery) -> TranslatedQuery:
    """Translate a pre-parsed query into a Redis search command.

    This avoids re-parsing SQL when the caller has already parsed it
    (e.g., AsyncExecutor extracts the index name before translation).

    Args:
        parsed: A ParsedQuery from SQLParser.parse().

    Returns:
        TranslatedQuery with command details.

    Raises:
        ValueError: If the index or a field is unknown.
    """
    # Get schema and analyze — raise early for missing indexes
    schema = self._schema_registry.get_schema(parsed.index)
    if not schema:
        raise ValueError(f"Unknown index: {parsed.index}")
    schemas = {parsed.index: schema}
    analyzer = Analyzer(schemas)
    analyzed = analyzer.analyze(parsed)

    # Build query
    return self._build_command(analyzed)

TranslatedQuery

TranslatedQuery dataclass

TranslatedQuery(
    command: str,
    index: str,
    query_string: str,
    args: list[str] = list(),
    params: dict[str, object] = dict(),
    score_alias: str | None = None,
)

Result of translating SQL to Redis.

to_command_list

to_command_list() -> list[str]

Return as a list suitable for redis.execute_command().

Source code in sql_redis/translator.py

def to_command_list(self) -> list[str]:
    """Return as a list suitable for redis.execute_command()."""
    return [self.command, self.index, self.query_string, *self.args]

to_command_string

to_command_string() -> str

Return as a human-readable command string.

Source code in sql_redis/translator.py

def to_command_string(self) -> str:
    """Return as a human-readable command string."""
    parts = [self.command, self.index, f'"{self.query_string}"']
    parts.extend(self.args)
    return " ".join(parts)