# sql/compiler.py # Copyright (C) 2005-2025 the SQLAlchemy authors and contributors # # # This module is part of SQLAlchemy and is released under # the MIT License: https://www.opensource.org/licenses/mit-license.php # mypy: allow-untyped-defs, allow-untyped-calls """Base SQL and DDL compiler implementations. Classes provided include: :class:`.compiler.SQLCompiler` - renders SQL strings :class:`.compiler.DDLCompiler` - renders DDL (data definition language) strings :class:`.compiler.GenericTypeCompiler` - renders type specification strings. To generate user-defined SQL strings, see :doc:`/ext/compiler`. """ from __future__ import annotations import collections import collections.abc as collections_abc import contextlib from enum import IntEnum import functools import itertools import operator import re from time import perf_counter import typing from typing import Any from typing import Callable from typing import cast from typing import ClassVar from typing import Dict from typing import FrozenSet from typing import Iterable from typing import Iterator from typing import List from typing import Mapping from typing import MutableMapping from typing import NamedTuple from typing import NoReturn from typing import Optional from typing import Pattern from typing import Sequence from typing import Set from typing import Tuple from typing import Type from typing import TYPE_CHECKING from typing import Union from . import base from . import coercions from . import crud from . import elements from . import functions from . import operators from . import roles from . import schema from . import selectable from . import sqltypes from . import util as sql_util from ._typing import is_column_element from ._typing import is_dml from .base import _de_clone from .base import _from_objects from .base import _NONE_NAME from .base import _SentinelDefaultCharacterization from .base import NO_ARG from .elements import quoted_name from .sqltypes import TupleType from .visitors import prefix_anon_map from .. import exc from .. import util from ..util import FastIntFlag from ..util.typing import Literal from ..util.typing import Protocol from ..util.typing import Self from ..util.typing import TypedDict if typing.TYPE_CHECKING: from .annotation import _AnnotationDict from .base import _AmbiguousTableNameMap from .base import CompileState from .base import Executable from .cache_key import CacheKey from .ddl import ExecutableDDLElement from .dml import Insert from .dml import Update from .dml import UpdateBase from .dml import UpdateDMLState from .dml import ValuesBase from .elements import _truncated_label from .elements import BinaryExpression from .elements import BindParameter from .elements import ClauseElement from .elements import ColumnClause from .elements import ColumnElement from .elements import False_ from .elements import Label from .elements import Null from .elements import True_ from .functions import Function from .schema import Column from .schema import Constraint from .schema import ForeignKeyConstraint from .schema import Index from .schema import PrimaryKeyConstraint from .schema import Table from .schema import UniqueConstraint from .selectable import _ColumnsClauseElement from .selectable import AliasedReturnsRows from .selectable import CompoundSelectState from .selectable import CTE from .selectable import FromClause from .selectable import NamedFromClause from .selectable import ReturnsRows from .selectable import Select from .selectable import SelectState from .type_api import _BindProcessorType from .type_api import TypeDecorator from .type_api import TypeEngine from .type_api import UserDefinedType from .visitors import Visitable from ..engine.cursor import CursorResultMetaData from ..engine.interfaces import _CoreSingleExecuteParams from ..engine.interfaces import _DBAPIAnyExecuteParams from ..engine.interfaces import _DBAPIMultiExecuteParams from ..engine.interfaces import _DBAPISingleExecuteParams from ..engine.interfaces import _ExecuteOptions from ..engine.interfaces import _GenericSetInputSizesType from ..engine.interfaces import _MutableCoreSingleExecuteParams from ..engine.interfaces import Dialect from ..engine.interfaces import SchemaTranslateMapType _FromHintsType = Dict["FromClause", str] RESERVED_WORDS = { "all", "analyse", "analyze", "and", "any", "array", "as", "asc", "asymmetric", "authorization", "between", "binary", "both", "case", "cast", "check", "collate", "column", "constraint", "create", "cross", "current_date", "current_role", "current_time", "current_timestamp", "current_user", "default", "deferrable", "desc", "distinct", "do", "else", "end", "except", "false", "for", "foreign", "freeze", "from", "full", "grant", "group", "having", "ilike", "in", "initially", "inner", "intersect", "into", "is", "isnull", "join", "leading", "left", "like", "limit", "localtime", "localtimestamp", "natural", "new", "not", "notnull", "null", "off", "offset", "old", "on", "only", "or", "order", "outer", "overlaps", "placing", "primary", "references", "right", "select", "session_user", "set", "similar", "some", "symmetric", "table", "then", "to", "trailing", "true", "union", "unique", "user", "using", "verbose", "when", "where", } LEGAL_CHARACTERS = re.compile(r"^[A-Z0-9_$]+$", re.I) LEGAL_CHARACTERS_PLUS_SPACE = re.compile(r"^[A-Z0-9_ $]+$", re.I) ILLEGAL_INITIAL_CHARACTERS = {str(x) for x in range(0, 10)}.union(["$"]) FK_ON_DELETE = re.compile( r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I ) FK_ON_UPDATE = re.compile( r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I ) FK_INITIALLY = re.compile(r"^(?:DEFERRED|IMMEDIATE)$", re.I) BIND_PARAMS = re.compile(r"(? ", operators.ge: " >= ", operators.eq: " = ", operators.is_distinct_from: " IS DISTINCT FROM ", operators.is_not_distinct_from: " IS NOT DISTINCT FROM ", operators.concat_op: " || ", operators.match_op: " MATCH ", operators.not_match_op: " NOT MATCH ", operators.in_op: " IN ", operators.not_in_op: " NOT IN ", operators.comma_op: ", ", operators.from_: " FROM ", operators.as_: " AS ", operators.is_: " IS ", operators.is_not: " IS NOT ", operators.collate: " COLLATE ", # unary operators.exists: "EXISTS ", operators.distinct_op: "DISTINCT ", operators.inv: "NOT ", operators.any_op: "ANY ", operators.all_op: "ALL ", # modifiers operators.desc_op: " DESC", operators.asc_op: " ASC", operators.nulls_first_op: " NULLS FIRST", operators.nulls_last_op: " NULLS LAST", # bitwise operators.bitwise_xor_op: " ^ ", operators.bitwise_or_op: " | ", operators.bitwise_and_op: " & ", operators.bitwise_not_op: "~", operators.bitwise_lshift_op: " << ", operators.bitwise_rshift_op: " >> ", } FUNCTIONS: Dict[Type[Function[Any]], str] = { functions.coalesce: "coalesce", functions.current_date: "CURRENT_DATE", functions.current_time: "CURRENT_TIME", functions.current_timestamp: "CURRENT_TIMESTAMP", functions.current_user: "CURRENT_USER", functions.localtime: "LOCALTIME", functions.localtimestamp: "LOCALTIMESTAMP", functions.random: "random", functions.sysdate: "sysdate", functions.session_user: "SESSION_USER", functions.user: "USER", functions.cube: "CUBE", functions.rollup: "ROLLUP", functions.grouping_sets: "GROUPING SETS", } EXTRACT_MAP = { "month": "month", "day": "day", "year": "year", "second": "second", "hour": "hour", "doy": "doy", "minute": "minute", "quarter": "quarter", "dow": "dow", "week": "week", "epoch": "epoch", "milliseconds": "milliseconds", "microseconds": "microseconds", "timezone_hour": "timezone_hour", "timezone_minute": "timezone_minute", } COMPOUND_KEYWORDS = { selectable._CompoundSelectKeyword.UNION: "UNION", selectable._CompoundSelectKeyword.UNION_ALL: "UNION ALL", selectable._CompoundSelectKeyword.EXCEPT: "EXCEPT", selectable._CompoundSelectKeyword.EXCEPT_ALL: "EXCEPT ALL", selectable._CompoundSelectKeyword.INTERSECT: "INTERSECT", selectable._CompoundSelectKeyword.INTERSECT_ALL: "INTERSECT ALL", } class ResultColumnsEntry(NamedTuple): """Tracks a column expression that is expected to be represented in the result rows for this statement. This normally refers to the columns clause of a SELECT statement but may also refer to a RETURNING clause, as well as for dialect-specific emulations. """ keyname: str """string name that's expected in cursor.description""" name: str """column name, may be labeled""" objects: Tuple[Any, ...] """sequence of objects that should be able to locate this column in a RowMapping. This is typically string names and aliases as well as Column objects. """ type: TypeEngine[Any] """Datatype to be associated with this column. This is where the "result processing" logic directly links the compiled statement to the rows that come back from the cursor. """ class _ResultMapAppender(Protocol): def __call__( self, keyname: str, name: str, objects: Sequence[Any], type_: TypeEngine[Any], ) -> None: ... # integer indexes into ResultColumnsEntry used by cursor.py. # some profiling showed integer access faster than named tuple RM_RENDERED_NAME: Literal[0] = 0 RM_NAME: Literal[1] = 1 RM_OBJECTS: Literal[2] = 2 RM_TYPE: Literal[3] = 3 class _BaseCompilerStackEntry(TypedDict): asfrom_froms: Set[FromClause] correlate_froms: Set[FromClause] selectable: ReturnsRows class _CompilerStackEntry(_BaseCompilerStackEntry, total=False): compile_state: CompileState need_result_map_for_nested: bool need_result_map_for_compound: bool select_0: ReturnsRows insert_from_select: Select[Any] class ExpandedState(NamedTuple): """represents state to use when producing "expanded" and "post compile" bound parameters for a statement. "expanded" parameters are parameters that are generated at statement execution time to suit a number of parameters passed, the most prominent example being the individual elements inside of an IN expression. "post compile" parameters are parameters where the SQL literal value will be rendered into the SQL statement at execution time, rather than being passed as separate parameters to the driver. To create an :class:`.ExpandedState` instance, use the :meth:`.SQLCompiler.construct_expanded_state` method on any :class:`.SQLCompiler` instance. """ statement: str """String SQL statement with parameters fully expanded""" parameters: _CoreSingleExecuteParams """Parameter dictionary with parameters fully expanded. For a statement that uses named parameters, this dictionary will map exactly to the names in the statement. For a statement that uses positional parameters, the :attr:`.ExpandedState.positional_parameters` will yield a tuple with the positional parameter set. """ processors: Mapping[str, _BindProcessorType[Any]] """mapping of bound value processors""" positiontup: Optional[Sequence[str]] """Sequence of string names indicating the order of positional parameters""" parameter_expansion: Mapping[str, List[str]] """Mapping representing the intermediary link from original parameter name to list of "expanded" parameter names, for those parameters that were expanded.""" @property def positional_parameters(self) -> Tuple[Any, ...]: """Tuple of positional parameters, for statements that were compiled using a positional paramstyle. """ if self.positiontup is None: raise exc.InvalidRequestError( "statement does not use a positional paramstyle" ) return tuple(self.parameters[key] for key in self.positiontup) @property def additional_parameters(self) -> _CoreSingleExecuteParams: """synonym for :attr:`.ExpandedState.parameters`.""" return self.parameters class _InsertManyValues(NamedTuple): """represents state to use for executing an "insertmanyvalues" statement. The primary consumers of this object are the :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods. .. versionadded:: 2.0 """ is_default_expr: bool """if True, the statement is of the form ``INSERT INTO TABLE DEFAULT VALUES``, and can't be rewritten as a "batch" """ single_values_expr: str """The rendered "values" clause of the INSERT statement. This is typically the parenthesized section e.g. "(?, ?, ?)" or similar. The insertmanyvalues logic uses this string as a search and replace target. """ insert_crud_params: List[crud._CrudParamElementStr] """List of Column / bind names etc. used while rewriting the statement""" num_positional_params_counted: int """the number of bound parameters in a single-row statement. This count may be larger or smaller than the actual number of columns targeted in the INSERT, as it accommodates for SQL expressions in the values list that may have zero or more parameters embedded within them. This count is part of what's used to organize rewritten parameter lists when batching. """ sort_by_parameter_order: bool = False """if the deterministic_returnined_order parameter were used on the insert. All of the attributes following this will only be used if this is True. """ includes_upsert_behaviors: bool = False """if True, we have to accommodate for upsert behaviors. This will in some cases downgrade "insertmanyvalues" that requests deterministic ordering. """ sentinel_columns: Optional[Sequence[Column[Any]]] = None """List of sentinel columns that were located. This list is only here if the INSERT asked for sort_by_parameter_order=True, and dialect-appropriate sentinel columns were located. .. versionadded:: 2.0.10 """ num_sentinel_columns: int = 0 """how many sentinel columns are in the above list, if any. This is the same as ``len(sentinel_columns) if sentinel_columns is not None else 0`` """ sentinel_param_keys: Optional[Sequence[str]] = None """parameter str keys in each param dictionary / tuple that would link to the client side "sentinel" values for that row, which we can use to match up parameter sets to result rows. This is only present if sentinel_columns is present and the INSERT statement actually refers to client side values for these sentinel columns. .. versionadded:: 2.0.10 .. versionchanged:: 2.0.29 - the sequence is now string dictionary keys only, used against the "compiled parameteters" collection before the parameters were converted by bound parameter processors """ implicit_sentinel: bool = False """if True, we have exactly one sentinel column and it uses a server side value, currently has to generate an incrementing integer value. The dialect in question would have asserted that it supports receiving these values back and sorting on that value as a means of guaranteeing correlation with the incoming parameter list. .. versionadded:: 2.0.10 """ embed_values_counter: bool = False """Whether to embed an incrementing integer counter in each parameter set within the VALUES clause as parameters are batched over. This is only used for a specific INSERT..SELECT..VALUES..RETURNING syntax where a subquery is used to produce value tuples. Current support includes PostgreSQL, Microsoft SQL Server. .. versionadded:: 2.0.10 """ class _InsertManyValuesBatch(NamedTuple): """represents an individual batch SQL statement for insertmanyvalues. This is passed through the :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods out to the :class:`.Connection` within the :meth:`.Connection._exec_insertmany_context` method. .. versionadded:: 2.0.10 """ replaced_statement: str replaced_parameters: _DBAPIAnyExecuteParams processed_setinputsizes: Optional[_GenericSetInputSizesType] batch: Sequence[_DBAPISingleExecuteParams] sentinel_values: Sequence[Tuple[Any, ...]] current_batch_size: int batchnum: int total_batches: int rows_sorted: bool is_downgraded: bool class InsertmanyvaluesSentinelOpts(FastIntFlag): """bitflag enum indicating styles of PK defaults which can work as implicit sentinel columns """ NOT_SUPPORTED = 1 AUTOINCREMENT = 2 IDENTITY = 4 SEQUENCE = 8 ANY_AUTOINCREMENT = AUTOINCREMENT | IDENTITY | SEQUENCE _SUPPORTED_OR_NOT = NOT_SUPPORTED | ANY_AUTOINCREMENT USE_INSERT_FROM_SELECT = 16 RENDER_SELECT_COL_CASTS = 64 class CompilerState(IntEnum): COMPILING = 0 """statement is present, compilation phase in progress""" STRING_APPLIED = 1 """statement is present, string form of the statement has been applied. Additional processors by subclasses may still be pending. """ NO_STATEMENT = 2 """compiler does not have a statement to compile, is used for method access""" class Linting(IntEnum): """represent preferences for the 'SQL linting' feature. this feature currently includes support for flagging cartesian products in SQL statements. """ NO_LINTING = 0 "Disable all linting." COLLECT_CARTESIAN_PRODUCTS = 1 """Collect data on FROMs and cartesian products and gather into 'self.from_linter'""" WARN_LINTING = 2 "Emit warnings for linters that find problems" FROM_LINTING = COLLECT_CARTESIAN_PRODUCTS | WARN_LINTING """Warn for cartesian products; combines COLLECT_CARTESIAN_PRODUCTS and WARN_LINTING""" NO_LINTING, COLLECT_CARTESIAN_PRODUCTS, WARN_LINTING, FROM_LINTING = tuple( Linting ) class FromLinter(collections.namedtuple("FromLinter", ["froms", "edges"])): """represents current state for the "cartesian product" detection feature.""" def lint(self, start=None): froms = self.froms if not froms: return None, None edges = set(self.edges) the_rest = set(froms) if start is not None: start_with = start the_rest.remove(start_with) else: start_with = the_rest.pop() stack = collections.deque([start_with]) while stack and the_rest: node = stack.popleft() the_rest.discard(node) # comparison of nodes in edges here is based on hash equality, as # there are "annotated" elements that match the non-annotated ones. # to remove the need for in-python hash() calls, use native # containment routines (e.g. "node in edge", "edge.index(node)") to_remove = {edge for edge in edges if node in edge} # appendleft the node in each edge that is not # the one that matched. stack.extendleft(edge[not edge.index(node)] for edge in to_remove) edges.difference_update(to_remove) # FROMS left over? boom if the_rest: return the_rest, start_with else: return None, None def warn(self, stmt_type="SELECT"): the_rest, start_with = self.lint() # FROMS left over? boom if the_rest: froms = the_rest if froms: template = ( "{stmt_type} statement has a cartesian product between " "FROM element(s) {froms} and " 'FROM element "{start}". Apply join condition(s) ' "between each element to resolve." ) froms_str = ", ".join( f'"{self.froms[from_]}"' for from_ in froms ) message = template.format( stmt_type=stmt_type, froms=froms_str, start=self.froms[start_with], ) util.warn(message) class Compiled: """Represent a compiled SQL or DDL expression. The ``__str__`` method of the ``Compiled`` object should produce the actual text of the statement. ``Compiled`` objects are specific to their underlying database dialect, and also may or may not be specific to the columns referenced within a particular set of bind parameters. In no case should the ``Compiled`` object be dependent on the actual values of those bind parameters, even though it may reference those values as defaults. """ statement: Optional[ClauseElement] = None "The statement to compile." string: str = "" "The string representation of the ``statement``" state: CompilerState """description of the compiler's state""" is_sql = False is_ddl = False _cached_metadata: Optional[CursorResultMetaData] = None _result_columns: Optional[List[ResultColumnsEntry]] = None schema_translate_map: Optional[SchemaTranslateMapType] = None execution_options: _ExecuteOptions = util.EMPTY_DICT """ Execution options propagated from the statement. In some cases, sub-elements of the statement can modify these. """ preparer: IdentifierPreparer _annotations: _AnnotationDict = util.EMPTY_DICT compile_state: Optional[CompileState] = None """Optional :class:`.CompileState` object that maintains additional state used by the compiler. Major executable objects such as :class:`_expression.Insert`, :class:`_expression.Update`, :class:`_expression.Delete`, :class:`_expression.Select` will generate this state when compiled in order to calculate additional information about the object. For the top level object that is to be executed, the state can be stored here where it can also have applicability towards result set processing. .. versionadded:: 1.4 """ dml_compile_state: Optional[CompileState] = None """Optional :class:`.CompileState` assigned at the same point that .isinsert, .isupdate, or .isdelete is assigned. This will normally be the same object as .compile_state, with the exception of cases like the :class:`.ORMFromStatementCompileState` object. .. versionadded:: 1.4.40 """ cache_key: Optional[CacheKey] = None """The :class:`.CacheKey` that was generated ahead of creating this :class:`.Compiled` object. This is used for routines that need access to the original :class:`.CacheKey` instance generated when the :class:`.Compiled` instance was first cached, typically in order to reconcile the original list of :class:`.BindParameter` objects with a per-statement list that's generated on each call. """ _gen_time: float """Generation time of this :class:`.Compiled`, used for reporting cache stats.""" def __init__( self, dialect: Dialect, statement: Optional[ClauseElement], schema_translate_map: Optional[SchemaTranslateMapType] = None, render_schema_translate: bool = False, compile_kwargs: Mapping[str, Any] = util.immutabledict(), ): """Construct a new :class:`.Compiled` object. :param dialect: :class:`.Dialect` to compile against. :param statement: :class:`_expression.ClauseElement` to be compiled. :param schema_translate_map: dictionary of schema names to be translated when forming the resultant SQL .. seealso:: :ref:`schema_translating` :param compile_kwargs: additional kwargs that will be passed to the initial call to :meth:`.Compiled.process`. """ self.dialect = dialect self.preparer = self.dialect.identifier_preparer if schema_translate_map: self.schema_translate_map = schema_translate_map self.preparer = self.preparer._with_schema_translate( schema_translate_map ) if statement is not None: self.state = CompilerState.COMPILING self.statement = statement self.can_execute = statement.supports_execution self._annotations = statement._annotations if self.can_execute: if TYPE_CHECKING: assert isinstance(statement, Executable) self.execution_options = statement._execution_options self.string = self.process(self.statement, **compile_kwargs) if render_schema_translate: assert schema_translate_map is not None self.string = self.preparer._render_schema_translates( self.string, schema_translate_map ) self.state = CompilerState.STRING_APPLIED else: self.state = CompilerState.NO_STATEMENT self._gen_time = perf_counter() def __init_subclass__(cls) -> None: cls._init_compiler_cls() return super().__init_subclass__() @classmethod def _init_compiler_cls(cls): pass def _execute_on_connection( self, connection, distilled_params, execution_options ): if self.can_execute: return connection._execute_compiled( self, distilled_params, execution_options ) else: raise exc.ObjectNotExecutableError(self.statement) def visit_unsupported_compilation(self, element, err, **kw): raise exc.UnsupportedCompilationError(self, type(element)) from err @property def sql_compiler(self) -> SQLCompiler: """Return a Compiled that is capable of processing SQL expressions. If this compiler is one, it would likely just return 'self'. """ raise NotImplementedError() def process(self, obj: Visitable, **kwargs: Any) -> str: return obj._compiler_dispatch(self, **kwargs) def __str__(self) -> str: """Return the string text of the generated SQL or DDL.""" if self.state is CompilerState.STRING_APPLIED: return self.string else: return "" def construct_params( self, params: Optional[_CoreSingleExecuteParams] = None, extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None, escape_names: bool = True, ) -> Optional[_MutableCoreSingleExecuteParams]: """Return the bind params for this compiled object. :param params: a dict of string/object pairs whose values will override bind values compiled in to the statement. """ raise NotImplementedError() @property def params(self): """Return the bind params for this compiled object.""" return self.construct_params() class TypeCompiler(util.EnsureKWArg): """Produces DDL specification for TypeEngine objects.""" ensure_kwarg = r"visit_\w+" def __init__(self, dialect: Dialect): self.dialect = dialect def process(self, type_: TypeEngine[Any], **kw: Any) -> str: if ( type_._variant_mapping and self.dialect.name in type_._variant_mapping ): type_ = type_._variant_mapping[self.dialect.name] return type_._compiler_dispatch(self, **kw) def visit_unsupported_compilation( self, element: Any, err: Exception, **kw: Any ) -> NoReturn: raise exc.UnsupportedCompilationError(self, element) from err # this was a Visitable, but to allow accurate detection of # column elements this is actually a column element class _CompileLabel( roles.BinaryElementRole[Any], elements.CompilerColumnElement ): """lightweight label object which acts as an expression.Label.""" __visit_name__ = "label" __slots__ = "element", "name", "_alt_names" def __init__(self, col, name, alt_names=()): self.element = col self.name = name self._alt_names = (col,) + alt_names @property def proxy_set(self): return self.element.proxy_set @property def type(self): return self.element.type def self_group(self, **kw): return self class ilike_case_insensitive( roles.BinaryElementRole[Any], elements.CompilerColumnElement ): """produce a wrapping element for a case-insensitive portion of an ILIKE construct. The construct usually renders the ``lower()`` function, but on PostgreSQL will pass silently with the assumption that "ILIKE" is being used. .. versionadded:: 2.0 """ __visit_name__ = "ilike_case_insensitive_operand" __slots__ = "element", "comparator" def __init__(self, element): self.element = element self.comparator = element.comparator @property def proxy_set(self): return self.element.proxy_set @property def type(self): return self.element.type def self_group(self, **kw): return self def _with_binary_element_type(self, type_): return ilike_case_insensitive( self.element._with_binary_element_type(type_) ) class SQLCompiler(Compiled): """Default implementation of :class:`.Compiled`. Compiles :class:`_expression.ClauseElement` objects into SQL strings. """ extract_map = EXTRACT_MAP bindname_escape_characters: ClassVar[Mapping[str, str]] = ( util.immutabledict( { "%": "P", "(": "A", ")": "Z", ":": "C", ".": "_", "[": "_", "]": "_", " ": "_", } ) ) """A mapping (e.g. dict or similar) containing a lookup of characters keyed to replacement characters which will be applied to all 'bind names' used in SQL statements as a form of 'escaping'; the given characters are replaced entirely with the 'replacement' character when rendered in the SQL statement, and a similar translation is performed on the incoming names used in parameter dictionaries passed to methods like :meth:`_engine.Connection.execute`. This allows bound parameter names used in :func:`_sql.bindparam` and other constructs to have any arbitrary characters present without any concern for characters that aren't allowed at all on the target database. Third party dialects can establish their own dictionary here to replace the default mapping, which will ensure that the particular characters in the mapping will never appear in a bound parameter name. The dictionary is evaluated at **class creation time**, so cannot be modified at runtime; it must be present on the class when the class is first declared. Note that for dialects that have additional bound parameter rules such as additional restrictions on leading characters, the :meth:`_sql.SQLCompiler.bindparam_string` method may need to be augmented. See the cx_Oracle compiler for an example of this. .. versionadded:: 2.0.0rc1 """ _bind_translate_re: ClassVar[Pattern[str]] _bind_translate_chars: ClassVar[Mapping[str, str]] is_sql = True compound_keywords = COMPOUND_KEYWORDS isdelete: bool = False isinsert: bool = False isupdate: bool = False """class-level defaults which can be set at the instance level to define if this Compiled instance represents INSERT/UPDATE/DELETE """ postfetch: Optional[List[Column[Any]]] """list of columns that can be post-fetched after INSERT or UPDATE to receive server-updated values""" insert_prefetch: Sequence[Column[Any]] = () """list of columns for which default values should be evaluated before an INSERT takes place""" update_prefetch: Sequence[Column[Any]] = () """list of columns for which onupdate default values should be evaluated before an UPDATE takes place""" implicit_returning: Optional[Sequence[ColumnElement[Any]]] = None """list of "implicit" returning columns for a toplevel INSERT or UPDATE statement, used to receive newly generated values of columns. .. versionadded:: 2.0 ``implicit_returning`` replaces the previous ``returning`` collection, which was not a generalized RETURNING collection and instead was in fact specific to the "implicit returning" feature. """ isplaintext: bool = False binds: Dict[str, BindParameter[Any]] """a dictionary of bind parameter keys to BindParameter instances.""" bind_names: Dict[BindParameter[Any], str] """a dictionary of BindParameter instances to "compiled" names that are actually present in the generated SQL""" stack: List[_CompilerStackEntry] """major statements such as SELECT, INSERT, UPDATE, DELETE are tracked in this stack using an entry format.""" returning_precedes_values: bool = False """set to True classwide to generate RETURNING clauses before the VALUES or WHERE clause (i.e. MSSQL) """ render_table_with_column_in_update_from: bool = False """set to True classwide to indicate the SET clause in a multi-table UPDATE statement should qualify columns with the table name (i.e. MySQL only) """ ansi_bind_rules: bool = False """SQL 92 doesn't allow bind parameters to be used in the columns clause of a SELECT, nor does it allow ambiguous expressions like "? = ?". A compiler subclass can set this flag to False if the target driver/DB enforces this """ bindtemplate: str """template to render bound parameters based on paramstyle.""" compilation_bindtemplate: str """template used by compiler to render parameters before positional paramstyle application""" _numeric_binds_identifier_char: str """Character that's used to as the identifier of a numerical bind param. For example if this char is set to ``$``, numerical binds will be rendered in the form ``$1, $2, $3``. """ _result_columns: List[ResultColumnsEntry] """relates label names in the final SQL to a tuple of local column/label name, ColumnElement object (if any) and TypeEngine. CursorResult uses this for type processing and column targeting""" _textual_ordered_columns: bool = False """tell the result object that the column names as rendered are important, but they are also "ordered" vs. what is in the compiled object here. As of 1.4.42 this condition is only present when the statement is a TextualSelect, e.g. text("....").columns(...), where it is required that the columns are considered positionally and not by name. """ _ad_hoc_textual: bool = False """tell the result that we encountered text() or '*' constructs in the middle of the result columns, but we also have compiled columns, so if the number of columns in cursor.description does not match how many expressions we have, that means we can't rely on positional at all and should match on name. """ _ordered_columns: bool = True """ if False, means we can't be sure the list of entries in _result_columns is actually the rendered order. Usually True unless using an unordered TextualSelect. """ _loose_column_name_matching: bool = False """tell the result object that the SQL statement is textual, wants to match up to Column objects, and may be using the ._tq_label in the SELECT rather than the base name. """ _numeric_binds: bool = False """ True if paramstyle is "numeric". This paramstyle is trickier than all the others. """ _render_postcompile: bool = False """ whether to render out POSTCOMPILE params during the compile phase. This attribute is used only for end-user invocation of stmt.compile(); it's never used for actual statement execution, where instead the dialect internals access and render the internal postcompile structure directly. """ _post_compile_expanded_state: Optional[ExpandedState] = None """When render_postcompile is used, the ``ExpandedState`` used to create the "expanded" SQL is assigned here, and then used by the ``.params`` accessor and ``.construct_params()`` methods for their return values. .. versionadded:: 2.0.0rc1 """ _pre_expanded_string: Optional[str] = None """Stores the original string SQL before 'post_compile' is applied, for cases where 'post_compile' were used. """ _pre_expanded_positiontup: Optional[List[str]] = None _insertmanyvalues: Optional[_InsertManyValues] = None _insert_crud_params: Optional[crud._CrudParamSequence] = None literal_execute_params: FrozenSet[BindParameter[Any]] = frozenset() """bindparameter objects that are rendered as literal values at statement execution time. """ post_compile_params: FrozenSet[BindParameter[Any]] = frozenset() """bindparameter objects that are rendered as bound parameter placeholders at statement execution time. """ escaped_bind_names: util.immutabledict[str, str] = util.EMPTY_DICT """Late escaping of bound parameter names that has to be converted to the original name when looking in the parameter dictionary. """ has_out_parameters = False """if True, there are bindparam() objects that have the isoutparam flag set.""" postfetch_lastrowid = False """if True, and this in insert, use cursor.lastrowid to populate result.inserted_primary_key. """ _cache_key_bind_match: Optional[ Tuple[ Dict[ BindParameter[Any], List[BindParameter[Any]], ], Dict[ str, BindParameter[Any], ], ] ] = None """a mapping that will relate the BindParameter object we compile to those that are part of the extracted collection of parameters in the cache key, if we were given a cache key. """ positiontup: Optional[List[str]] = None """for a compiled construct that uses a positional paramstyle, will be a sequence of strings, indicating the names of bound parameters in order. This is used in order to render bound parameters in their correct order, and is combined with the :attr:`_sql.Compiled.params` dictionary to render parameters. This sequence always contains the unescaped name of the parameters. .. seealso:: :ref:`faq_sql_expression_string` - includes a usage example for debugging use cases. """ _values_bindparam: Optional[List[str]] = None _visited_bindparam: Optional[List[str]] = None inline: bool = False ctes: Optional[MutableMapping[CTE, str]] # Detect same CTE references - Dict[(level, name), cte] # Level is required for supporting nesting ctes_by_level_name: Dict[Tuple[int, str], CTE] # To retrieve key/level in ctes_by_level_name - # Dict[cte_reference, (level, cte_name, cte_opts)] level_name_by_cte: Dict[CTE, Tuple[int, str, selectable._CTEOpts]] ctes_recursive: bool _post_compile_pattern = re.compile(r"__\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]") _pyformat_pattern = re.compile(r"%\(([^)]+?)\)s") _positional_pattern = re.compile( f"{_pyformat_pattern.pattern}|{_post_compile_pattern.pattern}" ) @classmethod def _init_compiler_cls(cls): cls._init_bind_translate() @classmethod def _init_bind_translate(cls): reg = re.escape("".join(cls.bindname_escape_characters)) cls._bind_translate_re = re.compile(f"[{reg}]") cls._bind_translate_chars = cls.bindname_escape_characters def __init__( self, dialect: Dialect, statement: Optional[ClauseElement], cache_key: Optional[CacheKey] = None, column_keys: Optional[Sequence[str]] = None, for_executemany: bool = False, linting: Linting = NO_LINTING, _supporting_against: Optional[SQLCompiler] = None, **kwargs: Any, ): """Construct a new :class:`.SQLCompiler` object. :param dialect: :class:`.Dialect` to be used :param statement: :class:`_expression.ClauseElement` to be compiled :param column_keys: a list of column names to be compiled into an INSERT or UPDATE statement. :param for_executemany: whether INSERT / UPDATE statements should expect that they are to be invoked in an "executemany" style, which may impact how the statement will be expected to return the values of defaults and autoincrement / sequences and similar. Depending on the backend and driver in use, support for retrieving these values may be disabled which means SQL expressions may be rendered inline, RETURNING may not be rendered, etc. :param kwargs: additional keyword arguments to be consumed by the superclass. """ self.column_keys = column_keys self.cache_key = cache_key if cache_key: cksm = {b.key: b for b in cache_key[1]} ckbm = {b: [b] for b in cache_key[1]} self._cache_key_bind_match = (ckbm, cksm) # compile INSERT/UPDATE defaults/sequences to expect executemany # style execution, which may mean no pre-execute of defaults, # or no RETURNING self.for_executemany = for_executemany self.linting = linting # a dictionary of bind parameter keys to BindParameter # instances. self.binds = {} # a dictionary of BindParameter instances to "compiled" names # that are actually present in the generated SQL self.bind_names = util.column_dict() # stack which keeps track of nested SELECT statements self.stack = [] self._result_columns = [] # true if the paramstyle is positional self.positional = dialect.positional if self.positional: self._numeric_binds = nb = dialect.paramstyle.startswith("numeric") if nb: self._numeric_binds_identifier_char = ( "$" if dialect.paramstyle == "numeric_dollar" else ":" ) self.compilation_bindtemplate = _pyformat_template else: self.compilation_bindtemplate = BIND_TEMPLATES[dialect.paramstyle] self.ctes = None self.label_length = ( dialect.label_length or dialect.max_identifier_length ) # a map which tracks "anonymous" identifiers that are created on # the fly here self.anon_map = prefix_anon_map() # a map which tracks "truncated" names based on # dialect.label_length or dialect.max_identifier_length self.truncated_names: Dict[Tuple[str, str], str] = {} self._truncated_counters: Dict[str, int] = {} Compiled.__init__(self, dialect, statement, **kwargs) if self.isinsert or self.isupdate or self.isdelete: if TYPE_CHECKING: assert isinstance(statement, UpdateBase) if self.isinsert or self.isupdate: if TYPE_CHECKING: assert isinstance(statement, ValuesBase) if statement._inline: self.inline = True elif self.for_executemany and ( not self.isinsert or ( self.dialect.insert_executemany_returning and statement._return_defaults ) ): self.inline = True self.bindtemplate = BIND_TEMPLATES[dialect.paramstyle] if _supporting_against: self.__dict__.update( { k: v for k, v in _supporting_against.__dict__.items() if k not in { "state", "dialect", "preparer", "positional", "_numeric_binds", "compilation_bindtemplate", "bindtemplate", } } ) if self.state is CompilerState.STRING_APPLIED: if self.positional: if self._numeric_binds: self._process_numeric() else: self._process_positional() if self._render_postcompile: parameters = self.construct_params( escape_names=False, _no_postcompile=True, ) self._process_parameters_for_postcompile( parameters, _populate_self=True ) @property def insert_single_values_expr(self) -> Optional[str]: """When an INSERT is compiled with a single set of parameters inside a VALUES expression, the string is assigned here, where it can be used for insert batching schemes to rewrite the VALUES expression. .. versionadded:: 1.3.8 .. versionchanged:: 2.0 This collection is no longer used by SQLAlchemy's built-in dialects, in favor of the currently internal ``_insertmanyvalues`` collection that is used only by :class:`.SQLCompiler`. """ if self._insertmanyvalues is None: return None else: return self._insertmanyvalues.single_values_expr @util.ro_memoized_property def effective_returning(self) -> Optional[Sequence[ColumnElement[Any]]]: """The effective "returning" columns for INSERT, UPDATE or DELETE. This is either the so-called "implicit returning" columns which are calculated by the compiler on the fly, or those present based on what's present in ``self.statement._returning`` (expanded into individual columns using the ``._all_selected_columns`` attribute) i.e. those set explicitly using the :meth:`.UpdateBase.returning` method. .. versionadded:: 2.0 """ if self.implicit_returning: return self.implicit_returning elif self.statement is not None and is_dml(self.statement): return [ c for c in self.statement._all_selected_columns if is_column_element(c) ] else: return None @property def returning(self): """backwards compatibility; returns the effective_returning collection. """ return self.effective_returning @property def current_executable(self): """Return the current 'executable' that is being compiled. This is currently the :class:`_sql.Select`, :class:`_sql.Insert`, :class:`_sql.Update`, :class:`_sql.Delete`, :class:`_sql.CompoundSelect` object that is being compiled. Specifically it's assigned to the ``self.stack`` list of elements. When a statement like the above is being compiled, it normally is also assigned to the ``.statement`` attribute of the :class:`_sql.Compiler` object. However, all SQL constructs are ultimately nestable, and this attribute should never be consulted by a ``visit_`` method, as it is not guaranteed to be assigned nor guaranteed to correspond to the current statement being compiled. .. versionadded:: 1.3.21 For compatibility with previous versions, use the following recipe:: statement = getattr(self, "current_executable", False) if statement is False: statement = self.stack[-1]["selectable"] For versions 1.4 and above, ensure only .current_executable is used; the format of "self.stack" may change. """ try: return self.stack[-1]["selectable"] except IndexError as ie: raise IndexError("Compiler does not have a stack entry") from ie @property def prefetch(self): return list(self.insert_prefetch) + list(self.update_prefetch) @util.memoized_property def _global_attributes(self) -> Dict[Any, Any]: return {} @util.memoized_instancemethod def _init_cte_state(self) -> MutableMapping[CTE, str]: """Initialize collections related to CTEs only if a CTE is located, to save on the overhead of these collections otherwise. """ # collect CTEs to tack on top of a SELECT # To store the query to print - Dict[cte, text_query] ctes: MutableMapping[CTE, str] = util.OrderedDict() self.ctes = ctes # Detect same CTE references - Dict[(level, name), cte] # Level is required for supporting nesting self.ctes_by_level_name = {} # To retrieve key/level in ctes_by_level_name - # Dict[cte_reference, (level, cte_name, cte_opts)] self.level_name_by_cte = {} self.ctes_recursive = False return ctes @contextlib.contextmanager def _nested_result(self): """special API to support the use case of 'nested result sets'""" result_columns, ordered_columns = ( self._result_columns, self._ordered_columns, ) self._result_columns, self._ordered_columns = [], False try: if self.stack: entry = self.stack[-1] entry["need_result_map_for_nested"] = True else: entry = None yield self._result_columns, self._ordered_columns finally: if entry: entry.pop("need_result_map_for_nested") self._result_columns, self._ordered_columns = ( result_columns, ordered_columns, ) def _process_positional(self): assert not self.positiontup assert self.state is CompilerState.STRING_APPLIED assert not self._numeric_binds if self.dialect.paramstyle == "format": placeholder = "%s" else: assert self.dialect.paramstyle == "qmark" placeholder = "?" positions = [] def find_position(m: re.Match[str]) -> str: normal_bind = m.group(1) if normal_bind: positions.append(normal_bind) return placeholder else: # this a post-compile bind positions.append(m.group(2)) return m.group(0) self.string = re.sub( self._positional_pattern, find_position, self.string ) if self.escaped_bind_names: reverse_escape = {v: k for k, v in self.escaped_bind_names.items()} assert len(self.escaped_bind_names) == len(reverse_escape) self.positiontup = [ reverse_escape.get(name, name) for name in positions ] else: self.positiontup = positions if self._insertmanyvalues: positions = [] single_values_expr = re.sub( self._positional_pattern, find_position, self._insertmanyvalues.single_values_expr, ) insert_crud_params = [ ( v[0], v[1], re.sub(self._positional_pattern, find_position, v[2]), v[3], ) for v in self._insertmanyvalues.insert_crud_params ] self._insertmanyvalues = self._insertmanyvalues._replace( single_values_expr=single_values_expr, insert_crud_params=insert_crud_params, ) def _process_numeric(self): assert self._numeric_binds assert self.state is CompilerState.STRING_APPLIED num = 1 param_pos: Dict[str, str] = {} order: Iterable[str] if self._insertmanyvalues and self._values_bindparam is not None: # bindparams that are not in values are always placed first. # this avoids the need of changing them when using executemany # values () () order = itertools.chain( ( name for name in self.bind_names.values() if name not in self._values_bindparam ), self.bind_names.values(), ) else: order = self.bind_names.values() for bind_name in order: if bind_name in param_pos: continue bind = self.binds[bind_name] if ( bind in self.post_compile_params or bind in self.literal_execute_params ): # set to None to just mark the in positiontup, it will not # be replaced below. param_pos[bind_name] = None # type: ignore else: ph = f"{self._numeric_binds_identifier_char}{num}" num += 1 param_pos[bind_name] = ph self.next_numeric_pos = num self.positiontup = list(param_pos) if self.escaped_bind_names: len_before = len(param_pos) param_pos = { self.escaped_bind_names.get(name, name): pos for name, pos in param_pos.items() } assert len(param_pos) == len_before # Can't use format here since % chars are not escaped. self.string = self._pyformat_pattern.sub( lambda m: param_pos[m.group(1)], self.string ) if self._insertmanyvalues: single_values_expr = ( # format is ok here since single_values_expr includes only # place-holders self._insertmanyvalues.single_values_expr % param_pos ) insert_crud_params = [ (v[0], v[1], "%s", v[3]) for v in self._insertmanyvalues.insert_crud_params ] self._insertmanyvalues = self._insertmanyvalues._replace( # This has the numbers (:1, :2) single_values_expr=single_values_expr, # The single binds are instead %s so they can be formatted insert_crud_params=insert_crud_params, ) @util.memoized_property def _bind_processors( self, ) -> MutableMapping[ str, Union[_BindProcessorType[Any], Sequence[_BindProcessorType[Any]]] ]: # mypy is not able to see the two value types as the above Union, # it just sees "object". don't know how to resolve return { key: value # type: ignore for key, value in ( ( self.bind_names[bindparam], ( bindparam.type._cached_bind_processor(self.dialect) if not bindparam.type._is_tuple_type else tuple( elem_type._cached_bind_processor(self.dialect) for elem_type in cast( TupleType, bindparam.type ).types ) ), ) for bindparam in self.bind_names ) if value is not None } def is_subquery(self): return len(self.stack) > 1 @property def sql_compiler(self) -> Self: return self def construct_expanded_state( self, params: Optional[_CoreSingleExecuteParams] = None, escape_names: bool = True, ) -> ExpandedState: """Return a new :class:`.ExpandedState` for a given parameter set. For queries that use "expanding" or other late-rendered parameters, this method will provide for both the finalized SQL string as well as the parameters that would be used for a particular parameter set. .. versionadded:: 2.0.0rc1 """ parameters = self.construct_params( params, escape_names=escape_names, _no_postcompile=True, ) return self._process_parameters_for_postcompile( parameters, ) def construct_params( self, params: Optional[_CoreSingleExecuteParams] = None, extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None, escape_names: bool = True, _group_number: Optional[int] = None, _check: bool = True, _no_postcompile: bool = False, ) -> _MutableCoreSingleExecuteParams: """return a dictionary of bind parameter keys and values""" if self._render_postcompile and not _no_postcompile: assert self._post_compile_expanded_state is not None if not params: return dict(self._post_compile_expanded_state.parameters) else: raise exc.InvalidRequestError( "can't construct new parameters when render_postcompile " "is used; the statement is hard-linked to the original " "parameters. Use construct_expanded_state to generate a " "new statement and parameters." ) has_escaped_names = escape_names and bool(self.escaped_bind_names) if extracted_parameters: # related the bound parameters collected in the original cache key # to those collected in the incoming cache key. They will not have # matching names but they will line up positionally in the same # way. The parameters present in self.bind_names may be clones of # these original cache key params in the case of DML but the .key # will be guaranteed to match. if self.cache_key is None: raise exc.CompileError( "This compiled object has no original cache key; " "can't pass extracted_parameters to construct_params" ) else: orig_extracted = self.cache_key[1] ckbm_tuple = self._cache_key_bind_match assert ckbm_tuple is not None ckbm, _ = ckbm_tuple resolved_extracted = { bind: extracted for b, extracted in zip(orig_extracted, extracted_parameters) for bind in ckbm[b] } else: resolved_extracted = None if params: pd = {} for bindparam, name in self.bind_names.items(): escaped_name = ( self.escaped_bind_names.get(name, name) if has_escaped_names else name ) if bindparam.key in params: pd[escaped_name] = params[bindparam.key] elif name in params: pd[escaped_name] = params[name] elif _check and bindparam.required: if _group_number: raise exc.InvalidRequestError( "A value is required for bind parameter %r, " "in parameter group %d" % (bindparam.key, _group_number), code="cd3x", ) else: raise exc.InvalidRequestError( "A value is required for bind parameter %r" % bindparam.key, code="cd3x", ) else: if resolved_extracted: value_param = resolved_extracted.get( bindparam, bindparam ) else: value_param = bindparam if bindparam.callable: pd[escaped_name] = value_param.effective_value else: pd[escaped_name] = value_param.value return pd else: pd = {} for bindparam, name in self.bind_names.items(): escaped_name = ( self.escaped_bind_names.get(name, name) if has_escaped_names else name ) if _check and bindparam.required: if _group_number: raise exc.InvalidRequestError( "A value is required for bind parameter %r, " "in parameter group %d" % (bindparam.key, _group_number), code="cd3x", ) else: raise exc.InvalidRequestError( "A value is required for bind parameter %r" % bindparam.key, code="cd3x", ) if resolved_extracted: value_param = resolved_extracted.get(bindparam, bindparam) else: value_param = bindparam if bindparam.callable: pd[escaped_name] = value_param.effective_value else: pd[escaped_name] = value_param.value return pd @util.memoized_instancemethod def _get_set_input_sizes_lookup(self): dialect = self.dialect include_types = dialect.include_set_input_sizes exclude_types = dialect.exclude_set_input_sizes dbapi = dialect.dbapi def lookup_type(typ): dbtype = typ._unwrapped_dialect_impl(dialect).get_dbapi_type(dbapi) if ( dbtype is not None and (exclude_types is None or dbtype not in exclude_types) and (include_types is None or dbtype in include_types) ): return dbtype else: return None inputsizes = {} literal_execute_params = self.literal_execute_params for bindparam in self.bind_names: if bindparam in literal_execute_params: continue if bindparam.type._is_tuple_type: inputsizes[bindparam] = [ lookup_type(typ) for typ in cast(TupleType, bindparam.type).types ] else: inputsizes[bindparam] = lookup_type(bindparam.type) return inputsizes @property def params(self): """Return the bind param dictionary embedded into this compiled object, for those values that are present. .. seealso:: :ref:`faq_sql_expression_string` - includes a usage example for debugging use cases. """ return self.construct_params(_check=False) def _process_parameters_for_postcompile( self, parameters: _MutableCoreSingleExecuteParams, _populate_self: bool = False, ) -> ExpandedState: """handle special post compile parameters. These include: * "expanding" parameters -typically IN tuples that are rendered on a per-parameter basis for an otherwise fixed SQL statement string. * literal_binds compiled with the literal_execute flag. Used for things like SQL Server "TOP N" where the driver does not accommodate N as a bound parameter. """ expanded_parameters = {} new_positiontup: Optional[List[str]] pre_expanded_string = self._pre_expanded_string if pre_expanded_string is None: pre_expanded_string = self.string if self.positional: new_positiontup = [] pre_expanded_positiontup = self._pre_expanded_positiontup if pre_expanded_positiontup is None: pre_expanded_positiontup = self.positiontup else: new_positiontup = pre_expanded_positiontup = None processors = self._bind_processors single_processors = cast( "Mapping[str, _BindProcessorType[Any]]", processors ) tuple_processors = cast( "Mapping[str, Sequence[_BindProcessorType[Any]]]", processors ) new_processors: Dict[str, _BindProcessorType[Any]] = {} replacement_expressions: Dict[str, Any] = {} to_update_sets: Dict[str, Any] = {} # notes: # *unescaped* parameter names in: # self.bind_names, self.binds, self._bind_processors, self.positiontup # # *escaped* parameter names in: # construct_params(), replacement_expressions numeric_positiontup: Optional[List[str]] = None if self.positional and pre_expanded_positiontup is not None: names: Iterable[str] = pre_expanded_positiontup if self._numeric_binds: numeric_positiontup = [] else: names = self.bind_names.values() ebn = self.escaped_bind_names for name in names: escaped_name = ebn.get(name, name) if ebn else name parameter = self.binds[name] if parameter in self.literal_execute_params: if escaped_name not in replacement_expressions: replacement_expressions[escaped_name] = ( self.render_literal_bindparam( parameter, render_literal_value=parameters.pop(escaped_name), ) ) continue if parameter in self.post_compile_params: if escaped_name in replacement_expressions: to_update = to_update_sets[escaped_name] values = None else: # we are removing the parameter from parameters # because it is a list value, which is not expected by # TypeEngine objects that would otherwise be asked to # process it. the single name is being replaced with # individual numbered parameters for each value in the # param. # # note we are also inserting *escaped* parameter names # into the given dictionary. default dialect will # use these param names directly as they will not be # in the escaped_bind_names dictionary. values = parameters.pop(name) leep_res = self._literal_execute_expanding_parameter( escaped_name, parameter, values ) (to_update, replacement_expr) = leep_res to_update_sets[escaped_name] = to_update replacement_expressions[escaped_name] = replacement_expr if not parameter.literal_execute: parameters.update(to_update) if parameter.type._is_tuple_type: assert values is not None new_processors.update( ( "%s_%s_%s" % (name, i, j), tuple_processors[name][j - 1], ) for i, tuple_element in enumerate(values, 1) for j, _ in enumerate(tuple_element, 1) if name in tuple_processors and tuple_processors[name][j - 1] is not None ) else: new_processors.update( (key, single_processors[name]) for key, _ in to_update if name in single_processors ) if numeric_positiontup is not None: numeric_positiontup.extend( name for name, _ in to_update ) elif new_positiontup is not None: # to_update has escaped names, but that's ok since # these are new names, that aren't in the # escaped_bind_names dict. new_positiontup.extend(name for name, _ in to_update) expanded_parameters[name] = [ expand_key for expand_key, _ in to_update ] elif new_positiontup is not None: new_positiontup.append(name) def process_expanding(m): key = m.group(1) expr = replacement_expressions[key] # if POSTCOMPILE included a bind_expression, render that # around each element if m.group(2): tok = m.group(2).split("~~") be_left, be_right = tok[1], tok[3] expr = ", ".join( "%s%s%s" % (be_left, exp, be_right) for exp in expr.split(", ") ) return expr statement = re.sub( self._post_compile_pattern, process_expanding, pre_expanded_string ) if numeric_positiontup is not None: assert new_positiontup is not None param_pos = { key: f"{self._numeric_binds_identifier_char}{num}" for num, key in enumerate( numeric_positiontup, self.next_numeric_pos ) } # Can't use format here since % chars are not escaped. statement = self._pyformat_pattern.sub( lambda m: param_pos[m.group(1)], statement ) new_positiontup.extend(numeric_positiontup) expanded_state = ExpandedState( statement, parameters, new_processors, new_positiontup, expanded_parameters, ) if _populate_self: # this is for the "render_postcompile" flag, which is not # otherwise used internally and is for end-user debugging and # special use cases. self._pre_expanded_string = pre_expanded_string self._pre_expanded_positiontup = pre_expanded_positiontup self.string = expanded_state.statement self.positiontup = ( list(expanded_state.positiontup or ()) if self.positional else None ) self._post_compile_expanded_state = expanded_state return expanded_state @util.preload_module("sqlalchemy.engine.cursor") def _create_result_map(self): """utility method used for unit tests only.""" cursor = util.preloaded.engine_cursor return cursor.CursorResultMetaData._create_description_match_map( self._result_columns ) # assigned by crud.py for insert/update statements _get_bind_name_for_col: _BindNameForColProtocol @util.memoized_property def _within_exec_param_key_getter(self) -> Callable[[Any], str]: getter = self._get_bind_name_for_col return getter @util.memoized_property @util.preload_module("sqlalchemy.engine.result") def _inserted_primary_key_from_lastrowid_getter(self): result = util.preloaded.engine_result param_key_getter = self._within_exec_param_key_getter assert self.compile_state is not None statement = self.compile_state.statement if TYPE_CHECKING: assert isinstance(statement, Insert) table = statement.table getters = [ (operator.methodcaller("get", param_key_getter(col), None), col) for col in table.primary_key ] autoinc_getter = None autoinc_col = table._autoincrement_column if autoinc_col is not None: # apply type post processors to the lastrowid lastrowid_processor = autoinc_col.type._cached_result_processor( self.dialect, None ) autoinc_key = param_key_getter(autoinc_col) # if a bind value is present for the autoincrement column # in the parameters, we need to do the logic dictated by # #7998; honor a non-None user-passed parameter over lastrowid. # previously in the 1.4 series we weren't fetching lastrowid # at all if the key were present in the parameters if autoinc_key in self.binds: def _autoinc_getter(lastrowid, parameters): param_value = parameters.get(autoinc_key, lastrowid) if param_value is not None: # they supplied non-None parameter, use that. # SQLite at least is observed to return the wrong # cursor.lastrowid for INSERT..ON CONFLICT so it # can't be used in all cases return param_value else: # use lastrowid return lastrowid # work around mypy https://github.com/python/mypy/issues/14027 autoinc_getter = _autoinc_getter else: lastrowid_processor = None row_fn = result.result_tuple([col.key for col in table.primary_key]) def get(lastrowid, parameters): """given cursor.lastrowid value and the parameters used for INSERT, return a "row" that represents the primary key, either by using the "lastrowid" or by extracting values from the parameters that were sent along with the INSERT. """ if lastrowid_processor is not None: lastrowid = lastrowid_processor(lastrowid) if lastrowid is None: return row_fn(getter(parameters) for getter, col in getters) else: return row_fn( ( ( autoinc_getter(lastrowid, parameters) if autoinc_getter is not None else lastrowid ) if col is autoinc_col else getter(parameters) ) for getter, col in getters ) return get @util.memoized_property @util.preload_module("sqlalchemy.engine.result") def _inserted_primary_key_from_returning_getter(self): result = util.preloaded.engine_result assert self.compile_state is not None statement = self.compile_state.statement if TYPE_CHECKING: assert isinstance(statement, Insert) param_key_getter = self._within_exec_param_key_getter table = statement.table returning = self.implicit_returning assert returning is not None ret = {col: idx for idx, col in enumerate(returning)} getters = cast( "List[Tuple[Callable[[Any], Any], bool]]", [ ( (operator.itemgetter(ret[col]), True) if col in ret else ( operator.methodcaller( "get", param_key_getter(col), None ), False, ) ) for col in table.primary_key ], ) row_fn = result.result_tuple([col.key for col in table.primary_key]) def get(row, parameters): return row_fn( getter(row) if use_row else getter(parameters) for getter, use_row in getters ) return get def default_from(self) -> str: """Called when a SELECT statement has no froms, and no FROM clause is to be appended. Gives Oracle Database a chance to tack on a ``FROM DUAL`` to the string output. """ return "" def visit_override_binds(self, override_binds, **kw): """SQL compile the nested element of an _OverrideBinds with bindparams swapped out. The _OverrideBinds is not normally expected to be compiled; it is meant to be used when an already cached statement is to be used, the compilation was already performed, and only the bound params should be swapped in at execution time. However, there are test cases that exericise this object, and additionally the ORM subquery loader is known to feed in expressions which include this construct into new queries (discovered in #11173), so it has to do the right thing at compile time as well. """ # get SQL text first sqltext = override_binds.element._compiler_dispatch(self, **kw) # for a test compile that is not for caching, change binds after the # fact. note that we don't try to # swap the bindparam as we compile, because our element may be # elsewhere in the statement already (e.g. a subquery or perhaps a # CTE) and was already visited / compiled. See # test_relationship_criteria.py -> # test_selectinload_local_criteria_subquery for k in override_binds.translate: if k not in self.binds: continue bp = self.binds[k] # so this would work, just change the value of bp in place. # but we dont want to mutate things outside. # bp.value = override_binds.translate[bp.key] # continue # instead, need to replace bp with new_bp or otherwise accommodate # in all internal collections new_bp = bp._with_value( override_binds.translate[bp.key], maintain_key=True, required=False, ) name = self.bind_names[bp] self.binds[k] = self.binds[name] = new_bp self.bind_names[new_bp] = name self.bind_names.pop(bp, None) if bp in self.post_compile_params: self.post_compile_params |= {new_bp} if bp in self.literal_execute_params: self.literal_execute_params |= {new_bp} ckbm_tuple = self._cache_key_bind_match if ckbm_tuple: ckbm, cksm = ckbm_tuple for bp in bp._cloned_set: if bp.key in cksm: cb = cksm[bp.key] ckbm[cb].append(new_bp) return sqltext def visit_grouping(self, grouping, asfrom=False, **kwargs): return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")" def visit_select_statement_grouping(self, grouping, **kwargs): return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")" def visit_label_reference( self, element, within_columns_clause=False, **kwargs ): if self.stack and self.dialect.supports_simple_order_by_label: try: compile_state = cast( "Union[SelectState, CompoundSelectState]", self.stack[-1]["compile_state"], ) except KeyError as ke: raise exc.CompileError( "Can't resolve label reference for ORDER BY / " "GROUP BY / DISTINCT etc." ) from ke ( with_cols, only_froms, only_cols, ) = compile_state._label_resolve_dict if within_columns_clause: resolve_dict = only_froms else: resolve_dict = only_cols # this can be None in the case that a _label_reference() # were subject to a replacement operation, in which case # the replacement of the Label element may have changed # to something else like a ColumnClause expression. order_by_elem = element.element._order_by_label_element if ( order_by_elem is not None and order_by_elem.name in resolve_dict and order_by_elem.shares_lineage( resolve_dict[order_by_elem.name] ) ): kwargs["render_label_as_label"] = ( element.element._order_by_label_element ) return self.process( element.element, within_columns_clause=within_columns_clause, **kwargs, ) def visit_textual_label_reference( self, element, within_columns_clause=False, **kwargs ): if not self.stack: # compiling the element outside of the context of a SELECT return self.process(element._text_clause) try: compile_state = cast( "Union[SelectState, CompoundSelectState]", self.stack[-1]["compile_state"], ) except KeyError as ke: coercions._no_text_coercion( element.element, extra=( "Can't resolve label reference for ORDER BY / " "GROUP BY / DISTINCT etc." ), exc_cls=exc.CompileError, err=ke, ) with_cols, only_froms, only_cols = compile_state._label_resolve_dict try: if within_columns_clause: col = only_froms[element.element] else: col = with_cols[element.element] except KeyError as err: coercions._no_text_coercion( element.element, extra=( "Can't resolve label reference for ORDER BY / " "GROUP BY / DISTINCT etc." ), exc_cls=exc.CompileError, err=err, ) else: kwargs["render_label_as_label"] = col return self.process( col, within_columns_clause=within_columns_clause, **kwargs ) def visit_label( self, label, add_to_result_map=None, within_label_clause=False, within_columns_clause=False, render_label_as_label=None, result_map_targets=(), **kw, ): # only render labels within the columns clause # or ORDER BY clause of a select. dialect-specific compilers # can modify this behavior. render_label_with_as = ( within_columns_clause and not within_label_clause ) render_label_only = render_label_as_label is label if render_label_only or render_label_with_as: if isinstance(label.name, elements._truncated_label): labelname = self._truncated_identifier("colident", label.name) else: labelname = label.name if render_label_with_as: if add_to_result_map is not None: add_to_result_map( labelname, label.name, (label, labelname) + label._alt_names + result_map_targets, label.type, ) return ( label.element._compiler_dispatch( self, within_columns_clause=True, within_label_clause=True, **kw, ) + OPERATORS[operators.as_] + self.preparer.format_label(label, labelname) ) elif render_label_only: return self.preparer.format_label(label, labelname) else: return label.element._compiler_dispatch( self, within_columns_clause=False, **kw ) def _fallback_column_name(self, column): raise exc.CompileError( "Cannot compile Column object until its 'name' is assigned." ) def visit_lambda_element(self, element, **kw): sql_element = element._resolved return self.process(sql_element, **kw) def visit_column( self, column: ColumnClause[Any], add_to_result_map: Optional[_ResultMapAppender] = None, include_table: bool = True, result_map_targets: Tuple[Any, ...] = (), ambiguous_table_name_map: Optional[_AmbiguousTableNameMap] = None, **kwargs: Any, ) -> str: name = orig_name = column.name if name is None: name = self._fallback_column_name(column) is_literal = column.is_literal if not is_literal and isinstance(name, elements._truncated_label): name = self._truncated_identifier("colident", name) if add_to_result_map is not None: targets = (column, name, column.key) + result_map_targets if column._tq_label: targets += (column._tq_label,) add_to_result_map(name, orig_name, targets, column.type) if is_literal: # note we are not currently accommodating for # literal_column(quoted_name('ident', True)) here name = self.escape_literal_column(name) else: name = self.preparer.quote(name) table = column.table if table is None or not include_table or not table.named_with_column: return name else: effective_schema = self.preparer.schema_for_object(table) if effective_schema: schema_prefix = ( self.preparer.quote_schema(effective_schema) + "." ) else: schema_prefix = "" if TYPE_CHECKING: assert isinstance(table, NamedFromClause) tablename = table.name if ( not effective_schema and ambiguous_table_name_map and tablename in ambiguous_table_name_map ): tablename = ambiguous_table_name_map[tablename] if isinstance(tablename, elements._truncated_label): tablename = self._truncated_identifier("alias", tablename) return schema_prefix + self.preparer.quote(tablename) + "." + name def visit_collation(self, element, **kw): return self.preparer.format_collation(element.collation) def visit_fromclause(self, fromclause, **kwargs): return fromclause.name def visit_index(self, index, **kwargs): return index.name def visit_typeclause(self, typeclause, **kw): kw["type_expression"] = typeclause kw["identifier_preparer"] = self.preparer return self.dialect.type_compiler_instance.process( typeclause.type, **kw ) def post_process_text(self, text): if self.preparer._double_percents: text = text.replace("%", "%%") return text def escape_literal_column(self, text): if self.preparer._double_percents: text = text.replace("%", "%%") return text def visit_textclause(self, textclause, add_to_result_map=None, **kw): def do_bindparam(m): name = m.group(1) if name in textclause._bindparams: return self.process(textclause._bindparams[name], **kw) else: return self.bindparam_string(name, **kw) if not self.stack: self.isplaintext = True if add_to_result_map: # text() object is present in the columns clause of a # select(). Add a no-name entry to the result map so that # row[text()] produces a result add_to_result_map(None, None, (textclause,), sqltypes.NULLTYPE) # un-escape any \:params return BIND_PARAMS_ESC.sub( lambda m: m.group(1), BIND_PARAMS.sub( do_bindparam, self.post_process_text(textclause.text) ), ) def visit_textual_select( self, taf, compound_index=None, asfrom=False, **kw ): toplevel = not self.stack entry = self._default_stack_entry if toplevel else self.stack[-1] new_entry: _CompilerStackEntry = { "correlate_froms": set(), "asfrom_froms": set(), "selectable": taf, } self.stack.append(new_entry) if taf._independent_ctes: self._dispatch_independent_ctes(taf, kw) populate_result_map = ( toplevel or ( compound_index == 0 and entry.get("need_result_map_for_compound", False) ) or entry.get("need_result_map_for_nested", False) ) if populate_result_map: self._ordered_columns = self._textual_ordered_columns = ( taf.positional ) # enable looser result column matching when the SQL text links to # Column objects by name only self._loose_column_name_matching = not taf.positional and bool( taf.column_args ) for c in taf.column_args: self.process( c, within_columns_clause=True, add_to_result_map=self._add_to_result_map, ) text = self.process(taf.element, **kw) if self.ctes: nesting_level = len(self.stack) if not toplevel else None text = self._render_cte_clause(nesting_level=nesting_level) + text self.stack.pop(-1) return text def visit_null(self, expr: Null, **kw: Any) -> str: return "NULL" def visit_true(self, expr: True_, **kw: Any) -> str: if self.dialect.supports_native_boolean: return "true" else: return "1" def visit_false(self, expr: False_, **kw: Any) -> str: if self.dialect.supports_native_boolean: return "false" else: return "0" def _generate_delimited_list(self, elements, separator, **kw): return separator.join( s for s in (c._compiler_dispatch(self, **kw) for c in elements) if s ) def _generate_delimited_and_list(self, clauses, **kw): lcc, clauses = elements.BooleanClauseList._process_clauses_for_boolean( operators.and_, elements.True_._singleton, elements.False_._singleton, clauses, ) if lcc == 1: return clauses[0]._compiler_dispatch(self, **kw) else: separator = OPERATORS[operators.and_] return separator.join( s for s in (c._compiler_dispatch(self, **kw) for c in clauses) if s ) def visit_tuple(self, clauselist, **kw): return "(%s)" % self.visit_clauselist(clauselist, **kw) def visit_clauselist(self, clauselist, **kw): sep = clauselist.operator if sep is None: sep = " " else: sep = OPERATORS[clauselist.operator] return self._generate_delimited_list(clauselist.clauses, sep, **kw) def visit_expression_clauselist(self, clauselist, **kw): operator_ = clauselist.operator disp = self._get_operator_dispatch( operator_, "expression_clauselist", None ) if disp: return disp(clauselist, operator_, **kw) try: opstring = OPERATORS[operator_] except KeyError as err: raise exc.UnsupportedCompilationError(self, operator_) from err else: kw["_in_operator_expression"] = True return self._generate_delimited_list( clauselist.clauses, opstring, **kw ) def visit_case(self, clause, **kwargs): x = "CASE " if clause.value is not None: x += clause.value._compiler_dispatch(self, **kwargs) + " " for cond, result in clause.whens: x += ( "WHEN " + cond._compiler_dispatch(self, **kwargs) + " THEN " + result._compiler_dispatch(self, **kwargs) + " " ) if clause.else_ is not None: x += ( "ELSE " + clause.else_._compiler_dispatch(self, **kwargs) + " " ) x += "END" return x def visit_type_coerce(self, type_coerce, **kw): return type_coerce.typed_expression._compiler_dispatch(self, **kw) def visit_cast(self, cast, **kwargs): type_clause = cast.typeclause._compiler_dispatch(self, **kwargs) match = re.match("(.*)( COLLATE .*)", type_clause) return "CAST(%s AS %s)%s" % ( cast.clause._compiler_dispatch(self, **kwargs), match.group(1) if match else type_clause, match.group(2) if match else "", ) def _format_frame_clause(self, range_, **kw): return "%s AND %s" % ( ( "UNBOUNDED PRECEDING" if range_[0] is elements.RANGE_UNBOUNDED else ( "CURRENT ROW" if range_[0] is elements.RANGE_CURRENT else ( "%s PRECEDING" % ( self.process( elements.literal(abs(range_[0])), **kw ), ) if range_[0] < 0 else "%s FOLLOWING" % (self.process(elements.literal(range_[0]), **kw),) ) ) ), ( "UNBOUNDED FOLLOWING" if range_[1] is elements.RANGE_UNBOUNDED else ( "CURRENT ROW" if range_[1] is elements.RANGE_CURRENT else ( "%s PRECEDING" % ( self.process( elements.literal(abs(range_[1])), **kw ), ) if range_[1] < 0 else "%s FOLLOWING" % (self.process(elements.literal(range_[1]), **kw),) ) ) ), ) def visit_over(self, over, **kwargs): text = over.element._compiler_dispatch(self, **kwargs) if over.range_ is not None: range_ = "RANGE BETWEEN %s" % self._format_frame_clause( over.range_, **kwargs ) elif over.rows is not None: range_ = "ROWS BETWEEN %s" % self._format_frame_clause( over.rows, **kwargs ) elif over.groups is not None: range_ = "GROUPS BETWEEN %s" % self._format_frame_clause( over.groups, **kwargs ) else: range_ = None return "%s OVER (%s)" % ( text, " ".join( [ "%s BY %s" % (word, clause._compiler_dispatch(self, **kwargs)) for word, clause in ( ("PARTITION", over.partition_by), ("ORDER", over.order_by), ) if clause is not None and len(clause) ] + ([range_] if range_ else []) ), ) def visit_withingroup(self, withingroup, **kwargs): return "%s WITHIN GROUP (ORDER BY %s)" % ( withingroup.element._compiler_dispatch(self, **kwargs), withingroup.order_by._compiler_dispatch(self, **kwargs), ) def visit_funcfilter(self, funcfilter, **kwargs): return "%s FILTER (WHERE %s)" % ( funcfilter.func._compiler_dispatch(self, **kwargs), funcfilter.criterion._compiler_dispatch(self, **kwargs), ) def visit_extract(self, extract, **kwargs): field = self.extract_map.get(extract.field, extract.field) return "EXTRACT(%s FROM %s)" % ( field, extract.expr._compiler_dispatch(self, **kwargs), ) def visit_scalar_function_column(self, element, **kw): compiled_fn = self.visit_function(element.fn, **kw) compiled_col = self.visit_column(element, **kw) return "(%s).%s" % (compiled_fn, compiled_col) def visit_function( self, func: Function[Any], add_to_result_map: Optional[_ResultMapAppender] = None, **kwargs: Any, ) -> str: if add_to_result_map is not None: add_to_result_map(func.name, func.name, (func.name,), func.type) disp = getattr(self, "visit_%s_func" % func.name.lower(), None) text: str if disp: text = disp(func, **kwargs) else: name = FUNCTIONS.get(func._deannotate().__class__, None) if name: if func._has_args: name += "%(expr)s" else: name = func.name name = ( self.preparer.quote(name) if self.preparer._requires_quotes_illegal_chars(name) or isinstance(name, elements.quoted_name) else name ) name = name + "%(expr)s" text = ".".join( [ ( self.preparer.quote(tok) if self.preparer._requires_quotes_illegal_chars(tok) or isinstance(name, elements.quoted_name) else tok ) for tok in func.packagenames ] + [name] ) % {"expr": self.function_argspec(func, **kwargs)} if func._with_ordinality: text += " WITH ORDINALITY" return text def visit_next_value_func(self, next_value, **kw): return self.visit_sequence(next_value.sequence) def visit_sequence(self, sequence, **kw): raise NotImplementedError( "Dialect '%s' does not support sequence increments." % self.dialect.name ) def function_argspec(self, func: Function[Any], **kwargs: Any) -> str: return func.clause_expr._compiler_dispatch(self, **kwargs) def visit_compound_select( self, cs, asfrom=False, compound_index=None, **kwargs ): toplevel = not self.stack compile_state = cs._compile_state_factory(cs, self, **kwargs) if toplevel and not self.compile_state: self.compile_state = compile_state compound_stmt = compile_state.statement entry = self._default_stack_entry if toplevel else self.stack[-1] need_result_map = toplevel or ( not compound_index and entry.get("need_result_map_for_compound", False) ) # indicates there is already a CompoundSelect in play if compound_index == 0: entry["select_0"] = cs self.stack.append( { "correlate_froms": entry["correlate_froms"], "asfrom_froms": entry["asfrom_froms"], "selectable": cs, "compile_state": compile_state, "need_result_map_for_compound": need_result_map, } ) if compound_stmt._independent_ctes: self._dispatch_independent_ctes(compound_stmt, kwargs) keyword = self.compound_keywords[cs.keyword] text = (" " + keyword + " ").join( ( c._compiler_dispatch( self, asfrom=asfrom, compound_index=i, **kwargs ) for i, c in enumerate(cs.selects) ) ) kwargs["include_table"] = False text += self.group_by_clause(cs, **dict(asfrom=asfrom, **kwargs)) text += self.order_by_clause(cs, **kwargs) if cs._has_row_limiting_clause: text += self._row_limit_clause(cs, **kwargs) if self.ctes: nesting_level = len(self.stack) if not toplevel else None text = ( self._render_cte_clause( nesting_level=nesting_level, include_following_stack=True, ) + text ) self.stack.pop(-1) return text def _row_limit_clause(self, cs, **kwargs): if cs._fetch_clause is not None: return self.fetch_clause(cs, **kwargs) else: return self.limit_clause(cs, **kwargs) def _get_operator_dispatch(self, operator_, qualifier1, qualifier2): attrname = "visit_%s_%s%s" % ( operator_.__name__, qualifier1, "_" + qualifier2 if qualifier2 else "", ) return getattr(self, attrname, None) def visit_unary( self, unary, add_to_result_map=None, result_map_targets=(), **kw ): if add_to_result_map is not None: result_map_targets += (unary,) kw["add_to_result_map"] = add_to_result_map kw["result_map_targets"] = result_map_targets if unary.operator: if unary.modifier: raise exc.CompileError( "Unary expression does not support operator " "and modifier simultaneously" ) disp = self._get_operator_dispatch( unary.operator, "unary", "operator" ) if disp: return disp(unary, unary.operator, **kw) else: return self._generate_generic_unary_operator( unary, OPERATORS[unary.operator], **kw ) elif unary.modifier: disp = self._get_operator_dispatch( unary.modifier, "unary", "modifier" ) if disp: return disp(unary, unary.modifier, **kw) else: return self._generate_generic_unary_modifier( unary, OPERATORS[unary.modifier], **kw ) else: raise exc.CompileError( "Unary expression has no operator or modifier" ) def visit_truediv_binary(self, binary, operator, **kw): if self.dialect.div_is_floordiv: return ( self.process(binary.left, **kw) + " / " # TODO: would need a fast cast again here, # unless we want to use an implicit cast like "+ 0.0" + self.process( elements.Cast( binary.right, ( binary.right.type if binary.right.type._type_affinity is sqltypes.Numeric else sqltypes.Numeric() ), ), **kw, ) ) else: return ( self.process(binary.left, **kw) + " / " + self.process(binary.right, **kw) ) def visit_floordiv_binary(self, binary, operator, **kw): if ( self.dialect.div_is_floordiv and binary.right.type._type_affinity is sqltypes.Integer ): return ( self.process(binary.left, **kw) + " / " + self.process(binary.right, **kw) ) else: return "FLOOR(%s)" % ( self.process(binary.left, **kw) + " / " + self.process(binary.right, **kw) ) def visit_is_true_unary_operator(self, element, operator, **kw): if ( element._is_implicitly_boolean or self.dialect.supports_native_boolean ): return self.process(element.element, **kw) else: return "%s = 1" % self.process(element.element, **kw) def visit_is_false_unary_operator(self, element, operator, **kw): if ( element._is_implicitly_boolean or self.dialect.supports_native_boolean ): return "NOT %s" % self.process(element.element, **kw) else: return "%s = 0" % self.process(element.element, **kw) def visit_not_match_op_binary(self, binary, operator, **kw): return "NOT %s" % self.visit_binary( binary, override_operator=operators.match_op ) def visit_not_in_op_binary(self, binary, operator, **kw): # The brackets are required in the NOT IN operation because the empty # case is handled using the form "(col NOT IN (null) OR 1 = 1)". # The presence of the OR makes the brackets required. return "(%s)" % self._generate_generic_binary( binary, OPERATORS[operator], **kw ) def visit_empty_set_op_expr(self, type_, expand_op, **kw): if expand_op is operators.not_in_op: if len(type_) > 1: return "(%s)) OR (1 = 1" % ( ", ".join("NULL" for element in type_) ) else: return "NULL) OR (1 = 1" elif expand_op is operators.in_op: if len(type_) > 1: return "(%s)) AND (1 != 1" % ( ", ".join("NULL" for element in type_) ) else: return "NULL) AND (1 != 1" else: return self.visit_empty_set_expr(type_) def visit_empty_set_expr(self, element_types, **kw): raise NotImplementedError( "Dialect '%s' does not support empty set expression." % self.dialect.name ) def _literal_execute_expanding_parameter_literal_binds( self, parameter, values, bind_expression_template=None ): typ_dialect_impl = parameter.type._unwrapped_dialect_impl(self.dialect) if not values: # empty IN expression. note we don't need to use # bind_expression_template here because there are no # expressions to render. if typ_dialect_impl._is_tuple_type: replacement_expression = ( "VALUES " if self.dialect.tuple_in_values else "" ) + self.visit_empty_set_op_expr( parameter.type.types, parameter.expand_op ) else: replacement_expression = self.visit_empty_set_op_expr( [parameter.type], parameter.expand_op ) elif typ_dialect_impl._is_tuple_type or ( typ_dialect_impl._isnull and isinstance(values[0], collections_abc.Sequence) and not isinstance(values[0], (str, bytes)) ): if typ_dialect_impl._has_bind_expression: raise NotImplementedError( "bind_expression() on TupleType not supported with " "literal_binds" ) replacement_expression = ( "VALUES " if self.dialect.tuple_in_values else "" ) + ", ".join( "(%s)" % ( ", ".join( self.render_literal_value(value, param_type) for value, param_type in zip( tuple_element, parameter.type.types ) ) ) for i, tuple_element in enumerate(values) ) else: if bind_expression_template: post_compile_pattern = self._post_compile_pattern m = post_compile_pattern.search(bind_expression_template) assert m and m.group( 2 ), "unexpected format for expanding parameter" tok = m.group(2).split("~~") be_left, be_right = tok[1], tok[3] replacement_expression = ", ".join( "%s%s%s" % ( be_left, self.render_literal_value(value, parameter.type), be_right, ) for value in values ) else: replacement_expression = ", ".join( self.render_literal_value(value, parameter.type) for value in values ) return (), replacement_expression def _literal_execute_expanding_parameter(self, name, parameter, values): if parameter.literal_execute: return self._literal_execute_expanding_parameter_literal_binds( parameter, values ) dialect = self.dialect typ_dialect_impl = parameter.type._unwrapped_dialect_impl(dialect) if self._numeric_binds: bind_template = self.compilation_bindtemplate else: bind_template = self.bindtemplate if ( self.dialect._bind_typing_render_casts and typ_dialect_impl.render_bind_cast ): def _render_bindtemplate(name): return self.render_bind_cast( parameter.type, typ_dialect_impl, bind_template % {"name": name}, ) else: def _render_bindtemplate(name): return bind_template % {"name": name} if not values: to_update = [] if typ_dialect_impl._is_tuple_type: replacement_expression = self.visit_empty_set_op_expr( parameter.type.types, parameter.expand_op ) else: replacement_expression = self.visit_empty_set_op_expr( [parameter.type], parameter.expand_op ) elif typ_dialect_impl._is_tuple_type or ( typ_dialect_impl._isnull and isinstance(values[0], collections_abc.Sequence) and not isinstance(values[0], (str, bytes)) ): assert not typ_dialect_impl._is_array to_update = [ ("%s_%s_%s" % (name, i, j), value) for i, tuple_element in enumerate(values, 1) for j, value in enumerate(tuple_element, 1) ] replacement_expression = ( "VALUES " if dialect.tuple_in_values else "" ) + ", ".join( "(%s)" % ( ", ".join( _render_bindtemplate( to_update[i * len(tuple_element) + j][0] ) for j, value in enumerate(tuple_element) ) ) for i, tuple_element in enumerate(values) ) else: to_update = [ ("%s_%s" % (name, i), value) for i, value in enumerate(values, 1) ] replacement_expression = ", ".join( _render_bindtemplate(key) for key, value in to_update ) return to_update, replacement_expression def visit_binary( self, binary, override_operator=None, eager_grouping=False, from_linter=None, lateral_from_linter=None, **kw, ): if from_linter and operators.is_comparison(binary.operator): if lateral_from_linter is not None: enclosing_lateral = kw["enclosing_lateral"] lateral_from_linter.edges.update( itertools.product( _de_clone( binary.left._from_objects + [enclosing_lateral] ), _de_clone( binary.right._from_objects + [enclosing_lateral] ), ) ) else: from_linter.edges.update( itertools.product( _de_clone(binary.left._from_objects), _de_clone(binary.right._from_objects), ) ) # don't allow "? = ?" to render if ( self.ansi_bind_rules and isinstance(binary.left, elements.BindParameter) and isinstance(binary.right, elements.BindParameter) ): kw["literal_execute"] = True operator_ = override_operator or binary.operator disp = self._get_operator_dispatch(operator_, "binary", None) if disp: return disp(binary, operator_, **kw) else: try: opstring = OPERATORS[operator_] except KeyError as err: raise exc.UnsupportedCompilationError(self, operator_) from err else: return self._generate_generic_binary( binary, opstring, from_linter=from_linter, lateral_from_linter=lateral_from_linter, **kw, ) def visit_function_as_comparison_op_binary(self, element, operator, **kw): return self.process(element.sql_function, **kw) def visit_mod_binary(self, binary, operator, **kw): if self.preparer._double_percents: return ( self.process(binary.left, **kw) + " %% " + self.process(binary.right, **kw) ) else: return ( self.process(binary.left, **kw) + " % " + self.process(binary.right, **kw) ) def visit_custom_op_binary(self, element, operator, **kw): kw["eager_grouping"] = operator.eager_grouping return self._generate_generic_binary( element, " " + self.escape_literal_column(operator.opstring) + " ", **kw, ) def visit_custom_op_unary_operator(self, element, operator, **kw): return self._generate_generic_unary_operator( element, self.escape_literal_column(operator.opstring) + " ", **kw ) def visit_custom_op_unary_modifier(self, element, operator, **kw): return self._generate_generic_unary_modifier( element, " " + self.escape_literal_column(operator.opstring), **kw ) def _generate_generic_binary( self, binary: BinaryExpression[Any], opstring: str, eager_grouping: bool = False, **kw: Any, ) -> str: _in_operator_expression = kw.get("_in_operator_expression", False) kw["_in_operator_expression"] = True kw["_binary_op"] = binary.operator text = ( binary.left._compiler_dispatch( self, eager_grouping=eager_grouping, **kw ) + opstring + binary.right._compiler_dispatch( self, eager_grouping=eager_grouping, **kw ) ) if _in_operator_expression and eager_grouping: text = "(%s)" % text return text def _generate_generic_unary_operator(self, unary, opstring, **kw): return opstring + unary.element._compiler_dispatch(self, **kw) def _generate_generic_unary_modifier(self, unary, opstring, **kw): return unary.element._compiler_dispatch(self, **kw) + opstring @util.memoized_property def _like_percent_literal(self): return elements.literal_column("'%'", type_=sqltypes.STRINGTYPE) def visit_ilike_case_insensitive_operand(self, element, **kw): return f"lower({element.element._compiler_dispatch(self, **kw)})" def visit_contains_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.right = percent.concat(binary.right).concat(percent) return self.visit_like_op_binary(binary, operator, **kw) def visit_not_contains_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.right = percent.concat(binary.right).concat(percent) return self.visit_not_like_op_binary(binary, operator, **kw) def visit_icontains_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.left = ilike_case_insensitive(binary.left) binary.right = percent.concat( ilike_case_insensitive(binary.right) ).concat(percent) return self.visit_ilike_op_binary(binary, operator, **kw) def visit_not_icontains_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.left = ilike_case_insensitive(binary.left) binary.right = percent.concat( ilike_case_insensitive(binary.right) ).concat(percent) return self.visit_not_ilike_op_binary(binary, operator, **kw) def visit_startswith_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.right = percent._rconcat(binary.right) return self.visit_like_op_binary(binary, operator, **kw) def visit_not_startswith_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.right = percent._rconcat(binary.right) return self.visit_not_like_op_binary(binary, operator, **kw) def visit_istartswith_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.left = ilike_case_insensitive(binary.left) binary.right = percent._rconcat(ilike_case_insensitive(binary.right)) return self.visit_ilike_op_binary(binary, operator, **kw) def visit_not_istartswith_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.left = ilike_case_insensitive(binary.left) binary.right = percent._rconcat(ilike_case_insensitive(binary.right)) return self.visit_not_ilike_op_binary(binary, operator, **kw) def visit_endswith_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.right = percent.concat(binary.right) return self.visit_like_op_binary(binary, operator, **kw) def visit_not_endswith_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.right = percent.concat(binary.right) return self.visit_not_like_op_binary(binary, operator, **kw) def visit_iendswith_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.left = ilike_case_insensitive(binary.left) binary.right = percent.concat(ilike_case_insensitive(binary.right)) return self.visit_ilike_op_binary(binary, operator, **kw) def visit_not_iendswith_op_binary(self, binary, operator, **kw): binary = binary._clone() percent = self._like_percent_literal binary.left = ilike_case_insensitive(binary.left) binary.right = percent.concat(ilike_case_insensitive(binary.right)) return self.visit_not_ilike_op_binary(binary, operator, **kw) def visit_like_op_binary(self, binary, operator, **kw): escape = binary.modifiers.get("escape", None) return "%s LIKE %s" % ( binary.left._compiler_dispatch(self, **kw), binary.right._compiler_dispatch(self, **kw), ) + ( " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE) if escape is not None else "" ) def visit_not_like_op_binary(self, binary, operator, **kw): escape = binary.modifiers.get("escape", None) return "%s NOT LIKE %s" % ( binary.left._compiler_dispatch(self, **kw), binary.right._compiler_dispatch(self, **kw), ) + ( " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE) if escape is not None else "" ) def visit_ilike_op_binary(self, binary, operator, **kw): if operator is operators.ilike_op: binary = binary._clone() binary.left = ilike_case_insensitive(binary.left) binary.right = ilike_case_insensitive(binary.right) # else we assume ilower() has been applied return self.visit_like_op_binary(binary, operator, **kw) def visit_not_ilike_op_binary(self, binary, operator, **kw): if operator is operators.not_ilike_op: binary = binary._clone() binary.left = ilike_case_insensitive(binary.left) binary.right = ilike_case_insensitive(binary.right) # else we assume ilower() has been applied return self.visit_not_like_op_binary(binary, operator, **kw) def visit_between_op_binary(self, binary, operator, **kw): symmetric = binary.modifiers.get("symmetric", False) return self._generate_generic_binary( binary, " BETWEEN SYMMETRIC " if symmetric else " BETWEEN ", **kw ) def visit_not_between_op_binary(self, binary, operator, **kw): symmetric = binary.modifiers.get("symmetric", False) return self._generate_generic_binary( binary, " NOT BETWEEN SYMMETRIC " if symmetric else " NOT BETWEEN ", **kw, ) def visit_regexp_match_op_binary( self, binary: BinaryExpression[Any], operator: Any, **kw: Any ) -> str: raise exc.CompileError( "%s dialect does not support regular expressions" % self.dialect.name ) def visit_not_regexp_match_op_binary( self, binary: BinaryExpression[Any], operator: Any, **kw: Any ) -> str: raise exc.CompileError( "%s dialect does not support regular expressions" % self.dialect.name ) def visit_regexp_replace_op_binary( self, binary: BinaryExpression[Any], operator: Any, **kw: Any ) -> str: raise exc.CompileError( "%s dialect does not support regular expression replacements" % self.dialect.name ) def visit_bindparam( self, bindparam, within_columns_clause=False, literal_binds=False, skip_bind_expression=False, literal_execute=False, render_postcompile=False, **kwargs, ): if not skip_bind_expression: impl = bindparam.type.dialect_impl(self.dialect) if impl._has_bind_expression: bind_expression = impl.bind_expression(bindparam) wrapped = self.process( bind_expression, skip_bind_expression=True, within_columns_clause=within_columns_clause, literal_binds=literal_binds and not bindparam.expanding, literal_execute=literal_execute, render_postcompile=render_postcompile, **kwargs, ) if bindparam.expanding: # for postcompile w/ expanding, move the "wrapped" part # of this into the inside m = re.match( r"^(.*)\(__\[POSTCOMPILE_(\S+?)\]\)(.*)$", wrapped ) assert m, "unexpected format for expanding parameter" wrapped = "(__[POSTCOMPILE_%s~~%s~~REPL~~%s~~])" % ( m.group(2), m.group(1), m.group(3), ) if literal_binds: ret = self.render_literal_bindparam( bindparam, within_columns_clause=True, bind_expression_template=wrapped, **kwargs, ) return "(%s)" % ret return wrapped if not literal_binds: literal_execute = ( literal_execute or bindparam.literal_execute or (within_columns_clause and self.ansi_bind_rules) ) post_compile = literal_execute or bindparam.expanding else: post_compile = False if literal_binds: ret = self.render_literal_bindparam( bindparam, within_columns_clause=True, **kwargs ) if bindparam.expanding: ret = "(%s)" % ret return ret name = self._truncate_bindparam(bindparam) if name in self.binds: existing = self.binds[name] if existing is not bindparam: if ( (existing.unique or bindparam.unique) and not existing.proxy_set.intersection( bindparam.proxy_set ) and not existing._cloned_set.intersection( bindparam._cloned_set ) ): raise exc.CompileError( "Bind parameter '%s' conflicts with " "unique bind parameter of the same name" % name ) elif existing.expanding != bindparam.expanding: raise exc.CompileError( "Can't reuse bound parameter name '%s' in both " "'expanding' (e.g. within an IN expression) and " "non-expanding contexts. If this parameter is to " "receive a list/array value, set 'expanding=True' on " "it for expressions that aren't IN, otherwise use " "a different parameter name." % (name,) ) elif existing._is_crud or bindparam._is_crud: if existing._is_crud and bindparam._is_crud: # TODO: this condition is not well understood. # see tests in test/sql/test_update.py raise exc.CompileError( "Encountered unsupported case when compiling an " "INSERT or UPDATE statement. If this is a " "multi-table " "UPDATE statement, please provide string-named " "arguments to the " "values() method with distinct names; support for " "multi-table UPDATE statements that " "target multiple tables for UPDATE is very " "limited", ) else: raise exc.CompileError( f"bindparam() name '{bindparam.key}' is reserved " "for automatic usage in the VALUES or SET " "clause of this " "insert/update statement. Please use a " "name other than column name when using " "bindparam() " "with insert() or update() (for example, " f"'b_{bindparam.key}')." ) self.binds[bindparam.key] = self.binds[name] = bindparam # if we are given a cache key that we're going to match against, # relate the bindparam here to one that is most likely present # in the "extracted params" portion of the cache key. this is used # to set up a positional mapping that is used to determine the # correct parameters for a subsequent use of this compiled with # a different set of parameter values. here, we accommodate for # parameters that may have been cloned both before and after the cache # key was been generated. ckbm_tuple = self._cache_key_bind_match if ckbm_tuple: ckbm, cksm = ckbm_tuple for bp in bindparam._cloned_set: if bp.key in cksm: cb = cksm[bp.key] ckbm[cb].append(bindparam) if bindparam.isoutparam: self.has_out_parameters = True if post_compile: if render_postcompile: self._render_postcompile = True if literal_execute: self.literal_execute_params |= {bindparam} else: self.post_compile_params |= {bindparam} ret = self.bindparam_string( name, post_compile=post_compile, expanding=bindparam.expanding, bindparam_type=bindparam.type, **kwargs, ) if bindparam.expanding: ret = "(%s)" % ret return ret def render_bind_cast(self, type_, dbapi_type, sqltext): raise NotImplementedError() def render_literal_bindparam( self, bindparam, render_literal_value=NO_ARG, bind_expression_template=None, **kw, ): if render_literal_value is not NO_ARG: value = render_literal_value else: if bindparam.value is None and bindparam.callable is None: op = kw.get("_binary_op", None) if op and op not in (operators.is_, operators.is_not): util.warn_limited( "Bound parameter '%s' rendering literal NULL in a SQL " "expression; comparisons to NULL should not use " "operators outside of 'is' or 'is not'", (bindparam.key,), ) return self.process(sqltypes.NULLTYPE, **kw) value = bindparam.effective_value if bindparam.expanding: leep = self._literal_execute_expanding_parameter_literal_binds to_update, replacement_expr = leep( bindparam, value, bind_expression_template=bind_expression_template, ) return replacement_expr else: return self.render_literal_value(value, bindparam.type) def render_literal_value( self, value: Any, type_: sqltypes.TypeEngine[Any] ) -> str: """Render the value of a bind parameter as a quoted literal. This is used for statement sections that do not accept bind parameters on the target driver/database. This should be implemented by subclasses using the quoting services of the DBAPI. """ if value is None and not type_.should_evaluate_none: # issue #10535 - handle NULL in the compiler without placing # this onto each type, except for "evaluate None" types # (e.g. JSON) return self.process(elements.Null._instance()) processor = type_._cached_literal_processor(self.dialect) if processor: try: return processor(value) except Exception as e: raise exc.CompileError( f"Could not render literal value " f'"{sql_util._repr_single_value(value)}" ' f"with datatype " f"{type_}; see parent stack trace for " "more detail." ) from e else: raise exc.CompileError( f"No literal value renderer is available for literal value " f'"{sql_util._repr_single_value(value)}" ' f"with datatype {type_}" ) def _truncate_bindparam(self, bindparam): if bindparam in self.bind_names: return self.bind_names[bindparam] bind_name = bindparam.key if isinstance(bind_name, elements._truncated_label): bind_name = self._truncated_identifier("bindparam", bind_name) # add to bind_names for translation self.bind_names[bindparam] = bind_name return bind_name def _truncated_identifier( self, ident_class: str, name: _truncated_label ) -> str: if (ident_class, name) in self.truncated_names: return self.truncated_names[(ident_class, name)] anonname = name.apply_map(self.anon_map) if len(anonname) > self.label_length - 6: counter = self._truncated_counters.get(ident_class, 1) truncname = ( anonname[0 : max(self.label_length - 6, 0)] + "_" + hex(counter)[2:] ) self._truncated_counters[ident_class] = counter + 1 else: truncname = anonname self.truncated_names[(ident_class, name)] = truncname return truncname def _anonymize(self, name: str) -> str: return name % self.anon_map def bindparam_string( self, name: str, post_compile: bool = False, expanding: bool = False, escaped_from: Optional[str] = None, bindparam_type: Optional[TypeEngine[Any]] = None, accumulate_bind_names: Optional[Set[str]] = None, visited_bindparam: Optional[List[str]] = None, **kw: Any, ) -> str: # TODO: accumulate_bind_names is passed by crud.py to gather # names on a per-value basis, visited_bindparam is passed by # visit_insert() to collect all parameters in the statement. # see if this gathering can be simplified somehow if accumulate_bind_names is not None: accumulate_bind_names.add(name) if visited_bindparam is not None: visited_bindparam.append(name) if not escaped_from: if self._bind_translate_re.search(name): # not quite the translate use case as we want to # also get a quick boolean if we even found # unusual characters in the name new_name = self._bind_translate_re.sub( lambda m: self._bind_translate_chars[m.group(0)], name, ) escaped_from = name name = new_name if escaped_from: self.escaped_bind_names = self.escaped_bind_names.union( {escaped_from: name} ) if post_compile: ret = "__[POSTCOMPILE_%s]" % name if expanding: # for expanding, bound parameters or literal values will be # rendered per item return ret # otherwise, for non-expanding "literal execute", apply # bind casts as determined by the datatype if bindparam_type is not None: type_impl = bindparam_type._unwrapped_dialect_impl( self.dialect ) if type_impl.render_literal_cast: ret = self.render_bind_cast(bindparam_type, type_impl, ret) return ret elif self.state is CompilerState.COMPILING: ret = self.compilation_bindtemplate % {"name": name} else: ret = self.bindtemplate % {"name": name} if ( bindparam_type is not None and self.dialect._bind_typing_render_casts ): type_impl = bindparam_type._unwrapped_dialect_impl(self.dialect) if type_impl.render_bind_cast: ret = self.render_bind_cast(bindparam_type, type_impl, ret) return ret def _dispatch_independent_ctes(self, stmt, kw): local_kw = kw.copy() local_kw.pop("cte_opts", None) for cte, opt in zip( stmt._independent_ctes, stmt._independent_ctes_opts ): cte._compiler_dispatch(self, cte_opts=opt, **local_kw) def visit_cte( self, cte: CTE, asfrom: bool = False, ashint: bool = False, fromhints: Optional[_FromHintsType] = None, visiting_cte: Optional[CTE] = None, from_linter: Optional[FromLinter] = None, cte_opts: selectable._CTEOpts = selectable._CTEOpts(False), **kwargs: Any, ) -> Optional[str]: self_ctes = self._init_cte_state() assert self_ctes is self.ctes kwargs["visiting_cte"] = cte cte_name = cte.name if isinstance(cte_name, elements._truncated_label): cte_name = self._truncated_identifier("alias", cte_name) is_new_cte = True embedded_in_current_named_cte = False _reference_cte = cte._get_reference_cte() nesting = cte.nesting or cte_opts.nesting # check for CTE already encountered if _reference_cte in self.level_name_by_cte: cte_level, _, existing_cte_opts = self.level_name_by_cte[ _reference_cte ] assert _ == cte_name cte_level_name = (cte_level, cte_name) existing_cte = self.ctes_by_level_name[cte_level_name] # check if we are receiving it here with a specific # "nest_here" location; if so, move it to this location if cte_opts.nesting: if existing_cte_opts.nesting: raise exc.CompileError( "CTE is stated as 'nest_here' in " "more than one location" ) old_level_name = (cte_level, cte_name) cte_level = len(self.stack) if nesting else 1 cte_level_name = new_level_name = (cte_level, cte_name) del self.ctes_by_level_name[old_level_name] self.ctes_by_level_name[new_level_name] = existing_cte self.level_name_by_cte[_reference_cte] = new_level_name + ( cte_opts, ) else: cte_level = len(self.stack) if nesting else 1 cte_level_name = (cte_level, cte_name) if cte_level_name in self.ctes_by_level_name: existing_cte = self.ctes_by_level_name[cte_level_name] else: existing_cte = None if existing_cte is not None: embedded_in_current_named_cte = visiting_cte is existing_cte # we've generated a same-named CTE that we are enclosed in, # or this is the same CTE. just return the name. if cte is existing_cte._restates or cte is existing_cte: is_new_cte = False elif existing_cte is cte._restates: # we've generated a same-named CTE that is # enclosed in us - we take precedence, so # discard the text for the "inner". del self_ctes[existing_cte] existing_cte_reference_cte = existing_cte._get_reference_cte() assert existing_cte_reference_cte is _reference_cte assert existing_cte_reference_cte is existing_cte del self.level_name_by_cte[existing_cte_reference_cte] else: if ( # if the two CTEs have the same hash, which we expect # here means that one/both is an annotated of the other (hash(cte) == hash(existing_cte)) # or... or ( ( # if they are clones, i.e. they came from the ORM # or some other visit method cte._is_clone_of is not None or existing_cte._is_clone_of is not None ) # and are deep-copy identical and cte.compare(existing_cte) ) ): # then consider these two CTEs the same is_new_cte = False else: # otherwise these are two CTEs that either will render # differently, or were indicated separately by the user, # with the same name raise exc.CompileError( "Multiple, unrelated CTEs found with " "the same name: %r" % cte_name ) if not asfrom and not is_new_cte: return None if cte._cte_alias is not None: pre_alias_cte = cte._cte_alias cte_pre_alias_name = cte._cte_alias.name if isinstance(cte_pre_alias_name, elements._truncated_label): cte_pre_alias_name = self._truncated_identifier( "alias", cte_pre_alias_name ) else: pre_alias_cte = cte cte_pre_alias_name = None if is_new_cte: self.ctes_by_level_name[cte_level_name] = cte self.level_name_by_cte[_reference_cte] = cte_level_name + ( cte_opts, ) if pre_alias_cte not in self.ctes: self.visit_cte(pre_alias_cte, **kwargs) if not cte_pre_alias_name and cte not in self_ctes: if cte.recursive: self.ctes_recursive = True text = self.preparer.format_alias(cte, cte_name) if cte.recursive or cte.element.name_cte_columns: col_source = cte.element # TODO: can we get at the .columns_plus_names collection # that is already (or will be?) generated for the SELECT # rather than calling twice? recur_cols = [ # TODO: proxy_name is not technically safe, # see test_cte-> # test_with_recursive_no_name_currently_buggy. not # clear what should be done with such a case fallback_label_name or proxy_name for ( _, proxy_name, fallback_label_name, c, repeated, ) in (col_source._generate_columns_plus_names(True)) if not repeated ] text += "(%s)" % ( ", ".join( self.preparer.format_label_name( ident, anon_map=self.anon_map ) for ident in recur_cols ) ) assert kwargs.get("subquery", False) is False if not self.stack: # toplevel, this is a stringify of the # cte directly. just compile the inner # the way alias() does. return cte.element._compiler_dispatch( self, asfrom=asfrom, **kwargs ) else: prefixes = self._generate_prefixes( cte, cte._prefixes, **kwargs ) inner = cte.element._compiler_dispatch( self, asfrom=True, **kwargs ) text += " AS %s\n(%s)" % (prefixes, inner) if cte._suffixes: text += " " + self._generate_prefixes( cte, cte._suffixes, **kwargs ) self_ctes[cte] = text if asfrom: if from_linter: from_linter.froms[cte._de_clone()] = cte_name if not is_new_cte and embedded_in_current_named_cte: return self.preparer.format_alias(cte, cte_name) if cte_pre_alias_name: text = self.preparer.format_alias(cte, cte_pre_alias_name) if self.preparer._requires_quotes(cte_name): cte_name = self.preparer.quote(cte_name) text += self.get_render_as_alias_suffix(cte_name) return text # type: ignore[no-any-return] else: return self.preparer.format_alias(cte, cte_name) return None def visit_table_valued_alias(self, element, **kw): if element.joins_implicitly: kw["from_linter"] = None if element._is_lateral: return self.visit_lateral(element, **kw) else: return self.visit_alias(element, **kw) def visit_table_valued_column(self, element, **kw): return self.visit_column(element, **kw) def visit_alias( self, alias, asfrom=False, ashint=False, iscrud=False, fromhints=None, subquery=False, lateral=False, enclosing_alias=None, from_linter=None, **kwargs, ): if lateral: if "enclosing_lateral" not in kwargs: # if lateral is set and enclosing_lateral is not # present, we assume we are being called directly # from visit_lateral() and we need to set enclosing_lateral. assert alias._is_lateral kwargs["enclosing_lateral"] = alias # for lateral objects, we track a second from_linter that is... # lateral! to the level above us. if ( from_linter and "lateral_from_linter" not in kwargs and "enclosing_lateral" in kwargs ): kwargs["lateral_from_linter"] = from_linter if enclosing_alias is not None and enclosing_alias.element is alias: inner = alias.element._compiler_dispatch( self, asfrom=asfrom, ashint=ashint, iscrud=iscrud, fromhints=fromhints, lateral=lateral, enclosing_alias=alias, **kwargs, ) if subquery and (asfrom or lateral): inner = "(%s)" % (inner,) return inner else: kwargs["enclosing_alias"] = alias if asfrom or ashint: if isinstance(alias.name, elements._truncated_label): alias_name = self._truncated_identifier("alias", alias.name) else: alias_name = alias.name if ashint: return self.preparer.format_alias(alias, alias_name) elif asfrom: if from_linter: from_linter.froms[alias._de_clone()] = alias_name inner = alias.element._compiler_dispatch( self, asfrom=True, lateral=lateral, **kwargs ) if subquery: inner = "(%s)" % (inner,) ret = inner + self.get_render_as_alias_suffix( self.preparer.format_alias(alias, alias_name) ) if alias._supports_derived_columns and alias._render_derived: ret += "(%s)" % ( ", ".join( "%s%s" % ( self.preparer.quote(col.name), ( " %s" % self.dialect.type_compiler_instance.process( col.type, **kwargs ) if alias._render_derived_w_types else "" ), ) for col in alias.c ) ) if fromhints and alias in fromhints: ret = self.format_from_hint_text( ret, alias, fromhints[alias], iscrud ) return ret else: # note we cancel the "subquery" flag here as well return alias.element._compiler_dispatch( self, lateral=lateral, **kwargs ) def visit_subquery(self, subquery, **kw): kw["subquery"] = True return self.visit_alias(subquery, **kw) def visit_lateral(self, lateral_, **kw): kw["lateral"] = True return "LATERAL %s" % self.visit_alias(lateral_, **kw) def visit_tablesample(self, tablesample, asfrom=False, **kw): text = "%s TABLESAMPLE %s" % ( self.visit_alias(tablesample, asfrom=True, **kw), tablesample._get_method()._compiler_dispatch(self, **kw), ) if tablesample.seed is not None: text += " REPEATABLE (%s)" % ( tablesample.seed._compiler_dispatch(self, **kw) ) return text def _render_values(self, element, **kw): kw.setdefault("literal_binds", element.literal_binds) tuples = ", ".join( self.process( elements.Tuple( types=element._column_types, *elem ).self_group(), **kw, ) for chunk in element._data for elem in chunk ) return f"VALUES {tuples}" def visit_values( self, element, asfrom=False, from_linter=None, visiting_cte=None, **kw ): if element._independent_ctes: self._dispatch_independent_ctes(element, kw) v = self._render_values(element, **kw) if element._unnamed: name = None elif isinstance(element.name, elements._truncated_label): name = self._truncated_identifier("values", element.name) else: name = element.name if element._is_lateral: lateral = "LATERAL " else: lateral = "" if asfrom: if from_linter: from_linter.froms[element._de_clone()] = ( name if name is not None else "(unnamed VALUES element)" ) if visiting_cte is not None and visiting_cte.element is element: if element._is_lateral: raise exc.CompileError( "Can't use a LATERAL VALUES expression inside of a CTE" ) elif name: kw["include_table"] = False v = "%s(%s)%s (%s)" % ( lateral, v, self.get_render_as_alias_suffix(self.preparer.quote(name)), ( ", ".join( c._compiler_dispatch(self, **kw) for c in element.columns ) ), ) else: v = "%s(%s)" % (lateral, v) return v def visit_scalar_values(self, element, **kw): return f"({self._render_values(element, **kw)})" def get_render_as_alias_suffix(self, alias_name_text): return " AS " + alias_name_text def _add_to_result_map( self, keyname: str, name: str, objects: Tuple[Any, ...], type_: TypeEngine[Any], ) -> None: # note objects must be non-empty for cursor.py to handle the # collection properly assert objects if keyname is None or keyname == "*": self._ordered_columns = False self._ad_hoc_textual = True if type_._is_tuple_type: raise exc.CompileError( "Most backends don't support SELECTing " "from a tuple() object. If this is an ORM query, " "consider using the Bundle object." ) self._result_columns.append( ResultColumnsEntry(keyname, name, objects, type_) ) def _label_returning_column( self, stmt, column, populate_result_map, column_clause_args=None, **kw ): """Render a column with necessary labels inside of a RETURNING clause. This method is provided for individual dialects in place of calling the _label_select_column method directly, so that the two use cases of RETURNING vs. SELECT can be disambiguated going forward. .. versionadded:: 1.4.21 """ return self._label_select_column( None, column, populate_result_map, False, {} if column_clause_args is None else column_clause_args, **kw, ) def _label_select_column( self, select, column, populate_result_map, asfrom, column_clause_args, name=None, proxy_name=None, fallback_label_name=None, within_columns_clause=True, column_is_repeated=False, need_column_expressions=False, include_table=True, ): """produce labeled columns present in a select().""" impl = column.type.dialect_impl(self.dialect) if impl._has_column_expression and ( need_column_expressions or populate_result_map ): col_expr = impl.column_expression(column) else: col_expr = column if populate_result_map: # pass an "add_to_result_map" callable into the compilation # of embedded columns. this collects information about the # column as it will be fetched in the result and is coordinated # with cursor.description when the query is executed. add_to_result_map = self._add_to_result_map # if the SELECT statement told us this column is a repeat, # wrap the callable with one that prevents the addition of the # targets if column_is_repeated: _add_to_result_map = add_to_result_map def add_to_result_map(keyname, name, objects, type_): _add_to_result_map(keyname, name, (keyname,), type_) # if we redefined col_expr for type expressions, wrap the # callable with one that adds the original column to the targets elif col_expr is not column: _add_to_result_map = add_to_result_map def add_to_result_map(keyname, name, objects, type_): _add_to_result_map( keyname, name, (column,) + objects, type_ ) else: add_to_result_map = None # this method is used by some of the dialects for RETURNING, # which has different inputs. _label_returning_column was added # as the better target for this now however for 1.4 we will keep # _label_select_column directly compatible with this use case. # these assertions right now set up the current expected inputs assert within_columns_clause, ( "_label_select_column is only relevant within " "the columns clause of a SELECT or RETURNING" ) if isinstance(column, elements.Label): if col_expr is not column: result_expr = _CompileLabel( col_expr, column.name, alt_names=(column.element,) ) else: result_expr = col_expr elif name: # here, _columns_plus_names has determined there's an explicit # label name we need to use. this is the default for # tablenames_plus_columnnames as well as when columns are being # deduplicated on name assert ( proxy_name is not None ), "proxy_name is required if 'name' is passed" result_expr = _CompileLabel( col_expr, name, alt_names=( proxy_name, # this is a hack to allow legacy result column lookups # to work as they did before; this goes away in 2.0. # TODO: this only seems to be tested indirectly # via test/orm/test_deprecations.py. should be a # resultset test for this column._tq_label, ), ) else: # determine here whether this column should be rendered in # a labelled context or not, as we were given no required label # name from the caller. Here we apply heuristics based on the kind # of SQL expression involved. if col_expr is not column: # type-specific expression wrapping the given column, # so we render a label render_with_label = True elif isinstance(column, elements.ColumnClause): # table-bound column, we render its name as a label if we are # inside of a subquery only render_with_label = ( asfrom and not column.is_literal and column.table is not None ) elif isinstance(column, elements.TextClause): render_with_label = False elif isinstance(column, elements.UnaryExpression): # unary expression. notes added as of #12681 # # By convention, the visit_unary() method # itself does not add an entry to the result map, and relies # upon either the inner expression creating a result map # entry, or if not, by creating a label here that produces # the result map entry. Where that happens is based on whether # or not the element immediately inside the unary is a # NamedColumn subclass or not. # # Now, this also impacts how the SELECT is written; if # we decide to generate a label here, we get the usual # "~(x+y) AS anon_1" thing in the columns clause. If we # don't, we don't get an AS at all, we get like # "~table.column". # # But here is the important thing as of modernish (like 1.4) # versions of SQLAlchemy - **whether or not the AS