Source code for sqlspec.builder._select

"""SELECT statement builder.

Provides a fluent interface for building SQL SELECT queries with
parameter binding and validation.
"""

# pyright: reportPrivateUsage=false, reportPrivateImportUsage=false

import re
from typing import TYPE_CHECKING, Any, Final, Union, cast

from mypy_extensions import trait
from sqlglot import exp
from typing_extensions import Self

from sqlspec.builder._base import BuiltQuery, QueryBuilder
from sqlspec.builder._explain import ExplainMixin
from sqlspec.builder._join import JoinClauseMixin, _attach_as_of_version
from sqlspec.builder._parsing_utils import (
    _PARAMETER_VALIDATOR,
    extract_column_name,
    extract_expression,
    parse_column_expression,
    parse_condition_expression,
    parse_order_expression,
    parse_table_expression,
    to_expression,
)
from sqlspec.core import SQL, ParameterStyle, SQLResult
from sqlspec.exceptions import SQLBuilderError
from sqlspec.utils.type_guards import (
    has_expression_and_sql,
    has_parameter_builder,
    has_sqlglot_expression,
    is_expression,
    is_iterable_parameters,
)

if TYPE_CHECKING:
    from collections.abc import Callable

    from sqlglot.dialects.dialect import DialectType

    from sqlspec.builder._column import Column, ColumnExpression, FunctionColumn
    from sqlspec.builder._expression_wrappers import ExpressionWrapper
    from sqlspec.protocols import SQLBuilderProtocol

__all__ = (
    "Case",
    "CaseBuilder",
    "CommonTableExpressionMixin",
    "HavingClauseMixin",
    "LimitOffsetClauseMixin",
    "OrderByClauseMixin",
    "PivotClauseMixin",
    "ReturningClauseMixin",
    "Select",
    "SelectClauseMixin",
    "SetOperationMixin",
    "SubqueryBuilder",
    "UnpivotClauseMixin",
    "WhereClauseMixin",
    "WindowFunctionBuilder",
)

BETWEEN_BOUND_COUNT = 2
PAIR_LENGTH = 2
TRIPLE_LENGTH = 3


def is_explicitly_quoted(identifier: Any) -> bool:
    """Detect if identifier was provided with explicit quotes."""
    if not isinstance(identifier, str):
        return False
    stripped = identifier.strip()
    return (stripped.startswith('"') and stripped.endswith('"')) or (
        stripped.startswith("`") and stripped.endswith("`")
    )


def _expr_eq(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return exp.EQ(this=col, expression=placeholder)


def _expr_neq(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return exp.NEQ(this=col, expression=placeholder)


def _expr_gt(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return exp.GT(this=col, expression=placeholder)


def _expr_gte(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return exp.GTE(this=col, expression=placeholder)


def _expr_lt(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return exp.LT(this=col, expression=placeholder)


def _expr_lte(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return exp.LTE(this=col, expression=placeholder)


def _expr_like_exp(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return exp.Like(this=col, expression=placeholder)


def _expr_like_method(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return cast("exp.Expr", col.like(placeholder))


def _expr_not_like(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return exp.Not(this=exp.Like(this=col, expression=placeholder))


def _expr_ilike(col: "exp.Expr", placeholder: "exp.Placeholder") -> "exp.Expr":
    return cast("exp.Expr", col.ilike(placeholder))


_SIMPLE_OPERATOR_MAP: dict[str, Any] = {
    "=": _expr_eq,
    "==": _expr_eq,
    "!=": _expr_neq,
    "<>": _expr_neq,
    ">": _expr_gt,
    ">=": _expr_gte,
    "<": _expr_lt,
    "<=": _expr_lte,
    "LIKE": _expr_like_exp,
    "NOT LIKE": _expr_not_like,
}



[docs]
class Case:
    """Represent a SQL CASE expression with structured components."""

    __slots__ = ("conditions", "default")


[docs]
    def __init__(self, *ifs: exp.Expr, default: exp.Expr | None = None) -> None:
        self.conditions = list(ifs)
        self.default = default