Source code for neosqlite.collection.query_helper.schema_compiler

"""
SQL Compiler for JSON Schema validation.
Translates MongoDB $jsonSchema rules into native SQLite SQL expressions.
"""

from typing import Any




def compile_schema_to_sql(
    schema: dict[str, Any], data_column: str = "data", jsonb: bool = False
) -> str:
    """
    Compile a JSON Schema into a SQL expression for use in a CHECK constraint.
    """
    json_func = "jsonb_extract" if jsonb else "json_extract"
    return _compile_node(schema, data_column, json_func)





def _safe_json_path(path: str) -> str:
    """Escape a dotted path for embedding in a SQL string literal (#123)."""
    from ..json_path_utils import parse_json_path

    rel = path[2:] if path.startswith("$.") else path
    return parse_json_path(rel) if rel else "$"





def _numeric_literal(value: Any, name: str) -> str:
    """Validate and render a numeric schema bound (#123)."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(
            f"JSON schema '{name}' must be a number, got {value!r}"
        )
    return repr(float(value))





def _compile_node(
    schema: Any, data_column: str, json_func: str, path: str = "$"
) -> str:
    """Recursively compile schema nodes into SQL."""
    if not isinstance(schema, dict):
        return "1"  # Always true

    clauses = []

    # 1. Handle required fields
    if "required" in schema:
        req_fields = schema["required"]
        for field in req_fields:
            field_path = f"{path}.{field}"
            clauses.append(
                f"{json_func}({data_column}, "
                f"'{_safe_json_path(field_path)}') IS NOT NULL"
            )

    # 2. Handle properties
    if "properties" in schema:
        for field, prop_schema in schema["properties"].items():
            field_path = f"{path}.{field}"
            # Only validate if the field exists (JSON Schema standard behavior)
            prop_sql = _compile_node(
                prop_schema, data_column, json_func, field_path
            )
            if prop_sql != "1":
                clauses.append(
                    f"({json_func}({data_column}, "
                    f"'{_safe_json_path(field_path)}') IS NULL OR ({prop_sql}))"
                )

    # 3. Handle type/bsonType
    # SQLite json_type returns: 'null', 'integer', 'real', 'text', 'array', 'object'
    target_type = schema.get("bsonType") or schema.get("type")
    if target_type:
        type_clause = _compile_type_check(
            data_column, path, target_type, json_func
        )
        if type_clause:
            clauses.append(type_clause)

    # 4. Handle numeric constraints
    val_expr = f"{json_func}({data_column}, '{_safe_json_path(path)}')"
    if "minimum" in schema:
        clauses.append(
            f"{val_expr} >= {_numeric_literal(schema['minimum'], 'minimum')}"
        )
    if "maximum" in schema:
        clauses.append(
            f"{val_expr} <= {_numeric_literal(schema['maximum'], 'maximum')}"
        )
    if "exclusiveMinimum" in schema:
        clauses.append(
            f"{val_expr} > {_numeric_literal(schema['exclusiveMinimum'], 'exclusiveMinimum')}"
        )
    if "exclusiveMaximum" in schema:
        clauses.append(
            f"{val_expr} < {_numeric_literal(schema['exclusiveMaximum'], 'exclusiveMaximum')}"
        )

    if not clauses:
        return "1"

    return " AND ".join(clauses)





def _compile_type_check(
    data_column: str, path: str, target_type: Any, json_func: str
) -> str | None:
    """Compile type checks into SQL using json_type."""
    # json_type(data, path)
    type_expr = f"json_type({data_column}, '{_safe_json_path(path)}')"

    types = target_type if isinstance(target_type, list) else [target_type]
    sql_types = []

    for t in types:
        match t:
            case "string":
                sql_types.append("'text'")
            case "number":
                sql_types.extend(["'integer'", "'real'"])
            case "integer" | "int" | "long":
                sql_types.append("'integer'")
            case "double" | "decimal":
                sql_types.append("'real'")
            case "object":
                sql_types.append("'object'")
            case "array":
                sql_types.append("'array'")
            case "bool" | "boolean":
                # SQLite stores bools as 0/1, json_type might not distinguish well
                # but usually it's handled.
                pass
            case "null":
                sql_types.append("'null'")

    if not sql_types:
        return None

    return f"{type_expr} IN ({', '.join(sql_types)})"