Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# sql/compiler.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9"""Base SQL and DDL compiler implementations.
11Classes provided include:
13:class:`.compiler.SQLCompiler` - renders SQL
14strings
16:class:`.compiler.DDLCompiler` - renders DDL
17(data definition language) strings
19:class:`.compiler.GenericTypeCompiler` - renders
20type specification strings.
22To generate user-defined SQL strings, see
23:doc:`/ext/compiler`.
25"""
26from __future__ import annotations
28import collections
29import collections.abc as collections_abc
30import contextlib
31from enum import IntEnum
32import functools
33import itertools
34import operator
35import re
36from time import perf_counter
37import typing
38from typing import Any
39from typing import Callable
40from typing import cast
41from typing import ClassVar
42from typing import Dict
43from typing import FrozenSet
44from typing import Iterable
45from typing import Iterator
46from typing import List
47from typing import Mapping
48from typing import MutableMapping
49from typing import NamedTuple
50from typing import NoReturn
51from typing import Optional
52from typing import Pattern
53from typing import Protocol
54from typing import Sequence
55from typing import Set
56from typing import Tuple
57from typing import Type
58from typing import TYPE_CHECKING
59from typing import TypedDict
60from typing import Union
62from . import base
63from . import coercions
64from . import crud
65from . import elements
66from . import functions
67from . import operators
68from . import roles
69from . import schema
70from . import selectable
71from . import sqltypes
72from . import util as sql_util
73from ._typing import is_column_element
74from ._typing import is_dml
75from .base import _de_clone
76from .base import _from_objects
77from .base import _NONE_NAME
78from .base import _SentinelDefaultCharacterization
79from .base import NO_ARG
80from .elements import quoted_name
81from .sqltypes import TupleType
82from .visitors import prefix_anon_map
83from .. import exc
84from .. import util
85from ..util import FastIntFlag
86from ..util.typing import Literal
87from ..util.typing import Self
88from ..util.typing import TupleAny
89from ..util.typing import Unpack
91if typing.TYPE_CHECKING:
92 from .annotation import _AnnotationDict
93 from .base import _AmbiguousTableNameMap
94 from .base import CompileState
95 from .base import Executable
96 from .cache_key import CacheKey
97 from .ddl import ExecutableDDLElement
98 from .dml import Delete
99 from .dml import Insert
100 from .dml import Update
101 from .dml import UpdateBase
102 from .dml import UpdateDMLState
103 from .dml import ValuesBase
104 from .elements import _truncated_label
105 from .elements import BinaryExpression
106 from .elements import BindParameter
107 from .elements import ClauseElement
108 from .elements import ColumnClause
109 from .elements import ColumnElement
110 from .elements import False_
111 from .elements import Label
112 from .elements import Null
113 from .elements import True_
114 from .functions import Function
115 from .schema import Column
116 from .schema import Constraint
117 from .schema import ForeignKeyConstraint
118 from .schema import Index
119 from .schema import PrimaryKeyConstraint
120 from .schema import Table
121 from .schema import UniqueConstraint
122 from .selectable import _ColumnsClauseElement
123 from .selectable import AliasedReturnsRows
124 from .selectable import CompoundSelectState
125 from .selectable import CTE
126 from .selectable import FromClause
127 from .selectable import NamedFromClause
128 from .selectable import ReturnsRows
129 from .selectable import Select
130 from .selectable import SelectState
131 from .type_api import _BindProcessorType
132 from .type_api import TypeDecorator
133 from .type_api import TypeEngine
134 from .type_api import UserDefinedType
135 from .visitors import Visitable
136 from ..engine.cursor import CursorResultMetaData
137 from ..engine.interfaces import _CoreSingleExecuteParams
138 from ..engine.interfaces import _DBAPIAnyExecuteParams
139 from ..engine.interfaces import _DBAPIMultiExecuteParams
140 from ..engine.interfaces import _DBAPISingleExecuteParams
141 from ..engine.interfaces import _ExecuteOptions
142 from ..engine.interfaces import _GenericSetInputSizesType
143 from ..engine.interfaces import _MutableCoreSingleExecuteParams
144 from ..engine.interfaces import Dialect
145 from ..engine.interfaces import SchemaTranslateMapType
148_FromHintsType = Dict["FromClause", str]
150RESERVED_WORDS = {
151 "all",
152 "analyse",
153 "analyze",
154 "and",
155 "any",
156 "array",
157 "as",
158 "asc",
159 "asymmetric",
160 "authorization",
161 "between",
162 "binary",
163 "both",
164 "case",
165 "cast",
166 "check",
167 "collate",
168 "column",
169 "constraint",
170 "create",
171 "cross",
172 "current_date",
173 "current_role",
174 "current_time",
175 "current_timestamp",
176 "current_user",
177 "default",
178 "deferrable",
179 "desc",
180 "distinct",
181 "do",
182 "else",
183 "end",
184 "except",
185 "false",
186 "for",
187 "foreign",
188 "freeze",
189 "from",
190 "full",
191 "grant",
192 "group",
193 "having",
194 "ilike",
195 "in",
196 "initially",
197 "inner",
198 "intersect",
199 "into",
200 "is",
201 "isnull",
202 "join",
203 "leading",
204 "left",
205 "like",
206 "limit",
207 "localtime",
208 "localtimestamp",
209 "natural",
210 "new",
211 "not",
212 "notnull",
213 "null",
214 "off",
215 "offset",
216 "old",
217 "on",
218 "only",
219 "or",
220 "order",
221 "outer",
222 "overlaps",
223 "placing",
224 "primary",
225 "references",
226 "right",
227 "select",
228 "session_user",
229 "set",
230 "similar",
231 "some",
232 "symmetric",
233 "table",
234 "then",
235 "to",
236 "trailing",
237 "true",
238 "union",
239 "unique",
240 "user",
241 "using",
242 "verbose",
243 "when",
244 "where",
245}
247LEGAL_CHARACTERS = re.compile(r"^[A-Z0-9_$]+$", re.I)
248LEGAL_CHARACTERS_PLUS_SPACE = re.compile(r"^[A-Z0-9_ $]+$", re.I)
249ILLEGAL_INITIAL_CHARACTERS = {str(x) for x in range(0, 10)}.union(["$"])
251FK_ON_DELETE = re.compile(
252 r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I
253)
254FK_ON_UPDATE = re.compile(
255 r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I
256)
257FK_INITIALLY = re.compile(r"^(?:DEFERRED|IMMEDIATE)$", re.I)
258BIND_PARAMS = re.compile(r"(?<![:\w\$\x5c]):([\w\$]+)(?![:\w\$])", re.UNICODE)
259BIND_PARAMS_ESC = re.compile(r"\x5c(:[\w\$]*)(?![:\w\$])", re.UNICODE)
261_pyformat_template = "%%(%(name)s)s"
262BIND_TEMPLATES = {
263 "pyformat": _pyformat_template,
264 "qmark": "?",
265 "format": "%%s",
266 "numeric": ":[_POSITION]",
267 "numeric_dollar": "$[_POSITION]",
268 "named": ":%(name)s",
269}
272OPERATORS = {
273 # binary
274 operators.and_: " AND ",
275 operators.or_: " OR ",
276 operators.add: " + ",
277 operators.mul: " * ",
278 operators.sub: " - ",
279 operators.mod: " % ",
280 operators.neg: "-",
281 operators.lt: " < ",
282 operators.le: " <= ",
283 operators.ne: " != ",
284 operators.gt: " > ",
285 operators.ge: " >= ",
286 operators.eq: " = ",
287 operators.is_distinct_from: " IS DISTINCT FROM ",
288 operators.is_not_distinct_from: " IS NOT DISTINCT FROM ",
289 operators.concat_op: " || ",
290 operators.match_op: " MATCH ",
291 operators.not_match_op: " NOT MATCH ",
292 operators.in_op: " IN ",
293 operators.not_in_op: " NOT IN ",
294 operators.comma_op: ", ",
295 operators.from_: " FROM ",
296 operators.as_: " AS ",
297 operators.is_: " IS ",
298 operators.is_not: " IS NOT ",
299 operators.collate: " COLLATE ",
300 # unary
301 operators.exists: "EXISTS ",
302 operators.distinct_op: "DISTINCT ",
303 operators.inv: "NOT ",
304 operators.any_op: "ANY ",
305 operators.all_op: "ALL ",
306 # modifiers
307 operators.desc_op: " DESC",
308 operators.asc_op: " ASC",
309 operators.nulls_first_op: " NULLS FIRST",
310 operators.nulls_last_op: " NULLS LAST",
311 # bitwise
312 operators.bitwise_xor_op: " ^ ",
313 operators.bitwise_or_op: " | ",
314 operators.bitwise_and_op: " & ",
315 operators.bitwise_not_op: "~",
316 operators.bitwise_lshift_op: " << ",
317 operators.bitwise_rshift_op: " >> ",
318}
320FUNCTIONS: Dict[Type[Function[Any]], str] = {
321 functions.coalesce: "coalesce",
322 functions.current_date: "CURRENT_DATE",
323 functions.current_time: "CURRENT_TIME",
324 functions.current_timestamp: "CURRENT_TIMESTAMP",
325 functions.current_user: "CURRENT_USER",
326 functions.localtime: "LOCALTIME",
327 functions.localtimestamp: "LOCALTIMESTAMP",
328 functions.random: "random",
329 functions.sysdate: "sysdate",
330 functions.session_user: "SESSION_USER",
331 functions.user: "USER",
332 functions.cube: "CUBE",
333 functions.rollup: "ROLLUP",
334 functions.grouping_sets: "GROUPING SETS",
335}
338EXTRACT_MAP = {
339 "month": "month",
340 "day": "day",
341 "year": "year",
342 "second": "second",
343 "hour": "hour",
344 "doy": "doy",
345 "minute": "minute",
346 "quarter": "quarter",
347 "dow": "dow",
348 "week": "week",
349 "epoch": "epoch",
350 "milliseconds": "milliseconds",
351 "microseconds": "microseconds",
352 "timezone_hour": "timezone_hour",
353 "timezone_minute": "timezone_minute",
354}
356COMPOUND_KEYWORDS = {
357 selectable._CompoundSelectKeyword.UNION: "UNION",
358 selectable._CompoundSelectKeyword.UNION_ALL: "UNION ALL",
359 selectable._CompoundSelectKeyword.EXCEPT: "EXCEPT",
360 selectable._CompoundSelectKeyword.EXCEPT_ALL: "EXCEPT ALL",
361 selectable._CompoundSelectKeyword.INTERSECT: "INTERSECT",
362 selectable._CompoundSelectKeyword.INTERSECT_ALL: "INTERSECT ALL",
363}
366class ResultColumnsEntry(NamedTuple):
367 """Tracks a column expression that is expected to be represented
368 in the result rows for this statement.
370 This normally refers to the columns clause of a SELECT statement
371 but may also refer to a RETURNING clause, as well as for dialect-specific
372 emulations.
374 """
376 keyname: str
377 """string name that's expected in cursor.description"""
379 name: str
380 """column name, may be labeled"""
382 objects: Tuple[Any, ...]
383 """sequence of objects that should be able to locate this column
384 in a RowMapping. This is typically string names and aliases
385 as well as Column objects.
387 """
389 type: TypeEngine[Any]
390 """Datatype to be associated with this column. This is where
391 the "result processing" logic directly links the compiled statement
392 to the rows that come back from the cursor.
394 """
397class _ResultMapAppender(Protocol):
398 def __call__(
399 self,
400 keyname: str,
401 name: str,
402 objects: Sequence[Any],
403 type_: TypeEngine[Any],
404 ) -> None: ...
407# integer indexes into ResultColumnsEntry used by cursor.py.
408# some profiling showed integer access faster than named tuple
409RM_RENDERED_NAME: Literal[0] = 0
410RM_NAME: Literal[1] = 1
411RM_OBJECTS: Literal[2] = 2
412RM_TYPE: Literal[3] = 3
415class _BaseCompilerStackEntry(TypedDict):
416 asfrom_froms: Set[FromClause]
417 correlate_froms: Set[FromClause]
418 selectable: ReturnsRows
421class _CompilerStackEntry(_BaseCompilerStackEntry, total=False):
422 compile_state: CompileState
423 need_result_map_for_nested: bool
424 need_result_map_for_compound: bool
425 select_0: ReturnsRows
426 insert_from_select: Select[Unpack[TupleAny]]
429class ExpandedState(NamedTuple):
430 """represents state to use when producing "expanded" and
431 "post compile" bound parameters for a statement.
433 "expanded" parameters are parameters that are generated at
434 statement execution time to suit a number of parameters passed, the most
435 prominent example being the individual elements inside of an IN expression.
437 "post compile" parameters are parameters where the SQL literal value
438 will be rendered into the SQL statement at execution time, rather than
439 being passed as separate parameters to the driver.
441 To create an :class:`.ExpandedState` instance, use the
442 :meth:`.SQLCompiler.construct_expanded_state` method on any
443 :class:`.SQLCompiler` instance.
445 """
447 statement: str
448 """String SQL statement with parameters fully expanded"""
450 parameters: _CoreSingleExecuteParams
451 """Parameter dictionary with parameters fully expanded.
453 For a statement that uses named parameters, this dictionary will map
454 exactly to the names in the statement. For a statement that uses
455 positional parameters, the :attr:`.ExpandedState.positional_parameters`
456 will yield a tuple with the positional parameter set.
458 """
460 processors: Mapping[str, _BindProcessorType[Any]]
461 """mapping of bound value processors"""
463 positiontup: Optional[Sequence[str]]
464 """Sequence of string names indicating the order of positional
465 parameters"""
467 parameter_expansion: Mapping[str, List[str]]
468 """Mapping representing the intermediary link from original parameter
469 name to list of "expanded" parameter names, for those parameters that
470 were expanded."""
472 @property
473 def positional_parameters(self) -> Tuple[Any, ...]:
474 """Tuple of positional parameters, for statements that were compiled
475 using a positional paramstyle.
477 """
478 if self.positiontup is None:
479 raise exc.InvalidRequestError(
480 "statement does not use a positional paramstyle"
481 )
482 return tuple(self.parameters[key] for key in self.positiontup)
484 @property
485 def additional_parameters(self) -> _CoreSingleExecuteParams:
486 """synonym for :attr:`.ExpandedState.parameters`."""
487 return self.parameters
490class _InsertManyValues(NamedTuple):
491 """represents state to use for executing an "insertmanyvalues" statement.
493 The primary consumers of this object are the
494 :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and
495 :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods.
497 .. versionadded:: 2.0
499 """
501 is_default_expr: bool
502 """if True, the statement is of the form
503 ``INSERT INTO TABLE DEFAULT VALUES``, and can't be rewritten as a "batch"
505 """
507 single_values_expr: str
508 """The rendered "values" clause of the INSERT statement.
510 This is typically the parenthesized section e.g. "(?, ?, ?)" or similar.
511 The insertmanyvalues logic uses this string as a search and replace
512 target.
514 """
516 insert_crud_params: List[crud._CrudParamElementStr]
517 """List of Column / bind names etc. used while rewriting the statement"""
519 num_positional_params_counted: int
520 """the number of bound parameters in a single-row statement.
522 This count may be larger or smaller than the actual number of columns
523 targeted in the INSERT, as it accommodates for SQL expressions
524 in the values list that may have zero or more parameters embedded
525 within them.
527 This count is part of what's used to organize rewritten parameter lists
528 when batching.
530 """
532 sort_by_parameter_order: bool = False
533 """if the deterministic_returnined_order parameter were used on the
534 insert.
536 All of the attributes following this will only be used if this is True.
538 """
540 includes_upsert_behaviors: bool = False
541 """if True, we have to accommodate for upsert behaviors.
543 This will in some cases downgrade "insertmanyvalues" that requests
544 deterministic ordering.
546 """
548 sentinel_columns: Optional[Sequence[Column[Any]]] = None
549 """List of sentinel columns that were located.
551 This list is only here if the INSERT asked for
552 sort_by_parameter_order=True,
553 and dialect-appropriate sentinel columns were located.
555 .. versionadded:: 2.0.10
557 """
559 num_sentinel_columns: int = 0
560 """how many sentinel columns are in the above list, if any.
562 This is the same as
563 ``len(sentinel_columns) if sentinel_columns is not None else 0``
565 """
567 sentinel_param_keys: Optional[Sequence[str]] = None
568 """parameter str keys in each param dictionary / tuple
569 that would link to the client side "sentinel" values for that row, which
570 we can use to match up parameter sets to result rows.
572 This is only present if sentinel_columns is present and the INSERT
573 statement actually refers to client side values for these sentinel
574 columns.
576 .. versionadded:: 2.0.10
578 .. versionchanged:: 2.0.29 - the sequence is now string dictionary keys
579 only, used against the "compiled parameteters" collection before
580 the parameters were converted by bound parameter processors
582 """
584 implicit_sentinel: bool = False
585 """if True, we have exactly one sentinel column and it uses a server side
586 value, currently has to generate an incrementing integer value.
588 The dialect in question would have asserted that it supports receiving
589 these values back and sorting on that value as a means of guaranteeing
590 correlation with the incoming parameter list.
592 .. versionadded:: 2.0.10
594 """
596 embed_values_counter: bool = False
597 """Whether to embed an incrementing integer counter in each parameter
598 set within the VALUES clause as parameters are batched over.
600 This is only used for a specific INSERT..SELECT..VALUES..RETURNING syntax
601 where a subquery is used to produce value tuples. Current support
602 includes PostgreSQL, Microsoft SQL Server.
604 .. versionadded:: 2.0.10
606 """
609class _InsertManyValuesBatch(NamedTuple):
610 """represents an individual batch SQL statement for insertmanyvalues.
612 This is passed through the
613 :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and
614 :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods out
615 to the :class:`.Connection` within the
616 :meth:`.Connection._exec_insertmany_context` method.
618 .. versionadded:: 2.0.10
620 """
622 replaced_statement: str
623 replaced_parameters: _DBAPIAnyExecuteParams
624 processed_setinputsizes: Optional[_GenericSetInputSizesType]
625 batch: Sequence[_DBAPISingleExecuteParams]
626 sentinel_values: Sequence[Tuple[Any, ...]]
627 current_batch_size: int
628 batchnum: int
629 total_batches: int
630 rows_sorted: bool
631 is_downgraded: bool
634class InsertmanyvaluesSentinelOpts(FastIntFlag):
635 """bitflag enum indicating styles of PK defaults
636 which can work as implicit sentinel columns
638 """
640 NOT_SUPPORTED = 1
641 AUTOINCREMENT = 2
642 IDENTITY = 4
643 SEQUENCE = 8
645 ANY_AUTOINCREMENT = AUTOINCREMENT | IDENTITY | SEQUENCE
646 _SUPPORTED_OR_NOT = NOT_SUPPORTED | ANY_AUTOINCREMENT
648 USE_INSERT_FROM_SELECT = 16
649 RENDER_SELECT_COL_CASTS = 64
652class CompilerState(IntEnum):
653 COMPILING = 0
654 """statement is present, compilation phase in progress"""
656 STRING_APPLIED = 1
657 """statement is present, string form of the statement has been applied.
659 Additional processors by subclasses may still be pending.
661 """
663 NO_STATEMENT = 2
664 """compiler does not have a statement to compile, is used
665 for method access"""
668class Linting(IntEnum):
669 """represent preferences for the 'SQL linting' feature.
671 this feature currently includes support for flagging cartesian products
672 in SQL statements.
674 """
676 NO_LINTING = 0
677 "Disable all linting."
679 COLLECT_CARTESIAN_PRODUCTS = 1
680 """Collect data on FROMs and cartesian products and gather into
681 'self.from_linter'"""
683 WARN_LINTING = 2
684 "Emit warnings for linters that find problems"
686 FROM_LINTING = COLLECT_CARTESIAN_PRODUCTS | WARN_LINTING
687 """Warn for cartesian products; combines COLLECT_CARTESIAN_PRODUCTS
688 and WARN_LINTING"""
691NO_LINTING, COLLECT_CARTESIAN_PRODUCTS, WARN_LINTING, FROM_LINTING = tuple(
692 Linting
693)
696class FromLinter(collections.namedtuple("FromLinter", ["froms", "edges"])):
697 """represents current state for the "cartesian product" detection
698 feature."""
700 def lint(self, start=None):
701 froms = self.froms
702 if not froms:
703 return None, None
705 edges = set(self.edges)
706 the_rest = set(froms)
708 if start is not None:
709 start_with = start
710 the_rest.remove(start_with)
711 else:
712 start_with = the_rest.pop()
714 stack = collections.deque([start_with])
716 while stack and the_rest:
717 node = stack.popleft()
718 the_rest.discard(node)
720 # comparison of nodes in edges here is based on hash equality, as
721 # there are "annotated" elements that match the non-annotated ones.
722 # to remove the need for in-python hash() calls, use native
723 # containment routines (e.g. "node in edge", "edge.index(node)")
724 to_remove = {edge for edge in edges if node in edge}
726 # appendleft the node in each edge that is not
727 # the one that matched.
728 stack.extendleft(edge[not edge.index(node)] for edge in to_remove)
729 edges.difference_update(to_remove)
731 # FROMS left over? boom
732 if the_rest:
733 return the_rest, start_with
734 else:
735 return None, None
737 def warn(self, stmt_type="SELECT"):
738 the_rest, start_with = self.lint()
740 # FROMS left over? boom
741 if the_rest:
742 froms = the_rest
743 if froms:
744 template = (
745 "{stmt_type} statement has a cartesian product between "
746 "FROM element(s) {froms} and "
747 'FROM element "{start}". Apply join condition(s) '
748 "between each element to resolve."
749 )
750 froms_str = ", ".join(
751 f'"{self.froms[from_]}"' for from_ in froms
752 )
753 message = template.format(
754 stmt_type=stmt_type,
755 froms=froms_str,
756 start=self.froms[start_with],
757 )
759 util.warn(message)
762class Compiled:
763 """Represent a compiled SQL or DDL expression.
765 The ``__str__`` method of the ``Compiled`` object should produce
766 the actual text of the statement. ``Compiled`` objects are
767 specific to their underlying database dialect, and also may
768 or may not be specific to the columns referenced within a
769 particular set of bind parameters. In no case should the
770 ``Compiled`` object be dependent on the actual values of those
771 bind parameters, even though it may reference those values as
772 defaults.
773 """
775 statement: Optional[ClauseElement] = None
776 "The statement to compile."
777 string: str = ""
778 "The string representation of the ``statement``"
780 state: CompilerState
781 """description of the compiler's state"""
783 is_sql = False
784 is_ddl = False
786 _cached_metadata: Optional[CursorResultMetaData] = None
788 _result_columns: Optional[List[ResultColumnsEntry]] = None
790 schema_translate_map: Optional[SchemaTranslateMapType] = None
792 execution_options: _ExecuteOptions = util.EMPTY_DICT
793 """
794 Execution options propagated from the statement. In some cases,
795 sub-elements of the statement can modify these.
796 """
798 preparer: IdentifierPreparer
800 _annotations: _AnnotationDict = util.EMPTY_DICT
802 compile_state: Optional[CompileState] = None
803 """Optional :class:`.CompileState` object that maintains additional
804 state used by the compiler.
806 Major executable objects such as :class:`_expression.Insert`,
807 :class:`_expression.Update`, :class:`_expression.Delete`,
808 :class:`_expression.Select` will generate this
809 state when compiled in order to calculate additional information about the
810 object. For the top level object that is to be executed, the state can be
811 stored here where it can also have applicability towards result set
812 processing.
814 .. versionadded:: 1.4
816 """
818 dml_compile_state: Optional[CompileState] = None
819 """Optional :class:`.CompileState` assigned at the same point that
820 .isinsert, .isupdate, or .isdelete is assigned.
822 This will normally be the same object as .compile_state, with the
823 exception of cases like the :class:`.ORMFromStatementCompileState`
824 object.
826 .. versionadded:: 1.4.40
828 """
830 cache_key: Optional[CacheKey] = None
831 """The :class:`.CacheKey` that was generated ahead of creating this
832 :class:`.Compiled` object.
834 This is used for routines that need access to the original
835 :class:`.CacheKey` instance generated when the :class:`.Compiled`
836 instance was first cached, typically in order to reconcile
837 the original list of :class:`.BindParameter` objects with a
838 per-statement list that's generated on each call.
840 """
842 _gen_time: float
843 """Generation time of this :class:`.Compiled`, used for reporting
844 cache stats."""
846 def __init__(
847 self,
848 dialect: Dialect,
849 statement: Optional[ClauseElement],
850 schema_translate_map: Optional[SchemaTranslateMapType] = None,
851 render_schema_translate: bool = False,
852 compile_kwargs: Mapping[str, Any] = util.immutabledict(),
853 ):
854 """Construct a new :class:`.Compiled` object.
856 :param dialect: :class:`.Dialect` to compile against.
858 :param statement: :class:`_expression.ClauseElement` to be compiled.
860 :param schema_translate_map: dictionary of schema names to be
861 translated when forming the resultant SQL
863 .. seealso::
865 :ref:`schema_translating`
867 :param compile_kwargs: additional kwargs that will be
868 passed to the initial call to :meth:`.Compiled.process`.
871 """
872 self.dialect = dialect
873 self.preparer = self.dialect.identifier_preparer
874 if schema_translate_map:
875 self.schema_translate_map = schema_translate_map
876 self.preparer = self.preparer._with_schema_translate(
877 schema_translate_map
878 )
880 if statement is not None:
881 self.state = CompilerState.COMPILING
882 self.statement = statement
883 self.can_execute = statement.supports_execution
884 self._annotations = statement._annotations
885 if self.can_execute:
886 if TYPE_CHECKING:
887 assert isinstance(statement, Executable)
888 self.execution_options = statement._execution_options
889 self.string = self.process(self.statement, **compile_kwargs)
891 if render_schema_translate:
892 assert schema_translate_map is not None
893 self.string = self.preparer._render_schema_translates(
894 self.string, schema_translate_map
895 )
897 self.state = CompilerState.STRING_APPLIED
898 else:
899 self.state = CompilerState.NO_STATEMENT
901 self._gen_time = perf_counter()
903 def __init_subclass__(cls) -> None:
904 cls._init_compiler_cls()
905 return super().__init_subclass__()
907 @classmethod
908 def _init_compiler_cls(cls):
909 pass
911 def _execute_on_connection(
912 self, connection, distilled_params, execution_options
913 ):
914 if self.can_execute:
915 return connection._execute_compiled(
916 self, distilled_params, execution_options
917 )
918 else:
919 raise exc.ObjectNotExecutableError(self.statement)
921 def visit_unsupported_compilation(self, element, err, **kw):
922 raise exc.UnsupportedCompilationError(self, type(element)) from err
924 @property
925 def sql_compiler(self) -> SQLCompiler:
926 """Return a Compiled that is capable of processing SQL expressions.
928 If this compiler is one, it would likely just return 'self'.
930 """
932 raise NotImplementedError()
934 def process(self, obj: Visitable, **kwargs: Any) -> str:
935 return obj._compiler_dispatch(self, **kwargs)
937 def __str__(self) -> str:
938 """Return the string text of the generated SQL or DDL."""
940 if self.state is CompilerState.STRING_APPLIED:
941 return self.string
942 else:
943 return ""
945 def construct_params(
946 self,
947 params: Optional[_CoreSingleExecuteParams] = None,
948 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
949 escape_names: bool = True,
950 ) -> Optional[_MutableCoreSingleExecuteParams]:
951 """Return the bind params for this compiled object.
953 :param params: a dict of string/object pairs whose values will
954 override bind values compiled in to the
955 statement.
956 """
958 raise NotImplementedError()
960 @property
961 def params(self):
962 """Return the bind params for this compiled object."""
963 return self.construct_params()
966class TypeCompiler(util.EnsureKWArg):
967 """Produces DDL specification for TypeEngine objects."""
969 ensure_kwarg = r"visit_\w+"
971 def __init__(self, dialect: Dialect):
972 self.dialect = dialect
974 def process(self, type_: TypeEngine[Any], **kw: Any) -> str:
975 if (
976 type_._variant_mapping
977 and self.dialect.name in type_._variant_mapping
978 ):
979 type_ = type_._variant_mapping[self.dialect.name]
980 return type_._compiler_dispatch(self, **kw)
982 def visit_unsupported_compilation(
983 self, element: Any, err: Exception, **kw: Any
984 ) -> NoReturn:
985 raise exc.UnsupportedCompilationError(self, element) from err
988# this was a Visitable, but to allow accurate detection of
989# column elements this is actually a column element
990class _CompileLabel(
991 roles.BinaryElementRole[Any], elements.CompilerColumnElement
992):
993 """lightweight label object which acts as an expression.Label."""
995 __visit_name__ = "label"
996 __slots__ = "element", "name", "_alt_names"
998 def __init__(self, col, name, alt_names=()):
999 self.element = col
1000 self.name = name
1001 self._alt_names = (col,) + alt_names
1003 @property
1004 def proxy_set(self):
1005 return self.element.proxy_set
1007 @property
1008 def type(self):
1009 return self.element.type
1011 def self_group(self, **kw):
1012 return self
1015class ilike_case_insensitive(
1016 roles.BinaryElementRole[Any], elements.CompilerColumnElement
1017):
1018 """produce a wrapping element for a case-insensitive portion of
1019 an ILIKE construct.
1021 The construct usually renders the ``lower()`` function, but on
1022 PostgreSQL will pass silently with the assumption that "ILIKE"
1023 is being used.
1025 .. versionadded:: 2.0
1027 """
1029 __visit_name__ = "ilike_case_insensitive_operand"
1030 __slots__ = "element", "comparator"
1032 def __init__(self, element):
1033 self.element = element
1034 self.comparator = element.comparator
1036 @property
1037 def proxy_set(self):
1038 return self.element.proxy_set
1040 @property
1041 def type(self):
1042 return self.element.type
1044 def self_group(self, **kw):
1045 return self
1047 def _with_binary_element_type(self, type_):
1048 return ilike_case_insensitive(
1049 self.element._with_binary_element_type(type_)
1050 )
1053class SQLCompiler(Compiled):
1054 """Default implementation of :class:`.Compiled`.
1056 Compiles :class:`_expression.ClauseElement` objects into SQL strings.
1058 """
1060 extract_map = EXTRACT_MAP
1062 bindname_escape_characters: ClassVar[Mapping[str, str]] = (
1063 util.immutabledict(
1064 {
1065 "%": "P",
1066 "(": "A",
1067 ")": "Z",
1068 ":": "C",
1069 ".": "_",
1070 "[": "_",
1071 "]": "_",
1072 " ": "_",
1073 }
1074 )
1075 )
1076 """A mapping (e.g. dict or similar) containing a lookup of
1077 characters keyed to replacement characters which will be applied to all
1078 'bind names' used in SQL statements as a form of 'escaping'; the given
1079 characters are replaced entirely with the 'replacement' character when
1080 rendered in the SQL statement, and a similar translation is performed
1081 on the incoming names used in parameter dictionaries passed to methods
1082 like :meth:`_engine.Connection.execute`.
1084 This allows bound parameter names used in :func:`_sql.bindparam` and
1085 other constructs to have any arbitrary characters present without any
1086 concern for characters that aren't allowed at all on the target database.
1088 Third party dialects can establish their own dictionary here to replace the
1089 default mapping, which will ensure that the particular characters in the
1090 mapping will never appear in a bound parameter name.
1092 The dictionary is evaluated at **class creation time**, so cannot be
1093 modified at runtime; it must be present on the class when the class
1094 is first declared.
1096 Note that for dialects that have additional bound parameter rules such
1097 as additional restrictions on leading characters, the
1098 :meth:`_sql.SQLCompiler.bindparam_string` method may need to be augmented.
1099 See the cx_Oracle compiler for an example of this.
1101 .. versionadded:: 2.0.0rc1
1103 """
1105 _bind_translate_re: ClassVar[Pattern[str]]
1106 _bind_translate_chars: ClassVar[Mapping[str, str]]
1108 is_sql = True
1110 compound_keywords = COMPOUND_KEYWORDS
1112 isdelete: bool = False
1113 isinsert: bool = False
1114 isupdate: bool = False
1115 """class-level defaults which can be set at the instance
1116 level to define if this Compiled instance represents
1117 INSERT/UPDATE/DELETE
1118 """
1120 postfetch: Optional[List[Column[Any]]]
1121 """list of columns that can be post-fetched after INSERT or UPDATE to
1122 receive server-updated values"""
1124 insert_prefetch: Sequence[Column[Any]] = ()
1125 """list of columns for which default values should be evaluated before
1126 an INSERT takes place"""
1128 update_prefetch: Sequence[Column[Any]] = ()
1129 """list of columns for which onupdate default values should be evaluated
1130 before an UPDATE takes place"""
1132 implicit_returning: Optional[Sequence[ColumnElement[Any]]] = None
1133 """list of "implicit" returning columns for a toplevel INSERT or UPDATE
1134 statement, used to receive newly generated values of columns.
1136 .. versionadded:: 2.0 ``implicit_returning`` replaces the previous
1137 ``returning`` collection, which was not a generalized RETURNING
1138 collection and instead was in fact specific to the "implicit returning"
1139 feature.
1141 """
1143 isplaintext: bool = False
1145 binds: Dict[str, BindParameter[Any]]
1146 """a dictionary of bind parameter keys to BindParameter instances."""
1148 bind_names: Dict[BindParameter[Any], str]
1149 """a dictionary of BindParameter instances to "compiled" names
1150 that are actually present in the generated SQL"""
1152 stack: List[_CompilerStackEntry]
1153 """major statements such as SELECT, INSERT, UPDATE, DELETE are
1154 tracked in this stack using an entry format."""
1156 returning_precedes_values: bool = False
1157 """set to True classwide to generate RETURNING
1158 clauses before the VALUES or WHERE clause (i.e. MSSQL)
1159 """
1161 render_table_with_column_in_update_from: bool = False
1162 """set to True classwide to indicate the SET clause
1163 in a multi-table UPDATE statement should qualify
1164 columns with the table name (i.e. MySQL only)
1165 """
1167 ansi_bind_rules: bool = False
1168 """SQL 92 doesn't allow bind parameters to be used
1169 in the columns clause of a SELECT, nor does it allow
1170 ambiguous expressions like "? = ?". A compiler
1171 subclass can set this flag to False if the target
1172 driver/DB enforces this
1173 """
1175 bindtemplate: str
1176 """template to render bound parameters based on paramstyle."""
1178 compilation_bindtemplate: str
1179 """template used by compiler to render parameters before positional
1180 paramstyle application"""
1182 _numeric_binds_identifier_char: str
1183 """Character that's used to as the identifier of a numerical bind param.
1184 For example if this char is set to ``$``, numerical binds will be rendered
1185 in the form ``$1, $2, $3``.
1186 """
1188 _result_columns: List[ResultColumnsEntry]
1189 """relates label names in the final SQL to a tuple of local
1190 column/label name, ColumnElement object (if any) and
1191 TypeEngine. CursorResult uses this for type processing and
1192 column targeting"""
1194 _textual_ordered_columns: bool = False
1195 """tell the result object that the column names as rendered are important,
1196 but they are also "ordered" vs. what is in the compiled object here.
1198 As of 1.4.42 this condition is only present when the statement is a
1199 TextualSelect, e.g. text("....").columns(...), where it is required
1200 that the columns are considered positionally and not by name.
1202 """
1204 _ad_hoc_textual: bool = False
1205 """tell the result that we encountered text() or '*' constructs in the
1206 middle of the result columns, but we also have compiled columns, so
1207 if the number of columns in cursor.description does not match how many
1208 expressions we have, that means we can't rely on positional at all and
1209 should match on name.
1211 """
1213 _ordered_columns: bool = True
1214 """
1215 if False, means we can't be sure the list of entries
1216 in _result_columns is actually the rendered order. Usually
1217 True unless using an unordered TextualSelect.
1218 """
1220 _loose_column_name_matching: bool = False
1221 """tell the result object that the SQL statement is textual, wants to match
1222 up to Column objects, and may be using the ._tq_label in the SELECT rather
1223 than the base name.
1225 """
1227 _numeric_binds: bool = False
1228 """
1229 True if paramstyle is "numeric". This paramstyle is trickier than
1230 all the others.
1232 """
1234 _render_postcompile: bool = False
1235 """
1236 whether to render out POSTCOMPILE params during the compile phase.
1238 This attribute is used only for end-user invocation of stmt.compile();
1239 it's never used for actual statement execution, where instead the
1240 dialect internals access and render the internal postcompile structure
1241 directly.
1243 """
1245 _post_compile_expanded_state: Optional[ExpandedState] = None
1246 """When render_postcompile is used, the ``ExpandedState`` used to create
1247 the "expanded" SQL is assigned here, and then used by the ``.params``
1248 accessor and ``.construct_params()`` methods for their return values.
1250 .. versionadded:: 2.0.0rc1
1252 """
1254 _pre_expanded_string: Optional[str] = None
1255 """Stores the original string SQL before 'post_compile' is applied,
1256 for cases where 'post_compile' were used.
1258 """
1260 _pre_expanded_positiontup: Optional[List[str]] = None
1262 _insertmanyvalues: Optional[_InsertManyValues] = None
1264 _insert_crud_params: Optional[crud._CrudParamSequence] = None
1266 literal_execute_params: FrozenSet[BindParameter[Any]] = frozenset()
1267 """bindparameter objects that are rendered as literal values at statement
1268 execution time.
1270 """
1272 post_compile_params: FrozenSet[BindParameter[Any]] = frozenset()
1273 """bindparameter objects that are rendered as bound parameter placeholders
1274 at statement execution time.
1276 """
1278 escaped_bind_names: util.immutabledict[str, str] = util.EMPTY_DICT
1279 """Late escaping of bound parameter names that has to be converted
1280 to the original name when looking in the parameter dictionary.
1282 """
1284 has_out_parameters = False
1285 """if True, there are bindparam() objects that have the isoutparam
1286 flag set."""
1288 postfetch_lastrowid = False
1289 """if True, and this in insert, use cursor.lastrowid to populate
1290 result.inserted_primary_key. """
1292 _cache_key_bind_match: Optional[
1293 Tuple[
1294 Dict[
1295 BindParameter[Any],
1296 List[BindParameter[Any]],
1297 ],
1298 Dict[
1299 str,
1300 BindParameter[Any],
1301 ],
1302 ]
1303 ] = None
1304 """a mapping that will relate the BindParameter object we compile
1305 to those that are part of the extracted collection of parameters
1306 in the cache key, if we were given a cache key.
1308 """
1310 positiontup: Optional[List[str]] = None
1311 """for a compiled construct that uses a positional paramstyle, will be
1312 a sequence of strings, indicating the names of bound parameters in order.
1314 This is used in order to render bound parameters in their correct order,
1315 and is combined with the :attr:`_sql.Compiled.params` dictionary to
1316 render parameters.
1318 This sequence always contains the unescaped name of the parameters.
1320 .. seealso::
1322 :ref:`faq_sql_expression_string` - includes a usage example for
1323 debugging use cases.
1325 """
1326 _values_bindparam: Optional[List[str]] = None
1328 _visited_bindparam: Optional[List[str]] = None
1330 inline: bool = False
1332 ctes: Optional[MutableMapping[CTE, str]]
1334 # Detect same CTE references - Dict[(level, name), cte]
1335 # Level is required for supporting nesting
1336 ctes_by_level_name: Dict[Tuple[int, str], CTE]
1338 # To retrieve key/level in ctes_by_level_name -
1339 # Dict[cte_reference, (level, cte_name, cte_opts)]
1340 level_name_by_cte: Dict[CTE, Tuple[int, str, selectable._CTEOpts]]
1342 ctes_recursive: bool
1344 _post_compile_pattern = re.compile(r"__\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]")
1345 _pyformat_pattern = re.compile(r"%\(([^)]+?)\)s")
1346 _positional_pattern = re.compile(
1347 f"{_pyformat_pattern.pattern}|{_post_compile_pattern.pattern}"
1348 )
1350 @classmethod
1351 def _init_compiler_cls(cls):
1352 cls._init_bind_translate()
1354 @classmethod
1355 def _init_bind_translate(cls):
1356 reg = re.escape("".join(cls.bindname_escape_characters))
1357 cls._bind_translate_re = re.compile(f"[{reg}]")
1358 cls._bind_translate_chars = cls.bindname_escape_characters
1360 def __init__(
1361 self,
1362 dialect: Dialect,
1363 statement: Optional[ClauseElement],
1364 cache_key: Optional[CacheKey] = None,
1365 column_keys: Optional[Sequence[str]] = None,
1366 for_executemany: bool = False,
1367 linting: Linting = NO_LINTING,
1368 _supporting_against: Optional[SQLCompiler] = None,
1369 **kwargs: Any,
1370 ):
1371 """Construct a new :class:`.SQLCompiler` object.
1373 :param dialect: :class:`.Dialect` to be used
1375 :param statement: :class:`_expression.ClauseElement` to be compiled
1377 :param column_keys: a list of column names to be compiled into an
1378 INSERT or UPDATE statement.
1380 :param for_executemany: whether INSERT / UPDATE statements should
1381 expect that they are to be invoked in an "executemany" style,
1382 which may impact how the statement will be expected to return the
1383 values of defaults and autoincrement / sequences and similar.
1384 Depending on the backend and driver in use, support for retrieving
1385 these values may be disabled which means SQL expressions may
1386 be rendered inline, RETURNING may not be rendered, etc.
1388 :param kwargs: additional keyword arguments to be consumed by the
1389 superclass.
1391 """
1392 self.column_keys = column_keys
1394 self.cache_key = cache_key
1396 if cache_key:
1397 cksm = {b.key: b for b in cache_key[1]}
1398 ckbm = {b: [b] for b in cache_key[1]}
1399 self._cache_key_bind_match = (ckbm, cksm)
1401 # compile INSERT/UPDATE defaults/sequences to expect executemany
1402 # style execution, which may mean no pre-execute of defaults,
1403 # or no RETURNING
1404 self.for_executemany = for_executemany
1406 self.linting = linting
1408 # a dictionary of bind parameter keys to BindParameter
1409 # instances.
1410 self.binds = {}
1412 # a dictionary of BindParameter instances to "compiled" names
1413 # that are actually present in the generated SQL
1414 self.bind_names = util.column_dict()
1416 # stack which keeps track of nested SELECT statements
1417 self.stack = []
1419 self._result_columns = []
1421 # true if the paramstyle is positional
1422 self.positional = dialect.positional
1423 if self.positional:
1424 self._numeric_binds = nb = dialect.paramstyle.startswith("numeric")
1425 if nb:
1426 self._numeric_binds_identifier_char = (
1427 "$" if dialect.paramstyle == "numeric_dollar" else ":"
1428 )
1430 self.compilation_bindtemplate = _pyformat_template
1431 else:
1432 self.compilation_bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
1434 self.ctes = None
1436 self.label_length = (
1437 dialect.label_length or dialect.max_identifier_length
1438 )
1440 # a map which tracks "anonymous" identifiers that are created on
1441 # the fly here
1442 self.anon_map = prefix_anon_map()
1444 # a map which tracks "truncated" names based on
1445 # dialect.label_length or dialect.max_identifier_length
1446 self.truncated_names: Dict[Tuple[str, str], str] = {}
1447 self._truncated_counters: Dict[str, int] = {}
1449 Compiled.__init__(self, dialect, statement, **kwargs)
1451 if self.isinsert or self.isupdate or self.isdelete:
1452 if TYPE_CHECKING:
1453 assert isinstance(statement, UpdateBase)
1455 if self.isinsert or self.isupdate:
1456 if TYPE_CHECKING:
1457 assert isinstance(statement, ValuesBase)
1458 if statement._inline:
1459 self.inline = True
1460 elif self.for_executemany and (
1461 not self.isinsert
1462 or (
1463 self.dialect.insert_executemany_returning
1464 and statement._return_defaults
1465 )
1466 ):
1467 self.inline = True
1469 self.bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
1471 if _supporting_against:
1472 self.__dict__.update(
1473 {
1474 k: v
1475 for k, v in _supporting_against.__dict__.items()
1476 if k
1477 not in {
1478 "state",
1479 "dialect",
1480 "preparer",
1481 "positional",
1482 "_numeric_binds",
1483 "compilation_bindtemplate",
1484 "bindtemplate",
1485 }
1486 }
1487 )
1489 if self.state is CompilerState.STRING_APPLIED:
1490 if self.positional:
1491 if self._numeric_binds:
1492 self._process_numeric()
1493 else:
1494 self._process_positional()
1496 if self._render_postcompile:
1497 parameters = self.construct_params(
1498 escape_names=False,
1499 _no_postcompile=True,
1500 )
1502 self._process_parameters_for_postcompile(
1503 parameters, _populate_self=True
1504 )
1506 @property
1507 def insert_single_values_expr(self) -> Optional[str]:
1508 """When an INSERT is compiled with a single set of parameters inside
1509 a VALUES expression, the string is assigned here, where it can be
1510 used for insert batching schemes to rewrite the VALUES expression.
1512 .. versionchanged:: 2.0 This collection is no longer used by
1513 SQLAlchemy's built-in dialects, in favor of the currently
1514 internal ``_insertmanyvalues`` collection that is used only by
1515 :class:`.SQLCompiler`.
1517 """
1518 if self._insertmanyvalues is None:
1519 return None
1520 else:
1521 return self._insertmanyvalues.single_values_expr
1523 @util.ro_memoized_property
1524 def effective_returning(self) -> Optional[Sequence[ColumnElement[Any]]]:
1525 """The effective "returning" columns for INSERT, UPDATE or DELETE.
1527 This is either the so-called "implicit returning" columns which are
1528 calculated by the compiler on the fly, or those present based on what's
1529 present in ``self.statement._returning`` (expanded into individual
1530 columns using the ``._all_selected_columns`` attribute) i.e. those set
1531 explicitly using the :meth:`.UpdateBase.returning` method.
1533 .. versionadded:: 2.0
1535 """
1536 if self.implicit_returning:
1537 return self.implicit_returning
1538 elif self.statement is not None and is_dml(self.statement):
1539 return [
1540 c
1541 for c in self.statement._all_selected_columns
1542 if is_column_element(c)
1543 ]
1545 else:
1546 return None
1548 @property
1549 def returning(self):
1550 """backwards compatibility; returns the
1551 effective_returning collection.
1553 """
1554 return self.effective_returning
1556 @property
1557 def current_executable(self):
1558 """Return the current 'executable' that is being compiled.
1560 This is currently the :class:`_sql.Select`, :class:`_sql.Insert`,
1561 :class:`_sql.Update`, :class:`_sql.Delete`,
1562 :class:`_sql.CompoundSelect` object that is being compiled.
1563 Specifically it's assigned to the ``self.stack`` list of elements.
1565 When a statement like the above is being compiled, it normally
1566 is also assigned to the ``.statement`` attribute of the
1567 :class:`_sql.Compiler` object. However, all SQL constructs are
1568 ultimately nestable, and this attribute should never be consulted
1569 by a ``visit_`` method, as it is not guaranteed to be assigned
1570 nor guaranteed to correspond to the current statement being compiled.
1572 """
1573 try:
1574 return self.stack[-1]["selectable"]
1575 except IndexError as ie:
1576 raise IndexError("Compiler does not have a stack entry") from ie
1578 @property
1579 def prefetch(self):
1580 return list(self.insert_prefetch) + list(self.update_prefetch)
1582 @util.memoized_property
1583 def _global_attributes(self) -> Dict[Any, Any]:
1584 return {}
1586 @util.memoized_instancemethod
1587 def _init_cte_state(self) -> MutableMapping[CTE, str]:
1588 """Initialize collections related to CTEs only if
1589 a CTE is located, to save on the overhead of
1590 these collections otherwise.
1592 """
1593 # collect CTEs to tack on top of a SELECT
1594 # To store the query to print - Dict[cte, text_query]
1595 ctes: MutableMapping[CTE, str] = util.OrderedDict()
1596 self.ctes = ctes
1598 # Detect same CTE references - Dict[(level, name), cte]
1599 # Level is required for supporting nesting
1600 self.ctes_by_level_name = {}
1602 # To retrieve key/level in ctes_by_level_name -
1603 # Dict[cte_reference, (level, cte_name, cte_opts)]
1604 self.level_name_by_cte = {}
1606 self.ctes_recursive = False
1608 return ctes
1610 @contextlib.contextmanager
1611 def _nested_result(self):
1612 """special API to support the use case of 'nested result sets'"""
1613 result_columns, ordered_columns = (
1614 self._result_columns,
1615 self._ordered_columns,
1616 )
1617 self._result_columns, self._ordered_columns = [], False
1619 try:
1620 if self.stack:
1621 entry = self.stack[-1]
1622 entry["need_result_map_for_nested"] = True
1623 else:
1624 entry = None
1625 yield self._result_columns, self._ordered_columns
1626 finally:
1627 if entry:
1628 entry.pop("need_result_map_for_nested")
1629 self._result_columns, self._ordered_columns = (
1630 result_columns,
1631 ordered_columns,
1632 )
1634 def _process_positional(self):
1635 assert not self.positiontup
1636 assert self.state is CompilerState.STRING_APPLIED
1637 assert not self._numeric_binds
1639 if self.dialect.paramstyle == "format":
1640 placeholder = "%s"
1641 else:
1642 assert self.dialect.paramstyle == "qmark"
1643 placeholder = "?"
1645 positions = []
1647 def find_position(m: re.Match[str]) -> str:
1648 normal_bind = m.group(1)
1649 if normal_bind:
1650 positions.append(normal_bind)
1651 return placeholder
1652 else:
1653 # this a post-compile bind
1654 positions.append(m.group(2))
1655 return m.group(0)
1657 self.string = re.sub(
1658 self._positional_pattern, find_position, self.string
1659 )
1661 if self.escaped_bind_names:
1662 reverse_escape = {v: k for k, v in self.escaped_bind_names.items()}
1663 assert len(self.escaped_bind_names) == len(reverse_escape)
1664 self.positiontup = [
1665 reverse_escape.get(name, name) for name in positions
1666 ]
1667 else:
1668 self.positiontup = positions
1670 if self._insertmanyvalues:
1671 positions = []
1673 single_values_expr = re.sub(
1674 self._positional_pattern,
1675 find_position,
1676 self._insertmanyvalues.single_values_expr,
1677 )
1678 insert_crud_params = [
1679 (
1680 v[0],
1681 v[1],
1682 re.sub(self._positional_pattern, find_position, v[2]),
1683 v[3],
1684 )
1685 for v in self._insertmanyvalues.insert_crud_params
1686 ]
1688 self._insertmanyvalues = self._insertmanyvalues._replace(
1689 single_values_expr=single_values_expr,
1690 insert_crud_params=insert_crud_params,
1691 )
1693 def _process_numeric(self):
1694 assert self._numeric_binds
1695 assert self.state is CompilerState.STRING_APPLIED
1697 num = 1
1698 param_pos: Dict[str, str] = {}
1699 order: Iterable[str]
1700 if self._insertmanyvalues and self._values_bindparam is not None:
1701 # bindparams that are not in values are always placed first.
1702 # this avoids the need of changing them when using executemany
1703 # values () ()
1704 order = itertools.chain(
1705 (
1706 name
1707 for name in self.bind_names.values()
1708 if name not in self._values_bindparam
1709 ),
1710 self.bind_names.values(),
1711 )
1712 else:
1713 order = self.bind_names.values()
1715 for bind_name in order:
1716 if bind_name in param_pos:
1717 continue
1718 bind = self.binds[bind_name]
1719 if (
1720 bind in self.post_compile_params
1721 or bind in self.literal_execute_params
1722 ):
1723 # set to None to just mark the in positiontup, it will not
1724 # be replaced below.
1725 param_pos[bind_name] = None # type: ignore
1726 else:
1727 ph = f"{self._numeric_binds_identifier_char}{num}"
1728 num += 1
1729 param_pos[bind_name] = ph
1731 self.next_numeric_pos = num
1733 self.positiontup = list(param_pos)
1734 if self.escaped_bind_names:
1735 len_before = len(param_pos)
1736 param_pos = {
1737 self.escaped_bind_names.get(name, name): pos
1738 for name, pos in param_pos.items()
1739 }
1740 assert len(param_pos) == len_before
1742 # Can't use format here since % chars are not escaped.
1743 self.string = self._pyformat_pattern.sub(
1744 lambda m: param_pos[m.group(1)], self.string
1745 )
1747 if self._insertmanyvalues:
1748 single_values_expr = (
1749 # format is ok here since single_values_expr includes only
1750 # place-holders
1751 self._insertmanyvalues.single_values_expr
1752 % param_pos
1753 )
1754 insert_crud_params = [
1755 (v[0], v[1], "%s", v[3])
1756 for v in self._insertmanyvalues.insert_crud_params
1757 ]
1759 self._insertmanyvalues = self._insertmanyvalues._replace(
1760 # This has the numbers (:1, :2)
1761 single_values_expr=single_values_expr,
1762 # The single binds are instead %s so they can be formatted
1763 insert_crud_params=insert_crud_params,
1764 )
1766 @util.memoized_property
1767 def _bind_processors(
1768 self,
1769 ) -> MutableMapping[
1770 str, Union[_BindProcessorType[Any], Sequence[_BindProcessorType[Any]]]
1771 ]:
1772 # mypy is not able to see the two value types as the above Union,
1773 # it just sees "object". don't know how to resolve
1774 return {
1775 key: value # type: ignore
1776 for key, value in (
1777 (
1778 self.bind_names[bindparam],
1779 (
1780 bindparam.type._cached_bind_processor(self.dialect)
1781 if not bindparam.type._is_tuple_type
1782 else tuple(
1783 elem_type._cached_bind_processor(self.dialect)
1784 for elem_type in cast(
1785 TupleType, bindparam.type
1786 ).types
1787 )
1788 ),
1789 )
1790 for bindparam in self.bind_names
1791 )
1792 if value is not None
1793 }
1795 def is_subquery(self):
1796 return len(self.stack) > 1
1798 @property
1799 def sql_compiler(self) -> Self:
1800 return self
1802 def construct_expanded_state(
1803 self,
1804 params: Optional[_CoreSingleExecuteParams] = None,
1805 escape_names: bool = True,
1806 ) -> ExpandedState:
1807 """Return a new :class:`.ExpandedState` for a given parameter set.
1809 For queries that use "expanding" or other late-rendered parameters,
1810 this method will provide for both the finalized SQL string as well
1811 as the parameters that would be used for a particular parameter set.
1813 .. versionadded:: 2.0.0rc1
1815 """
1816 parameters = self.construct_params(
1817 params,
1818 escape_names=escape_names,
1819 _no_postcompile=True,
1820 )
1821 return self._process_parameters_for_postcompile(
1822 parameters,
1823 )
1825 def construct_params(
1826 self,
1827 params: Optional[_CoreSingleExecuteParams] = None,
1828 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
1829 escape_names: bool = True,
1830 _group_number: Optional[int] = None,
1831 _check: bool = True,
1832 _no_postcompile: bool = False,
1833 ) -> _MutableCoreSingleExecuteParams:
1834 """return a dictionary of bind parameter keys and values"""
1836 if self._render_postcompile and not _no_postcompile:
1837 assert self._post_compile_expanded_state is not None
1838 if not params:
1839 return dict(self._post_compile_expanded_state.parameters)
1840 else:
1841 raise exc.InvalidRequestError(
1842 "can't construct new parameters when render_postcompile "
1843 "is used; the statement is hard-linked to the original "
1844 "parameters. Use construct_expanded_state to generate a "
1845 "new statement and parameters."
1846 )
1848 has_escaped_names = escape_names and bool(self.escaped_bind_names)
1850 if extracted_parameters:
1851 # related the bound parameters collected in the original cache key
1852 # to those collected in the incoming cache key. They will not have
1853 # matching names but they will line up positionally in the same
1854 # way. The parameters present in self.bind_names may be clones of
1855 # these original cache key params in the case of DML but the .key
1856 # will be guaranteed to match.
1857 if self.cache_key is None:
1858 raise exc.CompileError(
1859 "This compiled object has no original cache key; "
1860 "can't pass extracted_parameters to construct_params"
1861 )
1862 else:
1863 orig_extracted = self.cache_key[1]
1865 ckbm_tuple = self._cache_key_bind_match
1866 assert ckbm_tuple is not None
1867 ckbm, _ = ckbm_tuple
1868 resolved_extracted = {
1869 bind: extracted
1870 for b, extracted in zip(orig_extracted, extracted_parameters)
1871 for bind in ckbm[b]
1872 }
1873 else:
1874 resolved_extracted = None
1876 if params:
1877 pd = {}
1878 for bindparam, name in self.bind_names.items():
1879 escaped_name = (
1880 self.escaped_bind_names.get(name, name)
1881 if has_escaped_names
1882 else name
1883 )
1885 if bindparam.key in params:
1886 pd[escaped_name] = params[bindparam.key]
1887 elif name in params:
1888 pd[escaped_name] = params[name]
1890 elif _check and bindparam.required:
1891 if _group_number:
1892 raise exc.InvalidRequestError(
1893 "A value is required for bind parameter %r, "
1894 "in parameter group %d"
1895 % (bindparam.key, _group_number),
1896 code="cd3x",
1897 )
1898 else:
1899 raise exc.InvalidRequestError(
1900 "A value is required for bind parameter %r"
1901 % bindparam.key,
1902 code="cd3x",
1903 )
1904 else:
1905 if resolved_extracted:
1906 value_param = resolved_extracted.get(
1907 bindparam, bindparam
1908 )
1909 else:
1910 value_param = bindparam
1912 if bindparam.callable:
1913 pd[escaped_name] = value_param.effective_value
1914 else:
1915 pd[escaped_name] = value_param.value
1916 return pd
1917 else:
1918 pd = {}
1919 for bindparam, name in self.bind_names.items():
1920 escaped_name = (
1921 self.escaped_bind_names.get(name, name)
1922 if has_escaped_names
1923 else name
1924 )
1926 if _check and bindparam.required:
1927 if _group_number:
1928 raise exc.InvalidRequestError(
1929 "A value is required for bind parameter %r, "
1930 "in parameter group %d"
1931 % (bindparam.key, _group_number),
1932 code="cd3x",
1933 )
1934 else:
1935 raise exc.InvalidRequestError(
1936 "A value is required for bind parameter %r"
1937 % bindparam.key,
1938 code="cd3x",
1939 )
1941 if resolved_extracted:
1942 value_param = resolved_extracted.get(bindparam, bindparam)
1943 else:
1944 value_param = bindparam
1946 if bindparam.callable:
1947 pd[escaped_name] = value_param.effective_value
1948 else:
1949 pd[escaped_name] = value_param.value
1951 return pd
1953 @util.memoized_instancemethod
1954 def _get_set_input_sizes_lookup(self):
1955 dialect = self.dialect
1957 include_types = dialect.include_set_input_sizes
1958 exclude_types = dialect.exclude_set_input_sizes
1960 dbapi = dialect.dbapi
1962 def lookup_type(typ):
1963 dbtype = typ._unwrapped_dialect_impl(dialect).get_dbapi_type(dbapi)
1965 if (
1966 dbtype is not None
1967 and (exclude_types is None or dbtype not in exclude_types)
1968 and (include_types is None or dbtype in include_types)
1969 ):
1970 return dbtype
1971 else:
1972 return None
1974 inputsizes = {}
1976 literal_execute_params = self.literal_execute_params
1978 for bindparam in self.bind_names:
1979 if bindparam in literal_execute_params:
1980 continue
1982 if bindparam.type._is_tuple_type:
1983 inputsizes[bindparam] = [
1984 lookup_type(typ)
1985 for typ in cast(TupleType, bindparam.type).types
1986 ]
1987 else:
1988 inputsizes[bindparam] = lookup_type(bindparam.type)
1990 return inputsizes
1992 @property
1993 def params(self):
1994 """Return the bind param dictionary embedded into this
1995 compiled object, for those values that are present.
1997 .. seealso::
1999 :ref:`faq_sql_expression_string` - includes a usage example for
2000 debugging use cases.
2002 """
2003 return self.construct_params(_check=False)
2005 def _process_parameters_for_postcompile(
2006 self,
2007 parameters: _MutableCoreSingleExecuteParams,
2008 _populate_self: bool = False,
2009 ) -> ExpandedState:
2010 """handle special post compile parameters.
2012 These include:
2014 * "expanding" parameters -typically IN tuples that are rendered
2015 on a per-parameter basis for an otherwise fixed SQL statement string.
2017 * literal_binds compiled with the literal_execute flag. Used for
2018 things like SQL Server "TOP N" where the driver does not accommodate
2019 N as a bound parameter.
2021 """
2023 expanded_parameters = {}
2024 new_positiontup: Optional[List[str]]
2026 pre_expanded_string = self._pre_expanded_string
2027 if pre_expanded_string is None:
2028 pre_expanded_string = self.string
2030 if self.positional:
2031 new_positiontup = []
2033 pre_expanded_positiontup = self._pre_expanded_positiontup
2034 if pre_expanded_positiontup is None:
2035 pre_expanded_positiontup = self.positiontup
2037 else:
2038 new_positiontup = pre_expanded_positiontup = None
2040 processors = self._bind_processors
2041 single_processors = cast(
2042 "Mapping[str, _BindProcessorType[Any]]", processors
2043 )
2044 tuple_processors = cast(
2045 "Mapping[str, Sequence[_BindProcessorType[Any]]]", processors
2046 )
2048 new_processors: Dict[str, _BindProcessorType[Any]] = {}
2050 replacement_expressions: Dict[str, Any] = {}
2051 to_update_sets: Dict[str, Any] = {}
2053 # notes:
2054 # *unescaped* parameter names in:
2055 # self.bind_names, self.binds, self._bind_processors, self.positiontup
2056 #
2057 # *escaped* parameter names in:
2058 # construct_params(), replacement_expressions
2060 numeric_positiontup: Optional[List[str]] = None
2062 if self.positional and pre_expanded_positiontup is not None:
2063 names: Iterable[str] = pre_expanded_positiontup
2064 if self._numeric_binds:
2065 numeric_positiontup = []
2066 else:
2067 names = self.bind_names.values()
2069 ebn = self.escaped_bind_names
2070 for name in names:
2071 escaped_name = ebn.get(name, name) if ebn else name
2072 parameter = self.binds[name]
2074 if parameter in self.literal_execute_params:
2075 if escaped_name not in replacement_expressions:
2076 replacement_expressions[escaped_name] = (
2077 self.render_literal_bindparam(
2078 parameter,
2079 render_literal_value=parameters.pop(escaped_name),
2080 )
2081 )
2082 continue
2084 if parameter in self.post_compile_params:
2085 if escaped_name in replacement_expressions:
2086 to_update = to_update_sets[escaped_name]
2087 values = None
2088 else:
2089 # we are removing the parameter from parameters
2090 # because it is a list value, which is not expected by
2091 # TypeEngine objects that would otherwise be asked to
2092 # process it. the single name is being replaced with
2093 # individual numbered parameters for each value in the
2094 # param.
2095 #
2096 # note we are also inserting *escaped* parameter names
2097 # into the given dictionary. default dialect will
2098 # use these param names directly as they will not be
2099 # in the escaped_bind_names dictionary.
2100 values = parameters.pop(name)
2102 leep_res = self._literal_execute_expanding_parameter(
2103 escaped_name, parameter, values
2104 )
2105 (to_update, replacement_expr) = leep_res
2107 to_update_sets[escaped_name] = to_update
2108 replacement_expressions[escaped_name] = replacement_expr
2110 if not parameter.literal_execute:
2111 parameters.update(to_update)
2112 if parameter.type._is_tuple_type:
2113 assert values is not None
2114 new_processors.update(
2115 (
2116 "%s_%s_%s" % (name, i, j),
2117 tuple_processors[name][j - 1],
2118 )
2119 for i, tuple_element in enumerate(values, 1)
2120 for j, _ in enumerate(tuple_element, 1)
2121 if name in tuple_processors
2122 and tuple_processors[name][j - 1] is not None
2123 )
2124 else:
2125 new_processors.update(
2126 (key, single_processors[name])
2127 for key, _ in to_update
2128 if name in single_processors
2129 )
2130 if numeric_positiontup is not None:
2131 numeric_positiontup.extend(
2132 name for name, _ in to_update
2133 )
2134 elif new_positiontup is not None:
2135 # to_update has escaped names, but that's ok since
2136 # these are new names, that aren't in the
2137 # escaped_bind_names dict.
2138 new_positiontup.extend(name for name, _ in to_update)
2139 expanded_parameters[name] = [
2140 expand_key for expand_key, _ in to_update
2141 ]
2142 elif new_positiontup is not None:
2143 new_positiontup.append(name)
2145 def process_expanding(m):
2146 key = m.group(1)
2147 expr = replacement_expressions[key]
2149 # if POSTCOMPILE included a bind_expression, render that
2150 # around each element
2151 if m.group(2):
2152 tok = m.group(2).split("~~")
2153 be_left, be_right = tok[1], tok[3]
2154 expr = ", ".join(
2155 "%s%s%s" % (be_left, exp, be_right)
2156 for exp in expr.split(", ")
2157 )
2158 return expr
2160 statement = re.sub(
2161 self._post_compile_pattern, process_expanding, pre_expanded_string
2162 )
2164 if numeric_positiontup is not None:
2165 assert new_positiontup is not None
2166 param_pos = {
2167 key: f"{self._numeric_binds_identifier_char}{num}"
2168 for num, key in enumerate(
2169 numeric_positiontup, self.next_numeric_pos
2170 )
2171 }
2172 # Can't use format here since % chars are not escaped.
2173 statement = self._pyformat_pattern.sub(
2174 lambda m: param_pos[m.group(1)], statement
2175 )
2176 new_positiontup.extend(numeric_positiontup)
2178 expanded_state = ExpandedState(
2179 statement,
2180 parameters,
2181 new_processors,
2182 new_positiontup,
2183 expanded_parameters,
2184 )
2186 if _populate_self:
2187 # this is for the "render_postcompile" flag, which is not
2188 # otherwise used internally and is for end-user debugging and
2189 # special use cases.
2190 self._pre_expanded_string = pre_expanded_string
2191 self._pre_expanded_positiontup = pre_expanded_positiontup
2192 self.string = expanded_state.statement
2193 self.positiontup = (
2194 list(expanded_state.positiontup or ())
2195 if self.positional
2196 else None
2197 )
2198 self._post_compile_expanded_state = expanded_state
2200 return expanded_state
2202 @util.preload_module("sqlalchemy.engine.cursor")
2203 def _create_result_map(self):
2204 """utility method used for unit tests only."""
2205 cursor = util.preloaded.engine_cursor
2206 return cursor.CursorResultMetaData._create_description_match_map(
2207 self._result_columns
2208 )
2210 # assigned by crud.py for insert/update statements
2211 _get_bind_name_for_col: _BindNameForColProtocol
2213 @util.memoized_property
2214 def _within_exec_param_key_getter(self) -> Callable[[Any], str]:
2215 getter = self._get_bind_name_for_col
2216 return getter
2218 @util.memoized_property
2219 @util.preload_module("sqlalchemy.engine.result")
2220 def _inserted_primary_key_from_lastrowid_getter(self):
2221 result = util.preloaded.engine_result
2223 param_key_getter = self._within_exec_param_key_getter
2225 assert self.compile_state is not None
2226 statement = self.compile_state.statement
2228 if TYPE_CHECKING:
2229 assert isinstance(statement, Insert)
2231 table = statement.table
2233 getters = [
2234 (operator.methodcaller("get", param_key_getter(col), None), col)
2235 for col in table.primary_key
2236 ]
2238 autoinc_getter = None
2239 autoinc_col = table._autoincrement_column
2240 if autoinc_col is not None:
2241 # apply type post processors to the lastrowid
2242 lastrowid_processor = autoinc_col.type._cached_result_processor(
2243 self.dialect, None
2244 )
2245 autoinc_key = param_key_getter(autoinc_col)
2247 # if a bind value is present for the autoincrement column
2248 # in the parameters, we need to do the logic dictated by
2249 # #7998; honor a non-None user-passed parameter over lastrowid.
2250 # previously in the 1.4 series we weren't fetching lastrowid
2251 # at all if the key were present in the parameters
2252 if autoinc_key in self.binds:
2254 def _autoinc_getter(lastrowid, parameters):
2255 param_value = parameters.get(autoinc_key, lastrowid)
2256 if param_value is not None:
2257 # they supplied non-None parameter, use that.
2258 # SQLite at least is observed to return the wrong
2259 # cursor.lastrowid for INSERT..ON CONFLICT so it
2260 # can't be used in all cases
2261 return param_value
2262 else:
2263 # use lastrowid
2264 return lastrowid
2266 # work around mypy https://github.com/python/mypy/issues/14027
2267 autoinc_getter = _autoinc_getter
2269 else:
2270 lastrowid_processor = None
2272 row_fn = result.result_tuple([col.key for col in table.primary_key])
2274 def get(lastrowid, parameters):
2275 """given cursor.lastrowid value and the parameters used for INSERT,
2276 return a "row" that represents the primary key, either by
2277 using the "lastrowid" or by extracting values from the parameters
2278 that were sent along with the INSERT.
2280 """
2281 if lastrowid_processor is not None:
2282 lastrowid = lastrowid_processor(lastrowid)
2284 if lastrowid is None:
2285 return row_fn(getter(parameters) for getter, col in getters)
2286 else:
2287 return row_fn(
2288 (
2289 (
2290 autoinc_getter(lastrowid, parameters)
2291 if autoinc_getter is not None
2292 else lastrowid
2293 )
2294 if col is autoinc_col
2295 else getter(parameters)
2296 )
2297 for getter, col in getters
2298 )
2300 return get
2302 @util.memoized_property
2303 @util.preload_module("sqlalchemy.engine.result")
2304 def _inserted_primary_key_from_returning_getter(self):
2305 result = util.preloaded.engine_result
2307 assert self.compile_state is not None
2308 statement = self.compile_state.statement
2310 if TYPE_CHECKING:
2311 assert isinstance(statement, Insert)
2313 param_key_getter = self._within_exec_param_key_getter
2314 table = statement.table
2316 returning = self.implicit_returning
2317 assert returning is not None
2318 ret = {col: idx for idx, col in enumerate(returning)}
2320 getters = cast(
2321 "List[Tuple[Callable[[Any], Any], bool]]",
2322 [
2323 (
2324 (operator.itemgetter(ret[col]), True)
2325 if col in ret
2326 else (
2327 operator.methodcaller(
2328 "get", param_key_getter(col), None
2329 ),
2330 False,
2331 )
2332 )
2333 for col in table.primary_key
2334 ],
2335 )
2337 row_fn = result.result_tuple([col.key for col in table.primary_key])
2339 def get(row, parameters):
2340 return row_fn(
2341 getter(row) if use_row else getter(parameters)
2342 for getter, use_row in getters
2343 )
2345 return get
2347 def default_from(self) -> str:
2348 """Called when a SELECT statement has no froms, and no FROM clause is
2349 to be appended.
2351 Gives Oracle Database a chance to tack on a ``FROM DUAL`` to the string
2352 output.
2354 """
2355 return ""
2357 def visit_override_binds(self, override_binds, **kw):
2358 """SQL compile the nested element of an _OverrideBinds with
2359 bindparams swapped out.
2361 The _OverrideBinds is not normally expected to be compiled; it
2362 is meant to be used when an already cached statement is to be used,
2363 the compilation was already performed, and only the bound params should
2364 be swapped in at execution time.
2366 However, there are test cases that exericise this object, and
2367 additionally the ORM subquery loader is known to feed in expressions
2368 which include this construct into new queries (discovered in #11173),
2369 so it has to do the right thing at compile time as well.
2371 """
2373 # get SQL text first
2374 sqltext = override_binds.element._compiler_dispatch(self, **kw)
2376 # for a test compile that is not for caching, change binds after the
2377 # fact. note that we don't try to
2378 # swap the bindparam as we compile, because our element may be
2379 # elsewhere in the statement already (e.g. a subquery or perhaps a
2380 # CTE) and was already visited / compiled. See
2381 # test_relationship_criteria.py ->
2382 # test_selectinload_local_criteria_subquery
2383 for k in override_binds.translate:
2384 if k not in self.binds:
2385 continue
2386 bp = self.binds[k]
2388 # so this would work, just change the value of bp in place.
2389 # but we dont want to mutate things outside.
2390 # bp.value = override_binds.translate[bp.key]
2391 # continue
2393 # instead, need to replace bp with new_bp or otherwise accommodate
2394 # in all internal collections
2395 new_bp = bp._with_value(
2396 override_binds.translate[bp.key],
2397 maintain_key=True,
2398 required=False,
2399 )
2401 name = self.bind_names[bp]
2402 self.binds[k] = self.binds[name] = new_bp
2403 self.bind_names[new_bp] = name
2404 self.bind_names.pop(bp, None)
2406 if bp in self.post_compile_params:
2407 self.post_compile_params |= {new_bp}
2408 if bp in self.literal_execute_params:
2409 self.literal_execute_params |= {new_bp}
2411 ckbm_tuple = self._cache_key_bind_match
2412 if ckbm_tuple:
2413 ckbm, cksm = ckbm_tuple
2414 for bp in bp._cloned_set:
2415 if bp.key in cksm:
2416 cb = cksm[bp.key]
2417 ckbm[cb].append(new_bp)
2419 return sqltext
2421 def visit_grouping(self, grouping, asfrom=False, **kwargs):
2422 return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")"
2424 def visit_select_statement_grouping(self, grouping, **kwargs):
2425 return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")"
2427 def visit_label_reference(
2428 self, element, within_columns_clause=False, **kwargs
2429 ):
2430 if self.stack and self.dialect.supports_simple_order_by_label:
2431 try:
2432 compile_state = cast(
2433 "Union[SelectState, CompoundSelectState]",
2434 self.stack[-1]["compile_state"],
2435 )
2436 except KeyError as ke:
2437 raise exc.CompileError(
2438 "Can't resolve label reference for ORDER BY / "
2439 "GROUP BY / DISTINCT etc."
2440 ) from ke
2442 (
2443 with_cols,
2444 only_froms,
2445 only_cols,
2446 ) = compile_state._label_resolve_dict
2447 if within_columns_clause:
2448 resolve_dict = only_froms
2449 else:
2450 resolve_dict = only_cols
2452 # this can be None in the case that a _label_reference()
2453 # were subject to a replacement operation, in which case
2454 # the replacement of the Label element may have changed
2455 # to something else like a ColumnClause expression.
2456 order_by_elem = element.element._order_by_label_element
2458 if (
2459 order_by_elem is not None
2460 and order_by_elem.name in resolve_dict
2461 and order_by_elem.shares_lineage(
2462 resolve_dict[order_by_elem.name]
2463 )
2464 ):
2465 kwargs["render_label_as_label"] = (
2466 element.element._order_by_label_element
2467 )
2468 return self.process(
2469 element.element,
2470 within_columns_clause=within_columns_clause,
2471 **kwargs,
2472 )
2474 def visit_textual_label_reference(
2475 self, element, within_columns_clause=False, **kwargs
2476 ):
2477 if not self.stack:
2478 # compiling the element outside of the context of a SELECT
2479 return self.process(element._text_clause)
2481 try:
2482 compile_state = cast(
2483 "Union[SelectState, CompoundSelectState]",
2484 self.stack[-1]["compile_state"],
2485 )
2486 except KeyError as ke:
2487 coercions._no_text_coercion(
2488 element.element,
2489 extra=(
2490 "Can't resolve label reference for ORDER BY / "
2491 "GROUP BY / DISTINCT etc."
2492 ),
2493 exc_cls=exc.CompileError,
2494 err=ke,
2495 )
2497 with_cols, only_froms, only_cols = compile_state._label_resolve_dict
2498 try:
2499 if within_columns_clause:
2500 col = only_froms[element.element]
2501 else:
2502 col = with_cols[element.element]
2503 except KeyError as err:
2504 coercions._no_text_coercion(
2505 element.element,
2506 extra=(
2507 "Can't resolve label reference for ORDER BY / "
2508 "GROUP BY / DISTINCT etc."
2509 ),
2510 exc_cls=exc.CompileError,
2511 err=err,
2512 )
2513 else:
2514 kwargs["render_label_as_label"] = col
2515 return self.process(
2516 col, within_columns_clause=within_columns_clause, **kwargs
2517 )
2519 def visit_label(
2520 self,
2521 label,
2522 add_to_result_map=None,
2523 within_label_clause=False,
2524 within_columns_clause=False,
2525 render_label_as_label=None,
2526 result_map_targets=(),
2527 **kw,
2528 ):
2529 # only render labels within the columns clause
2530 # or ORDER BY clause of a select. dialect-specific compilers
2531 # can modify this behavior.
2532 render_label_with_as = (
2533 within_columns_clause and not within_label_clause
2534 )
2535 render_label_only = render_label_as_label is label
2537 if render_label_only or render_label_with_as:
2538 if isinstance(label.name, elements._truncated_label):
2539 labelname = self._truncated_identifier("colident", label.name)
2540 else:
2541 labelname = label.name
2543 if render_label_with_as:
2544 if add_to_result_map is not None:
2545 add_to_result_map(
2546 labelname,
2547 label.name,
2548 (label, labelname) + label._alt_names + result_map_targets,
2549 label.type,
2550 )
2551 return (
2552 label.element._compiler_dispatch(
2553 self,
2554 within_columns_clause=True,
2555 within_label_clause=True,
2556 **kw,
2557 )
2558 + OPERATORS[operators.as_]
2559 + self.preparer.format_label(label, labelname)
2560 )
2561 elif render_label_only:
2562 return self.preparer.format_label(label, labelname)
2563 else:
2564 return label.element._compiler_dispatch(
2565 self, within_columns_clause=False, **kw
2566 )
2568 def _fallback_column_name(self, column):
2569 raise exc.CompileError(
2570 "Cannot compile Column object until its 'name' is assigned."
2571 )
2573 def visit_lambda_element(self, element, **kw):
2574 sql_element = element._resolved
2575 return self.process(sql_element, **kw)
2577 def visit_column(
2578 self,
2579 column: ColumnClause[Any],
2580 add_to_result_map: Optional[_ResultMapAppender] = None,
2581 include_table: bool = True,
2582 result_map_targets: Tuple[Any, ...] = (),
2583 ambiguous_table_name_map: Optional[_AmbiguousTableNameMap] = None,
2584 **kwargs: Any,
2585 ) -> str:
2586 name = orig_name = column.name
2587 if name is None:
2588 name = self._fallback_column_name(column)
2590 is_literal = column.is_literal
2591 if not is_literal and isinstance(name, elements._truncated_label):
2592 name = self._truncated_identifier("colident", name)
2594 if add_to_result_map is not None:
2595 targets = (column, name, column.key) + result_map_targets
2596 if column._tq_label:
2597 targets += (column._tq_label,)
2599 add_to_result_map(name, orig_name, targets, column.type)
2601 if is_literal:
2602 # note we are not currently accommodating for
2603 # literal_column(quoted_name('ident', True)) here
2604 name = self.escape_literal_column(name)
2605 else:
2606 name = self.preparer.quote(name)
2607 table = column.table
2608 if table is None or not include_table or not table.named_with_column:
2609 return name
2610 else:
2611 effective_schema = self.preparer.schema_for_object(table)
2613 if effective_schema:
2614 schema_prefix = (
2615 self.preparer.quote_schema(effective_schema) + "."
2616 )
2617 else:
2618 schema_prefix = ""
2620 if TYPE_CHECKING:
2621 assert isinstance(table, NamedFromClause)
2622 tablename = table.name
2624 if (
2625 not effective_schema
2626 and ambiguous_table_name_map
2627 and tablename in ambiguous_table_name_map
2628 ):
2629 tablename = ambiguous_table_name_map[tablename]
2631 if isinstance(tablename, elements._truncated_label):
2632 tablename = self._truncated_identifier("alias", tablename)
2634 return schema_prefix + self.preparer.quote(tablename) + "." + name
2636 def visit_collation(self, element, **kw):
2637 return self.preparer.format_collation(element.collation)
2639 def visit_fromclause(self, fromclause, **kwargs):
2640 return fromclause.name
2642 def visit_index(self, index, **kwargs):
2643 return index.name
2645 def visit_typeclause(self, typeclause, **kw):
2646 kw["type_expression"] = typeclause
2647 kw["identifier_preparer"] = self.preparer
2648 return self.dialect.type_compiler_instance.process(
2649 typeclause.type, **kw
2650 )
2652 def post_process_text(self, text):
2653 if self.preparer._double_percents:
2654 text = text.replace("%", "%%")
2655 return text
2657 def escape_literal_column(self, text):
2658 if self.preparer._double_percents:
2659 text = text.replace("%", "%%")
2660 return text
2662 def visit_textclause(self, textclause, add_to_result_map=None, **kw):
2663 def do_bindparam(m):
2664 name = m.group(1)
2665 if name in textclause._bindparams:
2666 return self.process(textclause._bindparams[name], **kw)
2667 else:
2668 return self.bindparam_string(name, **kw)
2670 if not self.stack:
2671 self.isplaintext = True
2673 if add_to_result_map:
2674 # text() object is present in the columns clause of a
2675 # select(). Add a no-name entry to the result map so that
2676 # row[text()] produces a result
2677 add_to_result_map(None, None, (textclause,), sqltypes.NULLTYPE)
2679 # un-escape any \:params
2680 return BIND_PARAMS_ESC.sub(
2681 lambda m: m.group(1),
2682 BIND_PARAMS.sub(
2683 do_bindparam, self.post_process_text(textclause.text)
2684 ),
2685 )
2687 def visit_textual_select(
2688 self, taf, compound_index=None, asfrom=False, **kw
2689 ):
2690 toplevel = not self.stack
2691 entry = self._default_stack_entry if toplevel else self.stack[-1]
2693 new_entry: _CompilerStackEntry = {
2694 "correlate_froms": set(),
2695 "asfrom_froms": set(),
2696 "selectable": taf,
2697 }
2698 self.stack.append(new_entry)
2700 if taf._independent_ctes:
2701 self._dispatch_independent_ctes(taf, kw)
2703 populate_result_map = (
2704 toplevel
2705 or (
2706 compound_index == 0
2707 and entry.get("need_result_map_for_compound", False)
2708 )
2709 or entry.get("need_result_map_for_nested", False)
2710 )
2712 if populate_result_map:
2713 self._ordered_columns = self._textual_ordered_columns = (
2714 taf.positional
2715 )
2717 # enable looser result column matching when the SQL text links to
2718 # Column objects by name only
2719 self._loose_column_name_matching = not taf.positional and bool(
2720 taf.column_args
2721 )
2723 for c in taf.column_args:
2724 self.process(
2725 c,
2726 within_columns_clause=True,
2727 add_to_result_map=self._add_to_result_map,
2728 )
2730 text = self.process(taf.element, **kw)
2731 if self.ctes:
2732 nesting_level = len(self.stack) if not toplevel else None
2733 text = self._render_cte_clause(nesting_level=nesting_level) + text
2735 self.stack.pop(-1)
2737 return text
2739 def visit_null(self, expr: Null, **kw: Any) -> str:
2740 return "NULL"
2742 def visit_true(self, expr: True_, **kw: Any) -> str:
2743 if self.dialect.supports_native_boolean:
2744 return "true"
2745 else:
2746 return "1"
2748 def visit_false(self, expr: False_, **kw: Any) -> str:
2749 if self.dialect.supports_native_boolean:
2750 return "false"
2751 else:
2752 return "0"
2754 def _generate_delimited_list(self, elements, separator, **kw):
2755 return separator.join(
2756 s
2757 for s in (c._compiler_dispatch(self, **kw) for c in elements)
2758 if s
2759 )
2761 def _generate_delimited_and_list(self, clauses, **kw):
2762 lcc, clauses = elements.BooleanClauseList._process_clauses_for_boolean(
2763 operators.and_,
2764 elements.True_._singleton,
2765 elements.False_._singleton,
2766 clauses,
2767 )
2768 if lcc == 1:
2769 return clauses[0]._compiler_dispatch(self, **kw)
2770 else:
2771 separator = OPERATORS[operators.and_]
2772 return separator.join(
2773 s
2774 for s in (c._compiler_dispatch(self, **kw) for c in clauses)
2775 if s
2776 )
2778 def visit_tuple(self, clauselist, **kw):
2779 return "(%s)" % self.visit_clauselist(clauselist, **kw)
2781 def visit_element_list(self, element, **kw):
2782 return self._generate_delimited_list(element.clauses, " ", **kw)
2784 def visit_clauselist(self, clauselist, **kw):
2785 sep = clauselist.operator
2786 if sep is None:
2787 sep = " "
2788 else:
2789 sep = OPERATORS[clauselist.operator]
2791 return self._generate_delimited_list(clauselist.clauses, sep, **kw)
2793 def visit_expression_clauselist(self, clauselist, **kw):
2794 operator_ = clauselist.operator
2796 disp = self._get_operator_dispatch(
2797 operator_, "expression_clauselist", None
2798 )
2799 if disp:
2800 return disp(clauselist, operator_, **kw)
2802 try:
2803 opstring = OPERATORS[operator_]
2804 except KeyError as err:
2805 raise exc.UnsupportedCompilationError(self, operator_) from err
2806 else:
2807 kw["_in_operator_expression"] = True
2808 return self._generate_delimited_list(
2809 clauselist.clauses, opstring, **kw
2810 )
2812 def visit_case(self, clause, **kwargs):
2813 x = "CASE "
2814 if clause.value is not None:
2815 x += clause.value._compiler_dispatch(self, **kwargs) + " "
2816 for cond, result in clause.whens:
2817 x += (
2818 "WHEN "
2819 + cond._compiler_dispatch(self, **kwargs)
2820 + " THEN "
2821 + result._compiler_dispatch(self, **kwargs)
2822 + " "
2823 )
2824 if clause.else_ is not None:
2825 x += (
2826 "ELSE " + clause.else_._compiler_dispatch(self, **kwargs) + " "
2827 )
2828 x += "END"
2829 return x
2831 def visit_type_coerce(self, type_coerce, **kw):
2832 return type_coerce.typed_expression._compiler_dispatch(self, **kw)
2834 def visit_cast(self, cast, **kwargs):
2835 type_clause = cast.typeclause._compiler_dispatch(self, **kwargs)
2836 match = re.match("(.*)( COLLATE .*)", type_clause)
2837 return "CAST(%s AS %s)%s" % (
2838 cast.clause._compiler_dispatch(self, **kwargs),
2839 match.group(1) if match else type_clause,
2840 match.group(2) if match else "",
2841 )
2843 def visit_frame_clause(self, frameclause, **kw):
2845 if frameclause.lower_type is elements._FrameClauseType.RANGE_UNBOUNDED:
2846 left = "UNBOUNDED PRECEDING"
2847 elif frameclause.lower_type is elements._FrameClauseType.RANGE_CURRENT:
2848 left = "CURRENT ROW"
2849 else:
2850 val = self.process(frameclause.lower_integer_bind, **kw)
2851 if (
2852 frameclause.lower_type
2853 is elements._FrameClauseType.RANGE_PRECEDING
2854 ):
2855 left = f"{val} PRECEDING"
2856 else:
2857 left = f"{val} FOLLOWING"
2859 if frameclause.upper_type is elements._FrameClauseType.RANGE_UNBOUNDED:
2860 right = "UNBOUNDED FOLLOWING"
2861 elif frameclause.upper_type is elements._FrameClauseType.RANGE_CURRENT:
2862 right = "CURRENT ROW"
2863 else:
2864 val = self.process(frameclause.upper_integer_bind, **kw)
2865 if (
2866 frameclause.upper_type
2867 is elements._FrameClauseType.RANGE_PRECEDING
2868 ):
2869 right = f"{val} PRECEDING"
2870 else:
2871 right = f"{val} FOLLOWING"
2873 return f"{left} AND {right}"
2875 def visit_over(self, over, **kwargs):
2876 text = over.element._compiler_dispatch(self, **kwargs)
2877 if over.range_ is not None:
2878 range_ = f"RANGE BETWEEN {self.process(over.range_, **kwargs)}"
2879 elif over.rows is not None:
2880 range_ = f"ROWS BETWEEN {self.process(over.rows, **kwargs)}"
2881 elif over.groups is not None:
2882 range_ = f"GROUPS BETWEEN {self.process(over.groups, **kwargs)}"
2883 else:
2884 range_ = None
2886 return "%s OVER (%s)" % (
2887 text,
2888 " ".join(
2889 [
2890 "%s BY %s"
2891 % (word, clause._compiler_dispatch(self, **kwargs))
2892 for word, clause in (
2893 ("PARTITION", over.partition_by),
2894 ("ORDER", over.order_by),
2895 )
2896 if clause is not None and len(clause)
2897 ]
2898 + ([range_] if range_ else [])
2899 ),
2900 )
2902 def visit_withingroup(self, withingroup, **kwargs):
2903 return "%s WITHIN GROUP (ORDER BY %s)" % (
2904 withingroup.element._compiler_dispatch(self, **kwargs),
2905 withingroup.order_by._compiler_dispatch(self, **kwargs),
2906 )
2908 def visit_funcfilter(self, funcfilter, **kwargs):
2909 return "%s FILTER (WHERE %s)" % (
2910 funcfilter.func._compiler_dispatch(self, **kwargs),
2911 funcfilter.criterion._compiler_dispatch(self, **kwargs),
2912 )
2914 def visit_extract(self, extract, **kwargs):
2915 field = self.extract_map.get(extract.field, extract.field)
2916 return "EXTRACT(%s FROM %s)" % (
2917 field,
2918 extract.expr._compiler_dispatch(self, **kwargs),
2919 )
2921 def visit_scalar_function_column(self, element, **kw):
2922 compiled_fn = self.visit_function(element.fn, **kw)
2923 compiled_col = self.visit_column(element, **kw)
2924 return "(%s).%s" % (compiled_fn, compiled_col)
2926 def visit_function(
2927 self,
2928 func: Function[Any],
2929 add_to_result_map: Optional[_ResultMapAppender] = None,
2930 **kwargs: Any,
2931 ) -> str:
2932 if add_to_result_map is not None:
2933 add_to_result_map(func.name, func.name, (func.name,), func.type)
2935 disp = getattr(self, "visit_%s_func" % func.name.lower(), None)
2937 text: str
2939 if disp:
2940 text = disp(func, **kwargs)
2941 else:
2942 name = FUNCTIONS.get(func._deannotate().__class__, None)
2943 if name:
2944 if func._has_args:
2945 name += "%(expr)s"
2946 else:
2947 name = func.name
2948 name = (
2949 self.preparer.quote(name)
2950 if self.preparer._requires_quotes_illegal_chars(name)
2951 or isinstance(name, elements.quoted_name)
2952 else name
2953 )
2954 name = name + "%(expr)s"
2955 text = ".".join(
2956 [
2957 (
2958 self.preparer.quote(tok)
2959 if self.preparer._requires_quotes_illegal_chars(tok)
2960 or isinstance(name, elements.quoted_name)
2961 else tok
2962 )
2963 for tok in func.packagenames
2964 ]
2965 + [name]
2966 ) % {"expr": self.function_argspec(func, **kwargs)}
2968 if func._with_ordinality:
2969 text += " WITH ORDINALITY"
2970 return text
2972 def visit_next_value_func(self, next_value, **kw):
2973 return self.visit_sequence(next_value.sequence)
2975 def visit_sequence(self, sequence, **kw):
2976 raise NotImplementedError(
2977 "Dialect '%s' does not support sequence increments."
2978 % self.dialect.name
2979 )
2981 def function_argspec(self, func: Function[Any], **kwargs: Any) -> str:
2982 return func.clause_expr._compiler_dispatch(self, **kwargs)
2984 def visit_compound_select(
2985 self, cs, asfrom=False, compound_index=None, **kwargs
2986 ):
2987 toplevel = not self.stack
2989 compile_state = cs._compile_state_factory(cs, self, **kwargs)
2991 if toplevel and not self.compile_state:
2992 self.compile_state = compile_state
2994 compound_stmt = compile_state.statement
2996 entry = self._default_stack_entry if toplevel else self.stack[-1]
2997 need_result_map = toplevel or (
2998 not compound_index
2999 and entry.get("need_result_map_for_compound", False)
3000 )
3002 # indicates there is already a CompoundSelect in play
3003 if compound_index == 0:
3004 entry["select_0"] = cs
3006 self.stack.append(
3007 {
3008 "correlate_froms": entry["correlate_froms"],
3009 "asfrom_froms": entry["asfrom_froms"],
3010 "selectable": cs,
3011 "compile_state": compile_state,
3012 "need_result_map_for_compound": need_result_map,
3013 }
3014 )
3016 if compound_stmt._independent_ctes:
3017 self._dispatch_independent_ctes(compound_stmt, kwargs)
3019 keyword = self.compound_keywords[cs.keyword]
3021 text = (" " + keyword + " ").join(
3022 (
3023 c._compiler_dispatch(
3024 self, asfrom=asfrom, compound_index=i, **kwargs
3025 )
3026 for i, c in enumerate(cs.selects)
3027 )
3028 )
3030 kwargs["include_table"] = False
3031 text += self.group_by_clause(cs, **dict(asfrom=asfrom, **kwargs))
3032 text += self.order_by_clause(cs, **kwargs)
3033 if cs._has_row_limiting_clause:
3034 text += self._row_limit_clause(cs, **kwargs)
3036 if self.ctes:
3037 nesting_level = len(self.stack) if not toplevel else None
3038 text = (
3039 self._render_cte_clause(
3040 nesting_level=nesting_level,
3041 include_following_stack=True,
3042 )
3043 + text
3044 )
3046 self.stack.pop(-1)
3047 return text
3049 def _row_limit_clause(self, cs, **kwargs):
3050 if cs._fetch_clause is not None:
3051 return self.fetch_clause(cs, **kwargs)
3052 else:
3053 return self.limit_clause(cs, **kwargs)
3055 def _get_operator_dispatch(self, operator_, qualifier1, qualifier2):
3056 attrname = "visit_%s_%s%s" % (
3057 operator_.__name__,
3058 qualifier1,
3059 "_" + qualifier2 if qualifier2 else "",
3060 )
3061 return getattr(self, attrname, None)
3063 def visit_unary(
3064 self, unary, add_to_result_map=None, result_map_targets=(), **kw
3065 ):
3066 if add_to_result_map is not None:
3067 result_map_targets += (unary,)
3068 kw["add_to_result_map"] = add_to_result_map
3069 kw["result_map_targets"] = result_map_targets
3071 if unary.operator:
3072 if unary.modifier:
3073 raise exc.CompileError(
3074 "Unary expression does not support operator "
3075 "and modifier simultaneously"
3076 )
3077 disp = self._get_operator_dispatch(
3078 unary.operator, "unary", "operator"
3079 )
3080 if disp:
3081 return disp(unary, unary.operator, **kw)
3082 else:
3083 return self._generate_generic_unary_operator(
3084 unary, OPERATORS[unary.operator], **kw
3085 )
3086 elif unary.modifier:
3087 disp = self._get_operator_dispatch(
3088 unary.modifier, "unary", "modifier"
3089 )
3090 if disp:
3091 return disp(unary, unary.modifier, **kw)
3092 else:
3093 return self._generate_generic_unary_modifier(
3094 unary, OPERATORS[unary.modifier], **kw
3095 )
3096 else:
3097 raise exc.CompileError(
3098 "Unary expression has no operator or modifier"
3099 )
3101 def visit_truediv_binary(self, binary, operator, **kw):
3102 if self.dialect.div_is_floordiv:
3103 return (
3104 self.process(binary.left, **kw)
3105 + " / "
3106 # TODO: would need a fast cast again here,
3107 # unless we want to use an implicit cast like "+ 0.0"
3108 + self.process(
3109 elements.Cast(
3110 binary.right,
3111 (
3112 binary.right.type
3113 if binary.right.type._type_affinity
3114 in (sqltypes.Numeric, sqltypes.Float)
3115 else sqltypes.Numeric()
3116 ),
3117 ),
3118 **kw,
3119 )
3120 )
3121 else:
3122 return (
3123 self.process(binary.left, **kw)
3124 + " / "
3125 + self.process(binary.right, **kw)
3126 )
3128 def visit_floordiv_binary(self, binary, operator, **kw):
3129 if (
3130 self.dialect.div_is_floordiv
3131 and binary.right.type._type_affinity is sqltypes.Integer
3132 ):
3133 return (
3134 self.process(binary.left, **kw)
3135 + " / "
3136 + self.process(binary.right, **kw)
3137 )
3138 else:
3139 return "FLOOR(%s)" % (
3140 self.process(binary.left, **kw)
3141 + " / "
3142 + self.process(binary.right, **kw)
3143 )
3145 def visit_is_true_unary_operator(self, element, operator, **kw):
3146 if (
3147 element._is_implicitly_boolean
3148 or self.dialect.supports_native_boolean
3149 ):
3150 return self.process(element.element, **kw)
3151 else:
3152 return "%s = 1" % self.process(element.element, **kw)
3154 def visit_is_false_unary_operator(self, element, operator, **kw):
3155 if (
3156 element._is_implicitly_boolean
3157 or self.dialect.supports_native_boolean
3158 ):
3159 return "NOT %s" % self.process(element.element, **kw)
3160 else:
3161 return "%s = 0" % self.process(element.element, **kw)
3163 def visit_not_match_op_binary(self, binary, operator, **kw):
3164 return "NOT %s" % self.visit_binary(
3165 binary, override_operator=operators.match_op
3166 )
3168 def visit_not_in_op_binary(self, binary, operator, **kw):
3169 # The brackets are required in the NOT IN operation because the empty
3170 # case is handled using the form "(col NOT IN (null) OR 1 = 1)".
3171 # The presence of the OR makes the brackets required.
3172 return "(%s)" % self._generate_generic_binary(
3173 binary, OPERATORS[operator], **kw
3174 )
3176 def visit_empty_set_op_expr(self, type_, expand_op, **kw):
3177 if expand_op is operators.not_in_op:
3178 if len(type_) > 1:
3179 return "(%s)) OR (1 = 1" % (
3180 ", ".join("NULL" for element in type_)
3181 )
3182 else:
3183 return "NULL) OR (1 = 1"
3184 elif expand_op is operators.in_op:
3185 if len(type_) > 1:
3186 return "(%s)) AND (1 != 1" % (
3187 ", ".join("NULL" for element in type_)
3188 )
3189 else:
3190 return "NULL) AND (1 != 1"
3191 else:
3192 return self.visit_empty_set_expr(type_)
3194 def visit_empty_set_expr(self, element_types, **kw):
3195 raise NotImplementedError(
3196 "Dialect '%s' does not support empty set expression."
3197 % self.dialect.name
3198 )
3200 def _literal_execute_expanding_parameter_literal_binds(
3201 self, parameter, values, bind_expression_template=None
3202 ):
3203 typ_dialect_impl = parameter.type._unwrapped_dialect_impl(self.dialect)
3205 if not values:
3206 # empty IN expression. note we don't need to use
3207 # bind_expression_template here because there are no
3208 # expressions to render.
3210 if typ_dialect_impl._is_tuple_type:
3211 replacement_expression = (
3212 "VALUES " if self.dialect.tuple_in_values else ""
3213 ) + self.visit_empty_set_op_expr(
3214 parameter.type.types, parameter.expand_op
3215 )
3217 else:
3218 replacement_expression = self.visit_empty_set_op_expr(
3219 [parameter.type], parameter.expand_op
3220 )
3222 elif typ_dialect_impl._is_tuple_type or (
3223 typ_dialect_impl._isnull
3224 and isinstance(values[0], collections_abc.Sequence)
3225 and not isinstance(values[0], (str, bytes))
3226 ):
3227 if typ_dialect_impl._has_bind_expression:
3228 raise NotImplementedError(
3229 "bind_expression() on TupleType not supported with "
3230 "literal_binds"
3231 )
3233 replacement_expression = (
3234 "VALUES " if self.dialect.tuple_in_values else ""
3235 ) + ", ".join(
3236 "(%s)"
3237 % (
3238 ", ".join(
3239 self.render_literal_value(value, param_type)
3240 for value, param_type in zip(
3241 tuple_element, parameter.type.types
3242 )
3243 )
3244 )
3245 for i, tuple_element in enumerate(values)
3246 )
3247 else:
3248 if bind_expression_template:
3249 post_compile_pattern = self._post_compile_pattern
3250 m = post_compile_pattern.search(bind_expression_template)
3251 assert m and m.group(
3252 2
3253 ), "unexpected format for expanding parameter"
3255 tok = m.group(2).split("~~")
3256 be_left, be_right = tok[1], tok[3]
3257 replacement_expression = ", ".join(
3258 "%s%s%s"
3259 % (
3260 be_left,
3261 self.render_literal_value(value, parameter.type),
3262 be_right,
3263 )
3264 for value in values
3265 )
3266 else:
3267 replacement_expression = ", ".join(
3268 self.render_literal_value(value, parameter.type)
3269 for value in values
3270 )
3272 return (), replacement_expression
3274 def _literal_execute_expanding_parameter(self, name, parameter, values):
3275 if parameter.literal_execute:
3276 return self._literal_execute_expanding_parameter_literal_binds(
3277 parameter, values
3278 )
3280 dialect = self.dialect
3281 typ_dialect_impl = parameter.type._unwrapped_dialect_impl(dialect)
3283 if self._numeric_binds:
3284 bind_template = self.compilation_bindtemplate
3285 else:
3286 bind_template = self.bindtemplate
3288 if (
3289 self.dialect._bind_typing_render_casts
3290 and typ_dialect_impl.render_bind_cast
3291 ):
3293 def _render_bindtemplate(name):
3294 return self.render_bind_cast(
3295 parameter.type,
3296 typ_dialect_impl,
3297 bind_template % {"name": name},
3298 )
3300 else:
3302 def _render_bindtemplate(name):
3303 return bind_template % {"name": name}
3305 if not values:
3306 to_update = []
3307 if typ_dialect_impl._is_tuple_type:
3308 replacement_expression = self.visit_empty_set_op_expr(
3309 parameter.type.types, parameter.expand_op
3310 )
3311 else:
3312 replacement_expression = self.visit_empty_set_op_expr(
3313 [parameter.type], parameter.expand_op
3314 )
3316 elif typ_dialect_impl._is_tuple_type or (
3317 typ_dialect_impl._isnull
3318 and isinstance(values[0], collections_abc.Sequence)
3319 and not isinstance(values[0], (str, bytes))
3320 ):
3321 assert not typ_dialect_impl._is_array
3322 to_update = [
3323 ("%s_%s_%s" % (name, i, j), value)
3324 for i, tuple_element in enumerate(values, 1)
3325 for j, value in enumerate(tuple_element, 1)
3326 ]
3328 replacement_expression = (
3329 "VALUES " if dialect.tuple_in_values else ""
3330 ) + ", ".join(
3331 "(%s)"
3332 % (
3333 ", ".join(
3334 _render_bindtemplate(
3335 to_update[i * len(tuple_element) + j][0]
3336 )
3337 for j, value in enumerate(tuple_element)
3338 )
3339 )
3340 for i, tuple_element in enumerate(values)
3341 )
3342 else:
3343 to_update = [
3344 ("%s_%s" % (name, i), value)
3345 for i, value in enumerate(values, 1)
3346 ]
3347 replacement_expression = ", ".join(
3348 _render_bindtemplate(key) for key, value in to_update
3349 )
3351 return to_update, replacement_expression
3353 def visit_binary(
3354 self,
3355 binary,
3356 override_operator=None,
3357 eager_grouping=False,
3358 from_linter=None,
3359 lateral_from_linter=None,
3360 **kw,
3361 ):
3362 if from_linter and operators.is_comparison(binary.operator):
3363 if lateral_from_linter is not None:
3364 enclosing_lateral = kw["enclosing_lateral"]
3365 lateral_from_linter.edges.update(
3366 itertools.product(
3367 _de_clone(
3368 binary.left._from_objects + [enclosing_lateral]
3369 ),
3370 _de_clone(
3371 binary.right._from_objects + [enclosing_lateral]
3372 ),
3373 )
3374 )
3375 else:
3376 from_linter.edges.update(
3377 itertools.product(
3378 _de_clone(binary.left._from_objects),
3379 _de_clone(binary.right._from_objects),
3380 )
3381 )
3383 # don't allow "? = ?" to render
3384 if (
3385 self.ansi_bind_rules
3386 and isinstance(binary.left, elements.BindParameter)
3387 and isinstance(binary.right, elements.BindParameter)
3388 ):
3389 kw["literal_execute"] = True
3391 operator_ = override_operator or binary.operator
3392 disp = self._get_operator_dispatch(operator_, "binary", None)
3393 if disp:
3394 return disp(binary, operator_, **kw)
3395 else:
3396 try:
3397 opstring = OPERATORS[operator_]
3398 except KeyError as err:
3399 raise exc.UnsupportedCompilationError(self, operator_) from err
3400 else:
3401 return self._generate_generic_binary(
3402 binary,
3403 opstring,
3404 from_linter=from_linter,
3405 lateral_from_linter=lateral_from_linter,
3406 **kw,
3407 )
3409 def visit_function_as_comparison_op_binary(self, element, operator, **kw):
3410 return self.process(element.sql_function, **kw)
3412 def visit_mod_binary(self, binary, operator, **kw):
3413 if self.preparer._double_percents:
3414 return (
3415 self.process(binary.left, **kw)
3416 + " %% "
3417 + self.process(binary.right, **kw)
3418 )
3419 else:
3420 return (
3421 self.process(binary.left, **kw)
3422 + " % "
3423 + self.process(binary.right, **kw)
3424 )
3426 def visit_custom_op_binary(self, element, operator, **kw):
3427 kw["eager_grouping"] = operator.eager_grouping
3428 return self._generate_generic_binary(
3429 element,
3430 " " + self.escape_literal_column(operator.opstring) + " ",
3431 **kw,
3432 )
3434 def visit_custom_op_unary_operator(self, element, operator, **kw):
3435 return self._generate_generic_unary_operator(
3436 element, self.escape_literal_column(operator.opstring) + " ", **kw
3437 )
3439 def visit_custom_op_unary_modifier(self, element, operator, **kw):
3440 return self._generate_generic_unary_modifier(
3441 element, " " + self.escape_literal_column(operator.opstring), **kw
3442 )
3444 def _generate_generic_binary(
3445 self,
3446 binary: BinaryExpression[Any],
3447 opstring: str,
3448 eager_grouping: bool = False,
3449 **kw: Any,
3450 ) -> str:
3451 _in_operator_expression = kw.get("_in_operator_expression", False)
3453 kw["_in_operator_expression"] = True
3454 kw["_binary_op"] = binary.operator
3455 text = (
3456 binary.left._compiler_dispatch(
3457 self, eager_grouping=eager_grouping, **kw
3458 )
3459 + opstring
3460 + binary.right._compiler_dispatch(
3461 self, eager_grouping=eager_grouping, **kw
3462 )
3463 )
3465 if _in_operator_expression and eager_grouping:
3466 text = "(%s)" % text
3467 return text
3469 def _generate_generic_unary_operator(self, unary, opstring, **kw):
3470 return opstring + unary.element._compiler_dispatch(self, **kw)
3472 def _generate_generic_unary_modifier(self, unary, opstring, **kw):
3473 return unary.element._compiler_dispatch(self, **kw) + opstring
3475 @util.memoized_property
3476 def _like_percent_literal(self):
3477 return elements.literal_column("'%'", type_=sqltypes.STRINGTYPE)
3479 def visit_ilike_case_insensitive_operand(self, element, **kw):
3480 return f"lower({element.element._compiler_dispatch(self, **kw)})"
3482 def visit_contains_op_binary(self, binary, operator, **kw):
3483 binary = binary._clone()
3484 percent = self._like_percent_literal
3485 binary.right = percent.concat(binary.right).concat(percent)
3486 return self.visit_like_op_binary(binary, operator, **kw)
3488 def visit_not_contains_op_binary(self, binary, operator, **kw):
3489 binary = binary._clone()
3490 percent = self._like_percent_literal
3491 binary.right = percent.concat(binary.right).concat(percent)
3492 return self.visit_not_like_op_binary(binary, operator, **kw)
3494 def visit_icontains_op_binary(self, binary, operator, **kw):
3495 binary = binary._clone()
3496 percent = self._like_percent_literal
3497 binary.left = ilike_case_insensitive(binary.left)
3498 binary.right = percent.concat(
3499 ilike_case_insensitive(binary.right)
3500 ).concat(percent)
3501 return self.visit_ilike_op_binary(binary, operator, **kw)
3503 def visit_not_icontains_op_binary(self, binary, operator, **kw):
3504 binary = binary._clone()
3505 percent = self._like_percent_literal
3506 binary.left = ilike_case_insensitive(binary.left)
3507 binary.right = percent.concat(
3508 ilike_case_insensitive(binary.right)
3509 ).concat(percent)
3510 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3512 def visit_startswith_op_binary(self, binary, operator, **kw):
3513 binary = binary._clone()
3514 percent = self._like_percent_literal
3515 binary.right = percent._rconcat(binary.right)
3516 return self.visit_like_op_binary(binary, operator, **kw)
3518 def visit_not_startswith_op_binary(self, binary, operator, **kw):
3519 binary = binary._clone()
3520 percent = self._like_percent_literal
3521 binary.right = percent._rconcat(binary.right)
3522 return self.visit_not_like_op_binary(binary, operator, **kw)
3524 def visit_istartswith_op_binary(self, binary, operator, **kw):
3525 binary = binary._clone()
3526 percent = self._like_percent_literal
3527 binary.left = ilike_case_insensitive(binary.left)
3528 binary.right = percent._rconcat(ilike_case_insensitive(binary.right))
3529 return self.visit_ilike_op_binary(binary, operator, **kw)
3531 def visit_not_istartswith_op_binary(self, binary, operator, **kw):
3532 binary = binary._clone()
3533 percent = self._like_percent_literal
3534 binary.left = ilike_case_insensitive(binary.left)
3535 binary.right = percent._rconcat(ilike_case_insensitive(binary.right))
3536 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3538 def visit_endswith_op_binary(self, binary, operator, **kw):
3539 binary = binary._clone()
3540 percent = self._like_percent_literal
3541 binary.right = percent.concat(binary.right)
3542 return self.visit_like_op_binary(binary, operator, **kw)
3544 def visit_not_endswith_op_binary(self, binary, operator, **kw):
3545 binary = binary._clone()
3546 percent = self._like_percent_literal
3547 binary.right = percent.concat(binary.right)
3548 return self.visit_not_like_op_binary(binary, operator, **kw)
3550 def visit_iendswith_op_binary(self, binary, operator, **kw):
3551 binary = binary._clone()
3552 percent = self._like_percent_literal
3553 binary.left = ilike_case_insensitive(binary.left)
3554 binary.right = percent.concat(ilike_case_insensitive(binary.right))
3555 return self.visit_ilike_op_binary(binary, operator, **kw)
3557 def visit_not_iendswith_op_binary(self, binary, operator, **kw):
3558 binary = binary._clone()
3559 percent = self._like_percent_literal
3560 binary.left = ilike_case_insensitive(binary.left)
3561 binary.right = percent.concat(ilike_case_insensitive(binary.right))
3562 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3564 def visit_like_op_binary(self, binary, operator, **kw):
3565 escape = binary.modifiers.get("escape", None)
3567 return "%s LIKE %s" % (
3568 binary.left._compiler_dispatch(self, **kw),
3569 binary.right._compiler_dispatch(self, **kw),
3570 ) + (
3571 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
3572 if escape is not None
3573 else ""
3574 )
3576 def visit_not_like_op_binary(self, binary, operator, **kw):
3577 escape = binary.modifiers.get("escape", None)
3578 return "%s NOT LIKE %s" % (
3579 binary.left._compiler_dispatch(self, **kw),
3580 binary.right._compiler_dispatch(self, **kw),
3581 ) + (
3582 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
3583 if escape is not None
3584 else ""
3585 )
3587 def visit_ilike_op_binary(self, binary, operator, **kw):
3588 if operator is operators.ilike_op:
3589 binary = binary._clone()
3590 binary.left = ilike_case_insensitive(binary.left)
3591 binary.right = ilike_case_insensitive(binary.right)
3592 # else we assume ilower() has been applied
3594 return self.visit_like_op_binary(binary, operator, **kw)
3596 def visit_not_ilike_op_binary(self, binary, operator, **kw):
3597 if operator is operators.not_ilike_op:
3598 binary = binary._clone()
3599 binary.left = ilike_case_insensitive(binary.left)
3600 binary.right = ilike_case_insensitive(binary.right)
3601 # else we assume ilower() has been applied
3603 return self.visit_not_like_op_binary(binary, operator, **kw)
3605 def visit_between_op_binary(self, binary, operator, **kw):
3606 symmetric = binary.modifiers.get("symmetric", False)
3607 return self._generate_generic_binary(
3608 binary, " BETWEEN SYMMETRIC " if symmetric else " BETWEEN ", **kw
3609 )
3611 def visit_not_between_op_binary(self, binary, operator, **kw):
3612 symmetric = binary.modifiers.get("symmetric", False)
3613 return self._generate_generic_binary(
3614 binary,
3615 " NOT BETWEEN SYMMETRIC " if symmetric else " NOT BETWEEN ",
3616 **kw,
3617 )
3619 def visit_regexp_match_op_binary(
3620 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3621 ) -> str:
3622 raise exc.CompileError(
3623 "%s dialect does not support regular expressions"
3624 % self.dialect.name
3625 )
3627 def visit_not_regexp_match_op_binary(
3628 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3629 ) -> str:
3630 raise exc.CompileError(
3631 "%s dialect does not support regular expressions"
3632 % self.dialect.name
3633 )
3635 def visit_regexp_replace_op_binary(
3636 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3637 ) -> str:
3638 raise exc.CompileError(
3639 "%s dialect does not support regular expression replacements"
3640 % self.dialect.name
3641 )
3643 def visit_bindparam(
3644 self,
3645 bindparam,
3646 within_columns_clause=False,
3647 literal_binds=False,
3648 skip_bind_expression=False,
3649 literal_execute=False,
3650 render_postcompile=False,
3651 **kwargs,
3652 ):
3654 if not skip_bind_expression:
3655 impl = bindparam.type.dialect_impl(self.dialect)
3656 if impl._has_bind_expression:
3657 bind_expression = impl.bind_expression(bindparam)
3658 wrapped = self.process(
3659 bind_expression,
3660 skip_bind_expression=True,
3661 within_columns_clause=within_columns_clause,
3662 literal_binds=literal_binds and not bindparam.expanding,
3663 literal_execute=literal_execute,
3664 render_postcompile=render_postcompile,
3665 **kwargs,
3666 )
3667 if bindparam.expanding:
3668 # for postcompile w/ expanding, move the "wrapped" part
3669 # of this into the inside
3671 m = re.match(
3672 r"^(.*)\(__\[POSTCOMPILE_(\S+?)\]\)(.*)$", wrapped
3673 )
3674 assert m, "unexpected format for expanding parameter"
3675 wrapped = "(__[POSTCOMPILE_%s~~%s~~REPL~~%s~~])" % (
3676 m.group(2),
3677 m.group(1),
3678 m.group(3),
3679 )
3681 if literal_binds:
3682 ret = self.render_literal_bindparam(
3683 bindparam,
3684 within_columns_clause=True,
3685 bind_expression_template=wrapped,
3686 **kwargs,
3687 )
3688 return f"({ret})"
3690 return wrapped
3692 if not literal_binds:
3693 literal_execute = (
3694 literal_execute
3695 or bindparam.literal_execute
3696 or (within_columns_clause and self.ansi_bind_rules)
3697 )
3698 post_compile = literal_execute or bindparam.expanding
3699 else:
3700 post_compile = False
3702 if literal_binds:
3703 ret = self.render_literal_bindparam(
3704 bindparam, within_columns_clause=True, **kwargs
3705 )
3706 if bindparam.expanding:
3707 ret = f"({ret})"
3708 return ret
3710 name = self._truncate_bindparam(bindparam)
3712 if name in self.binds:
3713 existing = self.binds[name]
3714 if existing is not bindparam:
3715 if (
3716 (existing.unique or bindparam.unique)
3717 and not existing.proxy_set.intersection(
3718 bindparam.proxy_set
3719 )
3720 and not existing._cloned_set.intersection(
3721 bindparam._cloned_set
3722 )
3723 ):
3724 raise exc.CompileError(
3725 "Bind parameter '%s' conflicts with "
3726 "unique bind parameter of the same name" % name
3727 )
3728 elif existing.expanding != bindparam.expanding:
3729 raise exc.CompileError(
3730 "Can't reuse bound parameter name '%s' in both "
3731 "'expanding' (e.g. within an IN expression) and "
3732 "non-expanding contexts. If this parameter is to "
3733 "receive a list/array value, set 'expanding=True' on "
3734 "it for expressions that aren't IN, otherwise use "
3735 "a different parameter name." % (name,)
3736 )
3737 elif existing._is_crud or bindparam._is_crud:
3738 if existing._is_crud and bindparam._is_crud:
3739 # TODO: this condition is not well understood.
3740 # see tests in test/sql/test_update.py
3741 raise exc.CompileError(
3742 "Encountered unsupported case when compiling an "
3743 "INSERT or UPDATE statement. If this is a "
3744 "multi-table "
3745 "UPDATE statement, please provide string-named "
3746 "arguments to the "
3747 "values() method with distinct names; support for "
3748 "multi-table UPDATE statements that "
3749 "target multiple tables for UPDATE is very "
3750 "limited",
3751 )
3752 else:
3753 raise exc.CompileError(
3754 f"bindparam() name '{bindparam.key}' is reserved "
3755 "for automatic usage in the VALUES or SET "
3756 "clause of this "
3757 "insert/update statement. Please use a "
3758 "name other than column name when using "
3759 "bindparam() "
3760 "with insert() or update() (for example, "
3761 f"'b_{bindparam.key}')."
3762 )
3764 self.binds[bindparam.key] = self.binds[name] = bindparam
3766 # if we are given a cache key that we're going to match against,
3767 # relate the bindparam here to one that is most likely present
3768 # in the "extracted params" portion of the cache key. this is used
3769 # to set up a positional mapping that is used to determine the
3770 # correct parameters for a subsequent use of this compiled with
3771 # a different set of parameter values. here, we accommodate for
3772 # parameters that may have been cloned both before and after the cache
3773 # key was been generated.
3774 ckbm_tuple = self._cache_key_bind_match
3776 if ckbm_tuple:
3777 ckbm, cksm = ckbm_tuple
3778 for bp in bindparam._cloned_set:
3779 if bp.key in cksm:
3780 cb = cksm[bp.key]
3781 ckbm[cb].append(bindparam)
3783 if bindparam.isoutparam:
3784 self.has_out_parameters = True
3786 if post_compile:
3787 if render_postcompile:
3788 self._render_postcompile = True
3790 if literal_execute:
3791 self.literal_execute_params |= {bindparam}
3792 else:
3793 self.post_compile_params |= {bindparam}
3795 ret = self.bindparam_string(
3796 name,
3797 post_compile=post_compile,
3798 expanding=bindparam.expanding,
3799 bindparam_type=bindparam.type,
3800 **kwargs,
3801 )
3803 if bindparam.expanding:
3804 ret = f"({ret})"
3806 return ret
3808 def render_bind_cast(self, type_, dbapi_type, sqltext):
3809 raise NotImplementedError()
3811 def render_literal_bindparam(
3812 self,
3813 bindparam,
3814 render_literal_value=NO_ARG,
3815 bind_expression_template=None,
3816 **kw,
3817 ):
3818 if render_literal_value is not NO_ARG:
3819 value = render_literal_value
3820 else:
3821 if bindparam.value is None and bindparam.callable is None:
3822 op = kw.get("_binary_op", None)
3823 if op and op not in (operators.is_, operators.is_not):
3824 util.warn_limited(
3825 "Bound parameter '%s' rendering literal NULL in a SQL "
3826 "expression; comparisons to NULL should not use "
3827 "operators outside of 'is' or 'is not'",
3828 (bindparam.key,),
3829 )
3830 return self.process(sqltypes.NULLTYPE, **kw)
3831 value = bindparam.effective_value
3833 if bindparam.expanding:
3834 leep = self._literal_execute_expanding_parameter_literal_binds
3835 to_update, replacement_expr = leep(
3836 bindparam,
3837 value,
3838 bind_expression_template=bind_expression_template,
3839 )
3840 return replacement_expr
3841 else:
3842 return self.render_literal_value(value, bindparam.type)
3844 def render_literal_value(
3845 self, value: Any, type_: sqltypes.TypeEngine[Any]
3846 ) -> str:
3847 """Render the value of a bind parameter as a quoted literal.
3849 This is used for statement sections that do not accept bind parameters
3850 on the target driver/database.
3852 This should be implemented by subclasses using the quoting services
3853 of the DBAPI.
3855 """
3857 if value is None and not type_.should_evaluate_none:
3858 # issue #10535 - handle NULL in the compiler without placing
3859 # this onto each type, except for "evaluate None" types
3860 # (e.g. JSON)
3861 return self.process(elements.Null._instance())
3863 processor = type_._cached_literal_processor(self.dialect)
3864 if processor:
3865 try:
3866 return processor(value)
3867 except Exception as e:
3868 raise exc.CompileError(
3869 f"Could not render literal value "
3870 f'"{sql_util._repr_single_value(value)}" '
3871 f"with datatype "
3872 f"{type_}; see parent stack trace for "
3873 "more detail."
3874 ) from e
3876 else:
3877 raise exc.CompileError(
3878 f"No literal value renderer is available for literal value "
3879 f'"{sql_util._repr_single_value(value)}" '
3880 f"with datatype {type_}"
3881 )
3883 def _truncate_bindparam(self, bindparam):
3884 if bindparam in self.bind_names:
3885 return self.bind_names[bindparam]
3887 bind_name = bindparam.key
3888 if isinstance(bind_name, elements._truncated_label):
3889 bind_name = self._truncated_identifier("bindparam", bind_name)
3891 # add to bind_names for translation
3892 self.bind_names[bindparam] = bind_name
3894 return bind_name
3896 def _truncated_identifier(
3897 self, ident_class: str, name: _truncated_label
3898 ) -> str:
3899 if (ident_class, name) in self.truncated_names:
3900 return self.truncated_names[(ident_class, name)]
3902 anonname = name.apply_map(self.anon_map)
3904 if len(anonname) > self.label_length - 6:
3905 counter = self._truncated_counters.get(ident_class, 1)
3906 truncname = (
3907 anonname[0 : max(self.label_length - 6, 0)]
3908 + "_"
3909 + hex(counter)[2:]
3910 )
3911 self._truncated_counters[ident_class] = counter + 1
3912 else:
3913 truncname = anonname
3914 self.truncated_names[(ident_class, name)] = truncname
3915 return truncname
3917 def _anonymize(self, name: str) -> str:
3918 return name % self.anon_map
3920 def bindparam_string(
3921 self,
3922 name: str,
3923 post_compile: bool = False,
3924 expanding: bool = False,
3925 escaped_from: Optional[str] = None,
3926 bindparam_type: Optional[TypeEngine[Any]] = None,
3927 accumulate_bind_names: Optional[Set[str]] = None,
3928 visited_bindparam: Optional[List[str]] = None,
3929 **kw: Any,
3930 ) -> str:
3931 # TODO: accumulate_bind_names is passed by crud.py to gather
3932 # names on a per-value basis, visited_bindparam is passed by
3933 # visit_insert() to collect all parameters in the statement.
3934 # see if this gathering can be simplified somehow
3935 if accumulate_bind_names is not None:
3936 accumulate_bind_names.add(name)
3937 if visited_bindparam is not None:
3938 visited_bindparam.append(name)
3940 if not escaped_from:
3941 if self._bind_translate_re.search(name):
3942 # not quite the translate use case as we want to
3943 # also get a quick boolean if we even found
3944 # unusual characters in the name
3945 new_name = self._bind_translate_re.sub(
3946 lambda m: self._bind_translate_chars[m.group(0)],
3947 name,
3948 )
3949 escaped_from = name
3950 name = new_name
3952 if escaped_from:
3953 self.escaped_bind_names = self.escaped_bind_names.union(
3954 {escaped_from: name}
3955 )
3956 if post_compile:
3957 ret = "__[POSTCOMPILE_%s]" % name
3958 if expanding:
3959 # for expanding, bound parameters or literal values will be
3960 # rendered per item
3961 return ret
3963 # otherwise, for non-expanding "literal execute", apply
3964 # bind casts as determined by the datatype
3965 if bindparam_type is not None:
3966 type_impl = bindparam_type._unwrapped_dialect_impl(
3967 self.dialect
3968 )
3969 if type_impl.render_literal_cast:
3970 ret = self.render_bind_cast(bindparam_type, type_impl, ret)
3971 return ret
3972 elif self.state is CompilerState.COMPILING:
3973 ret = self.compilation_bindtemplate % {"name": name}
3974 else:
3975 ret = self.bindtemplate % {"name": name}
3977 if (
3978 bindparam_type is not None
3979 and self.dialect._bind_typing_render_casts
3980 ):
3981 type_impl = bindparam_type._unwrapped_dialect_impl(self.dialect)
3982 if type_impl.render_bind_cast:
3983 ret = self.render_bind_cast(bindparam_type, type_impl, ret)
3985 return ret
3987 def _dispatch_independent_ctes(self, stmt, kw):
3988 local_kw = kw.copy()
3989 local_kw.pop("cte_opts", None)
3990 for cte, opt in zip(
3991 stmt._independent_ctes, stmt._independent_ctes_opts
3992 ):
3993 cte._compiler_dispatch(self, cte_opts=opt, **local_kw)
3995 def visit_cte(
3996 self,
3997 cte: CTE,
3998 asfrom: bool = False,
3999 ashint: bool = False,
4000 fromhints: Optional[_FromHintsType] = None,
4001 visiting_cte: Optional[CTE] = None,
4002 from_linter: Optional[FromLinter] = None,
4003 cte_opts: selectable._CTEOpts = selectable._CTEOpts(False),
4004 **kwargs: Any,
4005 ) -> Optional[str]:
4006 self_ctes = self._init_cte_state()
4007 assert self_ctes is self.ctes
4009 kwargs["visiting_cte"] = cte
4011 cte_name = cte.name
4013 if isinstance(cte_name, elements._truncated_label):
4014 cte_name = self._truncated_identifier("alias", cte_name)
4016 is_new_cte = True
4017 embedded_in_current_named_cte = False
4019 _reference_cte = cte._get_reference_cte()
4021 nesting = cte.nesting or cte_opts.nesting
4023 # check for CTE already encountered
4024 if _reference_cte in self.level_name_by_cte:
4025 cte_level, _, existing_cte_opts = self.level_name_by_cte[
4026 _reference_cte
4027 ]
4028 assert _ == cte_name
4030 cte_level_name = (cte_level, cte_name)
4031 existing_cte = self.ctes_by_level_name[cte_level_name]
4033 # check if we are receiving it here with a specific
4034 # "nest_here" location; if so, move it to this location
4036 if cte_opts.nesting:
4037 if existing_cte_opts.nesting:
4038 raise exc.CompileError(
4039 "CTE is stated as 'nest_here' in "
4040 "more than one location"
4041 )
4043 old_level_name = (cte_level, cte_name)
4044 cte_level = len(self.stack) if nesting else 1
4045 cte_level_name = new_level_name = (cte_level, cte_name)
4047 del self.ctes_by_level_name[old_level_name]
4048 self.ctes_by_level_name[new_level_name] = existing_cte
4049 self.level_name_by_cte[_reference_cte] = new_level_name + (
4050 cte_opts,
4051 )
4053 else:
4054 cte_level = len(self.stack) if nesting else 1
4055 cte_level_name = (cte_level, cte_name)
4057 if cte_level_name in self.ctes_by_level_name:
4058 existing_cte = self.ctes_by_level_name[cte_level_name]
4059 else:
4060 existing_cte = None
4062 if existing_cte is not None:
4063 embedded_in_current_named_cte = visiting_cte is existing_cte
4065 # we've generated a same-named CTE that we are enclosed in,
4066 # or this is the same CTE. just return the name.
4067 if cte is existing_cte._restates or cte is existing_cte:
4068 is_new_cte = False
4069 elif existing_cte is cte._restates:
4070 # we've generated a same-named CTE that is
4071 # enclosed in us - we take precedence, so
4072 # discard the text for the "inner".
4073 del self_ctes[existing_cte]
4075 existing_cte_reference_cte = existing_cte._get_reference_cte()
4077 assert existing_cte_reference_cte is _reference_cte
4078 assert existing_cte_reference_cte is existing_cte
4080 del self.level_name_by_cte[existing_cte_reference_cte]
4081 else:
4082 if (
4083 # if the two CTEs have the same hash, which we expect
4084 # here means that one/both is an annotated of the other
4085 (hash(cte) == hash(existing_cte))
4086 # or...
4087 or (
4088 (
4089 # if they are clones, i.e. they came from the ORM
4090 # or some other visit method
4091 cte._is_clone_of is not None
4092 or existing_cte._is_clone_of is not None
4093 )
4094 # and are deep-copy identical
4095 and cte.compare(existing_cte)
4096 )
4097 ):
4098 # then consider these two CTEs the same
4099 is_new_cte = False
4100 else:
4101 # otherwise these are two CTEs that either will render
4102 # differently, or were indicated separately by the user,
4103 # with the same name
4104 raise exc.CompileError(
4105 "Multiple, unrelated CTEs found with "
4106 "the same name: %r" % cte_name
4107 )
4109 if not asfrom and not is_new_cte:
4110 return None
4112 if cte._cte_alias is not None:
4113 pre_alias_cte = cte._cte_alias
4114 cte_pre_alias_name = cte._cte_alias.name
4115 if isinstance(cte_pre_alias_name, elements._truncated_label):
4116 cte_pre_alias_name = self._truncated_identifier(
4117 "alias", cte_pre_alias_name
4118 )
4119 else:
4120 pre_alias_cte = cte
4121 cte_pre_alias_name = None
4123 if is_new_cte:
4124 self.ctes_by_level_name[cte_level_name] = cte
4125 self.level_name_by_cte[_reference_cte] = cte_level_name + (
4126 cte_opts,
4127 )
4129 if pre_alias_cte not in self.ctes:
4130 self.visit_cte(pre_alias_cte, **kwargs)
4132 if not cte_pre_alias_name and cte not in self_ctes:
4133 if cte.recursive:
4134 self.ctes_recursive = True
4135 text = self.preparer.format_alias(cte, cte_name)
4136 if cte.recursive:
4137 col_source = cte.element
4139 # TODO: can we get at the .columns_plus_names collection
4140 # that is already (or will be?) generated for the SELECT
4141 # rather than calling twice?
4142 recur_cols = [
4143 # TODO: proxy_name is not technically safe,
4144 # see test_cte->
4145 # test_with_recursive_no_name_currently_buggy. not
4146 # clear what should be done with such a case
4147 fallback_label_name or proxy_name
4148 for (
4149 _,
4150 proxy_name,
4151 fallback_label_name,
4152 c,
4153 repeated,
4154 ) in (col_source._generate_columns_plus_names(True))
4155 if not repeated
4156 ]
4158 text += "(%s)" % (
4159 ", ".join(
4160 self.preparer.format_label_name(
4161 ident, anon_map=self.anon_map
4162 )
4163 for ident in recur_cols
4164 )
4165 )
4167 assert kwargs.get("subquery", False) is False
4169 if not self.stack:
4170 # toplevel, this is a stringify of the
4171 # cte directly. just compile the inner
4172 # the way alias() does.
4173 return cte.element._compiler_dispatch(
4174 self, asfrom=asfrom, **kwargs
4175 )
4176 else:
4177 prefixes = self._generate_prefixes(
4178 cte, cte._prefixes, **kwargs
4179 )
4180 inner = cte.element._compiler_dispatch(
4181 self, asfrom=True, **kwargs
4182 )
4184 text += " AS %s\n(%s)" % (prefixes, inner)
4186 if cte._suffixes:
4187 text += " " + self._generate_prefixes(
4188 cte, cte._suffixes, **kwargs
4189 )
4191 self_ctes[cte] = text
4193 if asfrom:
4194 if from_linter:
4195 from_linter.froms[cte._de_clone()] = cte_name
4197 if not is_new_cte and embedded_in_current_named_cte:
4198 return self.preparer.format_alias(cte, cte_name)
4200 if cte_pre_alias_name:
4201 text = self.preparer.format_alias(cte, cte_pre_alias_name)
4202 if self.preparer._requires_quotes(cte_name):
4203 cte_name = self.preparer.quote(cte_name)
4204 text += self.get_render_as_alias_suffix(cte_name)
4205 return text # type: ignore[no-any-return]
4206 else:
4207 return self.preparer.format_alias(cte, cte_name)
4209 return None
4211 def visit_table_valued_alias(self, element, **kw):
4212 if element.joins_implicitly:
4213 kw["from_linter"] = None
4214 if element._is_lateral:
4215 return self.visit_lateral(element, **kw)
4216 else:
4217 return self.visit_alias(element, **kw)
4219 def visit_table_valued_column(self, element, **kw):
4220 return self.visit_column(element, **kw)
4222 def visit_alias(
4223 self,
4224 alias,
4225 asfrom=False,
4226 ashint=False,
4227 iscrud=False,
4228 fromhints=None,
4229 subquery=False,
4230 lateral=False,
4231 enclosing_alias=None,
4232 from_linter=None,
4233 **kwargs,
4234 ):
4235 if lateral:
4236 if "enclosing_lateral" not in kwargs:
4237 # if lateral is set and enclosing_lateral is not
4238 # present, we assume we are being called directly
4239 # from visit_lateral() and we need to set enclosing_lateral.
4240 assert alias._is_lateral
4241 kwargs["enclosing_lateral"] = alias
4243 # for lateral objects, we track a second from_linter that is...
4244 # lateral! to the level above us.
4245 if (
4246 from_linter
4247 and "lateral_from_linter" not in kwargs
4248 and "enclosing_lateral" in kwargs
4249 ):
4250 kwargs["lateral_from_linter"] = from_linter
4252 if enclosing_alias is not None and enclosing_alias.element is alias:
4253 inner = alias.element._compiler_dispatch(
4254 self,
4255 asfrom=asfrom,
4256 ashint=ashint,
4257 iscrud=iscrud,
4258 fromhints=fromhints,
4259 lateral=lateral,
4260 enclosing_alias=alias,
4261 **kwargs,
4262 )
4263 if subquery and (asfrom or lateral):
4264 inner = "(%s)" % (inner,)
4265 return inner
4266 else:
4267 kwargs["enclosing_alias"] = alias
4269 if asfrom or ashint:
4270 if isinstance(alias.name, elements._truncated_label):
4271 alias_name = self._truncated_identifier("alias", alias.name)
4272 else:
4273 alias_name = alias.name
4275 if ashint:
4276 return self.preparer.format_alias(alias, alias_name)
4277 elif asfrom:
4278 if from_linter:
4279 from_linter.froms[alias._de_clone()] = alias_name
4281 inner = alias.element._compiler_dispatch(
4282 self, asfrom=True, lateral=lateral, **kwargs
4283 )
4284 if subquery:
4285 inner = "(%s)" % (inner,)
4287 ret = inner + self.get_render_as_alias_suffix(
4288 self.preparer.format_alias(alias, alias_name)
4289 )
4291 if alias._supports_derived_columns and alias._render_derived:
4292 ret += "(%s)" % (
4293 ", ".join(
4294 "%s%s"
4295 % (
4296 self.preparer.quote(col.name),
4297 (
4298 " %s"
4299 % self.dialect.type_compiler_instance.process(
4300 col.type, **kwargs
4301 )
4302 if alias._render_derived_w_types
4303 else ""
4304 ),
4305 )
4306 for col in alias.c
4307 )
4308 )
4310 if fromhints and alias in fromhints:
4311 ret = self.format_from_hint_text(
4312 ret, alias, fromhints[alias], iscrud
4313 )
4315 return ret
4316 else:
4317 # note we cancel the "subquery" flag here as well
4318 return alias.element._compiler_dispatch(
4319 self, lateral=lateral, **kwargs
4320 )
4322 def visit_subquery(self, subquery, **kw):
4323 kw["subquery"] = True
4324 return self.visit_alias(subquery, **kw)
4326 def visit_lateral(self, lateral_, **kw):
4327 kw["lateral"] = True
4328 return "LATERAL %s" % self.visit_alias(lateral_, **kw)
4330 def visit_tablesample(self, tablesample, asfrom=False, **kw):
4331 text = "%s TABLESAMPLE %s" % (
4332 self.visit_alias(tablesample, asfrom=True, **kw),
4333 tablesample._get_method()._compiler_dispatch(self, **kw),
4334 )
4336 if tablesample.seed is not None:
4337 text += " REPEATABLE (%s)" % (
4338 tablesample.seed._compiler_dispatch(self, **kw)
4339 )
4341 return text
4343 def _render_values(self, element, **kw):
4344 kw.setdefault("literal_binds", element.literal_binds)
4345 tuples = ", ".join(
4346 self.process(
4347 elements.Tuple(
4348 types=element._column_types, *elem
4349 ).self_group(),
4350 **kw,
4351 )
4352 for chunk in element._data
4353 for elem in chunk
4354 )
4355 return f"VALUES {tuples}"
4357 def visit_values(self, element, asfrom=False, from_linter=None, **kw):
4358 v = self._render_values(element, **kw)
4360 if element._unnamed:
4361 name = None
4362 elif isinstance(element.name, elements._truncated_label):
4363 name = self._truncated_identifier("values", element.name)
4364 else:
4365 name = element.name
4367 if element._is_lateral:
4368 lateral = "LATERAL "
4369 else:
4370 lateral = ""
4372 if asfrom:
4373 if from_linter:
4374 from_linter.froms[element._de_clone()] = (
4375 name if name is not None else "(unnamed VALUES element)"
4376 )
4378 if name:
4379 kw["include_table"] = False
4380 v = "%s(%s)%s (%s)" % (
4381 lateral,
4382 v,
4383 self.get_render_as_alias_suffix(self.preparer.quote(name)),
4384 (
4385 ", ".join(
4386 c._compiler_dispatch(self, **kw)
4387 for c in element.columns
4388 )
4389 ),
4390 )
4391 else:
4392 v = "%s(%s)" % (lateral, v)
4393 return v
4395 def visit_scalar_values(self, element, **kw):
4396 return f"({self._render_values(element, **kw)})"
4398 def get_render_as_alias_suffix(self, alias_name_text):
4399 return " AS " + alias_name_text
4401 def _add_to_result_map(
4402 self,
4403 keyname: str,
4404 name: str,
4405 objects: Tuple[Any, ...],
4406 type_: TypeEngine[Any],
4407 ) -> None:
4409 # note objects must be non-empty for cursor.py to handle the
4410 # collection properly
4411 assert objects
4413 if keyname is None or keyname == "*":
4414 self._ordered_columns = False
4415 self._ad_hoc_textual = True
4416 if type_._is_tuple_type:
4417 raise exc.CompileError(
4418 "Most backends don't support SELECTing "
4419 "from a tuple() object. If this is an ORM query, "
4420 "consider using the Bundle object."
4421 )
4422 self._result_columns.append(
4423 ResultColumnsEntry(keyname, name, objects, type_)
4424 )
4426 def _label_returning_column(
4427 self, stmt, column, populate_result_map, column_clause_args=None, **kw
4428 ):
4429 """Render a column with necessary labels inside of a RETURNING clause.
4431 This method is provided for individual dialects in place of calling
4432 the _label_select_column method directly, so that the two use cases
4433 of RETURNING vs. SELECT can be disambiguated going forward.
4435 .. versionadded:: 1.4.21
4437 """
4438 return self._label_select_column(
4439 None,
4440 column,
4441 populate_result_map,
4442 False,
4443 {} if column_clause_args is None else column_clause_args,
4444 **kw,
4445 )
4447 def _label_select_column(
4448 self,
4449 select,
4450 column,
4451 populate_result_map,
4452 asfrom,
4453 column_clause_args,
4454 name=None,
4455 proxy_name=None,
4456 fallback_label_name=None,
4457 within_columns_clause=True,
4458 column_is_repeated=False,
4459 need_column_expressions=False,
4460 include_table=True,
4461 ):
4462 """produce labeled columns present in a select()."""
4463 impl = column.type.dialect_impl(self.dialect)
4465 if impl._has_column_expression and (
4466 need_column_expressions or populate_result_map
4467 ):
4468 col_expr = impl.column_expression(column)
4469 else:
4470 col_expr = column
4472 if populate_result_map:
4473 # pass an "add_to_result_map" callable into the compilation
4474 # of embedded columns. this collects information about the
4475 # column as it will be fetched in the result and is coordinated
4476 # with cursor.description when the query is executed.
4477 add_to_result_map = self._add_to_result_map
4479 # if the SELECT statement told us this column is a repeat,
4480 # wrap the callable with one that prevents the addition of the
4481 # targets
4482 if column_is_repeated:
4483 _add_to_result_map = add_to_result_map
4485 def add_to_result_map(keyname, name, objects, type_):
4486 _add_to_result_map(keyname, name, (keyname,), type_)
4488 # if we redefined col_expr for type expressions, wrap the
4489 # callable with one that adds the original column to the targets
4490 elif col_expr is not column:
4491 _add_to_result_map = add_to_result_map
4493 def add_to_result_map(keyname, name, objects, type_):
4494 _add_to_result_map(
4495 keyname, name, (column,) + objects, type_
4496 )
4498 else:
4499 add_to_result_map = None
4501 # this method is used by some of the dialects for RETURNING,
4502 # which has different inputs. _label_returning_column was added
4503 # as the better target for this now however for 1.4 we will keep
4504 # _label_select_column directly compatible with this use case.
4505 # these assertions right now set up the current expected inputs
4506 assert within_columns_clause, (
4507 "_label_select_column is only relevant within "
4508 "the columns clause of a SELECT or RETURNING"
4509 )
4510 if isinstance(column, elements.Label):
4511 if col_expr is not column:
4512 result_expr = _CompileLabel(
4513 col_expr, column.name, alt_names=(column.element,)
4514 )
4515 else:
4516 result_expr = col_expr
4518 elif name:
4519 # here, _columns_plus_names has determined there's an explicit
4520 # label name we need to use. this is the default for
4521 # tablenames_plus_columnnames as well as when columns are being
4522 # deduplicated on name
4524 assert (
4525 proxy_name is not None
4526 ), "proxy_name is required if 'name' is passed"
4528 result_expr = _CompileLabel(
4529 col_expr,
4530 name,
4531 alt_names=(
4532 proxy_name,
4533 # this is a hack to allow legacy result column lookups
4534 # to work as they did before; this goes away in 2.0.
4535 # TODO: this only seems to be tested indirectly
4536 # via test/orm/test_deprecations.py. should be a
4537 # resultset test for this
4538 column._tq_label,
4539 ),
4540 )
4541 else:
4542 # determine here whether this column should be rendered in
4543 # a labelled context or not, as we were given no required label
4544 # name from the caller. Here we apply heuristics based on the kind
4545 # of SQL expression involved.
4547 if col_expr is not column:
4548 # type-specific expression wrapping the given column,
4549 # so we render a label
4550 render_with_label = True
4551 elif isinstance(column, elements.ColumnClause):
4552 # table-bound column, we render its name as a label if we are
4553 # inside of a subquery only
4554 render_with_label = (
4555 asfrom
4556 and not column.is_literal
4557 and column.table is not None
4558 )
4559 elif isinstance(column, elements.TextClause):
4560 render_with_label = False
4561 elif isinstance(column, elements.UnaryExpression):
4562 # unary expression. notes added as of #12681
4563 #
4564 # By convention, the visit_unary() method
4565 # itself does not add an entry to the result map, and relies
4566 # upon either the inner expression creating a result map
4567 # entry, or if not, by creating a label here that produces
4568 # the result map entry. Where that happens is based on whether
4569 # or not the element immediately inside the unary is a
4570 # NamedColumn subclass or not.
4571 #
4572 # Now, this also impacts how the SELECT is written; if
4573 # we decide to generate a label here, we get the usual
4574 # "~(x+y) AS anon_1" thing in the columns clause. If we
4575 # don't, we don't get an AS at all, we get like
4576 # "~table.column".
4577 #
4578 # But here is the important thing as of modernish (like 1.4)
4579 # versions of SQLAlchemy - **whether or not the AS <label>
4580 # is present in the statement is not actually important**.
4581 # We target result columns **positionally** for a fully
4582 # compiled ``Select()`` object; before 1.4 we needed those
4583 # labels to match in cursor.description etc etc but now it
4584 # really doesn't matter.
4585 # So really, we could set render_with_label True in all cases.
4586 # Or we could just have visit_unary() populate the result map
4587 # in all cases.
4588 #
4589 # What we're doing here is strictly trying to not rock the
4590 # boat too much with when we do/don't render "AS label";
4591 # labels being present helps in the edge cases that we
4592 # "fall back" to named cursor.description matching, labels
4593 # not being present for columns keeps us from having awkward
4594 # phrases like "SELECT DISTINCT table.x AS x".
4595 render_with_label = (
4596 (
4597 # exception case to detect if we render "not boolean"
4598 # as "not <col>" for native boolean or "<col> = 1"
4599 # for non-native boolean. this is controlled by
4600 # visit_is_<true|false>_unary_operator
4601 column.operator
4602 in (operators.is_false, operators.is_true)
4603 and not self.dialect.supports_native_boolean
4604 )
4605 or column._wraps_unnamed_column()
4606 or asfrom
4607 )
4608 elif (
4609 # general class of expressions that don't have a SQL-column
4610 # addressible name. includes scalar selects, bind parameters,
4611 # SQL functions, others
4612 not isinstance(column, elements.NamedColumn)
4613 # deeper check that indicates there's no natural "name" to
4614 # this element, which accommodates for custom SQL constructs
4615 # that might have a ".name" attribute (but aren't SQL
4616 # functions) but are not implementing this more recently added
4617 # base class. in theory the "NamedColumn" check should be
4618 # enough, however here we seek to maintain legacy behaviors
4619 # as well.
4620 and column._non_anon_label is None
4621 ):
4622 render_with_label = True
4623 else:
4624 render_with_label = False
4626 if render_with_label:
4627 if not fallback_label_name:
4628 # used by the RETURNING case right now. we generate it
4629 # here as 3rd party dialects may be referring to
4630 # _label_select_column method directly instead of the
4631 # just-added _label_returning_column method
4632 assert not column_is_repeated
4633 fallback_label_name = column._anon_name_label
4635 fallback_label_name = (
4636 elements._truncated_label(fallback_label_name)
4637 if not isinstance(
4638 fallback_label_name, elements._truncated_label
4639 )
4640 else fallback_label_name
4641 )
4643 result_expr = _CompileLabel(
4644 col_expr, fallback_label_name, alt_names=(proxy_name,)
4645 )
4646 else:
4647 result_expr = col_expr
4649 column_clause_args.update(
4650 within_columns_clause=within_columns_clause,
4651 add_to_result_map=add_to_result_map,
4652 include_table=include_table,
4653 )
4654 return result_expr._compiler_dispatch(self, **column_clause_args)
4656 def format_from_hint_text(self, sqltext, table, hint, iscrud):
4657 hinttext = self.get_from_hint_text(table, hint)
4658 if hinttext:
4659 sqltext += " " + hinttext
4660 return sqltext
4662 def get_select_hint_text(self, byfroms):
4663 return None
4665 def get_from_hint_text(
4666 self, table: FromClause, text: Optional[str]
4667 ) -> Optional[str]:
4668 return None
4670 def get_crud_hint_text(self, table, text):
4671 return None
4673 def get_statement_hint_text(self, hint_texts):
4674 return " ".join(hint_texts)
4676 _default_stack_entry: _CompilerStackEntry
4678 if not typing.TYPE_CHECKING:
4679 _default_stack_entry = util.immutabledict(
4680 [("correlate_froms", frozenset()), ("asfrom_froms", frozenset())]
4681 )
4683 def _display_froms_for_select(
4684 self, select_stmt, asfrom, lateral=False, **kw
4685 ):
4686 # utility method to help external dialects
4687 # get the correct from list for a select.
4688 # specifically the oracle dialect needs this feature
4689 # right now.
4690 toplevel = not self.stack
4691 entry = self._default_stack_entry if toplevel else self.stack[-1]
4693 compile_state = select_stmt._compile_state_factory(select_stmt, self)
4695 correlate_froms = entry["correlate_froms"]
4696 asfrom_froms = entry["asfrom_froms"]
4698 if asfrom and not lateral:
4699 froms = compile_state._get_display_froms(
4700 explicit_correlate_froms=correlate_froms.difference(
4701 asfrom_froms
4702 ),
4703 implicit_correlate_froms=(),
4704 )
4705 else:
4706 froms = compile_state._get_display_froms(
4707 explicit_correlate_froms=correlate_froms,
4708 implicit_correlate_froms=asfrom_froms,
4709 )
4710 return froms
4712 translate_select_structure: Any = None
4713 """if not ``None``, should be a callable which accepts ``(select_stmt,
4714 **kw)`` and returns a select object. this is used for structural changes
4715 mostly to accommodate for LIMIT/OFFSET schemes
4717 """
4719 def visit_select(
4720 self,
4721 select_stmt,
4722 asfrom=False,
4723 insert_into=False,
4724 fromhints=None,
4725 compound_index=None,
4726 select_wraps_for=None,
4727 lateral=False,
4728 from_linter=None,
4729 **kwargs,
4730 ):
4731 assert select_wraps_for is None, (
4732 "SQLAlchemy 1.4 requires use of "
4733 "the translate_select_structure hook for structural "
4734 "translations of SELECT objects"
4735 )
4737 # initial setup of SELECT. the compile_state_factory may now
4738 # be creating a totally different SELECT from the one that was
4739 # passed in. for ORM use this will convert from an ORM-state
4740 # SELECT to a regular "Core" SELECT. other composed operations
4741 # such as computation of joins will be performed.
4743 kwargs["within_columns_clause"] = False
4745 compile_state = select_stmt._compile_state_factory(
4746 select_stmt, self, **kwargs
4747 )
4748 kwargs["ambiguous_table_name_map"] = (
4749 compile_state._ambiguous_table_name_map
4750 )
4752 select_stmt = compile_state.statement
4754 toplevel = not self.stack
4756 if toplevel and not self.compile_state:
4757 self.compile_state = compile_state
4759 is_embedded_select = compound_index is not None or insert_into
4761 # translate step for Oracle, SQL Server which often need to
4762 # restructure the SELECT to allow for LIMIT/OFFSET and possibly
4763 # other conditions
4764 if self.translate_select_structure:
4765 new_select_stmt = self.translate_select_structure(
4766 select_stmt, asfrom=asfrom, **kwargs
4767 )
4769 # if SELECT was restructured, maintain a link to the originals
4770 # and assemble a new compile state
4771 if new_select_stmt is not select_stmt:
4772 compile_state_wraps_for = compile_state
4773 select_wraps_for = select_stmt
4774 select_stmt = new_select_stmt
4776 compile_state = select_stmt._compile_state_factory(
4777 select_stmt, self, **kwargs
4778 )
4779 select_stmt = compile_state.statement
4781 entry = self._default_stack_entry if toplevel else self.stack[-1]
4783 populate_result_map = need_column_expressions = (
4784 toplevel
4785 or entry.get("need_result_map_for_compound", False)
4786 or entry.get("need_result_map_for_nested", False)
4787 )
4789 # indicates there is a CompoundSelect in play and we are not the
4790 # first select
4791 if compound_index:
4792 populate_result_map = False
4794 # this was first proposed as part of #3372; however, it is not
4795 # reached in current tests and could possibly be an assertion
4796 # instead.
4797 if not populate_result_map and "add_to_result_map" in kwargs:
4798 del kwargs["add_to_result_map"]
4800 froms = self._setup_select_stack(
4801 select_stmt, compile_state, entry, asfrom, lateral, compound_index
4802 )
4804 column_clause_args = kwargs.copy()
4805 column_clause_args.update(
4806 {"within_label_clause": False, "within_columns_clause": False}
4807 )
4809 text = "SELECT " # we're off to a good start !
4811 if select_stmt._post_select_clause is not None:
4812 psc = self.process(select_stmt._post_select_clause, **kwargs)
4813 if psc is not None:
4814 text += psc + " "
4816 if select_stmt._hints:
4817 hint_text, byfrom = self._setup_select_hints(select_stmt)
4818 if hint_text:
4819 text += hint_text + " "
4820 else:
4821 byfrom = None
4823 if select_stmt._independent_ctes:
4824 self._dispatch_independent_ctes(select_stmt, kwargs)
4826 if select_stmt._prefixes:
4827 text += self._generate_prefixes(
4828 select_stmt, select_stmt._prefixes, **kwargs
4829 )
4831 text += self.get_select_precolumns(select_stmt, **kwargs)
4833 if select_stmt._pre_columns_clause is not None:
4834 pcc = self.process(select_stmt._pre_columns_clause, **kwargs)
4835 if pcc is not None:
4836 text += pcc + " "
4838 # the actual list of columns to print in the SELECT column list.
4839 inner_columns = [
4840 c
4841 for c in [
4842 self._label_select_column(
4843 select_stmt,
4844 column,
4845 populate_result_map,
4846 asfrom,
4847 column_clause_args,
4848 name=name,
4849 proxy_name=proxy_name,
4850 fallback_label_name=fallback_label_name,
4851 column_is_repeated=repeated,
4852 need_column_expressions=need_column_expressions,
4853 )
4854 for (
4855 name,
4856 proxy_name,
4857 fallback_label_name,
4858 column,
4859 repeated,
4860 ) in compile_state.columns_plus_names
4861 ]
4862 if c is not None
4863 ]
4865 if populate_result_map and select_wraps_for is not None:
4866 # if this select was generated from translate_select,
4867 # rewrite the targeted columns in the result map
4869 translate = dict(
4870 zip(
4871 [
4872 name
4873 for (
4874 key,
4875 proxy_name,
4876 fallback_label_name,
4877 name,
4878 repeated,
4879 ) in compile_state.columns_plus_names
4880 ],
4881 [
4882 name
4883 for (
4884 key,
4885 proxy_name,
4886 fallback_label_name,
4887 name,
4888 repeated,
4889 ) in compile_state_wraps_for.columns_plus_names
4890 ],
4891 )
4892 )
4894 self._result_columns = [
4895 ResultColumnsEntry(
4896 key, name, tuple(translate.get(o, o) for o in obj), type_
4897 )
4898 for key, name, obj, type_ in self._result_columns
4899 ]
4901 text = self._compose_select_body(
4902 text,
4903 select_stmt,
4904 compile_state,
4905 inner_columns,
4906 froms,
4907 byfrom,
4908 toplevel,
4909 kwargs,
4910 )
4912 if select_stmt._post_body_clause is not None:
4913 pbc = self.process(select_stmt._post_body_clause, **kwargs)
4914 if pbc:
4915 text += " " + pbc
4917 if select_stmt._statement_hints:
4918 per_dialect = [
4919 ht
4920 for (dialect_name, ht) in select_stmt._statement_hints
4921 if dialect_name in ("*", self.dialect.name)
4922 ]
4923 if per_dialect:
4924 text += " " + self.get_statement_hint_text(per_dialect)
4926 # In compound query, CTEs are shared at the compound level
4927 if self.ctes and (not is_embedded_select or toplevel):
4928 nesting_level = len(self.stack) if not toplevel else None
4929 text = self._render_cte_clause(nesting_level=nesting_level) + text
4931 if select_stmt._suffixes:
4932 text += " " + self._generate_prefixes(
4933 select_stmt, select_stmt._suffixes, **kwargs
4934 )
4936 self.stack.pop(-1)
4938 return text
4940 def _setup_select_hints(
4941 self, select: Select[Unpack[TupleAny]]
4942 ) -> Tuple[str, _FromHintsType]:
4943 byfrom = {
4944 from_: hinttext
4945 % {"name": from_._compiler_dispatch(self, ashint=True)}
4946 for (from_, dialect), hinttext in select._hints.items()
4947 if dialect in ("*", self.dialect.name)
4948 }
4949 hint_text = self.get_select_hint_text(byfrom)
4950 return hint_text, byfrom
4952 def _setup_select_stack(
4953 self, select, compile_state, entry, asfrom, lateral, compound_index
4954 ):
4955 correlate_froms = entry["correlate_froms"]
4956 asfrom_froms = entry["asfrom_froms"]
4958 if compound_index == 0:
4959 entry["select_0"] = select
4960 elif compound_index:
4961 select_0 = entry["select_0"]
4962 numcols = len(select_0._all_selected_columns)
4964 if len(compile_state.columns_plus_names) != numcols:
4965 raise exc.CompileError(
4966 "All selectables passed to "
4967 "CompoundSelect must have identical numbers of "
4968 "columns; select #%d has %d columns, select "
4969 "#%d has %d"
4970 % (
4971 1,
4972 numcols,
4973 compound_index + 1,
4974 len(select._all_selected_columns),
4975 )
4976 )
4978 if asfrom and not lateral:
4979 froms = compile_state._get_display_froms(
4980 explicit_correlate_froms=correlate_froms.difference(
4981 asfrom_froms
4982 ),
4983 implicit_correlate_froms=(),
4984 )
4985 else:
4986 froms = compile_state._get_display_froms(
4987 explicit_correlate_froms=correlate_froms,
4988 implicit_correlate_froms=asfrom_froms,
4989 )
4991 new_correlate_froms = set(_from_objects(*froms))
4992 all_correlate_froms = new_correlate_froms.union(correlate_froms)
4994 new_entry: _CompilerStackEntry = {
4995 "asfrom_froms": new_correlate_froms,
4996 "correlate_froms": all_correlate_froms,
4997 "selectable": select,
4998 "compile_state": compile_state,
4999 }
5000 self.stack.append(new_entry)
5002 return froms
5004 def _compose_select_body(
5005 self,
5006 text,
5007 select,
5008 compile_state,
5009 inner_columns,
5010 froms,
5011 byfrom,
5012 toplevel,
5013 kwargs,
5014 ):
5015 text += ", ".join(inner_columns)
5017 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
5018 from_linter = FromLinter({}, set())
5019 warn_linting = self.linting & WARN_LINTING
5020 if toplevel:
5021 self.from_linter = from_linter
5022 else:
5023 from_linter = None
5024 warn_linting = False
5026 # adjust the whitespace for no inner columns, part of #9440,
5027 # so that a no-col SELECT comes out as "SELECT WHERE..." or
5028 # "SELECT FROM ...".
5029 # while it would be better to have built the SELECT starting string
5030 # without trailing whitespace first, then add whitespace only if inner
5031 # cols were present, this breaks compatibility with various custom
5032 # compilation schemes that are currently being tested.
5033 if not inner_columns:
5034 text = text.rstrip()
5036 if froms:
5037 text += " \nFROM "
5039 if select._hints:
5040 text += ", ".join(
5041 [
5042 f._compiler_dispatch(
5043 self,
5044 asfrom=True,
5045 fromhints=byfrom,
5046 from_linter=from_linter,
5047 **kwargs,
5048 )
5049 for f in froms
5050 ]
5051 )
5052 else:
5053 text += ", ".join(
5054 [
5055 f._compiler_dispatch(
5056 self,
5057 asfrom=True,
5058 from_linter=from_linter,
5059 **kwargs,
5060 )
5061 for f in froms
5062 ]
5063 )
5064 else:
5065 text += self.default_from()
5067 if select._where_criteria:
5068 t = self._generate_delimited_and_list(
5069 select._where_criteria, from_linter=from_linter, **kwargs
5070 )
5071 if t:
5072 text += " \nWHERE " + t
5074 if warn_linting:
5075 assert from_linter is not None
5076 from_linter.warn()
5078 if select._group_by_clauses:
5079 text += self.group_by_clause(select, **kwargs)
5081 if select._having_criteria:
5082 t = self._generate_delimited_and_list(
5083 select._having_criteria, **kwargs
5084 )
5085 if t:
5086 text += " \nHAVING " + t
5088 if select._post_criteria_clause is not None:
5089 pcc = self.process(select._post_criteria_clause, **kwargs)
5090 if pcc is not None:
5091 text += " \n" + pcc
5093 if select._order_by_clauses:
5094 text += self.order_by_clause(select, **kwargs)
5096 if select._has_row_limiting_clause:
5097 text += self._row_limit_clause(select, **kwargs)
5099 if select._for_update_arg is not None:
5100 text += self.for_update_clause(select, **kwargs)
5102 return text
5104 def _generate_prefixes(self, stmt, prefixes, **kw):
5105 clause = " ".join(
5106 prefix._compiler_dispatch(self, **kw)
5107 for prefix, dialect_name in prefixes
5108 if dialect_name in (None, "*") or dialect_name == self.dialect.name
5109 )
5110 if clause:
5111 clause += " "
5112 return clause
5114 def _render_cte_clause(
5115 self,
5116 nesting_level=None,
5117 include_following_stack=False,
5118 ):
5119 """
5120 include_following_stack
5121 Also render the nesting CTEs on the next stack. Useful for
5122 SQL structures like UNION or INSERT that can wrap SELECT
5123 statements containing nesting CTEs.
5124 """
5125 if not self.ctes:
5126 return ""
5128 ctes: MutableMapping[CTE, str]
5130 if nesting_level and nesting_level > 1:
5131 ctes = util.OrderedDict()
5132 for cte in list(self.ctes.keys()):
5133 cte_level, cte_name, cte_opts = self.level_name_by_cte[
5134 cte._get_reference_cte()
5135 ]
5136 nesting = cte.nesting or cte_opts.nesting
5137 is_rendered_level = cte_level == nesting_level or (
5138 include_following_stack and cte_level == nesting_level + 1
5139 )
5140 if not (nesting and is_rendered_level):
5141 continue
5143 ctes[cte] = self.ctes[cte]
5145 else:
5146 ctes = self.ctes
5148 if not ctes:
5149 return ""
5150 ctes_recursive = any([cte.recursive for cte in ctes])
5152 cte_text = self.get_cte_preamble(ctes_recursive) + " "
5153 cte_text += ", \n".join([txt for txt in ctes.values()])
5154 cte_text += "\n "
5156 if nesting_level and nesting_level > 1:
5157 for cte in list(ctes.keys()):
5158 cte_level, cte_name, cte_opts = self.level_name_by_cte[
5159 cte._get_reference_cte()
5160 ]
5161 del self.ctes[cte]
5162 del self.ctes_by_level_name[(cte_level, cte_name)]
5163 del self.level_name_by_cte[cte._get_reference_cte()]
5165 return cte_text
5167 def get_cte_preamble(self, recursive):
5168 if recursive:
5169 return "WITH RECURSIVE"
5170 else:
5171 return "WITH"
5173 def get_select_precolumns(self, select: Select[Any], **kw: Any) -> str:
5174 """Called when building a ``SELECT`` statement, position is just
5175 before column list.
5177 """
5178 if select._distinct_on:
5179 util.warn_deprecated(
5180 "DISTINCT ON is currently supported only by the PostgreSQL "
5181 "dialect. Use of DISTINCT ON for other backends is currently "
5182 "silently ignored, however this usage is deprecated, and will "
5183 "raise CompileError in a future release for all backends "
5184 "that do not support this syntax.",
5185 version="1.4",
5186 )
5187 return "DISTINCT " if select._distinct else ""
5189 def group_by_clause(self, select, **kw):
5190 """allow dialects to customize how GROUP BY is rendered."""
5192 group_by = self._generate_delimited_list(
5193 select._group_by_clauses, OPERATORS[operators.comma_op], **kw
5194 )
5195 if group_by:
5196 return " GROUP BY " + group_by
5197 else:
5198 return ""
5200 def order_by_clause(self, select, **kw):
5201 """allow dialects to customize how ORDER BY is rendered."""
5203 order_by = self._generate_delimited_list(
5204 select._order_by_clauses, OPERATORS[operators.comma_op], **kw
5205 )
5207 if order_by:
5208 return " ORDER BY " + order_by
5209 else:
5210 return ""
5212 def for_update_clause(self, select, **kw):
5213 return " FOR UPDATE"
5215 def returning_clause(
5216 self,
5217 stmt: UpdateBase,
5218 returning_cols: Sequence[_ColumnsClauseElement],
5219 *,
5220 populate_result_map: bool,
5221 **kw: Any,
5222 ) -> str:
5223 columns = [
5224 self._label_returning_column(
5225 stmt,
5226 column,
5227 populate_result_map,
5228 fallback_label_name=fallback_label_name,
5229 column_is_repeated=repeated,
5230 name=name,
5231 proxy_name=proxy_name,
5232 **kw,
5233 )
5234 for (
5235 name,
5236 proxy_name,
5237 fallback_label_name,
5238 column,
5239 repeated,
5240 ) in stmt._generate_columns_plus_names(
5241 True, cols=base._select_iterables(returning_cols)
5242 )
5243 ]
5245 return "RETURNING " + ", ".join(columns)
5247 def limit_clause(self, select, **kw):
5248 text = ""
5249 if select._limit_clause is not None:
5250 text += "\n LIMIT " + self.process(select._limit_clause, **kw)
5251 if select._offset_clause is not None:
5252 if select._limit_clause is None:
5253 text += "\n LIMIT -1"
5254 text += " OFFSET " + self.process(select._offset_clause, **kw)
5255 return text
5257 def fetch_clause(
5258 self,
5259 select,
5260 fetch_clause=None,
5261 require_offset=False,
5262 use_literal_execute_for_simple_int=False,
5263 **kw,
5264 ):
5265 if fetch_clause is None:
5266 fetch_clause = select._fetch_clause
5267 fetch_clause_options = select._fetch_clause_options
5268 else:
5269 fetch_clause_options = {"percent": False, "with_ties": False}
5271 text = ""
5273 if select._offset_clause is not None:
5274 offset_clause = select._offset_clause
5275 if (
5276 use_literal_execute_for_simple_int
5277 and select._simple_int_clause(offset_clause)
5278 ):
5279 offset_clause = offset_clause.render_literal_execute()
5280 offset_str = self.process(offset_clause, **kw)
5281 text += "\n OFFSET %s ROWS" % offset_str
5282 elif require_offset:
5283 text += "\n OFFSET 0 ROWS"
5285 if fetch_clause is not None:
5286 if (
5287 use_literal_execute_for_simple_int
5288 and select._simple_int_clause(fetch_clause)
5289 ):
5290 fetch_clause = fetch_clause.render_literal_execute()
5291 text += "\n FETCH FIRST %s%s ROWS %s" % (
5292 self.process(fetch_clause, **kw),
5293 " PERCENT" if fetch_clause_options["percent"] else "",
5294 "WITH TIES" if fetch_clause_options["with_ties"] else "ONLY",
5295 )
5296 return text
5298 def visit_table(
5299 self,
5300 table,
5301 asfrom=False,
5302 iscrud=False,
5303 ashint=False,
5304 fromhints=None,
5305 use_schema=True,
5306 from_linter=None,
5307 ambiguous_table_name_map=None,
5308 enclosing_alias=None,
5309 **kwargs,
5310 ):
5311 if from_linter:
5312 from_linter.froms[table] = table.fullname
5314 if asfrom or ashint:
5315 effective_schema = self.preparer.schema_for_object(table)
5317 if use_schema and effective_schema:
5318 ret = (
5319 self.preparer.quote_schema(effective_schema)
5320 + "."
5321 + self.preparer.quote(table.name)
5322 )
5323 else:
5324 ret = self.preparer.quote(table.name)
5326 if (
5327 (
5328 enclosing_alias is None
5329 or enclosing_alias.element is not table
5330 )
5331 and not effective_schema
5332 and ambiguous_table_name_map
5333 and table.name in ambiguous_table_name_map
5334 ):
5335 anon_name = self._truncated_identifier(
5336 "alias", ambiguous_table_name_map[table.name]
5337 )
5339 ret = ret + self.get_render_as_alias_suffix(
5340 self.preparer.format_alias(None, anon_name)
5341 )
5343 if fromhints and table in fromhints:
5344 ret = self.format_from_hint_text(
5345 ret, table, fromhints[table], iscrud
5346 )
5347 return ret
5348 else:
5349 return ""
5351 def visit_join(self, join, asfrom=False, from_linter=None, **kwargs):
5352 if from_linter:
5353 from_linter.edges.update(
5354 itertools.product(
5355 _de_clone(join.left._from_objects),
5356 _de_clone(join.right._from_objects),
5357 )
5358 )
5360 if join.full:
5361 join_type = " FULL OUTER JOIN "
5362 elif join.isouter:
5363 join_type = " LEFT OUTER JOIN "
5364 else:
5365 join_type = " JOIN "
5366 return (
5367 join.left._compiler_dispatch(
5368 self, asfrom=True, from_linter=from_linter, **kwargs
5369 )
5370 + join_type
5371 + join.right._compiler_dispatch(
5372 self, asfrom=True, from_linter=from_linter, **kwargs
5373 )
5374 + " ON "
5375 # TODO: likely need asfrom=True here?
5376 + join.onclause._compiler_dispatch(
5377 self, from_linter=from_linter, **kwargs
5378 )
5379 )
5381 def _setup_crud_hints(self, stmt, table_text):
5382 dialect_hints = {
5383 table: hint_text
5384 for (table, dialect), hint_text in stmt._hints.items()
5385 if dialect in ("*", self.dialect.name)
5386 }
5387 if stmt.table in dialect_hints:
5388 table_text = self.format_from_hint_text(
5389 table_text, stmt.table, dialect_hints[stmt.table], True
5390 )
5391 return dialect_hints, table_text
5393 # within the realm of "insertmanyvalues sentinel columns",
5394 # these lookups match different kinds of Column() configurations
5395 # to specific backend capabilities. they are broken into two
5396 # lookups, one for autoincrement columns and the other for non
5397 # autoincrement columns
5398 _sentinel_col_non_autoinc_lookup = util.immutabledict(
5399 {
5400 _SentinelDefaultCharacterization.CLIENTSIDE: (
5401 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5402 ),
5403 _SentinelDefaultCharacterization.SENTINEL_DEFAULT: (
5404 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5405 ),
5406 _SentinelDefaultCharacterization.NONE: (
5407 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5408 ),
5409 _SentinelDefaultCharacterization.IDENTITY: (
5410 InsertmanyvaluesSentinelOpts.IDENTITY
5411 ),
5412 _SentinelDefaultCharacterization.SEQUENCE: (
5413 InsertmanyvaluesSentinelOpts.SEQUENCE
5414 ),
5415 }
5416 )
5417 _sentinel_col_autoinc_lookup = _sentinel_col_non_autoinc_lookup.union(
5418 {
5419 _SentinelDefaultCharacterization.NONE: (
5420 InsertmanyvaluesSentinelOpts.AUTOINCREMENT
5421 ),
5422 }
5423 )
5425 def _get_sentinel_column_for_table(
5426 self, table: Table
5427 ) -> Optional[Sequence[Column[Any]]]:
5428 """given a :class:`.Table`, return a usable sentinel column or
5429 columns for this dialect if any.
5431 Return None if no sentinel columns could be identified, or raise an
5432 error if a column was marked as a sentinel explicitly but isn't
5433 compatible with this dialect.
5435 """
5437 sentinel_opts = self.dialect.insertmanyvalues_implicit_sentinel
5438 sentinel_characteristics = table._sentinel_column_characteristics
5440 sent_cols = sentinel_characteristics.columns
5442 if sent_cols is None:
5443 return None
5445 if sentinel_characteristics.is_autoinc:
5446 bitmask = self._sentinel_col_autoinc_lookup.get(
5447 sentinel_characteristics.default_characterization, 0
5448 )
5449 else:
5450 bitmask = self._sentinel_col_non_autoinc_lookup.get(
5451 sentinel_characteristics.default_characterization, 0
5452 )
5454 if sentinel_opts & bitmask:
5455 return sent_cols
5457 if sentinel_characteristics.is_explicit:
5458 # a column was explicitly marked as insert_sentinel=True,
5459 # however it is not compatible with this dialect. they should
5460 # not indicate this column as a sentinel if they need to include
5461 # this dialect.
5463 # TODO: do we want non-primary key explicit sentinel cols
5464 # that can gracefully degrade for some backends?
5465 # insert_sentinel="degrade" perhaps. not for the initial release.
5466 # I am hoping people are generally not dealing with this sentinel
5467 # business at all.
5469 # if is_explicit is True, there will be only one sentinel column.
5471 raise exc.InvalidRequestError(
5472 f"Column {sent_cols[0]} can't be explicitly "
5473 "marked as a sentinel column when using the "
5474 f"{self.dialect.name} dialect, as the "
5475 "particular type of default generation on this column is "
5476 "not currently compatible with this dialect's specific "
5477 f"INSERT..RETURNING syntax which can receive the "
5478 "server-generated value in "
5479 "a deterministic way. To remove this error, remove "
5480 "insert_sentinel=True from primary key autoincrement "
5481 "columns; these columns are automatically used as "
5482 "sentinels for supported dialects in any case."
5483 )
5485 return None
5487 def _deliver_insertmanyvalues_batches(
5488 self,
5489 statement: str,
5490 parameters: _DBAPIMultiExecuteParams,
5491 compiled_parameters: List[_MutableCoreSingleExecuteParams],
5492 generic_setinputsizes: Optional[_GenericSetInputSizesType],
5493 batch_size: int,
5494 sort_by_parameter_order: bool,
5495 schema_translate_map: Optional[SchemaTranslateMapType],
5496 ) -> Iterator[_InsertManyValuesBatch]:
5497 imv = self._insertmanyvalues
5498 assert imv is not None
5500 if not imv.sentinel_param_keys:
5501 _sentinel_from_params = None
5502 else:
5503 _sentinel_from_params = operator.itemgetter(
5504 *imv.sentinel_param_keys
5505 )
5507 lenparams = len(parameters)
5508 if imv.is_default_expr and not self.dialect.supports_default_metavalue:
5509 # backend doesn't support
5510 # INSERT INTO table (pk_col) VALUES (DEFAULT), (DEFAULT), ...
5511 # at the moment this is basically SQL Server due to
5512 # not being able to use DEFAULT for identity column
5513 # just yield out that many single statements! still
5514 # faster than a whole connection.execute() call ;)
5515 #
5516 # note we still are taking advantage of the fact that we know
5517 # we are using RETURNING. The generalized approach of fetching
5518 # cursor.lastrowid etc. still goes through the more heavyweight
5519 # "ExecutionContext per statement" system as it isn't usable
5520 # as a generic "RETURNING" approach
5521 use_row_at_a_time = True
5522 downgraded = False
5523 elif not self.dialect.supports_multivalues_insert or (
5524 sort_by_parameter_order
5525 and self._result_columns
5526 and (imv.sentinel_columns is None or imv.includes_upsert_behaviors)
5527 ):
5528 # deterministic order was requested and the compiler could
5529 # not organize sentinel columns for this dialect/statement.
5530 # use row at a time
5531 use_row_at_a_time = True
5532 downgraded = True
5533 else:
5534 use_row_at_a_time = False
5535 downgraded = False
5537 if use_row_at_a_time:
5538 for batchnum, (param, compiled_param) in enumerate(
5539 cast(
5540 "Sequence[Tuple[_DBAPISingleExecuteParams, _MutableCoreSingleExecuteParams]]", # noqa: E501
5541 zip(parameters, compiled_parameters),
5542 ),
5543 1,
5544 ):
5545 yield _InsertManyValuesBatch(
5546 statement,
5547 param,
5548 generic_setinputsizes,
5549 [param],
5550 (
5551 [_sentinel_from_params(compiled_param)]
5552 if _sentinel_from_params
5553 else []
5554 ),
5555 1,
5556 batchnum,
5557 lenparams,
5558 sort_by_parameter_order,
5559 downgraded,
5560 )
5561 return
5563 if schema_translate_map:
5564 rst = functools.partial(
5565 self.preparer._render_schema_translates,
5566 schema_translate_map=schema_translate_map,
5567 )
5568 else:
5569 rst = None
5571 imv_single_values_expr = imv.single_values_expr
5572 if rst:
5573 imv_single_values_expr = rst(imv_single_values_expr)
5575 executemany_values = f"({imv_single_values_expr})"
5576 statement = statement.replace(executemany_values, "__EXECMANY_TOKEN__")
5578 # Use optional insertmanyvalues_max_parameters
5579 # to further shrink the batch size so that there are no more than
5580 # insertmanyvalues_max_parameters params.
5581 # Currently used by SQL Server, which limits statements to 2100 bound
5582 # parameters (actually 2099).
5583 max_params = self.dialect.insertmanyvalues_max_parameters
5584 if max_params:
5585 total_num_of_params = len(self.bind_names)
5586 num_params_per_batch = len(imv.insert_crud_params)
5587 num_params_outside_of_batch = (
5588 total_num_of_params - num_params_per_batch
5589 )
5590 batch_size = min(
5591 batch_size,
5592 (
5593 (max_params - num_params_outside_of_batch)
5594 // num_params_per_batch
5595 ),
5596 )
5598 batches = cast("List[Sequence[Any]]", list(parameters))
5599 compiled_batches = cast(
5600 "List[Sequence[Any]]", list(compiled_parameters)
5601 )
5603 processed_setinputsizes: Optional[_GenericSetInputSizesType] = None
5604 batchnum = 1
5605 total_batches = lenparams // batch_size + (
5606 1 if lenparams % batch_size else 0
5607 )
5609 insert_crud_params = imv.insert_crud_params
5610 assert insert_crud_params is not None
5612 if rst:
5613 insert_crud_params = [
5614 (col, key, rst(expr), st)
5615 for col, key, expr, st in insert_crud_params
5616 ]
5618 escaped_bind_names: Mapping[str, str]
5619 expand_pos_lower_index = expand_pos_upper_index = 0
5621 if not self.positional:
5622 if self.escaped_bind_names:
5623 escaped_bind_names = self.escaped_bind_names
5624 else:
5625 escaped_bind_names = {}
5627 all_keys = set(parameters[0])
5629 def apply_placeholders(keys, formatted):
5630 for key in keys:
5631 key = escaped_bind_names.get(key, key)
5632 formatted = formatted.replace(
5633 self.bindtemplate % {"name": key},
5634 self.bindtemplate
5635 % {"name": f"{key}__EXECMANY_INDEX__"},
5636 )
5637 return formatted
5639 if imv.embed_values_counter:
5640 imv_values_counter = ", _IMV_VALUES_COUNTER"
5641 else:
5642 imv_values_counter = ""
5643 formatted_values_clause = f"""({', '.join(
5644 apply_placeholders(bind_keys, formatted)
5645 for _, _, formatted, bind_keys in insert_crud_params
5646 )}{imv_values_counter})"""
5648 keys_to_replace = all_keys.intersection(
5649 escaped_bind_names.get(key, key)
5650 for _, _, _, bind_keys in insert_crud_params
5651 for key in bind_keys
5652 )
5653 base_parameters = {
5654 key: parameters[0][key]
5655 for key in all_keys.difference(keys_to_replace)
5656 }
5657 executemany_values_w_comma = ""
5658 else:
5659 formatted_values_clause = ""
5660 keys_to_replace = set()
5661 base_parameters = {}
5663 if imv.embed_values_counter:
5664 executemany_values_w_comma = (
5665 f"({imv_single_values_expr}, _IMV_VALUES_COUNTER), "
5666 )
5667 else:
5668 executemany_values_w_comma = f"({imv_single_values_expr}), "
5670 all_names_we_will_expand: Set[str] = set()
5671 for elem in imv.insert_crud_params:
5672 all_names_we_will_expand.update(elem[3])
5674 # get the start and end position in a particular list
5675 # of parameters where we will be doing the "expanding".
5676 # statements can have params on either side or both sides,
5677 # given RETURNING and CTEs
5678 if all_names_we_will_expand:
5679 positiontup = self.positiontup
5680 assert positiontup is not None
5682 all_expand_positions = {
5683 idx
5684 for idx, name in enumerate(positiontup)
5685 if name in all_names_we_will_expand
5686 }
5687 expand_pos_lower_index = min(all_expand_positions)
5688 expand_pos_upper_index = max(all_expand_positions) + 1
5689 assert (
5690 len(all_expand_positions)
5691 == expand_pos_upper_index - expand_pos_lower_index
5692 )
5694 if self._numeric_binds:
5695 escaped = re.escape(self._numeric_binds_identifier_char)
5696 executemany_values_w_comma = re.sub(
5697 rf"{escaped}\d+", "%s", executemany_values_w_comma
5698 )
5700 while batches:
5701 batch = batches[0:batch_size]
5702 compiled_batch = compiled_batches[0:batch_size]
5704 batches[0:batch_size] = []
5705 compiled_batches[0:batch_size] = []
5707 if batches:
5708 current_batch_size = batch_size
5709 else:
5710 current_batch_size = len(batch)
5712 if generic_setinputsizes:
5713 # if setinputsizes is present, expand this collection to
5714 # suit the batch length as well
5715 # currently this will be mssql+pyodbc for internal dialects
5716 processed_setinputsizes = [
5717 (new_key, len_, typ)
5718 for new_key, len_, typ in (
5719 (f"{key}_{index}", len_, typ)
5720 for index in range(current_batch_size)
5721 for key, len_, typ in generic_setinputsizes
5722 )
5723 ]
5725 replaced_parameters: Any
5726 if self.positional:
5727 num_ins_params = imv.num_positional_params_counted
5729 batch_iterator: Iterable[Sequence[Any]]
5730 extra_params_left: Sequence[Any]
5731 extra_params_right: Sequence[Any]
5733 if num_ins_params == len(batch[0]):
5734 extra_params_left = extra_params_right = ()
5735 batch_iterator = batch
5736 else:
5737 extra_params_left = batch[0][:expand_pos_lower_index]
5738 extra_params_right = batch[0][expand_pos_upper_index:]
5739 batch_iterator = (
5740 b[expand_pos_lower_index:expand_pos_upper_index]
5741 for b in batch
5742 )
5744 if imv.embed_values_counter:
5745 expanded_values_string = (
5746 "".join(
5747 executemany_values_w_comma.replace(
5748 "_IMV_VALUES_COUNTER", str(i)
5749 )
5750 for i, _ in enumerate(batch)
5751 )
5752 )[:-2]
5753 else:
5754 expanded_values_string = (
5755 (executemany_values_w_comma * current_batch_size)
5756 )[:-2]
5758 if self._numeric_binds and num_ins_params > 0:
5759 # numeric will always number the parameters inside of
5760 # VALUES (and thus order self.positiontup) to be higher
5761 # than non-VALUES parameters, no matter where in the
5762 # statement those non-VALUES parameters appear (this is
5763 # ensured in _process_numeric by numbering first all
5764 # params that are not in _values_bindparam)
5765 # therefore all extra params are always
5766 # on the left side and numbered lower than the VALUES
5767 # parameters
5768 assert not extra_params_right
5770 start = expand_pos_lower_index + 1
5771 end = num_ins_params * (current_batch_size) + start
5773 # need to format here, since statement may contain
5774 # unescaped %, while values_string contains just (%s, %s)
5775 positions = tuple(
5776 f"{self._numeric_binds_identifier_char}{i}"
5777 for i in range(start, end)
5778 )
5779 expanded_values_string = expanded_values_string % positions
5781 replaced_statement = statement.replace(
5782 "__EXECMANY_TOKEN__", expanded_values_string
5783 )
5785 replaced_parameters = tuple(
5786 itertools.chain.from_iterable(batch_iterator)
5787 )
5789 replaced_parameters = (
5790 extra_params_left
5791 + replaced_parameters
5792 + extra_params_right
5793 )
5795 else:
5796 replaced_values_clauses = []
5797 replaced_parameters = base_parameters.copy()
5799 for i, param in enumerate(batch):
5800 fmv = formatted_values_clause.replace(
5801 "EXECMANY_INDEX__", str(i)
5802 )
5803 if imv.embed_values_counter:
5804 fmv = fmv.replace("_IMV_VALUES_COUNTER", str(i))
5806 replaced_values_clauses.append(fmv)
5807 replaced_parameters.update(
5808 {f"{key}__{i}": param[key] for key in keys_to_replace}
5809 )
5811 replaced_statement = statement.replace(
5812 "__EXECMANY_TOKEN__",
5813 ", ".join(replaced_values_clauses),
5814 )
5816 yield _InsertManyValuesBatch(
5817 replaced_statement,
5818 replaced_parameters,
5819 processed_setinputsizes,
5820 batch,
5821 (
5822 [_sentinel_from_params(cb) for cb in compiled_batch]
5823 if _sentinel_from_params
5824 else []
5825 ),
5826 current_batch_size,
5827 batchnum,
5828 total_batches,
5829 sort_by_parameter_order,
5830 False,
5831 )
5832 batchnum += 1
5834 def visit_insert(
5835 self, insert_stmt, visited_bindparam=None, visiting_cte=None, **kw
5836 ):
5837 compile_state = insert_stmt._compile_state_factory(
5838 insert_stmt, self, **kw
5839 )
5840 insert_stmt = compile_state.statement
5842 if visiting_cte is not None:
5843 kw["visiting_cte"] = visiting_cte
5844 toplevel = False
5845 else:
5846 toplevel = not self.stack
5848 if toplevel:
5849 self.isinsert = True
5850 if not self.dml_compile_state:
5851 self.dml_compile_state = compile_state
5852 if not self.compile_state:
5853 self.compile_state = compile_state
5855 self.stack.append(
5856 {
5857 "correlate_froms": set(),
5858 "asfrom_froms": set(),
5859 "selectable": insert_stmt,
5860 }
5861 )
5863 counted_bindparam = 0
5865 # reset any incoming "visited_bindparam" collection
5866 visited_bindparam = None
5868 # for positional, insertmanyvalues needs to know how many
5869 # bound parameters are in the VALUES sequence; there's no simple
5870 # rule because default expressions etc. can have zero or more
5871 # params inside them. After multiple attempts to figure this out,
5872 # this very simplistic "count after" works and is
5873 # likely the least amount of callcounts, though looks clumsy
5874 if self.positional and visiting_cte is None:
5875 # if we are inside a CTE, don't count parameters
5876 # here since they wont be for insertmanyvalues. keep
5877 # visited_bindparam at None so no counting happens.
5878 # see #9173
5879 visited_bindparam = []
5881 crud_params_struct = crud._get_crud_params(
5882 self,
5883 insert_stmt,
5884 compile_state,
5885 toplevel,
5886 visited_bindparam=visited_bindparam,
5887 **kw,
5888 )
5890 if self.positional and visited_bindparam is not None:
5891 counted_bindparam = len(visited_bindparam)
5892 if self._numeric_binds:
5893 if self._values_bindparam is not None:
5894 self._values_bindparam += visited_bindparam
5895 else:
5896 self._values_bindparam = visited_bindparam
5898 crud_params_single = crud_params_struct.single_params
5900 if (
5901 not crud_params_single
5902 and not self.dialect.supports_default_values
5903 and not self.dialect.supports_default_metavalue
5904 and not self.dialect.supports_empty_insert
5905 ):
5906 raise exc.CompileError(
5907 "The '%s' dialect with current database "
5908 "version settings does not support empty "
5909 "inserts." % self.dialect.name
5910 )
5912 if compile_state._has_multi_parameters:
5913 if not self.dialect.supports_multivalues_insert:
5914 raise exc.CompileError(
5915 "The '%s' dialect with current database "
5916 "version settings does not support "
5917 "in-place multirow inserts." % self.dialect.name
5918 )
5919 elif (
5920 self.implicit_returning or insert_stmt._returning
5921 ) and insert_stmt._sort_by_parameter_order:
5922 raise exc.CompileError(
5923 "RETURNING cannot be determinstically sorted when "
5924 "using an INSERT which includes multi-row values()."
5925 )
5926 crud_params_single = crud_params_struct.single_params
5927 else:
5928 crud_params_single = crud_params_struct.single_params
5930 preparer = self.preparer
5931 supports_default_values = self.dialect.supports_default_values
5933 text = "INSERT "
5935 if insert_stmt._prefixes:
5936 text += self._generate_prefixes(
5937 insert_stmt, insert_stmt._prefixes, **kw
5938 )
5940 text += "INTO "
5941 table_text = preparer.format_table(insert_stmt.table)
5943 if insert_stmt._hints:
5944 _, table_text = self._setup_crud_hints(insert_stmt, table_text)
5946 if insert_stmt._independent_ctes:
5947 self._dispatch_independent_ctes(insert_stmt, kw)
5949 text += table_text
5951 if crud_params_single or not supports_default_values:
5952 text += " (%s)" % ", ".join(
5953 [expr for _, expr, _, _ in crud_params_single]
5954 )
5956 # look for insertmanyvalues attributes that would have been configured
5957 # by crud.py as it scanned through the columns to be part of the
5958 # INSERT
5959 use_insertmanyvalues = crud_params_struct.use_insertmanyvalues
5960 named_sentinel_params: Optional[Sequence[str]] = None
5961 add_sentinel_cols = None
5962 implicit_sentinel = False
5964 returning_cols = self.implicit_returning or insert_stmt._returning
5965 if returning_cols:
5966 add_sentinel_cols = crud_params_struct.use_sentinel_columns
5967 if add_sentinel_cols is not None:
5968 assert use_insertmanyvalues
5970 # search for the sentinel column explicitly present
5971 # in the INSERT columns list, and additionally check that
5972 # this column has a bound parameter name set up that's in the
5973 # parameter list. If both of these cases are present, it means
5974 # we will have a client side value for the sentinel in each
5975 # parameter set.
5977 _params_by_col = {
5978 col: param_names
5979 for col, _, _, param_names in crud_params_single
5980 }
5981 named_sentinel_params = []
5982 for _add_sentinel_col in add_sentinel_cols:
5983 if _add_sentinel_col not in _params_by_col:
5984 named_sentinel_params = None
5985 break
5986 param_name = self._within_exec_param_key_getter(
5987 _add_sentinel_col
5988 )
5989 if param_name not in _params_by_col[_add_sentinel_col]:
5990 named_sentinel_params = None
5991 break
5992 named_sentinel_params.append(param_name)
5994 if named_sentinel_params is None:
5995 # if we are not going to have a client side value for
5996 # the sentinel in the parameter set, that means it's
5997 # an autoincrement, an IDENTITY, or a server-side SQL
5998 # expression like nextval('seqname'). So this is
5999 # an "implicit" sentinel; we will look for it in
6000 # RETURNING
6001 # only, and then sort on it. For this case on PG,
6002 # SQL Server we have to use a special INSERT form
6003 # that guarantees the server side function lines up with
6004 # the entries in the VALUES.
6005 if (
6006 self.dialect.insertmanyvalues_implicit_sentinel
6007 & InsertmanyvaluesSentinelOpts.ANY_AUTOINCREMENT
6008 ):
6009 implicit_sentinel = True
6010 else:
6011 # here, we are not using a sentinel at all
6012 # and we are likely the SQLite dialect.
6013 # The first add_sentinel_col that we have should not
6014 # be marked as "insert_sentinel=True". if it was,
6015 # an error should have been raised in
6016 # _get_sentinel_column_for_table.
6017 assert not add_sentinel_cols[0]._insert_sentinel, (
6018 "sentinel selection rules should have prevented "
6019 "us from getting here for this dialect"
6020 )
6022 # always put the sentinel columns last. even if they are
6023 # in the returning list already, they will be there twice
6024 # then.
6025 returning_cols = list(returning_cols) + list(add_sentinel_cols)
6027 returning_clause = self.returning_clause(
6028 insert_stmt,
6029 returning_cols,
6030 populate_result_map=toplevel,
6031 )
6033 if self.returning_precedes_values:
6034 text += " " + returning_clause
6036 else:
6037 returning_clause = None
6039 if insert_stmt.select is not None:
6040 # placed here by crud.py
6041 select_text = self.process(
6042 self.stack[-1]["insert_from_select"], insert_into=True, **kw
6043 )
6045 if self.ctes and self.dialect.cte_follows_insert:
6046 nesting_level = len(self.stack) if not toplevel else None
6047 text += " %s%s" % (
6048 self._render_cte_clause(
6049 nesting_level=nesting_level,
6050 include_following_stack=True,
6051 ),
6052 select_text,
6053 )
6054 else:
6055 text += " %s" % select_text
6056 elif not crud_params_single and supports_default_values:
6057 text += " DEFAULT VALUES"
6058 if use_insertmanyvalues:
6059 self._insertmanyvalues = _InsertManyValues(
6060 True,
6061 self.dialect.default_metavalue_token,
6062 cast(
6063 "List[crud._CrudParamElementStr]", crud_params_single
6064 ),
6065 counted_bindparam,
6066 sort_by_parameter_order=(
6067 insert_stmt._sort_by_parameter_order
6068 ),
6069 includes_upsert_behaviors=(
6070 insert_stmt._post_values_clause is not None
6071 ),
6072 sentinel_columns=add_sentinel_cols,
6073 num_sentinel_columns=(
6074 len(add_sentinel_cols) if add_sentinel_cols else 0
6075 ),
6076 implicit_sentinel=implicit_sentinel,
6077 )
6078 elif compile_state._has_multi_parameters:
6079 text += " VALUES %s" % (
6080 ", ".join(
6081 "(%s)"
6082 % (", ".join(value for _, _, value, _ in crud_param_set))
6083 for crud_param_set in crud_params_struct.all_multi_params
6084 ),
6085 )
6086 else:
6087 insert_single_values_expr = ", ".join(
6088 [
6089 value
6090 for _, _, value, _ in cast(
6091 "List[crud._CrudParamElementStr]",
6092 crud_params_single,
6093 )
6094 ]
6095 )
6097 if use_insertmanyvalues:
6098 if (
6099 implicit_sentinel
6100 and (
6101 self.dialect.insertmanyvalues_implicit_sentinel
6102 & InsertmanyvaluesSentinelOpts.USE_INSERT_FROM_SELECT
6103 )
6104 # this is checking if we have
6105 # INSERT INTO table (id) VALUES (DEFAULT).
6106 and not (crud_params_struct.is_default_metavalue_only)
6107 ):
6108 # if we have a sentinel column that is server generated,
6109 # then for selected backends render the VALUES list as a
6110 # subquery. This is the orderable form supported by
6111 # PostgreSQL and SQL Server.
6112 embed_sentinel_value = True
6114 render_bind_casts = (
6115 self.dialect.insertmanyvalues_implicit_sentinel
6116 & InsertmanyvaluesSentinelOpts.RENDER_SELECT_COL_CASTS
6117 )
6119 colnames = ", ".join(
6120 f"p{i}" for i, _ in enumerate(crud_params_single)
6121 )
6123 if render_bind_casts:
6124 # render casts for the SELECT list. For PG, we are
6125 # already rendering bind casts in the parameter list,
6126 # selectively for the more "tricky" types like ARRAY.
6127 # however, even for the "easy" types, if the parameter
6128 # is NULL for every entry, PG gives up and says
6129 # "it must be TEXT", which fails for other easy types
6130 # like ints. So we cast on this side too.
6131 colnames_w_cast = ", ".join(
6132 self.render_bind_cast(
6133 col.type,
6134 col.type._unwrapped_dialect_impl(self.dialect),
6135 f"p{i}",
6136 )
6137 for i, (col, *_) in enumerate(crud_params_single)
6138 )
6139 else:
6140 colnames_w_cast = colnames
6142 text += (
6143 f" SELECT {colnames_w_cast} FROM "
6144 f"(VALUES ({insert_single_values_expr})) "
6145 f"AS imp_sen({colnames}, sen_counter) "
6146 "ORDER BY sen_counter"
6147 )
6148 else:
6149 # otherwise, if no sentinel or backend doesn't support
6150 # orderable subquery form, use a plain VALUES list
6151 embed_sentinel_value = False
6152 text += f" VALUES ({insert_single_values_expr})"
6154 self._insertmanyvalues = _InsertManyValues(
6155 is_default_expr=False,
6156 single_values_expr=insert_single_values_expr,
6157 insert_crud_params=cast(
6158 "List[crud._CrudParamElementStr]",
6159 crud_params_single,
6160 ),
6161 num_positional_params_counted=counted_bindparam,
6162 sort_by_parameter_order=(
6163 insert_stmt._sort_by_parameter_order
6164 ),
6165 includes_upsert_behaviors=(
6166 insert_stmt._post_values_clause is not None
6167 ),
6168 sentinel_columns=add_sentinel_cols,
6169 num_sentinel_columns=(
6170 len(add_sentinel_cols) if add_sentinel_cols else 0
6171 ),
6172 sentinel_param_keys=named_sentinel_params,
6173 implicit_sentinel=implicit_sentinel,
6174 embed_values_counter=embed_sentinel_value,
6175 )
6177 else:
6178 text += f" VALUES ({insert_single_values_expr})"
6180 if insert_stmt._post_values_clause is not None:
6181 post_values_clause = self.process(
6182 insert_stmt._post_values_clause, **kw
6183 )
6184 if post_values_clause:
6185 text += " " + post_values_clause
6187 if returning_clause and not self.returning_precedes_values:
6188 text += " " + returning_clause
6190 if self.ctes and not self.dialect.cte_follows_insert:
6191 nesting_level = len(self.stack) if not toplevel else None
6192 text = (
6193 self._render_cte_clause(
6194 nesting_level=nesting_level,
6195 include_following_stack=True,
6196 )
6197 + text
6198 )
6200 self.stack.pop(-1)
6202 return text
6204 def update_tables_clause(self, update_stmt, from_table, extra_froms, **kw):
6205 """Provide a hook to override the initial table clause
6206 in an UPDATE statement.
6208 MySQL overrides this.
6210 """
6211 kw["asfrom"] = True
6212 return from_table._compiler_dispatch(self, iscrud=True, **kw)
6214 def update_from_clause(
6215 self, update_stmt, from_table, extra_froms, from_hints, **kw
6216 ):
6217 """Provide a hook to override the generation of an
6218 UPDATE..FROM clause.
6219 MySQL and MSSQL override this.
6220 """
6221 raise NotImplementedError(
6222 "This backend does not support multiple-table "
6223 "criteria within UPDATE"
6224 )
6226 def update_post_criteria_clause(
6227 self, update_stmt: Update, **kw: Any
6228 ) -> Optional[str]:
6229 """provide a hook to override generation after the WHERE criteria
6230 in an UPDATE statement
6232 .. versionadded:: 2.1
6234 """
6235 if update_stmt._post_criteria_clause is not None:
6236 return self.process(
6237 update_stmt._post_criteria_clause,
6238 **kw,
6239 )
6240 else:
6241 return None
6243 def delete_post_criteria_clause(
6244 self, delete_stmt: Delete, **kw: Any
6245 ) -> Optional[str]:
6246 """provide a hook to override generation after the WHERE criteria
6247 in a DELETE statement
6249 .. versionadded:: 2.1
6251 """
6252 if delete_stmt._post_criteria_clause is not None:
6253 return self.process(
6254 delete_stmt._post_criteria_clause,
6255 **kw,
6256 )
6257 else:
6258 return None
6260 def visit_update(
6261 self,
6262 update_stmt: Update,
6263 visiting_cte: Optional[CTE] = None,
6264 **kw: Any,
6265 ) -> str:
6266 compile_state = update_stmt._compile_state_factory(
6267 update_stmt, self, **kw
6268 )
6269 if TYPE_CHECKING:
6270 assert isinstance(compile_state, UpdateDMLState)
6271 update_stmt = compile_state.statement # type: ignore[assignment]
6273 if visiting_cte is not None:
6274 kw["visiting_cte"] = visiting_cte
6275 toplevel = False
6276 else:
6277 toplevel = not self.stack
6279 if toplevel:
6280 self.isupdate = True
6281 if not self.dml_compile_state:
6282 self.dml_compile_state = compile_state
6283 if not self.compile_state:
6284 self.compile_state = compile_state
6286 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
6287 from_linter = FromLinter({}, set())
6288 warn_linting = self.linting & WARN_LINTING
6289 if toplevel:
6290 self.from_linter = from_linter
6291 else:
6292 from_linter = None
6293 warn_linting = False
6295 extra_froms = compile_state._extra_froms
6296 is_multitable = bool(extra_froms)
6298 if is_multitable:
6299 # main table might be a JOIN
6300 main_froms = set(_from_objects(update_stmt.table))
6301 render_extra_froms = [
6302 f for f in extra_froms if f not in main_froms
6303 ]
6304 correlate_froms = main_froms.union(extra_froms)
6305 else:
6306 render_extra_froms = []
6307 correlate_froms = {update_stmt.table}
6309 self.stack.append(
6310 {
6311 "correlate_froms": correlate_froms,
6312 "asfrom_froms": correlate_froms,
6313 "selectable": update_stmt,
6314 }
6315 )
6317 text = "UPDATE "
6319 if update_stmt._prefixes:
6320 text += self._generate_prefixes(
6321 update_stmt, update_stmt._prefixes, **kw
6322 )
6324 table_text = self.update_tables_clause(
6325 update_stmt,
6326 update_stmt.table,
6327 render_extra_froms,
6328 from_linter=from_linter,
6329 **kw,
6330 )
6331 crud_params_struct = crud._get_crud_params(
6332 self, update_stmt, compile_state, toplevel, **kw
6333 )
6334 crud_params = crud_params_struct.single_params
6336 if update_stmt._hints:
6337 dialect_hints, table_text = self._setup_crud_hints(
6338 update_stmt, table_text
6339 )
6340 else:
6341 dialect_hints = None
6343 if update_stmt._independent_ctes:
6344 self._dispatch_independent_ctes(update_stmt, kw)
6346 text += table_text
6348 text += " SET "
6349 text += ", ".join(
6350 expr + "=" + value
6351 for _, expr, value, _ in cast(
6352 "List[Tuple[Any, str, str, Any]]", crud_params
6353 )
6354 )
6356 if self.implicit_returning or update_stmt._returning:
6357 if self.returning_precedes_values:
6358 text += " " + self.returning_clause(
6359 update_stmt,
6360 self.implicit_returning or update_stmt._returning,
6361 populate_result_map=toplevel,
6362 )
6364 if extra_froms:
6365 extra_from_text = self.update_from_clause(
6366 update_stmt,
6367 update_stmt.table,
6368 render_extra_froms,
6369 dialect_hints,
6370 from_linter=from_linter,
6371 **kw,
6372 )
6373 if extra_from_text:
6374 text += " " + extra_from_text
6376 if update_stmt._where_criteria:
6377 t = self._generate_delimited_and_list(
6378 update_stmt._where_criteria, from_linter=from_linter, **kw
6379 )
6380 if t:
6381 text += " WHERE " + t
6383 ulc = self.update_post_criteria_clause(
6384 update_stmt, from_linter=from_linter, **kw
6385 )
6386 if ulc:
6387 text += " " + ulc
6389 if (
6390 self.implicit_returning or update_stmt._returning
6391 ) and not self.returning_precedes_values:
6392 text += " " + self.returning_clause(
6393 update_stmt,
6394 self.implicit_returning or update_stmt._returning,
6395 populate_result_map=toplevel,
6396 )
6398 if self.ctes:
6399 nesting_level = len(self.stack) if not toplevel else None
6400 text = self._render_cte_clause(nesting_level=nesting_level) + text
6402 if warn_linting:
6403 assert from_linter is not None
6404 from_linter.warn(stmt_type="UPDATE")
6406 self.stack.pop(-1)
6408 return text # type: ignore[no-any-return]
6410 def delete_extra_from_clause(
6411 self, delete_stmt, from_table, extra_froms, from_hints, **kw
6412 ):
6413 """Provide a hook to override the generation of an
6414 DELETE..FROM clause.
6416 This can be used to implement DELETE..USING for example.
6418 MySQL and MSSQL override this.
6420 """
6421 raise NotImplementedError(
6422 "This backend does not support multiple-table "
6423 "criteria within DELETE"
6424 )
6426 def delete_table_clause(self, delete_stmt, from_table, extra_froms, **kw):
6427 return from_table._compiler_dispatch(
6428 self, asfrom=True, iscrud=True, **kw
6429 )
6431 def visit_delete(self, delete_stmt, visiting_cte=None, **kw):
6432 compile_state = delete_stmt._compile_state_factory(
6433 delete_stmt, self, **kw
6434 )
6435 delete_stmt = compile_state.statement
6437 if visiting_cte is not None:
6438 kw["visiting_cte"] = visiting_cte
6439 toplevel = False
6440 else:
6441 toplevel = not self.stack
6443 if toplevel:
6444 self.isdelete = True
6445 if not self.dml_compile_state:
6446 self.dml_compile_state = compile_state
6447 if not self.compile_state:
6448 self.compile_state = compile_state
6450 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
6451 from_linter = FromLinter({}, set())
6452 warn_linting = self.linting & WARN_LINTING
6453 if toplevel:
6454 self.from_linter = from_linter
6455 else:
6456 from_linter = None
6457 warn_linting = False
6459 extra_froms = compile_state._extra_froms
6461 correlate_froms = {delete_stmt.table}.union(extra_froms)
6462 self.stack.append(
6463 {
6464 "correlate_froms": correlate_froms,
6465 "asfrom_froms": correlate_froms,
6466 "selectable": delete_stmt,
6467 }
6468 )
6470 text = "DELETE "
6472 if delete_stmt._prefixes:
6473 text += self._generate_prefixes(
6474 delete_stmt, delete_stmt._prefixes, **kw
6475 )
6477 text += "FROM "
6479 try:
6480 table_text = self.delete_table_clause(
6481 delete_stmt,
6482 delete_stmt.table,
6483 extra_froms,
6484 from_linter=from_linter,
6485 )
6486 except TypeError:
6487 # anticipate 3rd party dialects that don't include **kw
6488 # TODO: remove in 2.1
6489 table_text = self.delete_table_clause(
6490 delete_stmt, delete_stmt.table, extra_froms
6491 )
6492 if from_linter:
6493 _ = self.process(delete_stmt.table, from_linter=from_linter)
6495 crud._get_crud_params(self, delete_stmt, compile_state, toplevel, **kw)
6497 if delete_stmt._hints:
6498 dialect_hints, table_text = self._setup_crud_hints(
6499 delete_stmt, table_text
6500 )
6501 else:
6502 dialect_hints = None
6504 if delete_stmt._independent_ctes:
6505 self._dispatch_independent_ctes(delete_stmt, kw)
6507 text += table_text
6509 if (
6510 self.implicit_returning or delete_stmt._returning
6511 ) and self.returning_precedes_values:
6512 text += " " + self.returning_clause(
6513 delete_stmt,
6514 self.implicit_returning or delete_stmt._returning,
6515 populate_result_map=toplevel,
6516 )
6518 if extra_froms:
6519 extra_from_text = self.delete_extra_from_clause(
6520 delete_stmt,
6521 delete_stmt.table,
6522 extra_froms,
6523 dialect_hints,
6524 from_linter=from_linter,
6525 **kw,
6526 )
6527 if extra_from_text:
6528 text += " " + extra_from_text
6530 if delete_stmt._where_criteria:
6531 t = self._generate_delimited_and_list(
6532 delete_stmt._where_criteria, from_linter=from_linter, **kw
6533 )
6534 if t:
6535 text += " WHERE " + t
6537 dlc = self.delete_post_criteria_clause(
6538 delete_stmt, from_linter=from_linter, **kw
6539 )
6540 if dlc:
6541 text += " " + dlc
6543 if (
6544 self.implicit_returning or delete_stmt._returning
6545 ) and not self.returning_precedes_values:
6546 text += " " + self.returning_clause(
6547 delete_stmt,
6548 self.implicit_returning or delete_stmt._returning,
6549 populate_result_map=toplevel,
6550 )
6552 if self.ctes:
6553 nesting_level = len(self.stack) if not toplevel else None
6554 text = self._render_cte_clause(nesting_level=nesting_level) + text
6556 if warn_linting:
6557 assert from_linter is not None
6558 from_linter.warn(stmt_type="DELETE")
6560 self.stack.pop(-1)
6562 return text
6564 def visit_savepoint(self, savepoint_stmt, **kw):
6565 return "SAVEPOINT %s" % self.preparer.format_savepoint(savepoint_stmt)
6567 def visit_rollback_to_savepoint(self, savepoint_stmt, **kw):
6568 return "ROLLBACK TO SAVEPOINT %s" % self.preparer.format_savepoint(
6569 savepoint_stmt
6570 )
6572 def visit_release_savepoint(self, savepoint_stmt, **kw):
6573 return "RELEASE SAVEPOINT %s" % self.preparer.format_savepoint(
6574 savepoint_stmt
6575 )
6578class StrSQLCompiler(SQLCompiler):
6579 """A :class:`.SQLCompiler` subclass which allows a small selection
6580 of non-standard SQL features to render into a string value.
6582 The :class:`.StrSQLCompiler` is invoked whenever a Core expression
6583 element is directly stringified without calling upon the
6584 :meth:`_expression.ClauseElement.compile` method.
6585 It can render a limited set
6586 of non-standard SQL constructs to assist in basic stringification,
6587 however for more substantial custom or dialect-specific SQL constructs,
6588 it will be necessary to make use of
6589 :meth:`_expression.ClauseElement.compile`
6590 directly.
6592 .. seealso::
6594 :ref:`faq_sql_expression_string`
6596 """
6598 def _fallback_column_name(self, column):
6599 return "<name unknown>"
6601 @util.preload_module("sqlalchemy.engine.url")
6602 def visit_unsupported_compilation(self, element, err, **kw):
6603 if element.stringify_dialect != "default":
6604 url = util.preloaded.engine_url
6605 dialect = url.URL.create(element.stringify_dialect).get_dialect()()
6607 compiler = dialect.statement_compiler(
6608 dialect, None, _supporting_against=self
6609 )
6610 if not isinstance(compiler, StrSQLCompiler):
6611 return compiler.process(element, **kw)
6613 return super().visit_unsupported_compilation(element, err)
6615 def visit_getitem_binary(self, binary, operator, **kw):
6616 return "%s[%s]" % (
6617 self.process(binary.left, **kw),
6618 self.process(binary.right, **kw),
6619 )
6621 def visit_json_getitem_op_binary(self, binary, operator, **kw):
6622 return self.visit_getitem_binary(binary, operator, **kw)
6624 def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
6625 return self.visit_getitem_binary(binary, operator, **kw)
6627 def visit_sequence(self, sequence, **kw):
6628 return (
6629 f"<next sequence value: {self.preparer.format_sequence(sequence)}>"
6630 )
6632 def returning_clause(
6633 self,
6634 stmt: UpdateBase,
6635 returning_cols: Sequence[_ColumnsClauseElement],
6636 *,
6637 populate_result_map: bool,
6638 **kw: Any,
6639 ) -> str:
6640 columns = [
6641 self._label_select_column(None, c, True, False, {})
6642 for c in base._select_iterables(returning_cols)
6643 ]
6644 return "RETURNING " + ", ".join(columns)
6646 def update_from_clause(
6647 self, update_stmt, from_table, extra_froms, from_hints, **kw
6648 ):
6649 kw["asfrom"] = True
6650 return "FROM " + ", ".join(
6651 t._compiler_dispatch(self, fromhints=from_hints, **kw)
6652 for t in extra_froms
6653 )
6655 def delete_extra_from_clause(
6656 self, delete_stmt, from_table, extra_froms, from_hints, **kw
6657 ):
6658 kw["asfrom"] = True
6659 return ", " + ", ".join(
6660 t._compiler_dispatch(self, fromhints=from_hints, **kw)
6661 for t in extra_froms
6662 )
6664 def visit_empty_set_expr(self, element_types, **kw):
6665 return "SELECT 1 WHERE 1!=1"
6667 def get_from_hint_text(self, table, text):
6668 return "[%s]" % text
6670 def visit_regexp_match_op_binary(self, binary, operator, **kw):
6671 return self._generate_generic_binary(binary, " <regexp> ", **kw)
6673 def visit_not_regexp_match_op_binary(self, binary, operator, **kw):
6674 return self._generate_generic_binary(binary, " <not regexp> ", **kw)
6676 def visit_regexp_replace_op_binary(self, binary, operator, **kw):
6677 return "<regexp replace>(%s, %s)" % (
6678 binary.left._compiler_dispatch(self, **kw),
6679 binary.right._compiler_dispatch(self, **kw),
6680 )
6682 def visit_try_cast(self, cast, **kwargs):
6683 return "TRY_CAST(%s AS %s)" % (
6684 cast.clause._compiler_dispatch(self, **kwargs),
6685 cast.typeclause._compiler_dispatch(self, **kwargs),
6686 )
6689class DDLCompiler(Compiled):
6690 is_ddl = True
6692 if TYPE_CHECKING:
6694 def __init__(
6695 self,
6696 dialect: Dialect,
6697 statement: ExecutableDDLElement,
6698 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
6699 render_schema_translate: bool = ...,
6700 compile_kwargs: Mapping[str, Any] = ...,
6701 ): ...
6703 @util.ro_memoized_property
6704 def sql_compiler(self) -> SQLCompiler:
6705 return self.dialect.statement_compiler(
6706 self.dialect, None, schema_translate_map=self.schema_translate_map
6707 )
6709 @util.memoized_property
6710 def type_compiler(self):
6711 return self.dialect.type_compiler_instance
6713 def construct_params(
6714 self,
6715 params: Optional[_CoreSingleExecuteParams] = None,
6716 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
6717 escape_names: bool = True,
6718 ) -> Optional[_MutableCoreSingleExecuteParams]:
6719 return None
6721 def visit_ddl(self, ddl, **kwargs):
6722 # table events can substitute table and schema name
6723 context = ddl.context
6724 if isinstance(ddl.target, schema.Table):
6725 context = context.copy()
6727 preparer = self.preparer
6728 path = preparer.format_table_seq(ddl.target)
6729 if len(path) == 1:
6730 table, sch = path[0], ""
6731 else:
6732 table, sch = path[-1], path[0]
6734 context.setdefault("table", table)
6735 context.setdefault("schema", sch)
6736 context.setdefault("fullname", preparer.format_table(ddl.target))
6738 return self.sql_compiler.post_process_text(ddl.statement % context)
6740 def visit_create_schema(self, create, **kw):
6741 text = "CREATE SCHEMA "
6742 if create.if_not_exists:
6743 text += "IF NOT EXISTS "
6744 return text + self.preparer.format_schema(create.element)
6746 def visit_drop_schema(self, drop, **kw):
6747 text = "DROP SCHEMA "
6748 if drop.if_exists:
6749 text += "IF EXISTS "
6750 text += self.preparer.format_schema(drop.element)
6751 if drop.cascade:
6752 text += " CASCADE"
6753 return text
6755 def visit_create_table(self, create, **kw):
6756 table = create.element
6757 preparer = self.preparer
6759 text = "\nCREATE "
6760 if table._prefixes:
6761 text += " ".join(table._prefixes) + " "
6763 text += "TABLE "
6764 if create.if_not_exists:
6765 text += "IF NOT EXISTS "
6767 text += preparer.format_table(table) + " "
6769 create_table_suffix = self.create_table_suffix(table)
6770 if create_table_suffix:
6771 text += create_table_suffix + " "
6773 text += "("
6775 separator = "\n"
6777 # if only one primary key, specify it along with the column
6778 first_pk = False
6779 for create_column in create.columns:
6780 column = create_column.element
6781 try:
6782 processed = self.process(
6783 create_column, first_pk=column.primary_key and not first_pk
6784 )
6785 if processed is not None:
6786 text += separator
6787 separator = ", \n"
6788 text += "\t" + processed
6789 if column.primary_key:
6790 first_pk = True
6791 except exc.CompileError as ce:
6792 raise exc.CompileError(
6793 "(in table '%s', column '%s'): %s"
6794 % (table.description, column.name, ce.args[0])
6795 ) from ce
6797 const = self.create_table_constraints(
6798 table,
6799 _include_foreign_key_constraints=create.include_foreign_key_constraints, # noqa
6800 )
6801 if const:
6802 text += separator + "\t" + const
6804 text += "\n)%s\n\n" % self.post_create_table(table)
6805 return text
6807 def visit_create_column(self, create, first_pk=False, **kw):
6808 column = create.element
6810 if column.system:
6811 return None
6813 text = self.get_column_specification(column, first_pk=first_pk)
6814 const = " ".join(
6815 self.process(constraint) for constraint in column.constraints
6816 )
6817 if const:
6818 text += " " + const
6820 return text
6822 def create_table_constraints(
6823 self, table, _include_foreign_key_constraints=None, **kw
6824 ):
6825 # On some DB order is significant: visit PK first, then the
6826 # other constraints (engine.ReflectionTest.testbasic failed on FB2)
6827 constraints = []
6828 if table.primary_key:
6829 constraints.append(table.primary_key)
6831 all_fkcs = table.foreign_key_constraints
6832 if _include_foreign_key_constraints is not None:
6833 omit_fkcs = all_fkcs.difference(_include_foreign_key_constraints)
6834 else:
6835 omit_fkcs = set()
6837 constraints.extend(
6838 [
6839 c
6840 for c in table._sorted_constraints
6841 if c is not table.primary_key and c not in omit_fkcs
6842 ]
6843 )
6845 return ", \n\t".join(
6846 p
6847 for p in (
6848 self.process(constraint)
6849 for constraint in constraints
6850 if (constraint._should_create_for_compiler(self))
6851 and (
6852 not self.dialect.supports_alter
6853 or not getattr(constraint, "use_alter", False)
6854 )
6855 )
6856 if p is not None
6857 )
6859 def visit_drop_table(self, drop, **kw):
6860 text = "\nDROP TABLE "
6861 if drop.if_exists:
6862 text += "IF EXISTS "
6863 return text + self.preparer.format_table(drop.element)
6865 def visit_drop_view(self, drop, **kw):
6866 return "\nDROP VIEW " + self.preparer.format_table(drop.element)
6868 def _verify_index_table(self, index: Index) -> None:
6869 if index.table is None:
6870 raise exc.CompileError(
6871 "Index '%s' is not associated with any table." % index.name
6872 )
6874 def visit_create_index(
6875 self, create, include_schema=False, include_table_schema=True, **kw
6876 ):
6877 index = create.element
6878 self._verify_index_table(index)
6879 preparer = self.preparer
6880 text = "CREATE "
6881 if index.unique:
6882 text += "UNIQUE "
6883 if index.name is None:
6884 raise exc.CompileError(
6885 "CREATE INDEX requires that the index have a name"
6886 )
6888 text += "INDEX "
6889 if create.if_not_exists:
6890 text += "IF NOT EXISTS "
6892 text += "%s ON %s (%s)" % (
6893 self._prepared_index_name(index, include_schema=include_schema),
6894 preparer.format_table(
6895 index.table, use_schema=include_table_schema
6896 ),
6897 ", ".join(
6898 self.sql_compiler.process(
6899 expr, include_table=False, literal_binds=True
6900 )
6901 for expr in index.expressions
6902 ),
6903 )
6904 return text
6906 def visit_drop_index(self, drop, **kw):
6907 index = drop.element
6909 if index.name is None:
6910 raise exc.CompileError(
6911 "DROP INDEX requires that the index have a name"
6912 )
6913 text = "\nDROP INDEX "
6914 if drop.if_exists:
6915 text += "IF EXISTS "
6917 return text + self._prepared_index_name(index, include_schema=True)
6919 def _prepared_index_name(
6920 self, index: Index, include_schema: bool = False
6921 ) -> str:
6922 if index.table is not None:
6923 effective_schema = self.preparer.schema_for_object(index.table)
6924 else:
6925 effective_schema = None
6926 if include_schema and effective_schema:
6927 schema_name = self.preparer.quote_schema(effective_schema)
6928 else:
6929 schema_name = None
6931 index_name: str = self.preparer.format_index(index)
6933 if schema_name:
6934 index_name = schema_name + "." + index_name
6935 return index_name
6937 def visit_add_constraint(self, create, **kw):
6938 return "ALTER TABLE %s ADD %s" % (
6939 self.preparer.format_table(create.element.table),
6940 self.process(create.element),
6941 )
6943 def visit_set_table_comment(self, create, **kw):
6944 return "COMMENT ON TABLE %s IS %s" % (
6945 self.preparer.format_table(create.element),
6946 self.sql_compiler.render_literal_value(
6947 create.element.comment, sqltypes.String()
6948 ),
6949 )
6951 def visit_drop_table_comment(self, drop, **kw):
6952 return "COMMENT ON TABLE %s IS NULL" % self.preparer.format_table(
6953 drop.element
6954 )
6956 def visit_set_column_comment(self, create, **kw):
6957 return "COMMENT ON COLUMN %s IS %s" % (
6958 self.preparer.format_column(
6959 create.element, use_table=True, use_schema=True
6960 ),
6961 self.sql_compiler.render_literal_value(
6962 create.element.comment, sqltypes.String()
6963 ),
6964 )
6966 def visit_drop_column_comment(self, drop, **kw):
6967 return "COMMENT ON COLUMN %s IS NULL" % self.preparer.format_column(
6968 drop.element, use_table=True
6969 )
6971 def visit_set_constraint_comment(self, create, **kw):
6972 raise exc.UnsupportedCompilationError(self, type(create))
6974 def visit_drop_constraint_comment(self, drop, **kw):
6975 raise exc.UnsupportedCompilationError(self, type(drop))
6977 def get_identity_options(self, identity_options):
6978 text = []
6979 if identity_options.increment is not None:
6980 text.append("INCREMENT BY %d" % identity_options.increment)
6981 if identity_options.start is not None:
6982 text.append("START WITH %d" % identity_options.start)
6983 if identity_options.minvalue is not None:
6984 text.append("MINVALUE %d" % identity_options.minvalue)
6985 if identity_options.maxvalue is not None:
6986 text.append("MAXVALUE %d" % identity_options.maxvalue)
6987 if identity_options.nominvalue is not None:
6988 text.append("NO MINVALUE")
6989 if identity_options.nomaxvalue is not None:
6990 text.append("NO MAXVALUE")
6991 if identity_options.cache is not None:
6992 text.append("CACHE %d" % identity_options.cache)
6993 if identity_options.cycle is not None:
6994 text.append("CYCLE" if identity_options.cycle else "NO CYCLE")
6995 return " ".join(text)
6997 def visit_create_sequence(self, create, prefix=None, **kw):
6998 text = "CREATE SEQUENCE "
6999 if create.if_not_exists:
7000 text += "IF NOT EXISTS "
7001 text += self.preparer.format_sequence(create.element)
7003 if prefix:
7004 text += prefix
7005 options = self.get_identity_options(create.element)
7006 if options:
7007 text += " " + options
7008 return text
7010 def visit_drop_sequence(self, drop, **kw):
7011 text = "DROP SEQUENCE "
7012 if drop.if_exists:
7013 text += "IF EXISTS "
7014 return text + self.preparer.format_sequence(drop.element)
7016 def visit_drop_constraint(self, drop, **kw):
7017 constraint = drop.element
7018 if constraint.name is not None:
7019 formatted_name = self.preparer.format_constraint(constraint)
7020 else:
7021 formatted_name = None
7023 if formatted_name is None:
7024 raise exc.CompileError(
7025 "Can't emit DROP CONSTRAINT for constraint %r; "
7026 "it has no name" % drop.element
7027 )
7028 return "ALTER TABLE %s DROP CONSTRAINT %s%s%s" % (
7029 self.preparer.format_table(drop.element.table),
7030 "IF EXISTS " if drop.if_exists else "",
7031 formatted_name,
7032 " CASCADE" if drop.cascade else "",
7033 )
7035 def get_column_specification(self, column, **kwargs):
7036 colspec = (
7037 self.preparer.format_column(column)
7038 + " "
7039 + self.dialect.type_compiler_instance.process(
7040 column.type, type_expression=column
7041 )
7042 )
7043 default = self.get_column_default_string(column)
7044 if default is not None:
7045 colspec += " DEFAULT " + default
7047 if column.computed is not None:
7048 colspec += " " + self.process(column.computed)
7050 if (
7051 column.identity is not None
7052 and self.dialect.supports_identity_columns
7053 ):
7054 colspec += " " + self.process(column.identity)
7056 if not column.nullable and (
7057 not column.identity or not self.dialect.supports_identity_columns
7058 ):
7059 colspec += " NOT NULL"
7060 return colspec
7062 def create_table_suffix(self, table):
7063 return ""
7065 def post_create_table(self, table):
7066 return ""
7068 def get_column_default_string(self, column: Column[Any]) -> Optional[str]:
7069 if isinstance(column.server_default, schema.DefaultClause):
7070 return self.render_default_string(column.server_default.arg)
7071 else:
7072 return None
7074 def render_default_string(self, default: Union[Visitable, str]) -> str:
7075 if isinstance(default, str):
7076 return self.sql_compiler.render_literal_value(
7077 default, sqltypes.STRINGTYPE
7078 )
7079 else:
7080 return self.sql_compiler.process(default, literal_binds=True)
7082 def visit_table_or_column_check_constraint(self, constraint, **kw):
7083 if constraint.is_column_level:
7084 return self.visit_column_check_constraint(constraint)
7085 else:
7086 return self.visit_check_constraint(constraint)
7088 def visit_check_constraint(self, constraint, **kw):
7089 text = ""
7090 if constraint.name is not None:
7091 formatted_name = self.preparer.format_constraint(constraint)
7092 if formatted_name is not None:
7093 text += "CONSTRAINT %s " % formatted_name
7094 text += "CHECK (%s)" % self.sql_compiler.process(
7095 constraint.sqltext, include_table=False, literal_binds=True
7096 )
7097 text += self.define_constraint_deferrability(constraint)
7098 return text
7100 def visit_column_check_constraint(self, constraint, **kw):
7101 text = ""
7102 if constraint.name is not None:
7103 formatted_name = self.preparer.format_constraint(constraint)
7104 if formatted_name is not None:
7105 text += "CONSTRAINT %s " % formatted_name
7106 text += "CHECK (%s)" % self.sql_compiler.process(
7107 constraint.sqltext, include_table=False, literal_binds=True
7108 )
7109 text += self.define_constraint_deferrability(constraint)
7110 return text
7112 def visit_primary_key_constraint(
7113 self, constraint: PrimaryKeyConstraint, **kw: Any
7114 ) -> str:
7115 if len(constraint) == 0:
7116 return ""
7117 text = ""
7118 if constraint.name is not None:
7119 formatted_name = self.preparer.format_constraint(constraint)
7120 if formatted_name is not None:
7121 text += "CONSTRAINT %s " % formatted_name
7122 text += "PRIMARY KEY "
7123 text += "(%s)" % ", ".join(
7124 self.preparer.quote(c.name)
7125 for c in (
7126 constraint.columns_autoinc_first
7127 if constraint._implicit_generated
7128 else constraint.columns
7129 )
7130 )
7131 text += self.define_constraint_deferrability(constraint)
7132 return text
7134 def visit_foreign_key_constraint(self, constraint, **kw):
7135 preparer = self.preparer
7136 text = ""
7137 if constraint.name is not None:
7138 formatted_name = self.preparer.format_constraint(constraint)
7139 if formatted_name is not None:
7140 text += "CONSTRAINT %s " % formatted_name
7141 remote_table = list(constraint.elements)[0].column.table
7142 text += "FOREIGN KEY(%s) REFERENCES %s (%s)" % (
7143 ", ".join(
7144 preparer.quote(f.parent.name) for f in constraint.elements
7145 ),
7146 self.define_constraint_remote_table(
7147 constraint, remote_table, preparer
7148 ),
7149 ", ".join(
7150 preparer.quote(f.column.name) for f in constraint.elements
7151 ),
7152 )
7153 text += self.define_constraint_match(constraint)
7154 text += self.define_constraint_cascades(constraint)
7155 text += self.define_constraint_deferrability(constraint)
7156 return text
7158 def define_constraint_remote_table(self, constraint, table, preparer):
7159 """Format the remote table clause of a CREATE CONSTRAINT clause."""
7161 return preparer.format_table(table)
7163 def visit_unique_constraint(
7164 self, constraint: UniqueConstraint, **kw: Any
7165 ) -> str:
7166 if len(constraint) == 0:
7167 return ""
7168 text = ""
7169 if constraint.name is not None:
7170 formatted_name = self.preparer.format_constraint(constraint)
7171 if formatted_name is not None:
7172 text += "CONSTRAINT %s " % formatted_name
7173 text += "UNIQUE %s(%s)" % (
7174 self.define_unique_constraint_distinct(constraint, **kw),
7175 ", ".join(self.preparer.quote(c.name) for c in constraint),
7176 )
7177 text += self.define_constraint_deferrability(constraint)
7178 return text
7180 def define_unique_constraint_distinct(
7181 self, constraint: UniqueConstraint, **kw: Any
7182 ) -> str:
7183 return ""
7185 def define_constraint_cascades(
7186 self, constraint: ForeignKeyConstraint
7187 ) -> str:
7188 text = ""
7189 if constraint.ondelete is not None:
7190 text += self.define_constraint_ondelete_cascade(constraint)
7192 if constraint.onupdate is not None:
7193 text += self.define_constraint_onupdate_cascade(constraint)
7194 return text
7196 def define_constraint_ondelete_cascade(
7197 self, constraint: ForeignKeyConstraint
7198 ) -> str:
7199 return " ON DELETE %s" % self.preparer.validate_sql_phrase(
7200 constraint.ondelete, FK_ON_DELETE
7201 )
7203 def define_constraint_onupdate_cascade(
7204 self, constraint: ForeignKeyConstraint
7205 ) -> str:
7206 return " ON UPDATE %s" % self.preparer.validate_sql_phrase(
7207 constraint.onupdate, FK_ON_UPDATE
7208 )
7210 def define_constraint_deferrability(self, constraint: Constraint) -> str:
7211 text = ""
7212 if constraint.deferrable is not None:
7213 if constraint.deferrable:
7214 text += " DEFERRABLE"
7215 else:
7216 text += " NOT DEFERRABLE"
7217 if constraint.initially is not None:
7218 text += " INITIALLY %s" % self.preparer.validate_sql_phrase(
7219 constraint.initially, FK_INITIALLY
7220 )
7221 return text
7223 def define_constraint_match(self, constraint):
7224 text = ""
7225 if constraint.match is not None:
7226 text += " MATCH %s" % constraint.match
7227 return text
7229 def visit_computed_column(self, generated, **kw):
7230 text = "GENERATED ALWAYS AS (%s)" % self.sql_compiler.process(
7231 generated.sqltext, include_table=False, literal_binds=True
7232 )
7233 if generated.persisted is True:
7234 text += " STORED"
7235 elif generated.persisted is False:
7236 text += " VIRTUAL"
7237 return text
7239 def visit_identity_column(self, identity, **kw):
7240 text = "GENERATED %s AS IDENTITY" % (
7241 "ALWAYS" if identity.always else "BY DEFAULT",
7242 )
7243 options = self.get_identity_options(identity)
7244 if options:
7245 text += " (%s)" % options
7246 return text
7249class GenericTypeCompiler(TypeCompiler):
7250 def visit_FLOAT(self, type_: sqltypes.Float[Any], **kw: Any) -> str:
7251 return "FLOAT"
7253 def visit_DOUBLE(self, type_: sqltypes.Double[Any], **kw: Any) -> str:
7254 return "DOUBLE"
7256 def visit_DOUBLE_PRECISION(
7257 self, type_: sqltypes.DOUBLE_PRECISION[Any], **kw: Any
7258 ) -> str:
7259 return "DOUBLE PRECISION"
7261 def visit_REAL(self, type_: sqltypes.REAL[Any], **kw: Any) -> str:
7262 return "REAL"
7264 def visit_NUMERIC(self, type_: sqltypes.Numeric[Any], **kw: Any) -> str:
7265 if type_.precision is None:
7266 return "NUMERIC"
7267 elif type_.scale is None:
7268 return "NUMERIC(%(precision)s)" % {"precision": type_.precision}
7269 else:
7270 return "NUMERIC(%(precision)s, %(scale)s)" % {
7271 "precision": type_.precision,
7272 "scale": type_.scale,
7273 }
7275 def visit_DECIMAL(self, type_: sqltypes.DECIMAL[Any], **kw: Any) -> str:
7276 if type_.precision is None:
7277 return "DECIMAL"
7278 elif type_.scale is None:
7279 return "DECIMAL(%(precision)s)" % {"precision": type_.precision}
7280 else:
7281 return "DECIMAL(%(precision)s, %(scale)s)" % {
7282 "precision": type_.precision,
7283 "scale": type_.scale,
7284 }
7286 def visit_INTEGER(self, type_: sqltypes.Integer, **kw: Any) -> str:
7287 return "INTEGER"
7289 def visit_SMALLINT(self, type_: sqltypes.SmallInteger, **kw: Any) -> str:
7290 return "SMALLINT"
7292 def visit_BIGINT(self, type_: sqltypes.BigInteger, **kw: Any) -> str:
7293 return "BIGINT"
7295 def visit_TIMESTAMP(self, type_: sqltypes.TIMESTAMP, **kw: Any) -> str:
7296 return "TIMESTAMP"
7298 def visit_DATETIME(self, type_: sqltypes.DateTime, **kw: Any) -> str:
7299 return "DATETIME"
7301 def visit_DATE(self, type_: sqltypes.Date, **kw: Any) -> str:
7302 return "DATE"
7304 def visit_TIME(self, type_: sqltypes.Time, **kw: Any) -> str:
7305 return "TIME"
7307 def visit_CLOB(self, type_: sqltypes.CLOB, **kw: Any) -> str:
7308 return "CLOB"
7310 def visit_NCLOB(self, type_: sqltypes.Text, **kw: Any) -> str:
7311 return "NCLOB"
7313 def _render_string_type(
7314 self, name: str, length: Optional[int], collation: Optional[str]
7315 ) -> str:
7316 text = name
7317 if length:
7318 text += f"({length})"
7319 if collation:
7320 text += f' COLLATE "{collation}"'
7321 return text
7323 def visit_CHAR(self, type_: sqltypes.CHAR, **kw: Any) -> str:
7324 return self._render_string_type("CHAR", type_.length, type_.collation)
7326 def visit_NCHAR(self, type_: sqltypes.NCHAR, **kw: Any) -> str:
7327 return self._render_string_type("NCHAR", type_.length, type_.collation)
7329 def visit_VARCHAR(self, type_: sqltypes.String, **kw: Any) -> str:
7330 return self._render_string_type(
7331 "VARCHAR", type_.length, type_.collation
7332 )
7334 def visit_NVARCHAR(self, type_: sqltypes.NVARCHAR, **kw: Any) -> str:
7335 return self._render_string_type(
7336 "NVARCHAR", type_.length, type_.collation
7337 )
7339 def visit_TEXT(self, type_: sqltypes.Text, **kw: Any) -> str:
7340 return self._render_string_type("TEXT", type_.length, type_.collation)
7342 def visit_UUID(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
7343 return "UUID"
7345 def visit_BLOB(self, type_: sqltypes.LargeBinary, **kw: Any) -> str:
7346 return "BLOB"
7348 def visit_BINARY(self, type_: sqltypes.BINARY, **kw: Any) -> str:
7349 return "BINARY" + (type_.length and "(%d)" % type_.length or "")
7351 def visit_VARBINARY(self, type_: sqltypes.VARBINARY, **kw: Any) -> str:
7352 return "VARBINARY" + (type_.length and "(%d)" % type_.length or "")
7354 def visit_BOOLEAN(self, type_: sqltypes.Boolean, **kw: Any) -> str:
7355 return "BOOLEAN"
7357 def visit_uuid(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
7358 if not type_.native_uuid or not self.dialect.supports_native_uuid:
7359 return self._render_string_type("CHAR", length=32, collation=None)
7360 else:
7361 return self.visit_UUID(type_, **kw)
7363 def visit_large_binary(
7364 self, type_: sqltypes.LargeBinary, **kw: Any
7365 ) -> str:
7366 return self.visit_BLOB(type_, **kw)
7368 def visit_boolean(self, type_: sqltypes.Boolean, **kw: Any) -> str:
7369 return self.visit_BOOLEAN(type_, **kw)
7371 def visit_time(self, type_: sqltypes.Time, **kw: Any) -> str:
7372 return self.visit_TIME(type_, **kw)
7374 def visit_datetime(self, type_: sqltypes.DateTime, **kw: Any) -> str:
7375 return self.visit_DATETIME(type_, **kw)
7377 def visit_date(self, type_: sqltypes.Date, **kw: Any) -> str:
7378 return self.visit_DATE(type_, **kw)
7380 def visit_big_integer(self, type_: sqltypes.BigInteger, **kw: Any) -> str:
7381 return self.visit_BIGINT(type_, **kw)
7383 def visit_small_integer(
7384 self, type_: sqltypes.SmallInteger, **kw: Any
7385 ) -> str:
7386 return self.visit_SMALLINT(type_, **kw)
7388 def visit_integer(self, type_: sqltypes.Integer, **kw: Any) -> str:
7389 return self.visit_INTEGER(type_, **kw)
7391 def visit_real(self, type_: sqltypes.REAL[Any], **kw: Any) -> str:
7392 return self.visit_REAL(type_, **kw)
7394 def visit_float(self, type_: sqltypes.Float[Any], **kw: Any) -> str:
7395 return self.visit_FLOAT(type_, **kw)
7397 def visit_double(self, type_: sqltypes.Double[Any], **kw: Any) -> str:
7398 return self.visit_DOUBLE(type_, **kw)
7400 def visit_numeric(self, type_: sqltypes.Numeric[Any], **kw: Any) -> str:
7401 return self.visit_NUMERIC(type_, **kw)
7403 def visit_string(self, type_: sqltypes.String, **kw: Any) -> str:
7404 return self.visit_VARCHAR(type_, **kw)
7406 def visit_unicode(self, type_: sqltypes.Unicode, **kw: Any) -> str:
7407 return self.visit_VARCHAR(type_, **kw)
7409 def visit_text(self, type_: sqltypes.Text, **kw: Any) -> str:
7410 return self.visit_TEXT(type_, **kw)
7412 def visit_unicode_text(
7413 self, type_: sqltypes.UnicodeText, **kw: Any
7414 ) -> str:
7415 return self.visit_TEXT(type_, **kw)
7417 def visit_enum(self, type_: sqltypes.Enum, **kw: Any) -> str:
7418 return self.visit_VARCHAR(type_, **kw)
7420 def visit_null(self, type_, **kw):
7421 raise exc.CompileError(
7422 "Can't generate DDL for %r; "
7423 "did you forget to specify a "
7424 "type on this Column?" % type_
7425 )
7427 def visit_type_decorator(
7428 self, type_: TypeDecorator[Any], **kw: Any
7429 ) -> str:
7430 return self.process(type_.type_engine(self.dialect), **kw)
7432 def visit_user_defined(
7433 self, type_: UserDefinedType[Any], **kw: Any
7434 ) -> str:
7435 return type_.get_col_spec(**kw)
7438class StrSQLTypeCompiler(GenericTypeCompiler):
7439 def process(self, type_, **kw):
7440 try:
7441 _compiler_dispatch = type_._compiler_dispatch
7442 except AttributeError:
7443 return self._visit_unknown(type_, **kw)
7444 else:
7445 return _compiler_dispatch(self, **kw)
7447 def __getattr__(self, key):
7448 if key.startswith("visit_"):
7449 return self._visit_unknown
7450 else:
7451 raise AttributeError(key)
7453 def _visit_unknown(self, type_, **kw):
7454 if type_.__class__.__name__ == type_.__class__.__name__.upper():
7455 return type_.__class__.__name__
7456 else:
7457 return repr(type_)
7459 def visit_null(self, type_, **kw):
7460 return "NULL"
7462 def visit_user_defined(self, type_, **kw):
7463 try:
7464 get_col_spec = type_.get_col_spec
7465 except AttributeError:
7466 return repr(type_)
7467 else:
7468 return get_col_spec(**kw)
7471class _SchemaForObjectCallable(Protocol):
7472 def __call__(self, obj: Any, /) -> str: ...
7475class _BindNameForColProtocol(Protocol):
7476 def __call__(self, col: ColumnClause[Any]) -> str: ...
7479class IdentifierPreparer:
7480 """Handle quoting and case-folding of identifiers based on options."""
7482 reserved_words = RESERVED_WORDS
7484 legal_characters = LEGAL_CHARACTERS
7486 illegal_initial_characters = ILLEGAL_INITIAL_CHARACTERS
7488 initial_quote: str
7490 final_quote: str
7492 _strings: MutableMapping[str, str]
7494 schema_for_object: _SchemaForObjectCallable = operator.attrgetter("schema")
7495 """Return the .schema attribute for an object.
7497 For the default IdentifierPreparer, the schema for an object is always
7498 the value of the ".schema" attribute. if the preparer is replaced
7499 with one that has a non-empty schema_translate_map, the value of the
7500 ".schema" attribute is rendered a symbol that will be converted to a
7501 real schema name from the mapping post-compile.
7503 """
7505 _includes_none_schema_translate: bool = False
7507 def __init__(
7508 self,
7509 dialect: Dialect,
7510 initial_quote: str = '"',
7511 final_quote: Optional[str] = None,
7512 escape_quote: str = '"',
7513 quote_case_sensitive_collations: bool = True,
7514 omit_schema: bool = False,
7515 ):
7516 """Construct a new ``IdentifierPreparer`` object.
7518 initial_quote
7519 Character that begins a delimited identifier.
7521 final_quote
7522 Character that ends a delimited identifier. Defaults to
7523 `initial_quote`.
7525 omit_schema
7526 Prevent prepending schema name. Useful for databases that do
7527 not support schemae.
7528 """
7530 self.dialect = dialect
7531 self.initial_quote = initial_quote
7532 self.final_quote = final_quote or self.initial_quote
7533 self.escape_quote = escape_quote
7534 self.escape_to_quote = self.escape_quote * 2
7535 self.omit_schema = omit_schema
7536 self.quote_case_sensitive_collations = quote_case_sensitive_collations
7537 self._strings = {}
7538 self._double_percents = self.dialect.paramstyle in (
7539 "format",
7540 "pyformat",
7541 )
7543 def _with_schema_translate(self, schema_translate_map):
7544 prep = self.__class__.__new__(self.__class__)
7545 prep.__dict__.update(self.__dict__)
7547 includes_none = None in schema_translate_map
7549 def symbol_getter(obj):
7550 name = obj.schema
7551 if obj._use_schema_map and (name is not None or includes_none):
7552 if name is not None and ("[" in name or "]" in name):
7553 raise exc.CompileError(
7554 "Square bracket characters ([]) not supported "
7555 "in schema translate name '%s'" % name
7556 )
7557 return quoted_name(
7558 "__[SCHEMA_%s]" % (name or "_none"), quote=False
7559 )
7560 else:
7561 return obj.schema
7563 prep.schema_for_object = symbol_getter
7564 prep._includes_none_schema_translate = includes_none
7565 return prep
7567 def _render_schema_translates(
7568 self, statement: str, schema_translate_map: SchemaTranslateMapType
7569 ) -> str:
7570 d = schema_translate_map
7571 if None in d:
7572 if not self._includes_none_schema_translate:
7573 raise exc.InvalidRequestError(
7574 "schema translate map which previously did not have "
7575 "`None` present as a key now has `None` present; compiled "
7576 "statement may lack adequate placeholders. Please use "
7577 "consistent keys in successive "
7578 "schema_translate_map dictionaries."
7579 )
7581 d["_none"] = d[None] # type: ignore[index]
7583 def replace(m):
7584 name = m.group(2)
7585 if name in d:
7586 effective_schema = d[name]
7587 else:
7588 if name in (None, "_none"):
7589 raise exc.InvalidRequestError(
7590 "schema translate map which previously had `None` "
7591 "present as a key now no longer has it present; don't "
7592 "know how to apply schema for compiled statement. "
7593 "Please use consistent keys in successive "
7594 "schema_translate_map dictionaries."
7595 )
7596 effective_schema = name
7598 if not effective_schema:
7599 effective_schema = self.dialect.default_schema_name
7600 if not effective_schema:
7601 # TODO: no coverage here
7602 raise exc.CompileError(
7603 "Dialect has no default schema name; can't "
7604 "use None as dynamic schema target."
7605 )
7606 return self.quote_schema(effective_schema)
7608 return re.sub(r"(__\[SCHEMA_([^\]]+)\])", replace, statement)
7610 def _escape_identifier(self, value: str) -> str:
7611 """Escape an identifier.
7613 Subclasses should override this to provide database-dependent
7614 escaping behavior.
7615 """
7617 value = value.replace(self.escape_quote, self.escape_to_quote)
7618 if self._double_percents:
7619 value = value.replace("%", "%%")
7620 return value
7622 def _unescape_identifier(self, value: str) -> str:
7623 """Canonicalize an escaped identifier.
7625 Subclasses should override this to provide database-dependent
7626 unescaping behavior that reverses _escape_identifier.
7627 """
7629 return value.replace(self.escape_to_quote, self.escape_quote)
7631 def validate_sql_phrase(self, element, reg):
7632 """keyword sequence filter.
7634 a filter for elements that are intended to represent keyword sequences,
7635 such as "INITIALLY", "INITIALLY DEFERRED", etc. no special characters
7636 should be present.
7638 """
7640 if element is not None and not reg.match(element):
7641 raise exc.CompileError(
7642 "Unexpected SQL phrase: %r (matching against %r)"
7643 % (element, reg.pattern)
7644 )
7645 return element
7647 def quote_identifier(self, value: str) -> str:
7648 """Quote an identifier.
7650 Subclasses should override this to provide database-dependent
7651 quoting behavior.
7652 """
7654 return (
7655 self.initial_quote
7656 + self._escape_identifier(value)
7657 + self.final_quote
7658 )
7660 def _requires_quotes(self, value: str) -> bool:
7661 """Return True if the given identifier requires quoting."""
7662 lc_value = value.lower()
7663 return (
7664 lc_value in self.reserved_words
7665 or value[0] in self.illegal_initial_characters
7666 or not self.legal_characters.match(str(value))
7667 or (lc_value != value)
7668 )
7670 def _requires_quotes_illegal_chars(self, value):
7671 """Return True if the given identifier requires quoting, but
7672 not taking case convention into account."""
7673 return not self.legal_characters.match(str(value))
7675 def quote_schema(self, schema: str) -> str:
7676 """Conditionally quote a schema name.
7679 The name is quoted if it is a reserved word, contains quote-necessary
7680 characters, or is an instance of :class:`.quoted_name` which includes
7681 ``quote`` set to ``True``.
7683 Subclasses can override this to provide database-dependent
7684 quoting behavior for schema names.
7686 :param schema: string schema name
7687 """
7688 return self.quote(schema)
7690 def quote(self, ident: str) -> str:
7691 """Conditionally quote an identifier.
7693 The identifier is quoted if it is a reserved word, contains
7694 quote-necessary characters, or is an instance of
7695 :class:`.quoted_name` which includes ``quote`` set to ``True``.
7697 Subclasses can override this to provide database-dependent
7698 quoting behavior for identifier names.
7700 :param ident: string identifier
7701 """
7702 force = getattr(ident, "quote", None)
7704 if force is None:
7705 if ident in self._strings:
7706 return self._strings[ident]
7707 else:
7708 if self._requires_quotes(ident):
7709 self._strings[ident] = self.quote_identifier(ident)
7710 else:
7711 self._strings[ident] = ident
7712 return self._strings[ident]
7713 elif force:
7714 return self.quote_identifier(ident)
7715 else:
7716 return ident
7718 def format_collation(self, collation_name):
7719 if self.quote_case_sensitive_collations:
7720 return self.quote(collation_name)
7721 else:
7722 return collation_name
7724 def format_sequence(
7725 self, sequence: schema.Sequence, use_schema: bool = True
7726 ) -> str:
7727 name = self.quote(sequence.name)
7729 effective_schema = self.schema_for_object(sequence)
7731 if (
7732 not self.omit_schema
7733 and use_schema
7734 and effective_schema is not None
7735 ):
7736 name = self.quote_schema(effective_schema) + "." + name
7737 return name
7739 def format_label(
7740 self, label: Label[Any], name: Optional[str] = None
7741 ) -> str:
7742 return self.quote(name or label.name)
7744 def format_alias(
7745 self, alias: Optional[AliasedReturnsRows], name: Optional[str] = None
7746 ) -> str:
7747 if name is None:
7748 assert alias is not None
7749 return self.quote(alias.name)
7750 else:
7751 return self.quote(name)
7753 def format_savepoint(self, savepoint, name=None):
7754 # Running the savepoint name through quoting is unnecessary
7755 # for all known dialects. This is here to support potential
7756 # third party use cases
7757 ident = name or savepoint.ident
7758 if self._requires_quotes(ident):
7759 ident = self.quote_identifier(ident)
7760 return ident
7762 @util.preload_module("sqlalchemy.sql.naming")
7763 def format_constraint(
7764 self, constraint: Union[Constraint, Index], _alembic_quote: bool = True
7765 ) -> Optional[str]:
7766 naming = util.preloaded.sql_naming
7768 if constraint.name is _NONE_NAME:
7769 name = naming._constraint_name_for_table(
7770 constraint, constraint.table
7771 )
7773 if name is None:
7774 return None
7775 else:
7776 name = constraint.name
7778 assert name is not None
7779 if constraint.__visit_name__ == "index":
7780 return self.truncate_and_render_index_name(
7781 name, _alembic_quote=_alembic_quote
7782 )
7783 else:
7784 return self.truncate_and_render_constraint_name(
7785 name, _alembic_quote=_alembic_quote
7786 )
7788 def truncate_and_render_index_name(
7789 self, name: str, _alembic_quote: bool = True
7790 ) -> str:
7791 # calculate these at format time so that ad-hoc changes
7792 # to dialect.max_identifier_length etc. can be reflected
7793 # as IdentifierPreparer is long lived
7794 max_ = (
7795 self.dialect.max_index_name_length
7796 or self.dialect.max_identifier_length
7797 )
7798 return self._truncate_and_render_maxlen_name(
7799 name, max_, _alembic_quote
7800 )
7802 def truncate_and_render_constraint_name(
7803 self, name: str, _alembic_quote: bool = True
7804 ) -> str:
7805 # calculate these at format time so that ad-hoc changes
7806 # to dialect.max_identifier_length etc. can be reflected
7807 # as IdentifierPreparer is long lived
7808 max_ = (
7809 self.dialect.max_constraint_name_length
7810 or self.dialect.max_identifier_length
7811 )
7812 return self._truncate_and_render_maxlen_name(
7813 name, max_, _alembic_quote
7814 )
7816 def _truncate_and_render_maxlen_name(
7817 self, name: str, max_: int, _alembic_quote: bool
7818 ) -> str:
7819 if isinstance(name, elements._truncated_label):
7820 if len(name) > max_:
7821 name = name[0 : max_ - 8] + "_" + util.md5_hex(name)[-4:]
7822 else:
7823 self.dialect.validate_identifier(name)
7825 if not _alembic_quote:
7826 return name
7827 else:
7828 return self.quote(name)
7830 def format_index(self, index: Index) -> str:
7831 name = self.format_constraint(index)
7832 assert name is not None
7833 return name
7835 def format_table(
7836 self,
7837 table: FromClause,
7838 use_schema: bool = True,
7839 name: Optional[str] = None,
7840 ) -> str:
7841 """Prepare a quoted table and schema name."""
7842 if name is None:
7843 if TYPE_CHECKING:
7844 assert isinstance(table, NamedFromClause)
7845 name = table.name
7847 result = self.quote(name)
7849 effective_schema = self.schema_for_object(table)
7851 if not self.omit_schema and use_schema and effective_schema:
7852 result = self.quote_schema(effective_schema) + "." + result
7853 return result
7855 def format_schema(self, name):
7856 """Prepare a quoted schema name."""
7858 return self.quote(name)
7860 def format_label_name(
7861 self,
7862 name,
7863 anon_map=None,
7864 ):
7865 """Prepare a quoted column name."""
7867 if anon_map is not None and isinstance(
7868 name, elements._truncated_label
7869 ):
7870 name = name.apply_map(anon_map)
7872 return self.quote(name)
7874 def format_column(
7875 self,
7876 column: ColumnElement[Any],
7877 use_table: bool = False,
7878 name: Optional[str] = None,
7879 table_name: Optional[str] = None,
7880 use_schema: bool = False,
7881 anon_map: Optional[Mapping[str, Any]] = None,
7882 ) -> str:
7883 """Prepare a quoted column name."""
7885 if name is None:
7886 name = column.name
7887 assert name is not None
7889 if anon_map is not None and isinstance(
7890 name, elements._truncated_label
7891 ):
7892 name = name.apply_map(anon_map)
7894 if not getattr(column, "is_literal", False):
7895 if use_table:
7896 return (
7897 self.format_table(
7898 column.table, use_schema=use_schema, name=table_name
7899 )
7900 + "."
7901 + self.quote(name)
7902 )
7903 else:
7904 return self.quote(name)
7905 else:
7906 # literal textual elements get stuck into ColumnClause a lot,
7907 # which shouldn't get quoted
7909 if use_table:
7910 return (
7911 self.format_table(
7912 column.table, use_schema=use_schema, name=table_name
7913 )
7914 + "."
7915 + name
7916 )
7917 else:
7918 return name
7920 def format_table_seq(self, table, use_schema=True):
7921 """Format table name and schema as a tuple."""
7923 # Dialects with more levels in their fully qualified references
7924 # ('database', 'owner', etc.) could override this and return
7925 # a longer sequence.
7927 effective_schema = self.schema_for_object(table)
7929 if not self.omit_schema and use_schema and effective_schema:
7930 return (
7931 self.quote_schema(effective_schema),
7932 self.format_table(table, use_schema=False),
7933 )
7934 else:
7935 return (self.format_table(table, use_schema=False),)
7937 @util.memoized_property
7938 def _r_identifiers(self):
7939 initial, final, escaped_final = (
7940 re.escape(s)
7941 for s in (
7942 self.initial_quote,
7943 self.final_quote,
7944 self._escape_identifier(self.final_quote),
7945 )
7946 )
7947 r = re.compile(
7948 r"(?:"
7949 r"(?:%(initial)s((?:%(escaped)s|[^%(final)s])+)%(final)s"
7950 r"|([^\.]+))(?=\.|$))+"
7951 % {"initial": initial, "final": final, "escaped": escaped_final}
7952 )
7953 return r
7955 def unformat_identifiers(self, identifiers: str) -> Sequence[str]:
7956 """Unpack 'schema.table.column'-like strings into components."""
7958 r = self._r_identifiers
7959 return [
7960 self._unescape_identifier(i)
7961 for i in [a or b for a, b in r.findall(identifiers)]
7962 ]