Source code for sqlspec.builder._insert

"""INSERT statement builder.

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

from typing import TYPE_CHECKING, Any, Final

from sqlglot import exp
from typing_extensions import Self

from sqlspec.builder._base import QueryBuilder
from sqlspec.builder._dml import InsertFromSelectMixin, InsertIntoClauseMixin, InsertValuesMixin
from sqlspec.builder._explain import ExplainMixin
from sqlspec.builder._parsing_utils import extract_sql_object_expression
from sqlspec.builder._select import ReturningClauseMixin
from sqlspec.core import SQLResult
from sqlspec.exceptions import SQLBuilderError
from sqlspec.utils.serializers import schema_dump, serialize_collection
from sqlspec.utils.type_guards import has_expression_and_sql

if TYPE_CHECKING:
    from collections.abc import Mapping, Sequence


__all__ = ("Insert",)

ERR_MSG_TABLE_NOT_SET: Final[str] = "The target table must be set using .into() before adding values."
ERR_MSG_INTERNAL_EXPRESSION_TYPE: Final[str] = "Internal error: expression is not an Insert instance as expected."
ERR_MSG_EXPRESSION_NOT_INITIALIZED: Final[str] = "Internal error: base expression not initialized."


class Insert(
    QueryBuilder, ReturningClauseMixin, InsertValuesMixin, InsertFromSelectMixin, InsertIntoClauseMixin, ExplainMixin
):
    """Builder for INSERT statements.

    Constructs SQL INSERT queries with parameter binding and validation.
    """

    __slots__ = ("_columns", "_values_added_count")



[docs]

    def __init__(self, table: str | None = None, **kwargs: Any) -> None:
        """Initialize INSERT with optional table.

        Args:
            table: Target table name
            **kwargs: Additional QueryBuilder arguments
        """
        (dialect, schema, enable_optimization, optimize_joins, optimize_predicates, simplify_expressions) = (
            self._parse_query_builder_kwargs(kwargs)
        )
        super().__init__(
            dialect=dialect,
            schema=schema,
            enable_optimization=enable_optimization,
            optimize_joins=optimize_joins,
            optimize_predicates=optimize_predicates,
            simplify_expressions=simplify_expressions,
        )

        self._columns: list[str] = []
        self._values_added_count: int = 0

        self._initialize_expression()

        if table:
            self.into(table)
def _create_base_expression(self) -> exp.Insert: """Create a base INSERT expression. This method is called by the base QueryBuilder during initialization. Returns: A new sqlglot Insert expression. """ return exp.Insert() @property def _expected_result_type(self) -> "type[SQLResult]": """Specifies the expected result type for an INSERT query. Returns: The type of result expected for INSERT operations. """ return SQLResult def _get_insert_expression(self) -> exp.Insert: """Safely gets and casts the internal expression to exp.Insert. Returns: The internal expression as exp.Insert. Raises: SQLBuilderError: If the expression is not initialized or is not an Insert. """ if self._expression is None: raise SQLBuilderError(ERR_MSG_EXPRESSION_NOT_INITIALIZED) if not isinstance(self._expression, exp.Insert): raise SQLBuilderError(ERR_MSG_INTERNAL_EXPRESSION_TYPE) return self._expression def get_insert_expression(self) -> exp.Insert: """Get the insert expression (public API).""" return self._get_insert_expression()