Source code for sqlspec.core.stack

"""Immutable builder utilities for multi-statement execution stacks."""

from collections.abc import Iterator, Mapping, Sequence
from types import MappingProxyType
from typing import TYPE_CHECKING, Any

from mypy_extensions import mypyc_attr

if TYPE_CHECKING:  # pragma: no cover
    from sqlspec.builder import QueryBuilder
    from sqlspec.core.filters import StatementFilter
    from sqlspec.core.statement import Statement, StatementConfig
    from sqlspec.typing import StatementParameters
__all__ = ("StackOperation", "StatementStack")


ALLOWED_METHODS: "tuple[str, ...]" = ("execute", "execute_many", "execute_script", "execute_arrow")


@mypyc_attr(allow_interpreted_subclasses=False)
class StackOperation:
    """Single SQL operation captured inside a statement stack."""

    __slots__ = ("arguments", "keyword_arguments", "method", "statement")

    def __init__(
        self,
        method: str,
        statement: Any,
        arguments: "tuple[Any, ...] | None" = None,
        keyword_arguments: "Mapping[str, Any] | None" = None,
    ) -> None:
        if method not in ALLOWED_METHODS:
            msg = f"Unsupported stack method {method!r}"
            raise ValueError(msg)
        self.method = method
        self.statement = statement
        self.arguments = arguments if arguments is not None else ()
        self.keyword_arguments = keyword_arguments