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-2026 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"""
27from __future__ import annotations
29import collections
30import collections.abc as collections_abc
31import contextlib
32from enum import IntEnum
33import functools
34import itertools
35import operator
36import re
37from time import perf_counter
38import typing
39from typing import AbstractSet
40from typing import Any
41from typing import Callable
42from typing import cast
43from typing import ClassVar
44from typing import Dict
45from typing import Final
46from typing import FrozenSet
47from typing import Iterable
48from typing import Iterator
49from typing import List
50from typing import Literal
51from typing import Mapping
52from typing import MutableMapping
53from typing import NamedTuple
54from typing import NoReturn
55from typing import Optional
56from typing import Pattern
57from typing import Protocol
58from typing import Sequence
59from typing import Set
60from typing import Tuple
61from typing import Type
62from typing import TYPE_CHECKING
63from typing import TypedDict
64from typing import Union
66from . import base
67from . import coercions
68from . import crud
69from . import elements
70from . import functions
71from . import operators
72from . import roles
73from . import schema
74from . import selectable
75from . import sqltypes
76from . import util as sql_util
77from ._typing import is_column_element
78from ._typing import is_dml
79from .base import _de_clone
80from .base import _from_objects
81from .base import _NONE_NAME
82from .base import _SentinelDefaultCharacterization
83from .base import NO_ARG
84from .elements import quoted_name
85from .sqltypes import TupleType
86from .visitors import prefix_anon_map
87from .. import exc
88from .. import util
89from ..util import FastIntFlag
90from ..util.typing import Self
91from ..util.typing import TupleAny
92from ..util.typing import Unpack
94if typing.TYPE_CHECKING:
95 from .annotation import _AnnotationDict
96 from .base import _AmbiguousTableNameMap
97 from .base import CompileState
98 from .base import Executable
99 from .base import ExecutableStatement
100 from .cache_key import CacheKey
101 from .ddl import _TableViaSelect
102 from .ddl import CreateTableAs
103 from .ddl import CreateView
104 from .ddl import ExecutableDDLElement
105 from .dml import Delete
106 from .dml import Insert
107 from .dml import Update
108 from .dml import UpdateBase
109 from .dml import UpdateDMLState
110 from .dml import ValuesBase
111 from .elements import _truncated_label
112 from .elements import BinaryExpression
113 from .elements import BindParameter
114 from .elements import ClauseElement
115 from .elements import ColumnClause
116 from .elements import ColumnElement
117 from .elements import False_
118 from .elements import Label
119 from .elements import Null
120 from .elements import True_
121 from .functions import Function
122 from .schema import CheckConstraint
123 from .schema import Column
124 from .schema import Constraint
125 from .schema import ForeignKeyConstraint
126 from .schema import IdentityOptions
127 from .schema import Index
128 from .schema import PrimaryKeyConstraint
129 from .schema import Table
130 from .schema import UniqueConstraint
131 from .selectable import _ColumnsClauseElement
132 from .selectable import AliasedReturnsRows
133 from .selectable import CompoundSelectState
134 from .selectable import CTE
135 from .selectable import FromClause
136 from .selectable import NamedFromClause
137 from .selectable import ReturnsRows
138 from .selectable import Select
139 from .selectable import SelectState
140 from .type_api import _BindProcessorType
141 from .type_api import TypeDecorator
142 from .type_api import TypeEngine
143 from .type_api import UserDefinedType
144 from .visitors import Visitable
145 from ..engine.cursor import CursorResultMetaData
146 from ..engine.interfaces import _CoreSingleExecuteParams
147 from ..engine.interfaces import _DBAPIAnyExecuteParams
148 from ..engine.interfaces import _DBAPIMultiExecuteParams
149 from ..engine.interfaces import _DBAPISingleExecuteParams
150 from ..engine.interfaces import _ExecuteOptions
151 from ..engine.interfaces import _GenericSetInputSizesType
152 from ..engine.interfaces import _MutableCoreSingleExecuteParams
153 from ..engine.interfaces import Dialect
154 from ..engine.interfaces import SchemaTranslateMapType
157_FromHintsType = Dict["FromClause", str]
159RESERVED_WORDS = {
160 "all",
161 "analyse",
162 "analyze",
163 "and",
164 "any",
165 "array",
166 "as",
167 "asc",
168 "asymmetric",
169 "authorization",
170 "between",
171 "binary",
172 "both",
173 "case",
174 "cast",
175 "check",
176 "collate",
177 "column",
178 "constraint",
179 "create",
180 "cross",
181 "current_date",
182 "current_role",
183 "current_time",
184 "current_timestamp",
185 "current_user",
186 "default",
187 "deferrable",
188 "desc",
189 "distinct",
190 "do",
191 "else",
192 "end",
193 "except",
194 "false",
195 "for",
196 "foreign",
197 "freeze",
198 "from",
199 "full",
200 "grant",
201 "group",
202 "having",
203 "ilike",
204 "in",
205 "initially",
206 "inner",
207 "intersect",
208 "into",
209 "is",
210 "isnull",
211 "join",
212 "leading",
213 "left",
214 "like",
215 "limit",
216 "localtime",
217 "localtimestamp",
218 "natural",
219 "new",
220 "not",
221 "notnull",
222 "null",
223 "off",
224 "offset",
225 "old",
226 "on",
227 "only",
228 "or",
229 "order",
230 "outer",
231 "overlaps",
232 "placing",
233 "primary",
234 "references",
235 "right",
236 "select",
237 "session_user",
238 "set",
239 "similar",
240 "some",
241 "symmetric",
242 "table",
243 "then",
244 "to",
245 "trailing",
246 "true",
247 "union",
248 "unique",
249 "user",
250 "using",
251 "verbose",
252 "when",
253 "where",
254}
256LEGAL_CHARACTERS = re.compile(r"^[A-Z0-9_$]+$", re.I)
257LEGAL_CHARACTERS_PLUS_SPACE = re.compile(r"^[A-Z0-9_ $]+$", re.I)
258ILLEGAL_INITIAL_CHARACTERS = {str(x) for x in range(0, 10)}.union(["$"])
260FK_ON_DELETE = re.compile(
261 r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I
262)
263FK_ON_UPDATE = re.compile(
264 r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I
265)
266FK_INITIALLY = re.compile(r"^(?:DEFERRED|IMMEDIATE)$", re.I)
267_WINDOW_EXCLUDE_RE = re.compile(
268 r"^(?:CURRENT ROW|GROUP|TIES|NO OTHERS)$", re.I
269)
270BIND_PARAMS = elements.TextClause._bind_params_regex
271BIND_PARAMS_ESC = re.compile(r"\x5c(:\w*)(?![:\w])", re.UNICODE)
273_pyformat_template = "%%(%(name)s)s"
274BIND_TEMPLATES = {
275 "pyformat": _pyformat_template,
276 "qmark": "?",
277 "format": "%%s",
278 "numeric": ":[_POSITION]",
279 "numeric_dollar": "$[_POSITION]",
280 "named": ":%(name)s",
281}
284OPERATORS = {
285 # binary
286 operators.and_: " AND ",
287 operators.or_: " OR ",
288 operators.add: " + ",
289 operators.mul: " * ",
290 operators.sub: " - ",
291 operators.mod: " % ",
292 operators.neg: "-",
293 operators.lt: " < ",
294 operators.le: " <= ",
295 operators.ne: " != ",
296 operators.gt: " > ",
297 operators.ge: " >= ",
298 operators.eq: " = ",
299 operators.is_distinct_from: " IS DISTINCT FROM ",
300 operators.is_not_distinct_from: " IS NOT DISTINCT FROM ",
301 operators.concat_op: " || ",
302 operators.match_op: " MATCH ",
303 operators.not_match_op: " NOT MATCH ",
304 operators.in_op: " IN ",
305 operators.not_in_op: " NOT IN ",
306 operators.comma_op: ", ",
307 operators.from_: " FROM ",
308 operators.as_: " AS ",
309 operators.is_: " IS ",
310 operators.is_not: " IS NOT ",
311 operators.collate: " COLLATE ",
312 # unary
313 operators.exists: "EXISTS ",
314 operators.distinct_op: "DISTINCT ",
315 operators.inv: "NOT ",
316 operators.any_op: "ANY ",
317 operators.all_op: "ALL ",
318 # modifiers
319 operators.desc_op: " DESC",
320 operators.asc_op: " ASC",
321 operators.nulls_first_op: " NULLS FIRST",
322 operators.nulls_last_op: " NULLS LAST",
323 # bitwise
324 operators.bitwise_xor_op: " ^ ",
325 operators.bitwise_or_op: " | ",
326 operators.bitwise_and_op: " & ",
327 operators.bitwise_not_op: "~",
328 operators.bitwise_lshift_op: " << ",
329 operators.bitwise_rshift_op: " >> ",
330}
332FUNCTIONS: Dict[Type[Function[Any]], str] = {
333 functions.coalesce: "coalesce",
334 functions.current_date: "CURRENT_DATE",
335 functions.current_time: "CURRENT_TIME",
336 functions.current_timestamp: "CURRENT_TIMESTAMP",
337 functions.current_user: "CURRENT_USER",
338 functions.localtime: "LOCALTIME",
339 functions.localtimestamp: "LOCALTIMESTAMP",
340 functions.random: "random",
341 functions.sysdate: "sysdate",
342 functions.session_user: "SESSION_USER",
343 functions.user: "USER",
344 functions.cube: "CUBE",
345 functions.rollup: "ROLLUP",
346 functions.grouping_sets: "GROUPING SETS",
347}
350EXTRACT_MAP = {
351 "month": "month",
352 "day": "day",
353 "year": "year",
354 "second": "second",
355 "hour": "hour",
356 "doy": "doy",
357 "minute": "minute",
358 "quarter": "quarter",
359 "dow": "dow",
360 "week": "week",
361 "epoch": "epoch",
362 "milliseconds": "milliseconds",
363 "microseconds": "microseconds",
364 "timezone_hour": "timezone_hour",
365 "timezone_minute": "timezone_minute",
366}
368COMPOUND_KEYWORDS = {
369 selectable._CompoundSelectKeyword.UNION: "UNION",
370 selectable._CompoundSelectKeyword.UNION_ALL: "UNION ALL",
371 selectable._CompoundSelectKeyword.EXCEPT: "EXCEPT",
372 selectable._CompoundSelectKeyword.EXCEPT_ALL: "EXCEPT ALL",
373 selectable._CompoundSelectKeyword.INTERSECT: "INTERSECT",
374 selectable._CompoundSelectKeyword.INTERSECT_ALL: "INTERSECT ALL",
375}
378class ResultColumnsEntry(NamedTuple):
379 """Tracks a column expression that is expected to be represented
380 in the result rows for this statement.
382 This normally refers to the columns clause of a SELECT statement
383 but may also refer to a RETURNING clause, as well as for dialect-specific
384 emulations.
386 """
388 keyname: str
389 """string name that's expected in cursor.description"""
391 name: str
392 """column name, may be labeled"""
394 objects: Tuple[Any, ...]
395 """sequence of objects that should be able to locate this column
396 in a RowMapping. This is typically string names and aliases
397 as well as Column objects.
399 """
401 type: TypeEngine[Any]
402 """Datatype to be associated with this column. This is where
403 the "result processing" logic directly links the compiled statement
404 to the rows that come back from the cursor.
406 """
409class _ResultMapAppender(Protocol):
410 def __call__(
411 self,
412 keyname: str,
413 name: str,
414 objects: Sequence[Any],
415 type_: TypeEngine[Any],
416 ) -> None: ...
419# integer indexes into ResultColumnsEntry used by cursor.py.
420# some profiling showed integer access faster than named tuple
421RM_RENDERED_NAME: Literal[0] = 0
422RM_NAME: Literal[1] = 1
423RM_OBJECTS: Literal[2] = 2
424RM_TYPE: Literal[3] = 3
427class _BaseCompilerStackEntry(TypedDict):
428 asfrom_froms: Set[FromClause]
429 correlate_froms: Set[FromClause]
430 selectable: ReturnsRows
433class _CompilerStackEntry(_BaseCompilerStackEntry, total=False):
434 compile_state: CompileState
435 need_result_map_for_nested: bool
436 need_result_map_for_compound: bool
437 select_0: ReturnsRows
438 insert_from_select: Select[Unpack[TupleAny]]
441class ExpandedState(NamedTuple):
442 """represents state to use when producing "expanded" and
443 "post compile" bound parameters for a statement.
445 "expanded" parameters are parameters that are generated at
446 statement execution time to suit a number of parameters passed, the most
447 prominent example being the individual elements inside of an IN expression.
449 "post compile" parameters are parameters where the SQL literal value
450 will be rendered into the SQL statement at execution time, rather than
451 being passed as separate parameters to the driver.
453 To create an :class:`.ExpandedState` instance, use the
454 :meth:`.SQLCompiler.construct_expanded_state` method on any
455 :class:`.SQLCompiler` instance.
457 """
459 statement: str
460 """String SQL statement with parameters fully expanded"""
462 parameters: _CoreSingleExecuteParams
463 """Parameter dictionary with parameters fully expanded.
465 For a statement that uses named parameters, this dictionary will map
466 exactly to the names in the statement. For a statement that uses
467 positional parameters, the :attr:`.ExpandedState.positional_parameters`
468 will yield a tuple with the positional parameter set.
470 """
472 processors: Mapping[str, _BindProcessorType[Any]]
473 """mapping of bound value processors"""
475 positiontup: Optional[Sequence[str]]
476 """Sequence of string names indicating the order of positional
477 parameters"""
479 parameter_expansion: Mapping[str, List[str]]
480 """Mapping representing the intermediary link from original parameter
481 name to list of "expanded" parameter names, for those parameters that
482 were expanded."""
484 @property
485 def positional_parameters(self) -> Tuple[Any, ...]:
486 """Tuple of positional parameters, for statements that were compiled
487 using a positional paramstyle.
489 """
490 if self.positiontup is None:
491 raise exc.InvalidRequestError(
492 "statement does not use a positional paramstyle"
493 )
494 return tuple(self.parameters[key] for key in self.positiontup)
496 @property
497 def additional_parameters(self) -> _CoreSingleExecuteParams:
498 """synonym for :attr:`.ExpandedState.parameters`."""
499 return self.parameters
502class _InsertManyValues(NamedTuple):
503 """represents state to use for executing an "insertmanyvalues" statement.
505 The primary consumers of this object are the
506 :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and
507 :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods.
509 .. versionadded:: 2.0
511 """
513 is_default_expr: bool
514 """if True, the statement is of the form
515 ``INSERT INTO TABLE DEFAULT VALUES``, and can't be rewritten as a "batch"
517 """
519 single_values_expr: str
520 """The rendered "values" clause of the INSERT statement.
522 This is typically the parenthesized section e.g. "(?, ?, ?)" or similar.
523 The insertmanyvalues logic uses this string as a search and replace
524 target.
526 """
528 insert_crud_params: List[crud._CrudParamElementStr]
529 """List of Column / bind names etc. used while rewriting the statement"""
531 num_positional_params_counted: int
532 """the number of bound parameters in a single-row statement.
534 This count may be larger or smaller than the actual number of columns
535 targeted in the INSERT, as it accommodates for SQL expressions
536 in the values list that may have zero or more parameters embedded
537 within them.
539 This count is part of what's used to organize rewritten parameter lists
540 when batching.
542 """
544 sort_by_parameter_order: bool = False
545 """if the deterministic_returnined_order parameter were used on the
546 insert.
548 All of the attributes following this will only be used if this is True.
550 """
552 includes_upsert_behaviors: bool = False
553 """if True, we have to accommodate for upsert behaviors.
555 This will in some cases downgrade "insertmanyvalues" that requests
556 deterministic ordering.
558 """
560 sentinel_columns: Optional[Sequence[Column[Any]]] = None
561 """List of sentinel columns that were located.
563 This list is only here if the INSERT asked for
564 sort_by_parameter_order=True,
565 and dialect-appropriate sentinel columns were located.
567 .. versionadded:: 2.0.10
569 """
571 num_sentinel_columns: int = 0
572 """how many sentinel columns are in the above list, if any.
574 This is the same as
575 ``len(sentinel_columns) if sentinel_columns is not None else 0``
577 """
579 sentinel_param_keys: Optional[Sequence[str]] = None
580 """parameter str keys in each param dictionary / tuple
581 that would link to the client side "sentinel" values for that row, which
582 we can use to match up parameter sets to result rows.
584 This is only present if sentinel_columns is present and the INSERT
585 statement actually refers to client side values for these sentinel
586 columns.
588 .. versionadded:: 2.0.10
590 .. versionchanged:: 2.0.29 - the sequence is now string dictionary keys
591 only, used against the "compiled parameteters" collection before
592 the parameters were converted by bound parameter processors
594 """
596 implicit_sentinel: bool = False
597 """if True, we have exactly one sentinel column and it uses a server side
598 value, currently has to generate an incrementing integer value.
600 The dialect in question would have asserted that it supports receiving
601 these values back and sorting on that value as a means of guaranteeing
602 correlation with the incoming parameter list.
604 .. versionadded:: 2.0.10
606 """
608 has_upsert_bound_parameters: bool = False
609 """if True, the upsert SET clause contains bound parameters that will
610 receive their values from the parameters dict (i.e., parametrized
611 bindparams where value is None and callable is None).
613 This means we can't batch multiple rows in a single statement, since
614 each row would need different values in the SET clause but there's only
615 one SET clause per statement. See issue #13130.
617 .. versionadded:: 2.0.37
619 """
621 embed_values_counter: bool = False
622 """Whether to embed an incrementing integer counter in each parameter
623 set within the VALUES clause as parameters are batched over.
625 This is only used for a specific INSERT..SELECT..VALUES..RETURNING syntax
626 where a subquery is used to produce value tuples. Current support
627 includes PostgreSQL, Microsoft SQL Server.
629 .. versionadded:: 2.0.10
631 """
634class _InsertManyValuesBatch(NamedTuple):
635 """represents an individual batch SQL statement for insertmanyvalues.
637 This is passed through the
638 :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and
639 :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods out
640 to the :class:`.Connection` within the
641 :meth:`.Connection._exec_insertmany_context` method.
643 .. versionadded:: 2.0.10
645 """
647 replaced_statement: str
648 replaced_parameters: _DBAPIAnyExecuteParams
649 processed_setinputsizes: Optional[_GenericSetInputSizesType]
650 batch: Sequence[_DBAPISingleExecuteParams]
651 sentinel_values: Sequence[Tuple[Any, ...]]
652 current_batch_size: int
653 batchnum: int
654 total_batches: int
655 rows_sorted: bool
656 is_downgraded: bool
659class InsertmanyvaluesSentinelOpts(FastIntFlag):
660 """bitflag enum indicating styles of PK defaults
661 which can work as implicit sentinel columns
663 """
665 NOT_SUPPORTED = 1
666 AUTOINCREMENT = 2
667 IDENTITY = 4
668 SEQUENCE = 8
669 MONOTONIC_FUNCTION = 16
671 ANY_AUTOINCREMENT = (
672 AUTOINCREMENT | IDENTITY | SEQUENCE | MONOTONIC_FUNCTION
673 )
674 _SUPPORTED_OR_NOT = NOT_SUPPORTED | ANY_AUTOINCREMENT
676 USE_INSERT_FROM_SELECT = 32
677 RENDER_SELECT_COL_CASTS = 64
680class AggregateOrderByStyle(IntEnum):
681 """Describes backend database's capabilities with ORDER BY for aggregate
682 functions
684 .. versionadded:: 2.1
686 """
688 NONE = 0
689 """database has no ORDER BY for aggregate functions"""
691 INLINE = 1
692 """ORDER BY is rendered inside the function's argument list, typically as
693 the last element"""
695 WITHIN_GROUP = 2
696 """the WITHIN GROUP (ORDER BY ...) phrase is used for all aggregate
697 functions (not just the ordered set ones)"""
700class CompilerState(IntEnum):
701 COMPILING = 0
702 """statement is present, compilation phase in progress"""
704 STRING_APPLIED = 1
705 """statement is present, string form of the statement has been applied.
707 Additional processors by subclasses may still be pending.
709 """
711 NO_STATEMENT = 2
712 """compiler does not have a statement to compile, is used
713 for method access"""
716class Linting(IntEnum):
717 """represent preferences for the 'SQL linting' feature.
719 this feature currently includes support for flagging cartesian products
720 in SQL statements.
722 """
724 NO_LINTING = 0
725 "Disable all linting."
727 COLLECT_CARTESIAN_PRODUCTS = 1
728 """Collect data on FROMs and cartesian products and gather into
729 'self.from_linter'"""
731 WARN_LINTING = 2
732 "Emit warnings for linters that find problems"
734 FROM_LINTING = COLLECT_CARTESIAN_PRODUCTS | WARN_LINTING
735 """Warn for cartesian products; combines COLLECT_CARTESIAN_PRODUCTS
736 and WARN_LINTING"""
739NO_LINTING, COLLECT_CARTESIAN_PRODUCTS, WARN_LINTING, FROM_LINTING = tuple(
740 Linting
741)
744class FromLinter(collections.namedtuple("FromLinter", ["froms", "edges"])):
745 """represents current state for the "cartesian product" detection
746 feature."""
748 def lint(self, start=None):
749 froms = self.froms
750 if not froms:
751 return None, None
753 edges = set(self.edges)
754 the_rest = set(froms)
756 if start is not None:
757 start_with = start
758 the_rest.remove(start_with)
759 else:
760 start_with = the_rest.pop()
762 stack = collections.deque([start_with])
764 while stack and the_rest:
765 node = stack.popleft()
766 the_rest.discard(node)
768 # comparison of nodes in edges here is based on hash equality, as
769 # there are "annotated" elements that match the non-annotated ones.
770 # to remove the need for in-python hash() calls, use native
771 # containment routines (e.g. "node in edge", "edge.index(node)")
772 to_remove = {edge for edge in edges if node in edge}
774 # appendleft the node in each edge that is not
775 # the one that matched.
776 stack.extendleft(edge[not edge.index(node)] for edge in to_remove)
777 edges.difference_update(to_remove)
779 # FROMS left over? boom
780 if the_rest:
781 return the_rest, start_with
782 else:
783 return None, None
785 def warn(self, stmt_type="SELECT"):
786 the_rest, start_with = self.lint()
788 # FROMS left over? boom
789 if the_rest:
790 froms = the_rest
791 if froms:
792 template = (
793 "{stmt_type} statement has a cartesian product between "
794 "FROM element(s) {froms} and "
795 'FROM element "{start}". Apply join condition(s) '
796 "between each element to resolve."
797 )
798 froms_str = ", ".join(
799 f'"{self.froms[from_]}"' for from_ in froms
800 )
801 message = template.format(
802 stmt_type=stmt_type,
803 froms=froms_str,
804 start=self.froms[start_with],
805 )
807 util.warn(message)
810class Compiled:
811 """Represent a compiled SQL or DDL expression.
813 The ``__str__`` method of the ``Compiled`` object should produce
814 the actual text of the statement. ``Compiled`` objects are
815 specific to their underlying database dialect, and also may
816 or may not be specific to the columns referenced within a
817 particular set of bind parameters. In no case should the
818 ``Compiled`` object be dependent on the actual values of those
819 bind parameters, even though it may reference those values as
820 defaults.
821 """
823 statement: Optional[ClauseElement] = None
824 "The statement to compile."
825 string: str = ""
826 "The string representation of the ``statement``"
828 state: CompilerState
829 """description of the compiler's state"""
831 is_sql = False
832 is_ddl = False
834 _cached_metadata: Optional[CursorResultMetaData] = None
836 _result_columns: Optional[List[ResultColumnsEntry]] = None
838 schema_translate_map: Optional[SchemaTranslateMapType] = None
840 execution_options: _ExecuteOptions = util.EMPTY_DICT
841 """
842 Execution options propagated from the statement. In some cases,
843 sub-elements of the statement can modify these.
844 """
846 preparer: IdentifierPreparer
848 _annotations: _AnnotationDict = util.EMPTY_DICT
850 compile_state: Optional[CompileState] = None
851 """Optional :class:`.CompileState` object that maintains additional
852 state used by the compiler.
854 Major executable objects such as :class:`_expression.Insert`,
855 :class:`_expression.Update`, :class:`_expression.Delete`,
856 :class:`_expression.Select` will generate this
857 state when compiled in order to calculate additional information about the
858 object. For the top level object that is to be executed, the state can be
859 stored here where it can also have applicability towards result set
860 processing.
862 .. versionadded:: 1.4
864 """
866 dml_compile_state: Optional[CompileState] = None
867 """Optional :class:`.CompileState` assigned at the same point that
868 .isinsert, .isupdate, or .isdelete is assigned.
870 This will normally be the same object as .compile_state, with the
871 exception of cases like the :class:`.ORMFromStatementCompileState`
872 object.
874 .. versionadded:: 1.4.40
876 """
878 cache_key: Optional[CacheKey] = None
879 """The :class:`.CacheKey` that was generated ahead of creating this
880 :class:`.Compiled` object.
882 This is used for routines that need access to the original
883 :class:`.CacheKey` instance generated when the :class:`.Compiled`
884 instance was first cached, typically in order to reconcile
885 the original list of :class:`.BindParameter` objects with a
886 per-statement list that's generated on each call.
888 """
890 _gen_time: float
891 """Generation time of this :class:`.Compiled`, used for reporting
892 cache stats."""
894 def __init__(
895 self,
896 dialect: Dialect,
897 statement: Optional[ClauseElement],
898 schema_translate_map: Optional[SchemaTranslateMapType] = None,
899 render_schema_translate: bool = False,
900 compile_kwargs: Mapping[str, Any] = util.immutabledict(),
901 ):
902 """Construct a new :class:`.Compiled` object.
904 :param dialect: :class:`.Dialect` to compile against.
906 :param statement: :class:`_expression.ClauseElement` to be compiled.
908 :param schema_translate_map: dictionary of schema names to be
909 translated when forming the resultant SQL
911 .. seealso::
913 :ref:`schema_translating`
915 :param compile_kwargs: additional kwargs that will be
916 passed to the initial call to :meth:`.Compiled.process`.
919 """
920 self.dialect = dialect
921 self.preparer = self.dialect.identifier_preparer
922 if schema_translate_map:
923 self.schema_translate_map = schema_translate_map
924 self.preparer = self.preparer._with_schema_translate(
925 schema_translate_map
926 )
928 if statement is not None:
929 self.state = CompilerState.COMPILING
930 self.statement = statement
931 self.can_execute = statement.supports_execution
932 self._annotations = statement._annotations
933 if self.can_execute:
934 if TYPE_CHECKING:
935 assert isinstance(statement, Executable)
936 self.execution_options = statement._execution_options
937 self.string = self.process(self.statement, **compile_kwargs)
939 if render_schema_translate:
940 assert schema_translate_map is not None
941 self.string = self.preparer._render_schema_translates(
942 self.string, schema_translate_map
943 )
945 self.state = CompilerState.STRING_APPLIED
946 else:
947 self.state = CompilerState.NO_STATEMENT
949 self._gen_time = perf_counter()
951 def __init_subclass__(cls) -> None:
952 cls._init_compiler_cls()
953 return super().__init_subclass__()
955 @classmethod
956 def _init_compiler_cls(cls):
957 pass
959 def visit_unsupported_compilation(self, element, err, **kw):
960 raise exc.UnsupportedCompilationError(self, type(element)) from err
962 @property
963 def sql_compiler(self) -> SQLCompiler:
964 """Return a Compiled that is capable of processing SQL expressions.
966 If this compiler is one, it would likely just return 'self'.
968 """
970 raise NotImplementedError()
972 def process(self, obj: Visitable, **kwargs: Any) -> str:
973 return obj._compiler_dispatch(self, **kwargs)
975 def __str__(self) -> str:
976 """Return the string text of the generated SQL or DDL."""
978 if self.state is CompilerState.STRING_APPLIED:
979 return self.string
980 else:
981 return ""
983 def construct_params(
984 self,
985 params: Optional[_CoreSingleExecuteParams] = None,
986 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
987 escape_names: bool = True,
988 ) -> Optional[_MutableCoreSingleExecuteParams]:
989 """Return the bind params for this compiled object.
991 :param params: a dict of string/object pairs whose values will
992 override bind values compiled in to the
993 statement.
994 """
996 raise NotImplementedError()
998 @property
999 def params(self):
1000 """Return the bind params for this compiled object."""
1001 return self.construct_params()
1004class TypeCompiler(util.EnsureKWArg):
1005 """Produces DDL specification for TypeEngine objects."""
1007 ensure_kwarg = r"visit_\w+"
1009 def __init__(self, dialect: Dialect):
1010 self.dialect = dialect
1012 def process(self, type_: TypeEngine[Any], **kw: Any) -> str:
1013 if (
1014 type_._variant_mapping
1015 and self.dialect.name in type_._variant_mapping
1016 ):
1017 type_ = type_._variant_mapping[self.dialect.name]
1018 return type_._compiler_dispatch(self, **kw)
1020 def visit_unsupported_compilation(
1021 self, element: Any, err: Exception, **kw: Any
1022 ) -> NoReturn:
1023 raise exc.UnsupportedCompilationError(self, element) from err
1026# this was a Visitable, but to allow accurate detection of
1027# column elements this is actually a column element
1028class _CompileLabel(
1029 roles.BinaryElementRole[Any], elements.CompilerColumnElement
1030):
1031 """lightweight label object which acts as an expression.Label."""
1033 __visit_name__ = "label"
1034 __slots__ = "element", "name", "_alt_names"
1036 def __init__(self, col, name, alt_names=()):
1037 self.element = col
1038 self.name = name
1039 self._alt_names = (col,) + alt_names
1041 @property
1042 def proxy_set(self):
1043 return self.element.proxy_set
1045 @property
1046 def type(self):
1047 return self.element.type
1049 def self_group(self, **kw):
1050 return self
1053class aggregate_orderby_inline(
1054 roles.BinaryElementRole[Any], elements.CompilerColumnElement
1055):
1056 """produce ORDER BY inside of function argument lists"""
1058 __visit_name__ = "aggregate_orderby_inline"
1059 __slots__ = "element", "aggregate_order_by"
1061 def __init__(self, element, orderby):
1062 self.element = element
1063 self.aggregate_order_by = orderby
1065 def __iter__(self):
1066 return iter(self.element)
1068 @property
1069 def proxy_set(self):
1070 return self.element.proxy_set
1072 @property
1073 def type(self):
1074 return self.element.type
1076 def self_group(self, **kw):
1077 return self
1079 def _with_binary_element_type(self, type_):
1080 return aggregate_orderby_inline(
1081 self.element._with_binary_element_type(type_),
1082 self.aggregate_order_by,
1083 )
1086class ilike_case_insensitive(
1087 roles.BinaryElementRole[Any], elements.CompilerColumnElement
1088):
1089 """produce a wrapping element for a case-insensitive portion of
1090 an ILIKE construct.
1092 The construct usually renders the ``lower()`` function, but on
1093 PostgreSQL will pass silently with the assumption that "ILIKE"
1094 is being used.
1096 .. versionadded:: 2.0
1098 """
1100 __visit_name__ = "ilike_case_insensitive_operand"
1101 __slots__ = "element", "comparator"
1103 def __init__(self, element):
1104 self.element = element
1105 self.comparator = element.comparator
1107 @property
1108 def proxy_set(self):
1109 return self.element.proxy_set
1111 @property
1112 def type(self):
1113 return self.element.type
1115 def self_group(self, **kw):
1116 return self
1118 def _with_binary_element_type(self, type_):
1119 return ilike_case_insensitive(
1120 self.element._with_binary_element_type(type_)
1121 )
1124class SQLCompiler(Compiled):
1125 """Default implementation of :class:`.Compiled`.
1127 Compiles :class:`_expression.ClauseElement` objects into SQL strings.
1129 """
1131 extract_map = EXTRACT_MAP
1133 bindname_escape_characters: ClassVar[Mapping[str, str]] = (
1134 util.immutabledict(
1135 {
1136 "%": "P",
1137 "(": "A",
1138 ")": "Z",
1139 ":": "C",
1140 ".": "_",
1141 "[": "_",
1142 "]": "_",
1143 " ": "_",
1144 }
1145 )
1146 )
1147 """A mapping (e.g. dict or similar) containing a lookup of
1148 characters keyed to replacement characters which will be applied to all
1149 'bind names' used in SQL statements as a form of 'escaping'; the given
1150 characters are replaced entirely with the 'replacement' character when
1151 rendered in the SQL statement, and a similar translation is performed
1152 on the incoming names used in parameter dictionaries passed to methods
1153 like :meth:`_engine.Connection.execute`.
1155 This allows bound parameter names used in :func:`_sql.bindparam` and
1156 other constructs to have any arbitrary characters present without any
1157 concern for characters that aren't allowed at all on the target database.
1159 Third party dialects can establish their own dictionary here to replace the
1160 default mapping, which will ensure that the particular characters in the
1161 mapping will never appear in a bound parameter name.
1163 The dictionary is evaluated at **class creation time**, so cannot be
1164 modified at runtime; it must be present on the class when the class
1165 is first declared.
1167 Note that for dialects that have additional bound parameter rules such
1168 as additional restrictions on leading characters, the
1169 :meth:`_sql.SQLCompiler.bindparam_string` method may need to be augmented.
1170 See the cx_Oracle compiler for an example of this.
1172 .. versionadded:: 2.0.0rc1
1174 """
1176 _bind_translate_re: ClassVar[Pattern[str]]
1177 _bind_translate_chars: ClassVar[Mapping[str, str]]
1179 is_sql = True
1181 compound_keywords = COMPOUND_KEYWORDS
1183 isdelete: bool = False
1184 isinsert: bool = False
1185 isupdate: bool = False
1186 """class-level defaults which can be set at the instance
1187 level to define if this Compiled instance represents
1188 INSERT/UPDATE/DELETE
1189 """
1191 postfetch: Optional[List[Column[Any]]]
1192 """list of columns that can be post-fetched after INSERT or UPDATE to
1193 receive server-updated values"""
1195 insert_prefetch: Sequence[Column[Any]] = ()
1196 """list of columns for which default values should be evaluated before
1197 an INSERT takes place"""
1199 update_prefetch: Sequence[Column[Any]] = ()
1200 """list of columns for which onupdate default values should be evaluated
1201 before an UPDATE takes place"""
1203 implicit_returning: Optional[Sequence[ColumnElement[Any]]] = None
1204 """list of "implicit" returning columns for a toplevel INSERT or UPDATE
1205 statement, used to receive newly generated values of columns.
1207 .. versionadded:: 2.0 ``implicit_returning`` replaces the previous
1208 ``returning`` collection, which was not a generalized RETURNING
1209 collection and instead was in fact specific to the "implicit returning"
1210 feature.
1212 """
1214 isplaintext: bool = False
1216 binds: Dict[str, BindParameter[Any]]
1217 """a dictionary of bind parameter keys to BindParameter instances."""
1219 bind_names: Dict[BindParameter[Any], str]
1220 """a dictionary of BindParameter instances to "compiled" names
1221 that are actually present in the generated SQL"""
1223 stack: List[_CompilerStackEntry]
1224 """major statements such as SELECT, INSERT, UPDATE, DELETE are
1225 tracked in this stack using an entry format."""
1227 returning_precedes_values: bool = False
1228 """set to True classwide to generate RETURNING
1229 clauses before the VALUES or WHERE clause (i.e. MSSQL)
1230 """
1232 render_table_with_column_in_update_from: bool = False
1233 """set to True classwide to indicate the SET clause
1234 in a multi-table UPDATE statement should qualify
1235 columns with the table name (i.e. MySQL only)
1236 """
1238 ansi_bind_rules: bool = False
1239 """SQL 92 doesn't allow bind parameters to be used
1240 in the columns clause of a SELECT, nor does it allow
1241 ambiguous expressions like "? = ?". A compiler
1242 subclass can set this flag to False if the target
1243 driver/DB enforces this
1244 """
1246 bindtemplate: str
1247 """template to render bound parameters based on paramstyle."""
1249 compilation_bindtemplate: str
1250 """template used by compiler to render parameters before positional
1251 paramstyle application"""
1253 _numeric_binds_identifier_char: str
1254 """Character that's used to as the identifier of a numerical bind param.
1255 For example if this char is set to ``$``, numerical binds will be rendered
1256 in the form ``$1, $2, $3``.
1257 """
1259 _result_columns: List[ResultColumnsEntry]
1260 """relates label names in the final SQL to a tuple of local
1261 column/label name, ColumnElement object (if any) and
1262 TypeEngine. CursorResult uses this for type processing and
1263 column targeting"""
1265 _textual_ordered_columns: bool = False
1266 """tell the result object that the column names as rendered are important,
1267 but they are also "ordered" vs. what is in the compiled object here.
1269 As of 1.4.42 this condition is only present when the statement is a
1270 TextualSelect, e.g. text("....").columns(...), where it is required
1271 that the columns are considered positionally and not by name.
1273 """
1275 _ad_hoc_textual: bool = False
1276 """tell the result that we encountered text() or '*' constructs in the
1277 middle of the result columns, but we also have compiled columns, so
1278 if the number of columns in cursor.description does not match how many
1279 expressions we have, that means we can't rely on positional at all and
1280 should match on name.
1282 """
1284 _ordered_columns: bool = True
1285 """
1286 if False, means we can't be sure the list of entries
1287 in _result_columns is actually the rendered order. Usually
1288 True unless using an unordered TextualSelect.
1289 """
1291 _loose_column_name_matching: bool = False
1292 """tell the result object that the SQL statement is textual, wants to match
1293 up to Column objects, and may be using the ._tq_label in the SELECT rather
1294 than the base name.
1296 """
1298 _numeric_binds: bool = False
1299 """
1300 True if paramstyle is "numeric". This paramstyle is trickier than
1301 all the others.
1303 """
1305 _render_postcompile: bool = False
1306 """
1307 whether to render out POSTCOMPILE params during the compile phase.
1309 This attribute is used only for end-user invocation of stmt.compile();
1310 it's never used for actual statement execution, where instead the
1311 dialect internals access and render the internal postcompile structure
1312 directly.
1314 """
1316 _post_compile_expanded_state: Optional[ExpandedState] = None
1317 """When render_postcompile is used, the ``ExpandedState`` used to create
1318 the "expanded" SQL is assigned here, and then used by the ``.params``
1319 accessor and ``.construct_params()`` methods for their return values.
1321 .. versionadded:: 2.0.0rc1
1323 """
1325 _pre_expanded_string: Optional[str] = None
1326 """Stores the original string SQL before 'post_compile' is applied,
1327 for cases where 'post_compile' were used.
1329 """
1331 _pre_expanded_positiontup: Optional[List[str]] = None
1333 _insertmanyvalues: Optional[_InsertManyValues] = None
1335 _insert_crud_params: Optional[crud._CrudParamSequence] = None
1337 literal_execute_params: FrozenSet[BindParameter[Any]] = frozenset()
1338 """bindparameter objects that are rendered as literal values at statement
1339 execution time.
1341 """
1343 post_compile_params: FrozenSet[BindParameter[Any]] = frozenset()
1344 """bindparameter objects that are rendered as bound parameter placeholders
1345 at statement execution time.
1347 """
1349 escaped_bind_names: util.immutabledict[str, str] = util.EMPTY_DICT
1351 # the names ``escaped_bind_names`` maps *to*, maintained alongside it so
1352 # that collision checks don't rebuild the value collection each time.
1353 # only assigned once a name actually needs escaping, which is rare;
1354 # ``escaped_bind_names`` being empty means this was never set.
1355 _escaped_bind_names_used: Set[str]
1356 """Late escaping of bound parameter names that has to be converted
1357 to the original name when looking in the parameter dictionary.
1359 """
1361 has_out_parameters = False
1362 """if True, there are bindparam() objects that have the isoutparam
1363 flag set."""
1365 postfetch_lastrowid = False
1366 """if True, and this in insert, use cursor.lastrowid to populate
1367 result.inserted_primary_key. """
1369 _cache_key_bind_match: Optional[
1370 Tuple[
1371 Dict[
1372 BindParameter[Any],
1373 List[BindParameter[Any]],
1374 ],
1375 Dict[
1376 str,
1377 BindParameter[Any],
1378 ],
1379 ]
1380 ] = None
1381 """a mapping that will relate the BindParameter object we compile
1382 to those that are part of the extracted collection of parameters
1383 in the cache key, if we were given a cache key.
1385 """
1387 positiontup: Optional[List[str]] = None
1388 """for a compiled construct that uses a positional paramstyle, will be
1389 a sequence of strings, indicating the names of bound parameters in order.
1391 This is used in order to render bound parameters in their correct order,
1392 and is combined with the :attr:`_sql.Compiled.params` dictionary to
1393 render parameters.
1395 This sequence always contains the unescaped name of the parameters.
1397 .. seealso::
1399 :ref:`faq_sql_expression_string` - includes a usage example for
1400 debugging use cases.
1402 """
1403 _values_bindparam: Optional[List[str]] = None
1405 _visited_bindparam: Optional[List[str]] = None
1407 inline: bool = False
1409 ctes: Optional[MutableMapping[CTE, str]]
1411 # Detect same CTE references - Dict[(level, name), cte]
1412 # Level is required for supporting nesting
1413 ctes_by_level_name: Dict[Tuple[int, str], CTE]
1415 # To retrieve key/level in ctes_by_level_name -
1416 # Dict[cte_reference, (level, cte_name, cte_opts)]
1417 level_name_by_cte: Dict[CTE, Tuple[int, str, selectable._CTEOpts]]
1419 ctes_recursive: bool
1421 _post_compile_pattern = re.compile(r"__\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]")
1422 _pyformat_pattern = re.compile(r"%\(([^)]+?)\)s")
1423 _positional_pattern = re.compile(
1424 f"{_pyformat_pattern.pattern}|{_post_compile_pattern.pattern}"
1425 )
1427 # variants of the above for dialects that double percent signs in
1428 # literal text; a "%%" pair is matched first so that an escaped
1429 # percent sign is never taken as the start of a bound parameter
1430 _pct_escaped_pyformat_pattern = re.compile(
1431 f"%%|{_pyformat_pattern.pattern}"
1432 )
1433 _pct_escaped_positional_pattern = re.compile(
1434 f"%%|{_positional_pattern.pattern}"
1435 )
1437 _collect_params: Final[bool]
1438 _collected_params: util.immutabledict[str, Any]
1440 @classmethod
1441 def _init_compiler_cls(cls):
1442 cls._init_bind_translate()
1444 @classmethod
1445 def _init_bind_translate(cls):
1446 reg = re.escape("".join(cls.bindname_escape_characters))
1447 cls._bind_translate_re = re.compile(f"[{reg}]")
1448 cls._bind_translate_chars = cls.bindname_escape_characters
1450 def __init__(
1451 self,
1452 dialect: Dialect,
1453 statement: Optional[ClauseElement],
1454 cache_key: Optional[CacheKey] = None,
1455 column_keys: Optional[Sequence[str]] = None,
1456 for_executemany: bool = False,
1457 linting: Linting = NO_LINTING,
1458 _supporting_against: Optional[SQLCompiler] = None,
1459 **kwargs: Any,
1460 ):
1461 """Construct a new :class:`.SQLCompiler` object.
1463 :param dialect: :class:`.Dialect` to be used
1465 :param statement: :class:`_expression.ClauseElement` to be compiled
1467 :param column_keys: a list of column names to be compiled into an
1468 INSERT or UPDATE statement.
1470 :param for_executemany: whether INSERT / UPDATE statements should
1471 expect that they are to be invoked in an "executemany" style,
1472 which may impact how the statement will be expected to return the
1473 values of defaults and autoincrement / sequences and similar.
1474 Depending on the backend and driver in use, support for retrieving
1475 these values may be disabled which means SQL expressions may
1476 be rendered inline, RETURNING may not be rendered, etc.
1478 :param kwargs: additional keyword arguments to be consumed by the
1479 superclass.
1481 """
1482 self.column_keys = column_keys
1484 self.cache_key = cache_key
1486 if cache_key:
1487 cksm = {b.key: b for b in cache_key[1]}
1488 ckbm = {b: [b] for b in cache_key[1]}
1489 self._cache_key_bind_match = (ckbm, cksm)
1491 # compile INSERT/UPDATE defaults/sequences to expect executemany
1492 # style execution, which may mean no pre-execute of defaults,
1493 # or no RETURNING
1494 self.for_executemany = for_executemany
1496 self.linting = linting
1498 # a dictionary of bind parameter keys to BindParameter
1499 # instances.
1500 self.binds = {}
1502 # a dictionary of BindParameter instances to "compiled" names
1503 # that are actually present in the generated SQL
1504 self.bind_names = util.column_dict()
1506 # stack which keeps track of nested SELECT statements
1507 self.stack = []
1509 self._result_columns = []
1511 # true if the paramstyle is positional
1512 self.positional = dialect.positional
1513 if self.positional:
1514 self._numeric_binds = nb = dialect.paramstyle.startswith("numeric")
1515 if nb:
1516 self._numeric_binds_identifier_char = (
1517 "$" if dialect.paramstyle == "numeric_dollar" else ":"
1518 )
1520 self.compilation_bindtemplate = _pyformat_template
1521 else:
1522 self.compilation_bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
1524 self.ctes = None
1526 self.label_length = (
1527 dialect.label_length or dialect.max_identifier_length
1528 )
1530 # a map which tracks "anonymous" identifiers that are created on
1531 # the fly here
1532 self.anon_map = prefix_anon_map()
1534 # a map which tracks "truncated" names based on
1535 # dialect.label_length or dialect.max_identifier_length
1536 self.truncated_names: Dict[Tuple[str, str], str] = {}
1537 self._truncated_counters: Dict[str, int] = {}
1538 if not cache_key:
1539 self._collect_params = True
1540 self._collected_params = util.EMPTY_DICT
1541 else:
1542 self._collect_params = False # type: ignore[misc]
1544 Compiled.__init__(self, dialect, statement, **kwargs)
1546 if self.isinsert or self.isupdate or self.isdelete:
1547 if TYPE_CHECKING:
1548 assert isinstance(statement, UpdateBase)
1550 if self.isinsert or self.isupdate:
1551 if TYPE_CHECKING:
1552 assert isinstance(statement, ValuesBase)
1553 if statement._inline:
1554 self.inline = True
1555 elif self.for_executemany and (
1556 not self.isinsert
1557 or (
1558 self.dialect.insert_executemany_returning
1559 and statement._return_defaults
1560 )
1561 ):
1562 self.inline = True
1564 self.bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
1566 if _supporting_against:
1567 self.__dict__.update(
1568 {
1569 k: v
1570 for k, v in _supporting_against.__dict__.items()
1571 if k
1572 not in {
1573 "state",
1574 "dialect",
1575 "preparer",
1576 "positional",
1577 "_numeric_binds",
1578 "compilation_bindtemplate",
1579 "bindtemplate",
1580 }
1581 }
1582 )
1584 if self.state is CompilerState.STRING_APPLIED:
1585 if self.positional:
1586 if self._numeric_binds:
1587 self._process_numeric()
1588 else:
1589 self._process_positional()
1591 if self._render_postcompile:
1592 parameters = self.construct_params(
1593 escape_names=False,
1594 _no_postcompile=True,
1595 )
1597 self._process_parameters_for_postcompile(
1598 parameters, _populate_self=True
1599 )
1601 @property
1602 def insert_single_values_expr(self) -> Optional[str]:
1603 """When an INSERT is compiled with a single set of parameters inside
1604 a VALUES expression, the string is assigned here, where it can be
1605 used for insert batching schemes to rewrite the VALUES expression.
1607 .. versionchanged:: 2.0 This collection is no longer used by
1608 SQLAlchemy's built-in dialects, in favor of the currently
1609 internal ``_insertmanyvalues`` collection that is used only by
1610 :class:`.SQLCompiler`.
1612 """
1613 if self._insertmanyvalues is None:
1614 return None
1615 else:
1616 return self._insertmanyvalues.single_values_expr
1618 @util.ro_memoized_property
1619 def effective_returning(self) -> Optional[Sequence[ColumnElement[Any]]]:
1620 """The effective "returning" columns for INSERT, UPDATE or DELETE.
1622 This is either the so-called "implicit returning" columns which are
1623 calculated by the compiler on the fly, or those present based on what's
1624 present in ``self.statement._returning`` (expanded into individual
1625 columns using the ``._all_selected_columns`` attribute) i.e. those set
1626 explicitly using the :meth:`.UpdateBase.returning` method.
1628 .. versionadded:: 2.0
1630 """
1631 if self.implicit_returning:
1632 return self.implicit_returning
1633 elif self.statement is not None and is_dml(self.statement):
1634 return [
1635 c
1636 for c in self.statement._all_selected_columns
1637 if is_column_element(c)
1638 ]
1640 else:
1641 return None
1643 @property
1644 def returning(self):
1645 """backwards compatibility; returns the
1646 effective_returning collection.
1648 """
1649 return self.effective_returning
1651 @property
1652 def current_executable(self):
1653 """Return the current 'executable' that is being compiled.
1655 This is currently the :class:`_sql.Select`, :class:`_sql.Insert`,
1656 :class:`_sql.Update`, :class:`_sql.Delete`,
1657 :class:`_sql.CompoundSelect` object that is being compiled.
1658 Specifically it's assigned to the ``self.stack`` list of elements.
1660 When a statement like the above is being compiled, it normally
1661 is also assigned to the ``.statement`` attribute of the
1662 :class:`_sql.Compiler` object. However, all SQL constructs are
1663 ultimately nestable, and this attribute should never be consulted
1664 by a ``visit_`` method, as it is not guaranteed to be assigned
1665 nor guaranteed to correspond to the current statement being compiled.
1667 """
1668 try:
1669 return self.stack[-1]["selectable"]
1670 except IndexError as ie:
1671 raise IndexError("Compiler does not have a stack entry") from ie
1673 @property
1674 def prefetch(self):
1675 return list(self.insert_prefetch) + list(self.update_prefetch)
1677 @util.memoized_property
1678 def _global_attributes(self) -> Dict[Any, Any]:
1679 return {}
1681 def _add_to_params(self, item: ExecutableStatement) -> None:
1682 # assumes that this is called before traversing the statement
1683 # so the call happens outer to inner, meaning that existing params
1684 # take precedence
1685 if item._params:
1686 self._collected_params = item._params | self._collected_params
1688 @util.memoized_instancemethod
1689 def _init_cte_state(self) -> MutableMapping[CTE, str]:
1690 """Initialize collections related to CTEs only if
1691 a CTE is located, to save on the overhead of
1692 these collections otherwise.
1694 """
1695 # collect CTEs to tack on top of a SELECT
1696 # To store the query to print - Dict[cte, text_query]
1697 ctes: MutableMapping[CTE, str] = util.OrderedDict()
1698 self.ctes = ctes
1700 # Detect same CTE references - Dict[(level, name), cte]
1701 # Level is required for supporting nesting
1702 self.ctes_by_level_name = {}
1704 # To retrieve key/level in ctes_by_level_name -
1705 # Dict[cte_reference, (level, cte_name, cte_opts)]
1706 self.level_name_by_cte = {}
1708 self.ctes_recursive = False
1710 return ctes
1712 @contextlib.contextmanager
1713 def _nested_result(self):
1714 """special API to support the use case of 'nested result sets'"""
1715 result_columns, ordered_columns = (
1716 self._result_columns,
1717 self._ordered_columns,
1718 )
1719 self._result_columns, self._ordered_columns = [], False
1721 try:
1722 if self.stack:
1723 entry = self.stack[-1]
1724 entry["need_result_map_for_nested"] = True
1725 else:
1726 entry = None
1727 yield self._result_columns, self._ordered_columns
1728 finally:
1729 if entry:
1730 entry.pop("need_result_map_for_nested")
1731 self._result_columns, self._ordered_columns = (
1732 result_columns,
1733 ordered_columns,
1734 )
1736 def _process_positional(self):
1737 assert not self.positiontup
1738 assert self.state is CompilerState.STRING_APPLIED
1739 assert not self._numeric_binds
1741 if self.dialect.paramstyle == "format":
1742 placeholder = "%s"
1743 else:
1744 assert self.dialect.paramstyle == "qmark"
1745 placeholder = "?"
1747 positions = []
1748 binds = self.binds
1750 if self.escaped_bind_names:
1751 reverse_escape = {v: k for k, v in self.escaped_bind_names.items()}
1752 assert len(self.escaped_bind_names) == len(reverse_escape)
1753 else:
1754 reverse_escape = {}
1756 if self.preparer._double_percents:
1757 positional_pattern = self._pct_escaped_positional_pattern
1758 else:
1759 positional_pattern = self._positional_pattern
1761 def find_position(m: re.Match[str]) -> str:
1762 normal_bind = m.group(1)
1763 if normal_bind:
1764 if reverse_escape.get(normal_bind, normal_bind) not in binds:
1765 # literal text that looks like a bound parameter
1766 return m.group(0)
1767 positions.append(normal_bind)
1768 return placeholder
1769 elif m.group(2):
1770 # this a post-compile bind
1771 positions.append(m.group(2))
1772 return m.group(0)
1773 else:
1774 # an escaped percent sign
1775 return m.group(0)
1777 self.string = re.sub(positional_pattern, find_position, self.string)
1779 if reverse_escape:
1780 self.positiontup = [
1781 reverse_escape.get(name, name) for name in positions
1782 ]
1783 else:
1784 self.positiontup = positions
1786 if self._insertmanyvalues:
1787 positions = []
1789 single_values_expr = re.sub(
1790 positional_pattern,
1791 find_position,
1792 self._insertmanyvalues.single_values_expr,
1793 )
1794 insert_crud_params = [
1795 (
1796 v[0],
1797 v[1],
1798 re.sub(positional_pattern, find_position, v[2]),
1799 v[3],
1800 )
1801 for v in self._insertmanyvalues.insert_crud_params
1802 ]
1804 self._insertmanyvalues = self._insertmanyvalues._replace(
1805 single_values_expr=single_values_expr,
1806 insert_crud_params=insert_crud_params,
1807 )
1809 def _process_numeric(self):
1810 assert self._numeric_binds
1811 assert self.state is CompilerState.STRING_APPLIED
1813 num = 1
1814 param_pos: Dict[str, str] = {}
1815 order: Iterable[str]
1816 if self._insertmanyvalues and self._values_bindparam is not None:
1817 # bindparams that are not in values are always placed first.
1818 # this avoids the need of changing them when using executemany
1819 # values () ()
1820 order = itertools.chain(
1821 (
1822 name
1823 for name in self.bind_names.values()
1824 if name not in self._values_bindparam
1825 ),
1826 self.bind_names.values(),
1827 )
1828 else:
1829 order = self.bind_names.values()
1831 for bind_name in order:
1832 if bind_name in param_pos:
1833 continue
1834 bind = self.binds[bind_name]
1835 if (
1836 bind in self.post_compile_params
1837 or bind in self.literal_execute_params
1838 ):
1839 # set to None to just mark the in positiontup, it will not
1840 # be replaced below.
1841 param_pos[bind_name] = None # type: ignore[assignment]
1842 else:
1843 ph = f"{self._numeric_binds_identifier_char}{num}"
1844 num += 1
1845 param_pos[bind_name] = ph
1847 self.next_numeric_pos = num
1849 self.positiontup = list(param_pos)
1850 if self.escaped_bind_names:
1851 len_before = len(param_pos)
1852 param_pos = {
1853 self.escaped_bind_names.get(name, name): pos
1854 for name, pos in param_pos.items()
1855 }
1856 assert len(param_pos) == len_before
1858 if self.preparer._double_percents:
1859 pyformat_pattern = self._pct_escaped_pyformat_pattern
1860 else:
1861 pyformat_pattern = self._pyformat_pattern
1863 # any match not in param_pos is an escaped percent sign or literal
1864 # text that looks like a bound parameter
1865 self.string = pyformat_pattern.sub(
1866 lambda m: param_pos.get(m.group(1), m.group(0)), self.string
1867 )
1869 if self._insertmanyvalues:
1870 single_values_expr = (
1871 # format is ok here since single_values_expr includes only
1872 # place-holders
1873 self._insertmanyvalues.single_values_expr
1874 % param_pos
1875 )
1876 insert_crud_params = [
1877 (v[0], v[1], "%s", v[3])
1878 for v in self._insertmanyvalues.insert_crud_params
1879 ]
1881 self._insertmanyvalues = self._insertmanyvalues._replace(
1882 # This has the numbers (:1, :2)
1883 single_values_expr=single_values_expr,
1884 # The single binds are instead %s so they can be formatted
1885 insert_crud_params=insert_crud_params,
1886 )
1888 @util.memoized_property
1889 def _bind_processors(
1890 self,
1891 ) -> MutableMapping[
1892 str, Union[_BindProcessorType[Any], Sequence[_BindProcessorType[Any]]]
1893 ]:
1894 # mypy is not able to see the two value types as the above Union,
1895 # it just sees "object". don't know how to resolve
1896 return {
1897 key: value # type: ignore[misc]
1898 for key, value in (
1899 (
1900 self.bind_names[bindparam],
1901 (
1902 bindparam.type._cached_bind_processor(self.dialect)
1903 if not bindparam.type._is_tuple_type
1904 else tuple(
1905 elem_type._cached_bind_processor(self.dialect)
1906 for elem_type in cast(
1907 TupleType, bindparam.type
1908 ).types
1909 )
1910 ),
1911 )
1912 for bindparam in self.bind_names
1913 # literal_execute parameters are rendered into the SQL string
1914 # via their literal_processor and never bound as values, so
1915 # they do not need a bind processor. Skipping them also avoids
1916 # invoking bind-processor construction that may require the
1917 # DBAPI to be present (see asyncpg, psycopgcffi cases),
1918 # facilitating testing.
1919 if bindparam not in self.literal_execute_params
1920 )
1921 if value is not None
1922 }
1924 @util.memoized_property
1925 def _expanded_bind_names_reserved(self) -> FrozenSet[str]:
1926 """Names that the individual elements of an expanding parameter
1927 may not take when the parameter is expanded at execution time.
1929 This is every bound parameter name in the statement, both in its
1930 compiled and in its escaped form. See :ticket:`13695`.
1932 """
1933 return frozenset(self.bind_names.values()).union(
1934 self.escaped_bind_names, self.escaped_bind_names.values()
1935 )
1937 def is_subquery(self):
1938 return len(self.stack) > 1
1940 @property
1941 def sql_compiler(self) -> Self:
1942 return self
1944 def construct_expanded_state(
1945 self,
1946 params: Optional[_CoreSingleExecuteParams] = None,
1947 escape_names: bool = True,
1948 ) -> ExpandedState:
1949 """Return a new :class:`.ExpandedState` for a given parameter set.
1951 For queries that use "expanding" or other late-rendered parameters,
1952 this method will provide for both the finalized SQL string as well
1953 as the parameters that would be used for a particular parameter set.
1955 .. versionadded:: 2.0.0rc1
1957 """
1958 parameters = self.construct_params(
1959 params,
1960 escape_names=escape_names,
1961 _no_postcompile=True,
1962 )
1963 return self._process_parameters_for_postcompile(
1964 parameters,
1965 )
1967 def construct_params(
1968 self,
1969 params: Optional[_CoreSingleExecuteParams] = None,
1970 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
1971 escape_names: bool = True,
1972 _group_number: Optional[int] = None,
1973 _check: bool = True,
1974 _no_postcompile: bool = False,
1975 _collected_params: _CoreSingleExecuteParams | None = None,
1976 ) -> _MutableCoreSingleExecuteParams:
1977 """return a dictionary of bind parameter keys and values"""
1978 if _collected_params is not None:
1979 assert not self._collect_params
1980 elif self._collect_params:
1981 _collected_params = self._collected_params
1983 if _collected_params:
1984 if not params:
1985 params = _collected_params
1986 else:
1987 params = {**_collected_params, **params}
1989 if self._render_postcompile and not _no_postcompile:
1990 assert self._post_compile_expanded_state is not None
1991 if not params:
1992 return dict(self._post_compile_expanded_state.parameters)
1993 else:
1994 raise exc.InvalidRequestError(
1995 "can't construct new parameters when render_postcompile "
1996 "is used; the statement is hard-linked to the original "
1997 "parameters. Use construct_expanded_state to generate a "
1998 "new statement and parameters."
1999 )
2001 has_escaped_names = escape_names and bool(self.escaped_bind_names)
2003 if extracted_parameters:
2004 # related the bound parameters collected in the original cache key
2005 # to those collected in the incoming cache key. They will not have
2006 # matching names but they will line up positionally in the same
2007 # way. The parameters present in self.bind_names may be clones of
2008 # these original cache key params in the case of DML but the .key
2009 # will be guaranteed to match.
2010 if self.cache_key is None:
2011 raise exc.CompileError(
2012 "This compiled object has no original cache key; "
2013 "can't pass extracted_parameters to construct_params"
2014 )
2015 else:
2016 orig_extracted = self.cache_key[1]
2018 ckbm_tuple = self._cache_key_bind_match
2019 assert ckbm_tuple is not None
2020 ckbm, _ = ckbm_tuple
2021 resolved_extracted = {
2022 bind: extracted
2023 for b, extracted in zip(orig_extracted, extracted_parameters)
2024 for bind in ckbm[b]
2025 }
2026 else:
2027 resolved_extracted = None
2029 if params:
2030 pd = {}
2031 for bindparam, name in self.bind_names.items():
2032 escaped_name = (
2033 self.escaped_bind_names.get(name, name)
2034 if has_escaped_names
2035 else name
2036 )
2038 if bindparam.key in params:
2039 pd[escaped_name] = params[bindparam.key]
2040 elif name in params:
2041 pd[escaped_name] = params[name]
2043 elif _check and bindparam.required:
2044 if _group_number:
2045 raise exc.InvalidRequestError(
2046 "A value is required for bind parameter %r, "
2047 "in parameter group %d"
2048 % (bindparam.key, _group_number),
2049 code="cd3x",
2050 )
2051 else:
2052 raise exc.InvalidRequestError(
2053 "A value is required for bind parameter %r"
2054 % bindparam.key,
2055 code="cd3x",
2056 )
2057 else:
2058 if resolved_extracted:
2059 value_param = resolved_extracted.get(
2060 bindparam, bindparam
2061 )
2062 else:
2063 value_param = bindparam
2065 if bindparam.callable:
2066 pd[escaped_name] = value_param.effective_value
2067 else:
2068 pd[escaped_name] = value_param.value
2069 return pd
2070 else:
2071 pd = {}
2072 for bindparam, name in self.bind_names.items():
2073 escaped_name = (
2074 self.escaped_bind_names.get(name, name)
2075 if has_escaped_names
2076 else name
2077 )
2079 if _check and bindparam.required:
2080 if _group_number:
2081 raise exc.InvalidRequestError(
2082 "A value is required for bind parameter %r, "
2083 "in parameter group %d"
2084 % (bindparam.key, _group_number),
2085 code="cd3x",
2086 )
2087 else:
2088 raise exc.InvalidRequestError(
2089 "A value is required for bind parameter %r"
2090 % bindparam.key,
2091 code="cd3x",
2092 )
2094 if resolved_extracted:
2095 value_param = resolved_extracted.get(bindparam, bindparam)
2096 else:
2097 value_param = bindparam
2099 if bindparam.callable:
2100 pd[escaped_name] = value_param.effective_value
2101 else:
2102 pd[escaped_name] = value_param.value
2104 return pd
2106 @util.memoized_instancemethod
2107 def _get_set_input_sizes_lookup(self):
2108 dialect = self.dialect
2110 include_types = dialect.include_set_input_sizes
2111 exclude_types = dialect.exclude_set_input_sizes
2113 dbapi = dialect.dbapi
2115 def lookup_type(typ):
2116 dbtype = typ._unwrapped_dialect_impl(dialect).get_dbapi_type(dbapi)
2118 if (
2119 dbtype is not None
2120 and (exclude_types is None or dbtype not in exclude_types)
2121 and (include_types is None or dbtype in include_types)
2122 ):
2123 return dbtype
2124 else:
2125 return None
2127 inputsizes = {}
2129 literal_execute_params = self.literal_execute_params
2131 for bindparam in self.bind_names:
2132 if bindparam in literal_execute_params:
2133 continue
2135 if bindparam.type._is_tuple_type:
2136 inputsizes[bindparam] = [
2137 lookup_type(typ)
2138 for typ in cast(TupleType, bindparam.type).types
2139 ]
2140 else:
2141 inputsizes[bindparam] = lookup_type(bindparam.type)
2143 return inputsizes
2145 @property
2146 def params(self):
2147 """Return the bind param dictionary embedded into this
2148 compiled object, for those values that are present.
2150 .. seealso::
2152 :ref:`faq_sql_expression_string` - includes a usage example for
2153 debugging use cases.
2155 """
2156 return self.construct_params(_check=False)
2158 def _process_parameters_for_postcompile(
2159 self,
2160 parameters: _MutableCoreSingleExecuteParams,
2161 _populate_self: bool = False,
2162 ) -> ExpandedState:
2163 """handle special post compile parameters.
2165 These include:
2167 * "expanding" parameters -typically IN tuples that are rendered
2168 on a per-parameter basis for an otherwise fixed SQL statement string.
2170 * literal_binds compiled with the literal_execute flag. Used for
2171 things like SQL Server "TOP N" where the driver does not accommodate
2172 N as a bound parameter.
2174 """
2176 expanded_parameters = {}
2177 new_positiontup: Optional[List[str]]
2179 pre_expanded_string = self._pre_expanded_string
2180 if pre_expanded_string is None:
2181 pre_expanded_string = self.string
2183 if self.positional:
2184 new_positiontup = []
2186 pre_expanded_positiontup = self._pre_expanded_positiontup
2187 if pre_expanded_positiontup is None:
2188 pre_expanded_positiontup = self.positiontup
2190 else:
2191 new_positiontup = pre_expanded_positiontup = None
2193 processors = self._bind_processors
2194 single_processors = cast(
2195 "Mapping[str, _BindProcessorType[Any]]", processors
2196 )
2197 tuple_processors = cast(
2198 "Mapping[str, Sequence[_BindProcessorType[Any]]]", processors
2199 )
2201 new_processors: Dict[str, _BindProcessorType[Any]] = {}
2203 replacement_expressions: Dict[str, Any] = {}
2204 to_update_sets: Dict[str, Any] = {}
2206 # names that the elements of expanding parameters can't take, as
2207 # they are in use either by other parameters in the statement or by
2208 # elements of expanding parameters that were already expanded.
2209 # copied into a mutable set, as it is added to within this method's
2210 # scope to track the additional element names generated for this
2211 # execution
2212 expanded_taken_names: Set[str] = set(
2213 self._expanded_bind_names_reserved
2214 )
2216 # notes:
2217 # *unescaped* parameter names in:
2218 # self.bind_names, self.binds, self._bind_processors, self.positiontup
2219 #
2220 # *escaped* parameter names in:
2221 # construct_params(), replacement_expressions
2223 numeric_positiontup: Optional[List[str]] = None
2225 if self.positional and pre_expanded_positiontup is not None:
2226 names: Iterable[str] = pre_expanded_positiontup
2227 if self._numeric_binds:
2228 numeric_positiontup = []
2229 else:
2230 names = self.bind_names.values()
2232 ebn = self.escaped_bind_names
2233 for name in names:
2234 escaped_name = ebn.get(name, name) if ebn else name
2235 parameter = self.binds[name]
2237 if parameter in self.literal_execute_params:
2238 if escaped_name not in replacement_expressions:
2239 replacement_expressions[escaped_name] = (
2240 self.render_literal_bindparam(
2241 parameter,
2242 render_literal_value=parameters.pop(escaped_name),
2243 )
2244 )
2245 continue
2247 if parameter in self.post_compile_params:
2248 if escaped_name in replacement_expressions:
2249 to_update = to_update_sets[escaped_name]
2250 values = None
2251 else:
2252 # we are removing the parameter from parameters
2253 # because it is a list value, which is not expected by
2254 # TypeEngine objects that would otherwise be asked to
2255 # process it. the single name is being replaced with
2256 # individual numbered parameters for each value in the
2257 # param.
2258 #
2259 # note we are also inserting *escaped* parameter names
2260 # into the given dictionary. default dialect will
2261 # use these param names directly as they will not be
2262 # in the escaped_bind_names dictionary.
2263 values = parameters.pop(name)
2265 leep_res = self._literal_execute_expanding_parameter(
2266 escaped_name,
2267 parameter,
2268 values,
2269 taken_names=expanded_taken_names,
2270 )
2271 to_update, replacement_expr = leep_res
2273 to_update_sets[escaped_name] = to_update
2274 replacement_expressions[escaped_name] = replacement_expr
2276 if not parameter.literal_execute:
2277 parameters.update(to_update)
2278 if parameter.type._is_tuple_type:
2279 if name in tuple_processors:
2280 # to_update is flattened row by row, one entry
2281 # per element of each tuple
2282 element_processors = tuple_processors[name]
2283 num = len(element_processors)
2284 new_processors.update(
2285 (key, element_processors[idx % num])
2286 for idx, (key, _) in enumerate(to_update)
2287 if element_processors[idx % num] is not None
2288 )
2289 else:
2290 new_processors.update(
2291 (key, single_processors[name])
2292 for key, _ in to_update
2293 if name in single_processors
2294 )
2295 if numeric_positiontup is not None:
2296 numeric_positiontup.extend(
2297 name for name, _ in to_update
2298 )
2299 elif new_positiontup is not None:
2300 # to_update has escaped names, but that's ok since
2301 # these are new names, that aren't in the
2302 # escaped_bind_names dict.
2303 new_positiontup.extend(name for name, _ in to_update)
2304 expanded_parameters[name] = [
2305 expand_key for expand_key, _ in to_update
2306 ]
2307 elif new_positiontup is not None:
2308 new_positiontup.append(name)
2310 def process_expanding(m):
2311 key = m.group(1)
2312 expr = replacement_expressions[key]
2314 # if POSTCOMPILE included a bind_expression, render that
2315 # around each element
2316 if m.group(2):
2317 tok = m.group(2).split("~~")
2318 be_left, be_right = tok[1], tok[3]
2319 expr = ", ".join(
2320 "%s%s%s" % (be_left, exp, be_right)
2321 for exp in expr.split(", ")
2322 )
2323 return expr
2325 statement = re.sub(
2326 self._post_compile_pattern, process_expanding, pre_expanded_string
2327 )
2329 if numeric_positiontup is not None:
2330 assert new_positiontup is not None
2331 param_pos = {
2332 key: f"{self._numeric_binds_identifier_char}{num}"
2333 for num, key in enumerate(
2334 numeric_positiontup, self.next_numeric_pos
2335 )
2336 }
2337 if self.preparer._double_percents:
2338 pyformat_pattern = self._pct_escaped_pyformat_pattern
2339 else:
2340 pyformat_pattern = self._pyformat_pattern
2342 # only the expanded parameters are replaced; any other
2343 # match is an escaped percent sign or literal text that looks
2344 # like a bound parameter
2345 statement = pyformat_pattern.sub(
2346 lambda m: param_pos.get(m.group(1), m.group(0)), statement
2347 )
2348 new_positiontup.extend(numeric_positiontup)
2350 expanded_state = ExpandedState(
2351 statement,
2352 parameters,
2353 new_processors,
2354 new_positiontup,
2355 expanded_parameters,
2356 )
2358 if _populate_self:
2359 # this is for the "render_postcompile" flag, which is not
2360 # otherwise used internally and is for end-user debugging and
2361 # special use cases.
2362 self._pre_expanded_string = pre_expanded_string
2363 self._pre_expanded_positiontup = pre_expanded_positiontup
2364 self.string = expanded_state.statement
2365 self.positiontup = (
2366 list(expanded_state.positiontup or ())
2367 if self.positional
2368 else None
2369 )
2370 self._post_compile_expanded_state = expanded_state
2372 return expanded_state
2374 @util.preload_module("sqlalchemy.engine.cursor")
2375 def _create_result_map(self):
2376 """utility method used for unit tests only."""
2377 cursor = util.preloaded.engine_cursor
2378 return cursor.CursorResultMetaData._create_description_match_map(
2379 self._result_columns
2380 )
2382 # assigned by crud.py for insert/update statements
2383 _get_bind_name_for_col: _BindNameForColProtocol
2385 @util.memoized_property
2386 def _within_exec_param_key_getter(self) -> Callable[[Any], str]:
2387 getter = self._get_bind_name_for_col
2388 return getter
2390 @util.memoized_property
2391 @util.preload_module("sqlalchemy.engine.result")
2392 def _inserted_primary_key_from_lastrowid_getter(self):
2393 result = util.preloaded.engine_result
2395 param_key_getter = self._within_exec_param_key_getter
2397 assert self.compile_state is not None
2398 statement = self.compile_state.statement
2400 if TYPE_CHECKING:
2401 assert isinstance(statement, Insert)
2403 table = statement.table
2405 getters = [
2406 (operator.methodcaller("get", param_key_getter(col), None), col)
2407 for col in table.primary_key
2408 ]
2410 autoinc_getter = None
2411 autoinc_col = table._autoincrement_column
2412 if autoinc_col is not None:
2413 # apply type post processors to the lastrowid
2414 lastrowid_processor = autoinc_col.type._cached_result_processor(
2415 self.dialect, None
2416 )
2417 autoinc_key = param_key_getter(autoinc_col)
2419 # if a bind value is present for the autoincrement column
2420 # in the parameters, we need to do the logic dictated by
2421 # #7998; honor a non-None user-passed parameter over lastrowid.
2422 # previously in the 1.4 series we weren't fetching lastrowid
2423 # at all if the key were present in the parameters
2424 if autoinc_key in self.binds:
2426 def _autoinc_getter(lastrowid, parameters):
2427 param_value = parameters.get(autoinc_key, lastrowid)
2428 if param_value is not None:
2429 # they supplied non-None parameter, use that.
2430 # SQLite at least is observed to return the wrong
2431 # cursor.lastrowid for INSERT..ON CONFLICT so it
2432 # can't be used in all cases
2433 return param_value
2434 else:
2435 # use lastrowid
2436 return lastrowid
2438 # work around mypy https://github.com/python/mypy/issues/14027
2439 autoinc_getter = _autoinc_getter
2441 else:
2442 lastrowid_processor = None
2444 row_fn = result.result_tuple([col.key for col in table.primary_key])
2446 def get(lastrowid, parameters):
2447 """given cursor.lastrowid value and the parameters used for INSERT,
2448 return a "row" that represents the primary key, either by
2449 using the "lastrowid" or by extracting values from the parameters
2450 that were sent along with the INSERT.
2452 """
2453 if lastrowid_processor is not None:
2454 lastrowid = lastrowid_processor(lastrowid)
2456 if lastrowid is None:
2457 return row_fn(getter(parameters) for getter, col in getters)
2458 else:
2459 return row_fn(
2460 (
2461 (
2462 autoinc_getter(lastrowid, parameters)
2463 if autoinc_getter is not None
2464 else lastrowid
2465 )
2466 if col is autoinc_col
2467 else getter(parameters)
2468 )
2469 for getter, col in getters
2470 )
2472 return get
2474 @util.memoized_property
2475 @util.preload_module("sqlalchemy.engine.result")
2476 def _inserted_primary_key_from_returning_getter(self):
2477 result = util.preloaded.engine_result
2479 assert self.compile_state is not None
2480 statement = self.compile_state.statement
2482 if TYPE_CHECKING:
2483 assert isinstance(statement, Insert)
2485 param_key_getter = self._within_exec_param_key_getter
2486 table = statement.table
2488 returning = self.implicit_returning
2489 assert returning is not None
2490 ret = {col: idx for idx, col in enumerate(returning)}
2492 getters = cast(
2493 "List[Tuple[Callable[[Any], Any], bool]]",
2494 [
2495 (
2496 (operator.itemgetter(ret[col]), True)
2497 if col in ret
2498 else (
2499 operator.methodcaller(
2500 "get", param_key_getter(col), None
2501 ),
2502 False,
2503 )
2504 )
2505 for col in table.primary_key
2506 ],
2507 )
2509 row_fn = result.result_tuple([col.key for col in table.primary_key])
2511 def get(row, parameters):
2512 return row_fn(
2513 getter(row) if use_row else getter(parameters)
2514 for getter, use_row in getters
2515 )
2517 return get
2519 def default_from(self) -> str:
2520 """Called when a SELECT statement has no froms, and no FROM clause is
2521 to be appended.
2523 Gives Oracle Database a chance to tack on a ``FROM DUAL`` to the string
2524 output.
2526 """
2527 return ""
2529 def visit_override_binds(
2530 self,
2531 override_binds,
2532 add_to_result_map=None,
2533 result_map_targets=(),
2534 **kw,
2535 ):
2536 """SQL compile the nested element of an _OverrideBinds with
2537 bindparams swapped out.
2539 The _OverrideBinds is not normally expected to be compiled; it
2540 is meant to be used when an already cached statement is to be used,
2541 the compilation was already performed, and only the bound params should
2542 be swapped in at execution time.
2544 However, there are test cases that exericise this object, and
2545 additionally the ORM subquery loader is known to feed in expressions
2546 which include this construct into new queries (discovered in #11173),
2547 so it has to do the right thing at compile time as well.
2549 """
2551 if add_to_result_map is not None:
2552 # the ORM looks up result columns using the objects it placed
2553 # into the loader option, which for a cached statement is the
2554 # _OverrideBinds wrapper rather than the element inside of it.
2555 # make sure the wrapper is present in the result map so that
2556 # lookup succeeds. See #13560
2557 result_map_targets += (override_binds,)
2558 kw["add_to_result_map"] = add_to_result_map
2559 kw["result_map_targets"] = result_map_targets
2561 # get SQL text first
2562 sqltext = override_binds.element._compiler_dispatch(self, **kw)
2564 # for a test compile that is not for caching, change binds after the
2565 # fact. note that we don't try to
2566 # swap the bindparam as we compile, because our element may be
2567 # elsewhere in the statement already (e.g. a subquery or perhaps a
2568 # CTE) and was already visited / compiled. See
2569 # test_relationship_criteria.py ->
2570 # test_selectinload_local_criteria_subquery
2571 for k in override_binds.translate:
2572 if k not in self.binds:
2573 continue
2574 bp = self.binds[k]
2576 # so this would work, just change the value of bp in place.
2577 # but we dont want to mutate things outside.
2578 # bp.value = override_binds.translate[bp.key]
2579 # continue
2581 # instead, need to replace bp with new_bp or otherwise accommodate
2582 # in all internal collections
2583 new_bp = bp._with_value(
2584 override_binds.translate[bp.key],
2585 maintain_key=True,
2586 required=False,
2587 )
2589 name = self.bind_names[bp]
2590 self.binds[k] = self.binds[name] = new_bp
2591 self.bind_names[new_bp] = name
2592 self.bind_names.pop(bp, None)
2594 if bp in self.post_compile_params:
2595 self.post_compile_params |= {new_bp}
2596 if bp in self.literal_execute_params:
2597 self.literal_execute_params |= {new_bp}
2599 ckbm_tuple = self._cache_key_bind_match
2600 if ckbm_tuple:
2601 ckbm, cksm = ckbm_tuple
2602 for bp in bp._cloned_set:
2603 if bp.key in cksm:
2604 cb = cksm[bp.key]
2605 ckbm[cb].append(new_bp)
2607 return sqltext
2609 def visit_grouping(self, grouping, asfrom=False, **kwargs):
2610 return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")"
2612 def visit_select_statement_grouping(self, grouping, **kwargs):
2613 return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")"
2615 def visit_label_reference(
2616 self, element, within_columns_clause=False, **kwargs
2617 ):
2618 if self.stack and self.dialect.supports_simple_order_by_label:
2619 try:
2620 compile_state = cast(
2621 "Union[SelectState, CompoundSelectState]",
2622 self.stack[-1]["compile_state"],
2623 )
2624 except KeyError as ke:
2625 raise exc.CompileError(
2626 "Can't resolve label reference for ORDER BY / "
2627 "GROUP BY / DISTINCT etc."
2628 ) from ke
2630 (
2631 with_cols,
2632 only_froms,
2633 only_cols,
2634 ) = compile_state._label_resolve_dict
2635 if within_columns_clause:
2636 resolve_dict = only_froms
2637 else:
2638 resolve_dict = only_cols
2640 # this can be None in the case that a _label_reference()
2641 # were subject to a replacement operation, in which case
2642 # the replacement of the Label element may have changed
2643 # to something else like a ColumnClause expression.
2644 order_by_elem = element.element._order_by_label_element
2646 if (
2647 order_by_elem is not None
2648 and order_by_elem.name in resolve_dict
2649 and order_by_elem.shares_lineage(
2650 resolve_dict[order_by_elem.name]
2651 )
2652 ):
2653 kwargs["render_label_as_label"] = (
2654 element.element._order_by_label_element
2655 )
2656 return self.process(
2657 element.element,
2658 within_columns_clause=within_columns_clause,
2659 **kwargs,
2660 )
2662 def visit_textual_label_reference(
2663 self, element, within_columns_clause=False, **kwargs
2664 ):
2665 if not self.stack:
2666 # compiling the element outside of the context of a SELECT
2667 return self.process(element._text_clause)
2669 try:
2670 compile_state = cast(
2671 "Union[SelectState, CompoundSelectState]",
2672 self.stack[-1]["compile_state"],
2673 )
2674 except KeyError as ke:
2675 coercions._no_text_coercion(
2676 element.element,
2677 extra=(
2678 "Can't resolve label reference for ORDER BY / "
2679 "GROUP BY / DISTINCT etc."
2680 ),
2681 exc_cls=exc.CompileError,
2682 err=ke,
2683 )
2685 with_cols, only_froms, only_cols = compile_state._label_resolve_dict
2686 try:
2687 if within_columns_clause:
2688 col = only_froms[element.element]
2689 else:
2690 col = with_cols[element.element]
2691 except KeyError as err:
2692 coercions._no_text_coercion(
2693 element.element,
2694 extra=(
2695 "Can't resolve label reference for ORDER BY / "
2696 "GROUP BY / DISTINCT etc."
2697 ),
2698 exc_cls=exc.CompileError,
2699 err=err,
2700 )
2701 else:
2702 kwargs["render_label_as_label"] = col
2703 return self.process(
2704 col, within_columns_clause=within_columns_clause, **kwargs
2705 )
2707 def visit_label(
2708 self,
2709 label,
2710 add_to_result_map=None,
2711 within_label_clause=False,
2712 within_columns_clause=False,
2713 render_label_as_label=None,
2714 result_map_targets=(),
2715 within_tstring=False,
2716 **kw,
2717 ):
2718 if within_tstring:
2719 raise exc.CompileError(
2720 "Using label() directly inside tstring is not supported "
2721 "as it is ambiguous how the label expression should be "
2722 "rendered without knowledge of how it's being used in SQL"
2723 )
2724 # only render labels within the columns clause
2725 # or ORDER BY clause of a select. dialect-specific compilers
2726 # can modify this behavior.
2727 render_label_with_as = (
2728 within_columns_clause and not within_label_clause
2729 )
2730 render_label_only = render_label_as_label is label
2732 if render_label_only or render_label_with_as:
2733 if isinstance(label.name, elements._truncated_label):
2734 labelname = self._truncated_identifier("colident", label.name)
2735 else:
2736 labelname = label.name
2738 if render_label_with_as:
2739 if add_to_result_map is not None:
2740 add_to_result_map(
2741 labelname,
2742 label.name,
2743 (label, labelname) + label._alt_names + result_map_targets,
2744 label.type,
2745 )
2746 return (
2747 label.element._compiler_dispatch(
2748 self,
2749 within_columns_clause=True,
2750 within_label_clause=True,
2751 **kw,
2752 )
2753 + OPERATORS[operators.as_]
2754 + self.preparer.format_label(label, labelname)
2755 )
2756 elif render_label_only:
2757 return self.preparer.format_label(label, labelname)
2758 else:
2759 return label.element._compiler_dispatch(
2760 self, within_columns_clause=False, **kw
2761 )
2763 def _fallback_column_name(self, column):
2764 raise exc.CompileError(
2765 "Cannot compile Column object until its 'name' is assigned."
2766 )
2768 def visit_lambda_element(self, element, **kw):
2769 sql_element = element._resolved
2770 return self.process(sql_element, **kw)
2772 def visit_column(
2773 self,
2774 column: ColumnClause[Any],
2775 add_to_result_map: Optional[_ResultMapAppender] = None,
2776 include_table: bool = True,
2777 result_map_targets: Tuple[Any, ...] = (),
2778 ambiguous_table_name_map: Optional[_AmbiguousTableNameMap] = None,
2779 **kwargs: Any,
2780 ) -> str:
2781 name = orig_name = column.name
2782 if name is None:
2783 name = self._fallback_column_name(column)
2785 is_literal = column.is_literal
2786 if not is_literal and isinstance(name, elements._truncated_label):
2787 name = self._truncated_identifier("colident", name)
2789 if add_to_result_map is not None:
2790 targets = (column, name, column.key) + result_map_targets
2791 if column._tq_label:
2792 targets += (column._tq_label,)
2794 add_to_result_map(name, orig_name, targets, column.type)
2796 if is_literal:
2797 # note we are not currently accommodating for
2798 # literal_column(quoted_name('ident', True)) here
2799 name = self.escape_literal_column(name)
2800 else:
2801 name = self.preparer.quote(name)
2802 table = column.table
2803 if table is None or not include_table or not table.named_with_column:
2804 return name
2805 else:
2806 effective_schema = self.preparer.schema_for_object(table)
2808 if effective_schema:
2809 schema_prefix = (
2810 self.preparer.quote_schema(effective_schema) + "."
2811 )
2812 else:
2813 schema_prefix = ""
2815 if TYPE_CHECKING:
2816 assert isinstance(table, NamedFromClause)
2817 tablename = table.name
2819 if (
2820 not effective_schema
2821 and ambiguous_table_name_map
2822 and tablename in ambiguous_table_name_map
2823 ):
2824 tablename = ambiguous_table_name_map[tablename]
2826 if isinstance(tablename, elements._truncated_label):
2827 tablename = self._truncated_identifier("alias", tablename)
2829 return schema_prefix + self.preparer.quote(tablename) + "." + name
2831 def visit_collation(self, element, **kw):
2832 return self.preparer.format_collation(
2833 element.collation, element.collation_schema
2834 )
2836 def visit_fromclause(self, fromclause, **kwargs):
2837 return fromclause.name
2839 def visit_index(self, index, **kwargs):
2840 return index.name
2842 def visit_typeclause(self, typeclause, **kw):
2843 kw["type_expression"] = typeclause
2844 kw["identifier_preparer"] = self.preparer
2845 return self.dialect.type_compiler_instance.process(
2846 typeclause.type, **kw
2847 )
2849 def post_process_text(self, text):
2850 if self.preparer._double_percents:
2851 text = text.replace("%", "%%")
2852 return text
2854 def escape_literal_column(self, text):
2855 if self.preparer._double_percents:
2856 text = text.replace("%", "%%")
2857 return text
2859 def visit_textclause(self, textclause, add_to_result_map=None, **kw):
2860 if self._collect_params:
2861 self._add_to_params(textclause)
2863 def do_bindparam(m):
2864 name = m.group(1)
2865 if name in textclause._bindparams:
2866 return self.process(textclause._bindparams[name], **kw)
2867 else:
2868 return self.bindparam_string(name, **kw)
2870 if not self.stack:
2871 self.isplaintext = True
2873 if add_to_result_map:
2874 # text() object is present in the columns clause of a
2875 # select(). Add a no-name entry to the result map so that
2876 # row[text()] produces a result
2877 add_to_result_map(None, None, (textclause,), sqltypes.NULLTYPE)
2879 # un-escape any \:params
2880 return BIND_PARAMS_ESC.sub(
2881 lambda m: m.group(1),
2882 BIND_PARAMS.sub(
2883 do_bindparam, self.post_process_text(textclause.text)
2884 ),
2885 )
2887 def visit_tstring_text(self, element, **kw):
2888 return self.post_process_text(element.text)
2890 def visit_tstring(self, tstring, add_to_result_map=None, **kw):
2891 if self._collect_params:
2892 self._add_to_params(tstring)
2894 if not self.stack:
2895 self.isplaintext = True
2897 if add_to_result_map:
2898 # tstring() object is present in the columns clause of a
2899 # select(). Add a no-name entry to the result map so that
2900 # row[tstring()] produces a result
2901 add_to_result_map(None, None, (tstring,), sqltypes.NULLTYPE)
2903 # Process each part and concatenate
2904 kw["within_tstring"] = True
2905 return "".join(self.process(part, **kw) for part in tstring.parts)
2907 def visit_textual_select(
2908 self, taf, compound_index=None, asfrom=False, **kw
2909 ):
2910 if self._collect_params:
2911 self._add_to_params(taf)
2912 toplevel = not self.stack
2913 entry = self._default_stack_entry if toplevel else self.stack[-1]
2915 new_entry: _CompilerStackEntry = {
2916 "correlate_froms": set(),
2917 "asfrom_froms": set(),
2918 "selectable": taf,
2919 }
2920 self.stack.append(new_entry)
2922 if taf._independent_ctes:
2923 self._dispatch_independent_ctes(taf, kw)
2925 populate_result_map = (
2926 toplevel
2927 or (
2928 compound_index == 0
2929 and entry.get("need_result_map_for_compound", False)
2930 )
2931 or entry.get("need_result_map_for_nested", False)
2932 )
2934 if populate_result_map:
2935 self._ordered_columns = self._textual_ordered_columns = (
2936 taf.positional
2937 )
2939 # enable looser result column matching when the SQL text links to
2940 # Column objects by name only
2941 self._loose_column_name_matching = not taf.positional and bool(
2942 taf.column_args
2943 )
2945 for c in taf.column_args:
2946 self.process(
2947 c,
2948 within_columns_clause=True,
2949 add_to_result_map=self._add_to_result_map,
2950 )
2952 text = self.process(taf.element, **kw)
2953 if self.ctes:
2954 nesting_level = len(self.stack) if not toplevel else None
2955 text = self._render_cte_clause(nesting_level=nesting_level) + text
2957 self.stack.pop(-1)
2959 return text
2961 def visit_null(self, expr: Null, **kw: Any) -> str:
2962 return "NULL"
2964 def visit_true(self, expr: True_, **kw: Any) -> str:
2965 if self.dialect.supports_native_boolean:
2966 return "true"
2967 else:
2968 return "1"
2970 def visit_false(self, expr: False_, **kw: Any) -> str:
2971 if self.dialect.supports_native_boolean:
2972 return "false"
2973 else:
2974 return "0"
2976 def _generate_delimited_list(self, elements, separator, **kw):
2977 return separator.join(
2978 s
2979 for s in (c._compiler_dispatch(self, **kw) for c in elements)
2980 if s
2981 )
2983 def _generate_delimited_and_list(self, clauses, **kw):
2984 lcc, clauses = elements.BooleanClauseList._process_clauses_for_boolean(
2985 operators.and_,
2986 elements.True_._singleton,
2987 elements.False_._singleton,
2988 clauses,
2989 )
2990 if lcc == 1:
2991 return clauses[0]._compiler_dispatch(self, **kw)
2992 else:
2993 separator = OPERATORS[operators.and_]
2994 return separator.join(
2995 s
2996 for s in (c._compiler_dispatch(self, **kw) for c in clauses)
2997 if s
2998 )
3000 def visit_tuple(self, clauselist, **kw):
3001 return "(%s)" % self.visit_clauselist(clauselist, **kw)
3003 def visit_element_list(self, element, **kw):
3004 return self._generate_delimited_list(element.clauses, " ", **kw)
3006 def visit_order_by_list(self, element, **kw):
3007 return self._generate_delimited_list(element.clauses, ", ", **kw)
3009 def visit_clauselist(self, clauselist, **kw):
3010 sep = clauselist.operator
3011 if sep is None:
3012 sep = " "
3013 else:
3014 sep = OPERATORS[clauselist.operator]
3016 return self._generate_delimited_list(clauselist.clauses, sep, **kw)
3018 def visit_expression_clauselist(self, clauselist, **kw):
3019 operator_ = clauselist.operator
3021 disp = self._get_operator_dispatch(
3022 operator_, "expression_clauselist", None
3023 )
3024 if disp:
3025 return disp(clauselist, operator_, **kw)
3027 try:
3028 opstring = OPERATORS[operator_]
3029 except KeyError as err:
3030 raise exc.UnsupportedCompilationError(self, operator_) from err
3031 else:
3032 kw["_in_operator_expression"] = True
3033 return self._generate_delimited_list(
3034 clauselist.clauses, opstring, **kw
3035 )
3037 def visit_case(self, clause, **kwargs):
3038 x = "CASE "
3039 if clause.value is not None:
3040 x += clause.value._compiler_dispatch(self, **kwargs) + " "
3041 for cond, result in clause.whens:
3042 x += (
3043 "WHEN "
3044 + cond._compiler_dispatch(self, **kwargs)
3045 + " THEN "
3046 + result._compiler_dispatch(self, **kwargs)
3047 + " "
3048 )
3049 if clause.else_ is not None:
3050 x += (
3051 "ELSE " + clause.else_._compiler_dispatch(self, **kwargs) + " "
3052 )
3053 x += "END"
3054 return x
3056 def visit_type_coerce(self, type_coerce, **kw):
3057 return type_coerce.typed_expression._compiler_dispatch(self, **kw)
3059 def visit_cast(self, cast, **kwargs):
3060 type_clause = cast.typeclause._compiler_dispatch(self, **kwargs)
3061 match = re.match("(.*)( COLLATE .*)", type_clause)
3062 return "CAST(%s AS %s)%s" % (
3063 cast.clause._compiler_dispatch(self, **kwargs),
3064 match.group(1) if match else type_clause,
3065 match.group(2) if match else "",
3066 )
3068 def visit_frame_clause(self, frameclause, **kw):
3070 if frameclause.lower_type is elements.FrameClauseType.UNBOUNDED:
3071 left = "UNBOUNDED PRECEDING"
3072 elif frameclause.lower_type is elements.FrameClauseType.CURRENT:
3073 left = "CURRENT ROW"
3074 else:
3075 val = self.process(frameclause.lower_bind, **kw)
3076 if frameclause.lower_type is elements.FrameClauseType.PRECEDING:
3077 left = f"{val} PRECEDING"
3078 else:
3079 left = f"{val} FOLLOWING"
3081 if frameclause.upper_type is elements.FrameClauseType.UNBOUNDED:
3082 right = "UNBOUNDED FOLLOWING"
3083 elif frameclause.upper_type is elements.FrameClauseType.CURRENT:
3084 right = "CURRENT ROW"
3085 else:
3086 val = self.process(frameclause.upper_bind, **kw)
3087 if frameclause.upper_type is elements.FrameClauseType.PRECEDING:
3088 right = f"{val} PRECEDING"
3089 else:
3090 right = f"{val} FOLLOWING"
3092 return f"{left} AND {right}"
3094 def visit_over(self, over, **kwargs):
3095 text = over.element._compiler_dispatch(self, **kwargs)
3096 if over.range_ is not None:
3097 range_ = f"RANGE BETWEEN {self.process(over.range_, **kwargs)}"
3098 elif over.rows is not None:
3099 range_ = f"ROWS BETWEEN {self.process(over.rows, **kwargs)}"
3100 elif over.groups is not None:
3101 range_ = f"GROUPS BETWEEN {self.process(over.groups, **kwargs)}"
3102 else:
3103 range_ = None
3105 if range_ is not None and over.exclude is not None:
3106 range_ += " EXCLUDE " + self.preparer.validate_sql_phrase(
3107 over.exclude, _WINDOW_EXCLUDE_RE
3108 )
3110 return "%s OVER (%s)" % (
3111 text,
3112 " ".join(
3113 [
3114 "%s BY %s"
3115 % (word, clause._compiler_dispatch(self, **kwargs))
3116 for word, clause in (
3117 ("PARTITION", over.partition_by),
3118 ("ORDER", over.order_by),
3119 )
3120 if clause is not None and len(clause)
3121 ]
3122 + ([range_] if range_ else [])
3123 ),
3124 )
3126 def visit_withingroup(self, withingroup, **kwargs):
3127 return "%s WITHIN GROUP (ORDER BY %s)" % (
3128 withingroup.element._compiler_dispatch(self, **kwargs),
3129 withingroup.order_by._compiler_dispatch(self, **kwargs),
3130 )
3132 def visit_funcfilter(self, funcfilter, **kwargs):
3133 return "%s FILTER (WHERE %s)" % (
3134 funcfilter.func._compiler_dispatch(self, **kwargs),
3135 funcfilter.criterion._compiler_dispatch(self, **kwargs),
3136 )
3138 def visit_aggregateorderby(self, aggregateorderby, **kwargs):
3139 if self.dialect.aggregate_order_by_style is AggregateOrderByStyle.NONE:
3140 raise exc.CompileError(
3141 "this dialect does not support "
3142 "ORDER BY within an aggregate function"
3143 )
3144 elif (
3145 self.dialect.aggregate_order_by_style
3146 is AggregateOrderByStyle.INLINE
3147 ):
3148 new_fn = aggregateorderby.element._clone()
3149 new_fn.clause_expr = elements.Grouping(
3150 aggregate_orderby_inline(
3151 new_fn.clause_expr.element, aggregateorderby.order_by
3152 )
3153 )
3155 return new_fn._compiler_dispatch(self, **kwargs)
3156 else:
3157 return self.visit_withingroup(aggregateorderby, **kwargs)
3159 def visit_aggregate_orderby_inline(self, element, **kw):
3160 return "%s ORDER BY %s" % (
3161 self.process(element.element, **kw),
3162 self.process(element.aggregate_order_by, **kw),
3163 )
3165 def visit_aggregate_strings_func(
3166 self, fn, *, use_function_name=None, **kw
3167 ):
3168 # aggreagate_order_by attribute is present if visit_function
3169 # gave us a Function with aggregate_orderby_inline() as the inner
3170 # contents
3171 order_by = getattr(fn.clauses, "aggregate_order_by", None)
3173 if use_function_name is None:
3174 raise exc.CompileError(
3175 "%s dialect does not support the aggregate_strings function"
3176 % self.dialect.name
3177 )
3179 literal_exec = dict(kw)
3180 literal_exec["literal_execute"] = True
3182 # break up the function into its components so we can apply
3183 # literal_execute to the second argument (the delimiter)
3184 cl = list(fn.clauses)
3185 expr, delimiter = cl[0:2]
3186 if (
3187 order_by is not None
3188 and self.dialect.aggregate_order_by_style
3189 is AggregateOrderByStyle.INLINE
3190 ):
3191 return (
3192 f"{use_function_name}({expr._compiler_dispatch(self, **kw)}, "
3193 f"{delimiter._compiler_dispatch(self, **literal_exec)} "
3194 f"ORDER BY {order_by._compiler_dispatch(self, **kw)})"
3195 )
3196 else:
3197 return (
3198 f"{use_function_name}({expr._compiler_dispatch(self, **kw)}, "
3199 f"{delimiter._compiler_dispatch(self, **literal_exec)})"
3200 )
3202 def visit_extract(self, extract, **kwargs):
3203 field = self.extract_map.get(extract.field, extract.field)
3204 return "EXTRACT(%s FROM %s)" % (
3205 field,
3206 extract.expr._compiler_dispatch(self, **kwargs),
3207 )
3209 def visit_scalar_function_column(self, element, **kw):
3210 compiled_fn = self.visit_function(element.fn, **kw)
3211 compiled_col = self.visit_column(element, **kw)
3212 return "(%s).%s" % (compiled_fn, compiled_col)
3214 def visit_function(
3215 self,
3216 func: Function[Any],
3217 add_to_result_map: Optional[_ResultMapAppender] = None,
3218 **kwargs: Any,
3219 ) -> str:
3220 if self._collect_params:
3221 self._add_to_params(func)
3222 if add_to_result_map is not None:
3223 add_to_result_map(func.name, func.name, (func.name,), func.type)
3225 disp = getattr(self, "visit_%s_func" % func.name.lower(), None)
3227 text: str
3229 kwargs["within_aggregate_function"] = True
3231 if disp:
3232 text = disp(func, **kwargs)
3233 else:
3234 name = FUNCTIONS.get(func._deannotate().__class__, None)
3235 if name:
3236 if func._has_args:
3237 name += "%(expr)s"
3238 else:
3239 name = func.name
3240 name = (
3241 self.preparer.quote(name)
3242 if self.preparer._requires_quotes_illegal_chars(name)
3243 or isinstance(name, elements.quoted_name)
3244 else name
3245 )
3246 name = name + "%(expr)s"
3247 text = ".".join(
3248 [
3249 (
3250 self.preparer.quote(tok)
3251 if self.preparer._requires_quotes_illegal_chars(tok)
3252 or isinstance(name, elements.quoted_name)
3253 else tok
3254 )
3255 for tok in func.packagenames
3256 ]
3257 + [name]
3258 ) % {"expr": self.function_argspec(func, **kwargs)}
3260 if func._with_ordinality:
3261 text += " WITH ORDINALITY"
3262 return text
3264 def visit_next_value_func(self, next_value, **kw):
3265 return self.visit_sequence(next_value.sequence)
3267 def visit_sequence(self, sequence, **kw):
3268 raise NotImplementedError(
3269 "Dialect '%s' does not support sequence increments."
3270 % self.dialect.name
3271 )
3273 def function_argspec(self, func: Function[Any], **kwargs: Any) -> str:
3274 return func.clause_expr._compiler_dispatch(self, **kwargs)
3276 def visit_compound_select(
3277 self, cs, asfrom=False, compound_index=None, **kwargs
3278 ):
3279 if self._collect_params:
3280 self._add_to_params(cs)
3281 toplevel = not self.stack
3283 compile_state = cs._compile_state_factory(cs, self, **kwargs)
3285 if toplevel and not self.compile_state:
3286 self.compile_state = compile_state
3288 compound_stmt = compile_state.statement
3290 entry = self._default_stack_entry if toplevel else self.stack[-1]
3291 need_result_map = toplevel or (
3292 not compound_index
3293 and entry.get("need_result_map_for_compound", False)
3294 )
3296 # indicates there is already a CompoundSelect in play
3297 if compound_index == 0:
3298 entry["select_0"] = cs
3300 self.stack.append(
3301 {
3302 "correlate_froms": entry["correlate_froms"],
3303 "asfrom_froms": entry["asfrom_froms"],
3304 "selectable": cs,
3305 "compile_state": compile_state,
3306 "need_result_map_for_compound": need_result_map,
3307 }
3308 )
3310 if compound_stmt._independent_ctes:
3311 self._dispatch_independent_ctes(compound_stmt, kwargs)
3313 keyword = self.compound_keywords[cs.keyword]
3315 text = (" " + keyword + " ").join(
3316 (
3317 c._compiler_dispatch(
3318 self, asfrom=asfrom, compound_index=i, **kwargs
3319 )
3320 for i, c in enumerate(cs.selects)
3321 )
3322 )
3324 kwargs["include_table"] = False
3325 text += self.group_by_clause(cs, **dict(asfrom=asfrom, **kwargs))
3326 text += self.order_by_clause(cs, **kwargs)
3327 if cs._has_row_limiting_clause:
3328 text += self._row_limit_clause(cs, **kwargs)
3330 if self.ctes:
3331 nesting_level = len(self.stack) if not toplevel else None
3332 text = (
3333 self._render_cte_clause(
3334 nesting_level=nesting_level,
3335 include_following_stack=True,
3336 )
3337 + text
3338 )
3340 self.stack.pop(-1)
3341 return text
3343 def _row_limit_clause(self, cs, **kwargs):
3344 if cs._fetch_clause is not None:
3345 return self.fetch_clause(cs, **kwargs)
3346 else:
3347 return self.limit_clause(cs, **kwargs)
3349 def _get_operator_dispatch(self, operator_, qualifier1, qualifier2):
3350 attrname = "visit_%s_%s%s" % (
3351 operator_.__name__,
3352 qualifier1,
3353 "_" + qualifier2 if qualifier2 else "",
3354 )
3355 return getattr(self, attrname, None)
3357 def _get_custom_operator_dispatch(self, operator_, qualifier1):
3358 attrname = "visit_%s_op_%s" % (operator_.visit_name, qualifier1)
3359 return getattr(self, attrname, None)
3361 def visit_unary(
3362 self, unary, add_to_result_map=None, result_map_targets=(), **kw
3363 ):
3364 if add_to_result_map is not None:
3365 result_map_targets += (unary,)
3366 kw["add_to_result_map"] = add_to_result_map
3367 kw["result_map_targets"] = result_map_targets
3369 if unary.operator is operators.distinct_op and not kw.get(
3370 "within_aggregate_function", False
3371 ):
3372 util.warn(
3373 "Column-expression-level unary distinct() "
3374 "should not be used outside of an aggregate "
3375 "function. For general 'SELECT DISTINCT' support "
3376 "use select().distinct()."
3377 )
3379 if unary.operator:
3380 if unary.modifier:
3381 raise exc.CompileError(
3382 "Unary expression does not support operator "
3383 "and modifier simultaneously"
3384 )
3385 disp = self._get_operator_dispatch(
3386 unary.operator, "unary", "operator"
3387 )
3388 if disp:
3389 return disp(unary, unary.operator, **kw)
3390 else:
3391 return self._generate_generic_unary_operator(
3392 unary, OPERATORS[unary.operator], **kw
3393 )
3394 elif unary.modifier:
3395 disp = self._get_operator_dispatch(
3396 unary.modifier, "unary", "modifier"
3397 )
3398 if disp:
3399 return disp(unary, unary.modifier, **kw)
3400 else:
3401 return self._generate_generic_unary_modifier(
3402 unary, OPERATORS[unary.modifier], **kw
3403 )
3404 else:
3405 raise exc.CompileError(
3406 "Unary expression has no operator or modifier"
3407 )
3409 def visit_truediv_binary(self, binary, operator, **kw):
3410 if self.dialect.div_is_floordiv:
3411 return (
3412 self.process(binary.left, **kw)
3413 + " / "
3414 # TODO: would need a fast cast again here,
3415 # unless we want to use an implicit cast like "+ 0.0"
3416 + self.process(
3417 elements.Cast(
3418 binary.right,
3419 (
3420 binary.right.type
3421 if binary.right.type._type_affinity
3422 in (sqltypes.Numeric, sqltypes.Float)
3423 else sqltypes.Numeric()
3424 ),
3425 ),
3426 **kw,
3427 )
3428 )
3429 else:
3430 return (
3431 self.process(binary.left, **kw)
3432 + " / "
3433 + self.process(binary.right, **kw)
3434 )
3436 def visit_floordiv_binary(self, binary, operator, **kw):
3437 if (
3438 self.dialect.div_is_floordiv
3439 and binary.right.type._type_affinity is sqltypes.Integer
3440 and binary.left.type._type_affinity is sqltypes.Integer
3441 ):
3442 return (
3443 self.process(binary.left, **kw)
3444 + " / "
3445 + self.process(binary.right, **kw)
3446 )
3447 else:
3448 return "FLOOR(%s)" % (
3449 self.process(binary.left, **kw)
3450 + " / "
3451 + self.process(binary.right, **kw)
3452 )
3454 def visit_is_true_unary_operator(self, element, operator, **kw):
3455 if (
3456 element._is_implicitly_boolean
3457 or self.dialect.supports_native_boolean
3458 ):
3459 return self.process(element.element, **kw)
3460 else:
3461 return "%s = 1" % self.process(element.element, **kw)
3463 def visit_is_false_unary_operator(self, element, operator, **kw):
3464 if (
3465 element._is_implicitly_boolean
3466 or self.dialect.supports_native_boolean
3467 ):
3468 return "NOT %s" % self.process(element.element, **kw)
3469 else:
3470 return "%s = 0" % self.process(element.element, **kw)
3472 def visit_not_match_op_binary(self, binary, operator, **kw):
3473 return "NOT %s" % self.visit_binary(
3474 binary, override_operator=operators.match_op
3475 )
3477 def visit_not_in_op_binary(self, binary, operator, **kw):
3478 # The brackets are required in the NOT IN operation because the empty
3479 # case is handled using the form "(col NOT IN (null) OR 1 = 1)".
3480 # The presence of the OR makes the brackets required.
3481 return "(%s)" % self._generate_generic_binary(
3482 binary, OPERATORS[operator], **kw
3483 )
3485 def visit_empty_set_op_expr(self, type_, expand_op, **kw):
3486 if expand_op is operators.not_in_op:
3487 if len(type_) > 1:
3488 return "(%s)) OR (1 = 1" % (
3489 ", ".join("NULL" for element in type_)
3490 )
3491 else:
3492 return "NULL) OR (1 = 1"
3493 elif expand_op is operators.in_op:
3494 if len(type_) > 1:
3495 return "(%s)) AND (1 != 1" % (
3496 ", ".join("NULL" for element in type_)
3497 )
3498 else:
3499 return "NULL) AND (1 != 1"
3500 else:
3501 return self.visit_empty_set_expr(type_)
3503 def visit_empty_set_expr(self, element_types, **kw):
3504 raise NotImplementedError(
3505 "Dialect '%s' does not support empty set expression."
3506 % self.dialect.name
3507 )
3509 def _literal_execute_expanding_parameter_literal_binds(
3510 self, parameter, values, bind_expression_template=None
3511 ):
3512 typ_dialect_impl = parameter.type._unwrapped_dialect_impl(self.dialect)
3514 if not values:
3515 # empty IN expression. note we don't need to use
3516 # bind_expression_template here because there are no
3517 # expressions to render.
3519 if typ_dialect_impl._is_tuple_type:
3520 replacement_expression = (
3521 "VALUES " if self.dialect.tuple_in_values else ""
3522 ) + self.visit_empty_set_op_expr(
3523 parameter.type.types, parameter.expand_op
3524 )
3526 else:
3527 replacement_expression = self.visit_empty_set_op_expr(
3528 [parameter.type], parameter.expand_op
3529 )
3531 elif typ_dialect_impl._is_tuple_type or (
3532 typ_dialect_impl._isnull
3533 and isinstance(values[0], collections_abc.Sequence)
3534 and not isinstance(values[0], (str, bytes))
3535 ):
3536 if typ_dialect_impl._has_bind_expression:
3537 raise NotImplementedError(
3538 "bind_expression() on TupleType not supported with "
3539 "literal_binds"
3540 )
3542 replacement_expression = (
3543 "VALUES " if self.dialect.tuple_in_values else ""
3544 ) + ", ".join(
3545 "(%s)"
3546 % (
3547 ", ".join(
3548 self.render_literal_value(value, param_type)
3549 for value, param_type in zip(
3550 tuple_element, parameter.type.types
3551 )
3552 )
3553 )
3554 for i, tuple_element in enumerate(values)
3555 )
3556 else:
3557 if bind_expression_template:
3558 post_compile_pattern = self._post_compile_pattern
3559 m = post_compile_pattern.search(bind_expression_template)
3560 assert m and m.group(
3561 2
3562 ), "unexpected format for expanding parameter"
3564 tok = m.group(2).split("~~")
3565 be_left, be_right = tok[1], tok[3]
3566 replacement_expression = ", ".join(
3567 "%s%s%s"
3568 % (
3569 be_left,
3570 self.render_literal_value(value, parameter.type),
3571 be_right,
3572 )
3573 for value in values
3574 )
3575 else:
3576 replacement_expression = ", ".join(
3577 self.render_literal_value(value, parameter.type)
3578 for value in values
3579 )
3581 return (), replacement_expression
3583 def _literal_execute_expanding_parameter(
3584 self, name, parameter, values, *, taken_names
3585 ):
3586 """Render the per-element bound parameters of an expanding
3587 parameter.
3589 ``taken_names`` is a set of names the elements may not use; each
3590 name generated is checked against it and added to it. The element
3591 names are normally ``<name>_<n>``, or ``<name>_<n>_<m>`` for tuples,
3592 which may be the name of another parameter in the same statement,
3593 e.g. a parameter ``x_1`` alongside the expanding parameter ``x``, or
3594 the element of another expanding parameter. A colliding name is
3595 uniquified with a counter, rather than the element value replacing
3596 the value of the other parameter. See :ticket:`13695`.
3598 """
3599 if parameter.literal_execute:
3600 return self._literal_execute_expanding_parameter_literal_binds(
3601 parameter, values
3602 )
3604 dialect = self.dialect
3605 typ_dialect_impl = parameter.type._unwrapped_dialect_impl(dialect)
3607 if self._numeric_binds:
3608 bind_template = self.compilation_bindtemplate
3609 else:
3610 bind_template = self.bindtemplate
3612 if (
3613 self.dialect._bind_typing_render_casts
3614 and typ_dialect_impl.render_bind_cast
3615 ):
3617 def _render_bindtemplate(name):
3618 return self.render_bind_cast(
3619 parameter.type,
3620 typ_dialect_impl,
3621 bind_template % {"name": name},
3622 )
3624 else:
3626 def _render_bindtemplate(name):
3627 return bind_template % {"name": name}
3629 def _unique_name(key):
3630 if key in taken_names:
3631 base = key
3632 counter = 0
3633 while key in taken_names:
3634 counter += 1
3635 # double underscore keeps the suffix out of the
3636 # <name>_<n> form generated for anonymous names and
3637 # for the elements themselves, so a renamed element
3638 # won't take a name another element would generate
3639 key = "%s__%d" % (base, counter)
3640 taken_names.add(key)
3641 return key
3643 if not values:
3644 to_update = []
3645 if typ_dialect_impl._is_tuple_type:
3646 replacement_expression = self.visit_empty_set_op_expr(
3647 parameter.type.types, parameter.expand_op
3648 )
3649 else:
3650 replacement_expression = self.visit_empty_set_op_expr(
3651 [parameter.type], parameter.expand_op
3652 )
3654 elif typ_dialect_impl._is_tuple_type or (
3655 typ_dialect_impl._isnull
3656 and isinstance(values[0], collections_abc.Sequence)
3657 and not isinstance(values[0], (str, bytes))
3658 ):
3659 assert not typ_dialect_impl._is_array
3660 to_update = [
3661 (_unique_name("%s_%s_%s" % (name, i, j)), value)
3662 for i, tuple_element in enumerate(values, 1)
3663 for j, value in enumerate(tuple_element, 1)
3664 ]
3666 replacement_expression = (
3667 "VALUES " if dialect.tuple_in_values else ""
3668 ) + ", ".join(
3669 "(%s)"
3670 % (
3671 ", ".join(
3672 _render_bindtemplate(
3673 to_update[i * len(tuple_element) + j][0]
3674 )
3675 for j, value in enumerate(tuple_element)
3676 )
3677 )
3678 for i, tuple_element in enumerate(values)
3679 )
3680 else:
3681 to_update = [
3682 (_unique_name("%s_%s" % (name, i)), value)
3683 for i, value in enumerate(values, 1)
3684 ]
3685 replacement_expression = ", ".join(
3686 _render_bindtemplate(key) for key, value in to_update
3687 )
3689 return to_update, replacement_expression
3691 def visit_binary(
3692 self,
3693 binary,
3694 override_operator=None,
3695 eager_grouping=False,
3696 from_linter=None,
3697 lateral_from_linter=None,
3698 **kw,
3699 ):
3700 if from_linter and operators.is_comparison(binary.operator):
3701 if lateral_from_linter is not None:
3702 enclosing_lateral = kw["enclosing_lateral"]
3703 lateral_from_linter.edges.update(
3704 itertools.product(
3705 _de_clone(
3706 binary.left._from_objects + [enclosing_lateral]
3707 ),
3708 _de_clone(
3709 binary.right._from_objects + [enclosing_lateral]
3710 ),
3711 )
3712 )
3713 else:
3714 from_linter.edges.update(
3715 itertools.product(
3716 _de_clone(binary.left._from_objects),
3717 _de_clone(binary.right._from_objects),
3718 )
3719 )
3721 # don't allow "? = ?" to render
3722 if (
3723 self.ansi_bind_rules
3724 and isinstance(binary.left, elements.BindParameter)
3725 and isinstance(binary.right, elements.BindParameter)
3726 ):
3727 kw["literal_execute"] = True
3729 operator_ = override_operator or binary.operator
3730 disp = self._get_operator_dispatch(operator_, "binary", None)
3731 if disp:
3732 return disp(binary, operator_, **kw)
3733 else:
3734 try:
3735 opstring = OPERATORS[operator_]
3736 except KeyError as err:
3737 raise exc.UnsupportedCompilationError(self, operator_) from err
3738 else:
3739 return self._generate_generic_binary(
3740 binary,
3741 opstring,
3742 from_linter=from_linter,
3743 lateral_from_linter=lateral_from_linter,
3744 **kw,
3745 )
3747 def visit_function_as_comparison_op_binary(self, element, operator, **kw):
3748 return self.process(element.sql_function, **kw)
3750 def visit_mod_binary(self, binary, operator, **kw):
3751 if self.preparer._double_percents:
3752 return (
3753 self.process(binary.left, **kw)
3754 + " %% "
3755 + self.process(binary.right, **kw)
3756 )
3757 else:
3758 return (
3759 self.process(binary.left, **kw)
3760 + " % "
3761 + self.process(binary.right, **kw)
3762 )
3764 def visit_custom_op_binary(self, element, operator, **kw):
3765 if operator.visit_name:
3766 disp = self._get_custom_operator_dispatch(operator, "binary")
3767 if disp:
3768 return disp(element, operator, **kw)
3770 kw["eager_grouping"] = operator.eager_grouping
3771 return self._generate_generic_binary(
3772 element,
3773 " " + self.escape_literal_column(operator.opstring) + " ",
3774 **kw,
3775 )
3777 def visit_custom_op_unary_operator(self, element, operator, **kw):
3778 if operator.visit_name:
3779 disp = self._get_custom_operator_dispatch(operator, "unary")
3780 if disp:
3781 return disp(element, operator, **kw)
3783 return self._generate_generic_unary_operator(
3784 element, self.escape_literal_column(operator.opstring) + " ", **kw
3785 )
3787 def visit_custom_op_unary_modifier(self, element, operator, **kw):
3788 if operator.visit_name:
3789 disp = self._get_custom_operator_dispatch(operator, "unary")
3790 if disp:
3791 return disp(element, operator, **kw)
3793 return self._generate_generic_unary_modifier(
3794 element, " " + self.escape_literal_column(operator.opstring), **kw
3795 )
3797 def _generate_generic_binary(
3798 self,
3799 binary: BinaryExpression[Any],
3800 opstring: str,
3801 eager_grouping: bool = False,
3802 **kw: Any,
3803 ) -> str:
3804 _in_operator_expression = kw.get("_in_operator_expression", False)
3806 kw["_in_operator_expression"] = True
3807 kw["_binary_op"] = binary.operator
3808 text = (
3809 binary.left._compiler_dispatch(
3810 self, eager_grouping=eager_grouping, **kw
3811 )
3812 + opstring
3813 + binary.right._compiler_dispatch(
3814 self, eager_grouping=eager_grouping, **kw
3815 )
3816 )
3818 if _in_operator_expression and eager_grouping:
3819 text = "(%s)" % text
3820 return text
3822 def _generate_generic_unary_operator(self, unary, opstring, **kw):
3823 return opstring + unary.element._compiler_dispatch(self, **kw)
3825 def _generate_generic_unary_modifier(self, unary, opstring, **kw):
3826 return unary.element._compiler_dispatch(self, **kw) + opstring
3828 @util.memoized_property
3829 def _like_percent_literal(self):
3830 return elements.literal_column("'%'", type_=sqltypes.STRINGTYPE)
3832 def visit_ilike_case_insensitive_operand(self, element, **kw):
3833 return f"lower({element.element._compiler_dispatch(self, **kw)})"
3835 def visit_contains_op_binary(self, binary, operator, **kw):
3836 binary = binary._clone()
3837 percent = self._like_percent_literal
3838 binary.right = percent.concat(binary.right).concat(percent)
3839 return self.visit_like_op_binary(binary, operator, **kw)
3841 def visit_not_contains_op_binary(self, binary, operator, **kw):
3842 binary = binary._clone()
3843 percent = self._like_percent_literal
3844 binary.right = percent.concat(binary.right).concat(percent)
3845 return self.visit_not_like_op_binary(binary, operator, **kw)
3847 def visit_icontains_op_binary(self, binary, operator, **kw):
3848 binary = binary._clone()
3849 percent = self._like_percent_literal
3850 binary.left = ilike_case_insensitive(binary.left)
3851 binary.right = percent.concat(
3852 ilike_case_insensitive(binary.right)
3853 ).concat(percent)
3854 return self.visit_ilike_op_binary(binary, operator, **kw)
3856 def visit_not_icontains_op_binary(self, binary, operator, **kw):
3857 binary = binary._clone()
3858 percent = self._like_percent_literal
3859 binary.left = ilike_case_insensitive(binary.left)
3860 binary.right = percent.concat(
3861 ilike_case_insensitive(binary.right)
3862 ).concat(percent)
3863 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3865 def visit_startswith_op_binary(self, binary, operator, **kw):
3866 binary = binary._clone()
3867 percent = self._like_percent_literal
3868 binary.right = percent._rconcat(binary.right)
3869 return self.visit_like_op_binary(binary, operator, **kw)
3871 def visit_not_startswith_op_binary(self, binary, operator, **kw):
3872 binary = binary._clone()
3873 percent = self._like_percent_literal
3874 binary.right = percent._rconcat(binary.right)
3875 return self.visit_not_like_op_binary(binary, operator, **kw)
3877 def visit_istartswith_op_binary(self, binary, operator, **kw):
3878 binary = binary._clone()
3879 percent = self._like_percent_literal
3880 binary.left = ilike_case_insensitive(binary.left)
3881 binary.right = percent._rconcat(ilike_case_insensitive(binary.right))
3882 return self.visit_ilike_op_binary(binary, operator, **kw)
3884 def visit_not_istartswith_op_binary(self, binary, operator, **kw):
3885 binary = binary._clone()
3886 percent = self._like_percent_literal
3887 binary.left = ilike_case_insensitive(binary.left)
3888 binary.right = percent._rconcat(ilike_case_insensitive(binary.right))
3889 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3891 def visit_endswith_op_binary(self, binary, operator, **kw):
3892 binary = binary._clone()
3893 percent = self._like_percent_literal
3894 binary.right = percent.concat(binary.right)
3895 return self.visit_like_op_binary(binary, operator, **kw)
3897 def visit_not_endswith_op_binary(self, binary, operator, **kw):
3898 binary = binary._clone()
3899 percent = self._like_percent_literal
3900 binary.right = percent.concat(binary.right)
3901 return self.visit_not_like_op_binary(binary, operator, **kw)
3903 def visit_iendswith_op_binary(self, binary, operator, **kw):
3904 binary = binary._clone()
3905 percent = self._like_percent_literal
3906 binary.left = ilike_case_insensitive(binary.left)
3907 binary.right = percent.concat(ilike_case_insensitive(binary.right))
3908 return self.visit_ilike_op_binary(binary, operator, **kw)
3910 def visit_not_iendswith_op_binary(self, binary, operator, **kw):
3911 binary = binary._clone()
3912 percent = self._like_percent_literal
3913 binary.left = ilike_case_insensitive(binary.left)
3914 binary.right = percent.concat(ilike_case_insensitive(binary.right))
3915 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3917 def visit_like_op_binary(self, binary, operator, **kw):
3918 escape = binary.modifiers.get("escape", None)
3920 return "%s LIKE %s" % (
3921 binary.left._compiler_dispatch(self, **kw),
3922 binary.right._compiler_dispatch(self, **kw),
3923 ) + (
3924 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
3925 if escape is not None
3926 else ""
3927 )
3929 def visit_not_like_op_binary(self, binary, operator, **kw):
3930 escape = binary.modifiers.get("escape", None)
3931 return "%s NOT LIKE %s" % (
3932 binary.left._compiler_dispatch(self, **kw),
3933 binary.right._compiler_dispatch(self, **kw),
3934 ) + (
3935 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
3936 if escape is not None
3937 else ""
3938 )
3940 def visit_ilike_op_binary(self, binary, operator, **kw):
3941 if operator is operators.ilike_op:
3942 binary = binary._clone()
3943 binary.left = ilike_case_insensitive(binary.left)
3944 binary.right = ilike_case_insensitive(binary.right)
3945 # else we assume ilower() has been applied
3947 return self.visit_like_op_binary(binary, operator, **kw)
3949 def visit_not_ilike_op_binary(self, binary, operator, **kw):
3950 if operator is operators.not_ilike_op:
3951 binary = binary._clone()
3952 binary.left = ilike_case_insensitive(binary.left)
3953 binary.right = ilike_case_insensitive(binary.right)
3954 # else we assume ilower() has been applied
3956 return self.visit_not_like_op_binary(binary, operator, **kw)
3958 def visit_between_op_binary(self, binary, operator, **kw):
3959 symmetric = binary.modifiers.get("symmetric", False)
3960 return self._generate_generic_binary(
3961 binary, " BETWEEN SYMMETRIC " if symmetric else " BETWEEN ", **kw
3962 )
3964 def visit_not_between_op_binary(self, binary, operator, **kw):
3965 symmetric = binary.modifiers.get("symmetric", False)
3966 return self._generate_generic_binary(
3967 binary,
3968 " NOT BETWEEN SYMMETRIC " if symmetric else " NOT BETWEEN ",
3969 **kw,
3970 )
3972 def visit_regexp_match_op_binary(
3973 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3974 ) -> str:
3975 raise exc.CompileError(
3976 "%s dialect does not support regular expressions"
3977 % self.dialect.name
3978 )
3980 def visit_not_regexp_match_op_binary(
3981 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3982 ) -> str:
3983 raise exc.CompileError(
3984 "%s dialect does not support regular expressions"
3985 % self.dialect.name
3986 )
3988 def visit_regexp_replace_op_binary(
3989 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3990 ) -> str:
3991 raise exc.CompileError(
3992 "%s dialect does not support regular expression replacements"
3993 % self.dialect.name
3994 )
3996 def visit_dmltargetcopy(self, element, *, bindmarkers=None, **kw):
3997 if bindmarkers is None:
3998 raise exc.CompileError(
3999 "DML target objects may only be used with "
4000 "compiled INSERT or UPDATE statements"
4001 )
4003 bindmarkers[element.column.key] = element
4004 return f"__BINDMARKER_~~{element.column.key}~~"
4006 def visit_bindparam(
4007 self,
4008 bindparam,
4009 within_columns_clause=False,
4010 literal_binds=False,
4011 skip_bind_expression=False,
4012 literal_execute=False,
4013 render_postcompile=False,
4014 is_upsert_set=False,
4015 **kwargs,
4016 ):
4017 # Detect parametrized bindparams in upsert SET clause for issue #13130
4018 if (
4019 is_upsert_set
4020 and bindparam.value is None
4021 and bindparam.callable is None
4022 and self._insertmanyvalues is not None
4023 ):
4024 self._insertmanyvalues = self._insertmanyvalues._replace(
4025 has_upsert_bound_parameters=True
4026 )
4028 if not skip_bind_expression:
4029 impl = bindparam.type.dialect_impl(self.dialect)
4030 if impl._has_bind_expression:
4031 bind_expression = impl.bind_expression(bindparam)
4032 wrapped = self.process(
4033 bind_expression,
4034 skip_bind_expression=True,
4035 within_columns_clause=within_columns_clause,
4036 literal_binds=literal_binds and not bindparam.expanding,
4037 literal_execute=literal_execute,
4038 render_postcompile=render_postcompile,
4039 **kwargs,
4040 )
4041 if bindparam.expanding:
4042 # for postcompile w/ expanding, move the "wrapped" part
4043 # of this into the inside
4045 m = re.match(
4046 r"^(.*)\(__\[POSTCOMPILE_(\S+?)\]\)(.*)$", wrapped
4047 )
4048 assert m, "unexpected format for expanding parameter"
4049 wrapped = "(__[POSTCOMPILE_%s~~%s~~REPL~~%s~~])" % (
4050 m.group(2),
4051 m.group(1),
4052 m.group(3),
4053 )
4055 if literal_binds:
4056 ret = self.render_literal_bindparam(
4057 bindparam,
4058 within_columns_clause=True,
4059 bind_expression_template=wrapped,
4060 **kwargs,
4061 )
4062 return f"({ret})"
4064 return wrapped
4066 if not literal_binds:
4067 literal_execute = (
4068 literal_execute
4069 or bindparam.literal_execute
4070 or (within_columns_clause and self.ansi_bind_rules)
4071 )
4072 post_compile = literal_execute or bindparam.expanding
4073 else:
4074 post_compile = False
4076 if literal_binds:
4077 ret = self.render_literal_bindparam(
4078 bindparam, within_columns_clause=True, **kwargs
4079 )
4080 if bindparam.expanding:
4081 ret = f"({ret})"
4082 return ret
4084 name = self._truncate_bindparam(bindparam)
4086 if name in self.binds:
4087 existing = self.binds[name]
4088 if existing is not bindparam:
4089 if (
4090 (existing.unique or bindparam.unique)
4091 and not existing.proxy_set.intersection(
4092 bindparam.proxy_set
4093 )
4094 and not existing._cloned_set.intersection(
4095 bindparam._cloned_set
4096 )
4097 ):
4098 raise exc.CompileError(
4099 "Bind parameter '%s' conflicts with "
4100 "unique bind parameter of the same name" % name
4101 )
4102 elif existing.expanding != bindparam.expanding:
4103 raise exc.CompileError(
4104 "Can't reuse bound parameter name '%s' in both "
4105 "'expanding' (e.g. within an IN expression) and "
4106 "non-expanding contexts. If this parameter is to "
4107 "receive a list/array value, set 'expanding=True' on "
4108 "it for expressions that aren't IN, otherwise use "
4109 "a different parameter name." % (name,)
4110 )
4111 elif existing._is_crud or bindparam._is_crud:
4112 if existing._is_crud and bindparam._is_crud:
4113 # TODO: this condition is not well understood.
4114 # see tests in test/sql/test_update.py
4115 raise exc.CompileError(
4116 "Encountered unsupported case when compiling an "
4117 "INSERT or UPDATE statement. If this is a "
4118 "multi-table "
4119 "UPDATE statement, please provide string-named "
4120 "arguments to the "
4121 "values() method with distinct names; support for "
4122 "multi-table UPDATE statements that "
4123 "target multiple tables for UPDATE is very "
4124 "limited",
4125 )
4126 else:
4127 raise exc.CompileError(
4128 f"bindparam() name '{bindparam.key}' is reserved "
4129 "for automatic usage in the VALUES or SET "
4130 "clause of this "
4131 "insert/update statement. Please use a "
4132 "name other than column name when using "
4133 "bindparam() "
4134 "with insert() or update() (for example, "
4135 f"'b_{bindparam.key}')."
4136 )
4138 self.binds[bindparam.key] = self.binds[name] = bindparam
4140 # if we are given a cache key that we're going to match against,
4141 # relate the bindparam here to one that is most likely present
4142 # in the "extracted params" portion of the cache key. this is used
4143 # to set up a positional mapping that is used to determine the
4144 # correct parameters for a subsequent use of this compiled with
4145 # a different set of parameter values. here, we accommodate for
4146 # parameters that may have been cloned both before and after the cache
4147 # key was been generated.
4148 ckbm_tuple = self._cache_key_bind_match
4150 if ckbm_tuple:
4151 ckbm, cksm = ckbm_tuple
4152 for bp in bindparam._cloned_set:
4153 if bp.key in cksm:
4154 cb = cksm[bp.key]
4155 ckbm[cb].append(bindparam)
4157 if bindparam.isoutparam:
4158 self.has_out_parameters = True
4160 if post_compile:
4161 if render_postcompile:
4162 self._render_postcompile = True
4164 if literal_execute:
4165 self.literal_execute_params |= {bindparam}
4166 else:
4167 self.post_compile_params |= {bindparam}
4169 ret = self.bindparam_string(
4170 name,
4171 post_compile=post_compile,
4172 expanding=bindparam.expanding,
4173 bindparam_type=bindparam.type,
4174 **kwargs,
4175 )
4177 if bindparam.expanding:
4178 ret = f"({ret})"
4180 return ret
4182 def render_bind_cast(self, type_, dbapi_type, sqltext):
4183 raise NotImplementedError()
4185 def render_literal_bindparam(
4186 self,
4187 bindparam,
4188 render_literal_value=NO_ARG,
4189 bind_expression_template=None,
4190 **kw,
4191 ):
4192 if render_literal_value is not NO_ARG:
4193 value = render_literal_value
4194 elif self._collect_params and bindparam.key in self._collected_params:
4195 # values passed to ExecutableStatement.params() are not
4196 # present on the bindparam itself
4197 value = self._collected_params[bindparam.key]
4198 else:
4199 if bindparam.value is None and bindparam.callable is None:
4200 op = kw.get("_binary_op", None)
4201 if op and op not in (operators.is_, operators.is_not):
4202 util.warn_limited(
4203 "Bound parameter '%s' rendering literal NULL in a SQL "
4204 "expression; comparisons to NULL should not use "
4205 "operators outside of 'is' or 'is not'",
4206 (bindparam.key,),
4207 )
4208 return self.process(sqltypes.NULLTYPE, **kw)
4209 value = bindparam.effective_value
4211 if bindparam.expanding:
4212 leep = self._literal_execute_expanding_parameter_literal_binds
4213 to_update, replacement_expr = leep(
4214 bindparam,
4215 value,
4216 bind_expression_template=bind_expression_template,
4217 )
4218 return replacement_expr
4219 else:
4220 return self.render_literal_value(value, bindparam.type)
4222 def render_literal_value(
4223 self, value: Any, type_: sqltypes.TypeEngine[Any]
4224 ) -> str:
4225 """Render the value of a bind parameter as a quoted literal.
4227 This is used for statement sections that do not accept bind parameters
4228 on the target driver/database.
4230 This should be implemented by subclasses using the quoting services
4231 of the DBAPI.
4233 """
4235 if value is None and not type_.should_evaluate_none:
4236 # issue #10535 - handle NULL in the compiler without placing
4237 # this onto each type, except for "evaluate None" types
4238 # (e.g. JSON)
4239 return self.process(elements.Null._instance())
4241 processor = type_._cached_literal_processor(self.dialect)
4242 if processor:
4243 try:
4244 return processor(value)
4245 except Exception as e:
4246 raise exc.CompileError(
4247 f"Could not render literal value "
4248 f'"{sql_util._repr_single_value(value)}" '
4249 f"with datatype "
4250 f"{type_}; see parent stack trace for "
4251 "more detail."
4252 ) from e
4254 else:
4255 raise exc.CompileError(
4256 f"No literal value renderer is available for literal value "
4257 f'"{sql_util._repr_single_value(value)}" '
4258 f"with datatype {type_}"
4259 )
4261 def _truncate_bindparam(self, bindparam):
4262 if bindparam in self.bind_names:
4263 return self.bind_names[bindparam]
4265 bind_name = bindparam.key
4266 if isinstance(bind_name, elements._truncated_label):
4267 bind_name = self._truncated_identifier("bindparam", bind_name)
4269 # add to bind_names for translation
4270 self.bind_names[bindparam] = bind_name
4272 return bind_name
4274 def _truncated_identifier(
4275 self, ident_class: str, name: _truncated_label
4276 ) -> str:
4277 if (ident_class, name) in self.truncated_names:
4278 return self.truncated_names[(ident_class, name)]
4280 anonname = name.apply_map(self.anon_map)
4282 if len(anonname) > self.label_length - 6:
4283 counter = self._truncated_counters.get(ident_class, 1)
4284 truncname = (
4285 anonname[0 : max(self.label_length - 6, 0)]
4286 + "_"
4287 + hex(counter)[2:]
4288 )
4289 self._truncated_counters[ident_class] = counter + 1
4290 else:
4291 truncname = anonname
4292 self.truncated_names[(ident_class, name)] = truncname
4293 return truncname
4295 def _anonymize(self, name: str) -> str:
4296 return name % self.anon_map
4298 def _escaped_bind_name(self, escaped_from: str, name: str) -> str:
4299 """Record the escaped form of a bind name, uniquifying it against
4300 the names already in use.
4302 Distinct bind names can escape to the same string, e.g. parameters
4303 named ``"a.b"`` and ``"a_b"`` both escape to ``a_b``. As the escaped
4304 name is what keys the parameter dictionary handed to the driver, the
4305 second parameter would otherwise overwrite the first and its value be
4306 silently discarded. A counter is appended so that the two remain
4307 distinct. The ``.key`` of each :class:`.BindParameter` is untouched,
4308 so parameter dictionaries passed by the caller are unaffected.
4310 See :ticket:`13534`.
4312 """
4313 try:
4314 return self.escaped_bind_names[escaped_from]
4315 except KeyError:
4316 pass
4318 if self.escaped_bind_names:
4319 used = self._escaped_bind_names_used
4320 else:
4321 self._escaped_bind_names_used = used = set()
4323 if name in used or (name != escaped_from and name in self.binds):
4324 base = name
4325 counter = 0
4326 while name in used or name in self.binds:
4327 counter += 1
4328 name = "%s__%d" % (base, counter)
4330 used.add(name)
4331 self.escaped_bind_names = self.escaped_bind_names.union(
4332 {escaped_from: name}
4333 )
4334 return name
4336 def bindparam_string(
4337 self,
4338 name: str,
4339 post_compile: bool = False,
4340 expanding: bool = False,
4341 escaped_from: Optional[str] = None,
4342 bindparam_type: Optional[TypeEngine[Any]] = None,
4343 accumulate_bind_names: Optional[Set[str]] = None,
4344 visited_bindparam: Optional[List[str]] = None,
4345 **kw: Any,
4346 ) -> str:
4347 # TODO: accumulate_bind_names is passed by crud.py to gather
4348 # names on a per-value basis, visited_bindparam is passed by
4349 # visit_insert() to collect all parameters in the statement.
4350 # see if this gathering can be simplified somehow
4351 if accumulate_bind_names is not None:
4352 accumulate_bind_names.add(name)
4353 if visited_bindparam is not None:
4354 visited_bindparam.append(name)
4356 if not escaped_from:
4357 if self._bind_translate_re.search(name):
4358 # not quite the translate use case as we want to
4359 # also get a quick boolean if we even found
4360 # unusual characters in the name
4361 new_name = self._bind_translate_re.sub(
4362 lambda m: self._bind_translate_chars[m.group(0)],
4363 name,
4364 )
4365 escaped_from = name
4366 name = new_name
4367 elif (
4368 self.escaped_bind_names
4369 and name in self._escaped_bind_names_used
4370 ):
4371 # this name needs no escaping of its own, but another
4372 # parameter has already escaped to it; #13534
4373 escaped_from = name
4375 if escaped_from:
4376 name = self._escaped_bind_name(escaped_from, name)
4377 if post_compile:
4378 ret = "__[POSTCOMPILE_%s]" % name
4379 if expanding:
4380 # for expanding, bound parameters or literal values will be
4381 # rendered per item
4382 return ret
4384 # otherwise, for non-expanding "literal execute", apply
4385 # bind casts as determined by the datatype
4386 if bindparam_type is not None:
4387 type_impl = bindparam_type._unwrapped_dialect_impl(
4388 self.dialect
4389 )
4390 if type_impl.render_literal_cast:
4391 ret = self.render_bind_cast(bindparam_type, type_impl, ret)
4392 return ret
4393 elif self.state is CompilerState.COMPILING:
4394 ret = self.compilation_bindtemplate % {"name": name}
4395 else:
4396 ret = self.bindtemplate % {"name": name}
4398 if (
4399 bindparam_type is not None
4400 and self.dialect._bind_typing_render_casts
4401 ):
4402 type_impl = bindparam_type._unwrapped_dialect_impl(self.dialect)
4403 if type_impl.render_bind_cast:
4404 ret = self.render_bind_cast(bindparam_type, type_impl, ret)
4406 return ret
4408 def _dispatch_independent_ctes(self, stmt, kw):
4409 local_kw = kw.copy()
4410 local_kw.pop("cte_opts", None)
4411 for cte, opt in zip(
4412 stmt._independent_ctes, stmt._independent_ctes_opts
4413 ):
4414 cte._compiler_dispatch(self, cte_opts=opt, **local_kw)
4416 def visit_cte(
4417 self,
4418 cte: CTE,
4419 asfrom: bool = False,
4420 ashint: bool = False,
4421 fromhints: Optional[_FromHintsType] = None,
4422 visiting_cte: Optional[CTE] = None,
4423 from_linter: Optional[FromLinter] = None,
4424 cte_opts: selectable._CTEOpts = selectable._CTEOpts(False),
4425 **kwargs: Any,
4426 ) -> Optional[str]:
4427 self_ctes = self._init_cte_state()
4428 assert self_ctes is self.ctes
4430 kwargs["visiting_cte"] = cte
4432 cte_name = cte.name
4434 if isinstance(cte_name, elements._truncated_label):
4435 cte_name = self._truncated_identifier("alias", cte_name)
4437 is_new_cte = True
4438 embedded_in_current_named_cte = False
4440 _reference_cte = cte._get_reference_cte()
4442 nesting = cte.nesting or cte_opts.nesting
4444 # check for CTE already encountered
4445 if _reference_cte in self.level_name_by_cte:
4446 cte_level, _, existing_cte_opts = self.level_name_by_cte[
4447 _reference_cte
4448 ]
4449 assert _ == cte_name
4451 cte_level_name = (cte_level, cte_name)
4452 existing_cte = self.ctes_by_level_name[cte_level_name]
4454 # check if we are receiving it here with a specific
4455 # "nest_here" location; if so, move it to this location
4457 if cte_opts.nesting:
4458 if existing_cte_opts.nesting:
4459 raise exc.CompileError(
4460 "CTE is stated as 'nest_here' in "
4461 "more than one location"
4462 )
4464 old_level_name = (cte_level, cte_name)
4465 cte_level = len(self.stack) if nesting else 1
4466 cte_level_name = new_level_name = (cte_level, cte_name)
4468 del self.ctes_by_level_name[old_level_name]
4469 self.ctes_by_level_name[new_level_name] = existing_cte
4470 self.level_name_by_cte[_reference_cte] = new_level_name + (
4471 cte_opts,
4472 )
4474 else:
4475 cte_level = len(self.stack) if nesting else 1
4476 cte_level_name = (cte_level, cte_name)
4478 if cte_level_name in self.ctes_by_level_name:
4479 existing_cte = self.ctes_by_level_name[cte_level_name]
4480 else:
4481 existing_cte = None
4483 if existing_cte is not None:
4484 embedded_in_current_named_cte = visiting_cte is existing_cte
4486 # we've generated a same-named CTE that we are enclosed in,
4487 # or this is the same CTE. just return the name.
4488 if cte is existing_cte._restates or cte is existing_cte:
4489 is_new_cte = False
4490 elif existing_cte is cte._restates:
4491 # we've generated a same-named CTE that is
4492 # enclosed in us - we take precedence, so
4493 # discard the text for the "inner".
4494 del self_ctes[existing_cte]
4496 existing_cte_reference_cte = existing_cte._get_reference_cte()
4498 assert existing_cte_reference_cte is _reference_cte
4499 assert existing_cte_reference_cte is existing_cte
4501 del self.level_name_by_cte[existing_cte_reference_cte]
4502 else:
4503 if (
4504 # if the two CTEs have the same hash, which we expect
4505 # here means that one/both is an annotated of the other
4506 (hash(cte) == hash(existing_cte))
4507 # or...
4508 or (
4509 (
4510 # if they are clones, i.e. they came from the ORM
4511 # or some other visit method
4512 cte._is_clone_of is not None
4513 or existing_cte._is_clone_of is not None
4514 )
4515 # and are deep-copy identical
4516 and cte.compare(existing_cte)
4517 )
4518 ):
4519 # then consider these two CTEs the same
4520 is_new_cte = False
4521 else:
4522 # otherwise these are two CTEs that either will render
4523 # differently, or were indicated separately by the user,
4524 # with the same name
4525 raise exc.CompileError(
4526 "Multiple, unrelated CTEs found with "
4527 "the same name: %r" % cte_name
4528 )
4530 if not asfrom and not is_new_cte:
4531 return None
4533 if cte._cte_alias is not None:
4534 pre_alias_cte = cte._cte_alias
4535 cte_pre_alias_name = cte._cte_alias.name
4536 if isinstance(cte_pre_alias_name, elements._truncated_label):
4537 cte_pre_alias_name = self._truncated_identifier(
4538 "alias", cte_pre_alias_name
4539 )
4540 else:
4541 pre_alias_cte = cte
4542 cte_pre_alias_name = None
4544 if is_new_cte:
4545 self.ctes_by_level_name[cte_level_name] = cte
4546 self.level_name_by_cte[_reference_cte] = cte_level_name + (
4547 cte_opts,
4548 )
4550 if pre_alias_cte not in self.ctes:
4551 self.visit_cte(pre_alias_cte, **kwargs)
4553 if not cte_pre_alias_name and cte not in self_ctes:
4554 if cte.recursive:
4555 self.ctes_recursive = True
4556 text = self.preparer.format_alias(cte, cte_name)
4557 if cte.recursive or cte.element.name_cte_columns:
4558 col_source = cte.element
4560 # TODO: can we get at the .columns_plus_names collection
4561 # that is already (or will be?) generated for the SELECT
4562 # rather than calling twice?
4563 recur_cols = [
4564 # TODO: proxy_name is not technically safe,
4565 # see test_cte->
4566 # test_with_recursive_no_name_currently_buggy. not
4567 # clear what should be done with such a case
4568 fallback_label_name or proxy_name
4569 for (
4570 _,
4571 proxy_name,
4572 fallback_label_name,
4573 c,
4574 repeated,
4575 ) in (col_source._generate_columns_plus_names(True))
4576 if not repeated
4577 ]
4579 text += "(%s)" % (
4580 ", ".join(
4581 self.preparer.format_label_name(
4582 ident, anon_map=self.anon_map
4583 )
4584 for ident in recur_cols
4585 )
4586 )
4588 assert kwargs.get("subquery", False) is False
4590 if not self.stack:
4591 # toplevel, this is a stringify of the
4592 # cte directly. just compile the inner
4593 # the way alias() does.
4594 return cte.element._compiler_dispatch(
4595 self, asfrom=asfrom, **kwargs
4596 )
4597 else:
4598 prefixes = self._generate_prefixes(
4599 cte, cte._prefixes, **kwargs
4600 )
4601 inner = cte.element._compiler_dispatch(
4602 self, asfrom=True, **kwargs
4603 )
4605 text += " AS %s\n(%s)" % (prefixes, inner)
4607 if cte._suffixes:
4608 text += " " + self._generate_prefixes(
4609 cte, cte._suffixes, **kwargs
4610 )
4612 self_ctes[cte] = text
4614 if asfrom:
4615 if from_linter:
4616 from_linter.froms[cte._de_clone()] = cte_name
4618 if not is_new_cte and embedded_in_current_named_cte:
4619 return self.preparer.format_alias(cte, cte_name)
4621 if cte_pre_alias_name:
4622 text = self.preparer.format_alias(cte, cte_pre_alias_name)
4623 if self.preparer._requires_quotes(cte_name):
4624 cte_name = self.preparer.quote(cte_name)
4625 text += self.get_render_as_alias_suffix(cte_name)
4626 return text # type: ignore[no-any-return]
4627 else:
4628 return self.preparer.format_alias(cte, cte_name)
4630 return None
4632 def visit_table_valued_alias(self, element, **kw):
4633 if element.joins_implicitly:
4634 kw["from_linter"] = None
4635 if element._is_lateral:
4636 return self.visit_lateral(element, **kw)
4637 else:
4638 return self.visit_alias(element, **kw)
4640 def visit_table_valued_column(self, element, **kw):
4641 return self.visit_column(element, **kw)
4643 def visit_alias(
4644 self,
4645 alias,
4646 asfrom=False,
4647 ashint=False,
4648 iscrud=False,
4649 fromhints=None,
4650 subquery=False,
4651 lateral=False,
4652 enclosing_alias=None,
4653 from_linter=None,
4654 within_tstring=False,
4655 **kwargs,
4656 ):
4657 if lateral:
4658 if "enclosing_lateral" not in kwargs:
4659 # if lateral is set and enclosing_lateral is not
4660 # present, we assume we are being called directly
4661 # from visit_lateral() and we need to set enclosing_lateral.
4662 assert alias._is_lateral
4663 kwargs["enclosing_lateral"] = alias
4665 # for lateral objects, we track a second from_linter that is...
4666 # lateral! to the level above us.
4667 if (
4668 from_linter
4669 and "lateral_from_linter" not in kwargs
4670 and "enclosing_lateral" in kwargs
4671 ):
4672 kwargs["lateral_from_linter"] = from_linter
4674 if enclosing_alias is not None and enclosing_alias.element is alias:
4675 inner = alias.element._compiler_dispatch(
4676 self,
4677 asfrom=asfrom,
4678 ashint=ashint,
4679 iscrud=iscrud,
4680 fromhints=fromhints,
4681 lateral=lateral,
4682 enclosing_alias=alias,
4683 **kwargs,
4684 )
4685 if subquery and (asfrom or lateral):
4686 inner = "(%s)" % (inner,)
4687 return inner
4688 else:
4689 kwargs["enclosing_alias"] = alias
4691 if asfrom or ashint or within_tstring:
4692 if isinstance(alias.name, elements._truncated_label):
4693 alias_name = self._truncated_identifier("alias", alias.name)
4694 else:
4695 alias_name = alias.name
4697 if ashint:
4698 return self.preparer.format_alias(alias, alias_name)
4699 elif asfrom or within_tstring:
4700 if from_linter:
4701 from_linter.froms[alias._de_clone()] = alias_name
4703 inner = alias.element._compiler_dispatch(
4704 self, asfrom=True, lateral=lateral, **kwargs
4705 )
4706 if subquery:
4707 inner = "(%s)" % (inner,)
4709 ret = inner + self.get_render_as_alias_suffix(
4710 self.preparer.format_alias(alias, alias_name)
4711 )
4713 if alias._supports_derived_columns and alias._render_derived:
4714 ret += "(%s)" % (
4715 ", ".join(
4716 "%s%s"
4717 % (
4718 self.preparer.quote(col.name),
4719 (
4720 " %s"
4721 % self.dialect.type_compiler_instance.process(
4722 col.type, **kwargs
4723 )
4724 if alias._render_derived_w_types
4725 else ""
4726 ),
4727 )
4728 for col in alias.c
4729 )
4730 )
4732 if fromhints and alias in fromhints:
4733 ret = self.format_from_hint_text(
4734 ret, alias, fromhints[alias], iscrud
4735 )
4737 return ret
4738 else:
4739 # note we cancel the "subquery" flag here as well
4740 return alias.element._compiler_dispatch(
4741 self, lateral=lateral, **kwargs
4742 )
4744 def visit_subquery(self, subquery, **kw):
4745 kw["subquery"] = True
4746 return self.visit_alias(subquery, **kw)
4748 def visit_lateral(self, lateral_, **kw):
4749 kw["lateral"] = True
4750 return "LATERAL %s" % self.visit_alias(lateral_, **kw)
4752 def visit_tablesample(self, tablesample, asfrom=False, **kw):
4753 text = "%s TABLESAMPLE %s" % (
4754 self.visit_alias(tablesample, asfrom=True, **kw),
4755 tablesample._get_method()._compiler_dispatch(self, **kw),
4756 )
4758 if tablesample.seed is not None:
4759 text += " REPEATABLE (%s)" % (
4760 tablesample.seed._compiler_dispatch(self, **kw)
4761 )
4763 return text
4765 def _render_values(self, element, **kw):
4766 kw.setdefault("literal_binds", element.literal_binds)
4767 tuples = ", ".join(
4768 self.process(
4769 elements.Tuple(
4770 types=element._column_types, *elem
4771 ).self_group(),
4772 **kw,
4773 )
4774 for chunk in element._data
4775 for elem in chunk
4776 )
4777 return f"VALUES {tuples}"
4779 def visit_values(
4780 self, element, asfrom=False, from_linter=None, visiting_cte=None, **kw
4781 ):
4783 if element._independent_ctes:
4784 self._dispatch_independent_ctes(element, kw)
4786 v = self._render_values(element, **kw)
4788 if element._unnamed:
4789 name = None
4790 elif isinstance(element.name, elements._truncated_label):
4791 name = self._truncated_identifier("values", element.name)
4792 else:
4793 name = element.name
4795 if element._is_lateral:
4796 lateral = "LATERAL "
4797 else:
4798 lateral = ""
4800 if asfrom:
4801 if from_linter:
4802 from_linter.froms[element._de_clone()] = (
4803 name if name is not None else "(unnamed VALUES element)"
4804 )
4806 if visiting_cte is not None and visiting_cte.element is element:
4807 if element._is_lateral:
4808 raise exc.CompileError(
4809 "Can't use a LATERAL VALUES expression inside of a CTE"
4810 )
4811 elif name:
4812 kw["include_table"] = False
4813 v = "%s(%s)%s (%s)" % (
4814 lateral,
4815 v,
4816 self.get_render_as_alias_suffix(self.preparer.quote(name)),
4817 (
4818 ", ".join(
4819 c._compiler_dispatch(self, **kw)
4820 for c in element.columns
4821 )
4822 ),
4823 )
4824 else:
4825 v = "%s(%s)" % (lateral, v)
4826 return v
4828 def visit_scalar_values(self, element, **kw):
4829 return f"({self._render_values(element, **kw)})"
4831 def get_render_as_alias_suffix(self, alias_name_text):
4832 return " AS " + alias_name_text
4834 def _add_to_result_map(
4835 self,
4836 keyname: str,
4837 name: str,
4838 objects: Tuple[Any, ...],
4839 type_: TypeEngine[Any],
4840 ) -> None:
4842 # note objects must be non-empty for cursor.py to handle the
4843 # collection properly
4844 assert objects
4846 if keyname is None or keyname == "*":
4847 self._ordered_columns = False
4848 self._ad_hoc_textual = True
4849 if type_._is_tuple_type:
4850 raise exc.CompileError(
4851 "Most backends don't support SELECTing "
4852 "from a tuple() object. If this is an ORM query, "
4853 "consider using the Bundle object."
4854 )
4855 self._result_columns.append(
4856 ResultColumnsEntry(keyname, name, objects, type_)
4857 )
4859 def _label_returning_column(
4860 self, stmt, column, populate_result_map, column_clause_args=None, **kw
4861 ):
4862 """Render a column with necessary labels inside of a RETURNING clause.
4864 This method is provided for individual dialects in place of calling
4865 the _label_select_column method directly, so that the two use cases
4866 of RETURNING vs. SELECT can be disambiguated going forward.
4868 .. versionadded:: 1.4.21
4870 """
4871 return self._label_select_column(
4872 None,
4873 column,
4874 populate_result_map,
4875 False,
4876 {} if column_clause_args is None else column_clause_args,
4877 **kw,
4878 )
4880 def _label_select_column(
4881 self,
4882 select,
4883 column,
4884 populate_result_map,
4885 asfrom,
4886 column_clause_args,
4887 name=None,
4888 proxy_name=None,
4889 fallback_label_name=None,
4890 within_columns_clause=True,
4891 column_is_repeated=False,
4892 need_column_expressions=False,
4893 include_table=True,
4894 ):
4895 """produce labeled columns present in a select()."""
4896 impl = column.type.dialect_impl(self.dialect)
4898 if impl._has_column_expression and (
4899 need_column_expressions or populate_result_map
4900 ):
4901 col_expr = impl.column_expression(column)
4902 else:
4903 col_expr = column
4905 if populate_result_map:
4906 # pass an "add_to_result_map" callable into the compilation
4907 # of embedded columns. this collects information about the
4908 # column as it will be fetched in the result and is coordinated
4909 # with cursor.description when the query is executed.
4910 add_to_result_map = self._add_to_result_map
4912 # if the SELECT statement told us this column is a repeat,
4913 # wrap the callable with one that prevents the addition of the
4914 # targets
4915 if column_is_repeated:
4916 _add_to_result_map = add_to_result_map
4918 def add_to_result_map(keyname, name, objects, type_):
4919 _add_to_result_map(keyname, name, (keyname,), type_)
4921 # if we redefined col_expr for type expressions, wrap the
4922 # callable with one that adds the original column to the targets
4923 elif col_expr is not column:
4924 _add_to_result_map = add_to_result_map
4926 def add_to_result_map(keyname, name, objects, type_):
4927 _add_to_result_map(
4928 keyname, name, (column,) + objects, type_
4929 )
4931 else:
4932 add_to_result_map = None
4934 # this method is used by some of the dialects for RETURNING,
4935 # which has different inputs. _label_returning_column was added
4936 # as the better target for this now however for 1.4 we will keep
4937 # _label_select_column directly compatible with this use case.
4938 # these assertions right now set up the current expected inputs
4939 assert within_columns_clause, (
4940 "_label_select_column is only relevant within "
4941 "the columns clause of a SELECT or RETURNING"
4942 )
4943 result_expr: elements.Label[Any] | _CompileLabel
4945 if isinstance(column, elements.Label):
4946 if col_expr is not column:
4947 result_expr = _CompileLabel(
4948 col_expr, column.name, alt_names=(column.element,)
4949 )
4950 else:
4951 result_expr = col_expr
4953 elif name:
4954 # here, _columns_plus_names has determined there's an explicit
4955 # label name we need to use. this is the default for
4956 # tablenames_plus_columnnames as well as when columns are being
4957 # deduplicated on name
4959 assert (
4960 proxy_name is not None
4961 ), "proxy_name is required if 'name' is passed"
4963 result_expr = _CompileLabel(
4964 col_expr,
4965 name,
4966 alt_names=(
4967 proxy_name,
4968 # this is a hack to allow legacy result column lookups
4969 # to work as they did before; this goes away in 2.0.
4970 # TODO: this only seems to be tested indirectly
4971 # via test/orm/test_deprecations.py. should be a
4972 # resultset test for this
4973 column._tq_label,
4974 ),
4975 )
4976 else:
4977 # determine here whether this column should be rendered in
4978 # a labelled context or not, as we were given no required label
4979 # name from the caller. Here we apply heuristics based on the kind
4980 # of SQL expression involved.
4982 if col_expr is not column:
4983 # type-specific expression wrapping the given column,
4984 # so we render a label
4985 render_with_label = True
4986 elif isinstance(column, elements.ColumnClause):
4987 # table-bound column, we render its name as a label if we are
4988 # inside of a subquery only
4989 render_with_label = (
4990 asfrom
4991 and not column.is_literal
4992 and column.table is not None
4993 )
4994 elif isinstance(column, elements.TextClause):
4995 render_with_label = False
4996 elif isinstance(column, elements.UnaryExpression):
4997 # unary expression. notes added as of #12681
4998 #
4999 # By convention, the visit_unary() method
5000 # itself does not add an entry to the result map, and relies
5001 # upon either the inner expression creating a result map
5002 # entry, or if not, by creating a label here that produces
5003 # the result map entry. Where that happens is based on whether
5004 # or not the element immediately inside the unary is a
5005 # NamedColumn subclass or not.
5006 #
5007 # Now, this also impacts how the SELECT is written; if
5008 # we decide to generate a label here, we get the usual
5009 # "~(x+y) AS anon_1" thing in the columns clause. If we
5010 # don't, we don't get an AS at all, we get like
5011 # "~table.column".
5012 #
5013 # But here is the important thing as of modernish (like 1.4)
5014 # versions of SQLAlchemy - **whether or not the AS <label>
5015 # is present in the statement is not actually important**.
5016 # We target result columns **positionally** for a fully
5017 # compiled ``Select()`` object; before 1.4 we needed those
5018 # labels to match in cursor.description etc etc but now it
5019 # really doesn't matter.
5020 # So really, we could set render_with_label True in all cases.
5021 # Or we could just have visit_unary() populate the result map
5022 # in all cases.
5023 #
5024 # What we're doing here is strictly trying to not rock the
5025 # boat too much with when we do/don't render "AS label";
5026 # labels being present helps in the edge cases that we
5027 # "fall back" to named cursor.description matching, labels
5028 # not being present for columns keeps us from having awkward
5029 # phrases like "SELECT DISTINCT table.x AS x".
5030 render_with_label = (
5031 (
5032 # exception case to detect if we render "not boolean"
5033 # as "not <col>" for native boolean or "<col> = 1"
5034 # for non-native boolean. this is controlled by
5035 # visit_is_<true|false>_unary_operator
5036 column.operator
5037 in (operators.is_false, operators.is_true)
5038 and not self.dialect.supports_native_boolean
5039 )
5040 or column._wraps_unnamed_column()
5041 or asfrom
5042 )
5043 elif (
5044 # general class of expressions that don't have a SQL-column
5045 # addressable name. includes scalar selects, bind parameters,
5046 # SQL functions, others
5047 not isinstance(column, elements.NamedColumn)
5048 # deeper check that indicates there's no natural "name" to
5049 # this element, which accommodates for custom SQL constructs
5050 # that might have a ".name" attribute (but aren't SQL
5051 # functions) but are not implementing this more recently added
5052 # base class. in theory the "NamedColumn" check should be
5053 # enough, however here we seek to maintain legacy behaviors
5054 # as well.
5055 and column._non_anon_label is None
5056 ):
5057 render_with_label = True
5058 else:
5059 render_with_label = False
5061 if render_with_label:
5062 if not fallback_label_name:
5063 # used by the RETURNING case right now. we generate it
5064 # here as 3rd party dialects may be referring to
5065 # _label_select_column method directly instead of the
5066 # just-added _label_returning_column method
5067 assert not column_is_repeated
5068 fallback_label_name = column._anon_name_label
5070 fallback_label_name = (
5071 elements._truncated_label(fallback_label_name)
5072 if not isinstance(
5073 fallback_label_name, elements._truncated_label
5074 )
5075 else fallback_label_name
5076 )
5078 result_expr = _CompileLabel(
5079 col_expr, fallback_label_name, alt_names=(proxy_name,)
5080 )
5081 else:
5082 result_expr = col_expr
5084 column_clause_args.update(
5085 within_columns_clause=within_columns_clause,
5086 add_to_result_map=add_to_result_map,
5087 include_table=include_table,
5088 within_tstring=False,
5089 )
5090 return result_expr._compiler_dispatch(self, **column_clause_args)
5092 def format_from_hint_text(self, sqltext, table, hint, iscrud):
5093 hinttext = self.get_from_hint_text(table, hint)
5094 if hinttext:
5095 sqltext += " " + hinttext
5096 return sqltext
5098 def get_select_hint_text(self, byfroms):
5099 return None
5101 def get_from_hint_text(
5102 self, table: FromClause, text: Optional[str]
5103 ) -> Optional[str]:
5104 return None
5106 def get_crud_hint_text(self, table, text):
5107 return None
5109 def get_statement_hint_text(self, hint_texts):
5110 return " ".join(hint_texts)
5112 _default_stack_entry: _CompilerStackEntry
5114 if not typing.TYPE_CHECKING:
5115 _default_stack_entry = util.immutabledict(
5116 [("correlate_froms", frozenset()), ("asfrom_froms", frozenset())]
5117 )
5119 def _display_froms_for_select(
5120 self, select_stmt, asfrom, lateral=False, **kw
5121 ):
5122 # utility method to help external dialects
5123 # get the correct from list for a select.
5124 # specifically the oracle dialect needs this feature
5125 # right now.
5126 toplevel = not self.stack
5127 entry = self._default_stack_entry if toplevel else self.stack[-1]
5129 compile_state = select_stmt._compile_state_factory(
5130 select_stmt, self, asfrom=asfrom, lateral=lateral
5131 )
5133 explicit, implicit = self._enclosing_correlate_froms(
5134 entry, asfrom, lateral
5135 )
5136 return compile_state._get_display_froms(
5137 explicit_correlate_froms=explicit,
5138 implicit_correlate_froms=implicit,
5139 )
5141 def _enclosing_correlate_froms(
5142 self, entry: _CompilerStackEntry, asfrom: bool, lateral: bool
5143 ) -> Tuple[AbstractSet[FromClause], AbstractSet[FromClause]]:
5144 """Given the stack entry of an enclosing statement, return the
5145 explicit and implicit correlate FROMs for a SELECT that's embedded
5146 within it, as passed to :meth:`.SelectState._get_display_froms`.
5148 explicit_correlate_froms: FROMs of enclosing SELECTs that this SELECT
5149 may correlate to when explicit correlation is used, i.e.
5150 :meth:`.Select.correlate` or :meth:`.Select.correlate_except`.
5151 Includes the FROMs of all enclosing SELECTs.
5153 implicit_correlate_froms: FROMs of enclosing SELECTs that this SELECT
5154 will correlate to automatically, assuming auto-correlation is in
5155 effect. Includes only the FROMs of the immediately enclosing SELECT,
5156 and is empty if this SELECT is a subquery in a FROM clause that's not
5157 LATERAL. Is always a subset of explicit_correlate_froms.
5159 """
5160 correlate_froms = entry["correlate_froms"]
5161 asfrom_froms = entry["asfrom_froms"]
5163 if asfrom and not lateral:
5164 return correlate_froms.difference(asfrom_froms), util.EMPTY_SET
5165 else:
5166 return correlate_froms, asfrom_froms
5168 translate_select_structure: Any = None
5169 """if not ``None``, should be a callable which accepts ``(select_stmt,
5170 **kw)`` and returns a select object. this is used for structural changes
5171 mostly to accommodate for LIMIT/OFFSET schemes
5173 """
5175 def visit_select(
5176 self,
5177 select_stmt,
5178 asfrom=False,
5179 insert_into=False,
5180 fromhints=None,
5181 compound_index=None,
5182 select_wraps_for=None,
5183 lateral=False,
5184 from_linter=None,
5185 **kwargs,
5186 ):
5187 assert select_wraps_for is None, (
5188 "SQLAlchemy 1.4 requires use of "
5189 "the translate_select_structure hook for structural "
5190 "translations of SELECT objects"
5191 )
5192 if self._collect_params:
5193 self._add_to_params(select_stmt)
5195 # initial setup of SELECT. the compile_state_factory may now
5196 # be creating a totally different SELECT from the one that was
5197 # passed in. for ORM use this will convert from an ORM-state
5198 # SELECT to a regular "Core" SELECT. other composed operations
5199 # such as computation of joins will be performed.
5201 kwargs["within_columns_clause"] = False
5203 compile_state = select_stmt._compile_state_factory(
5204 select_stmt, self, asfrom=asfrom, lateral=lateral, **kwargs
5205 )
5206 kwargs["ambiguous_table_name_map"] = (
5207 compile_state._ambiguous_table_name_map
5208 )
5210 select_stmt = compile_state.statement
5212 toplevel = not self.stack
5214 if toplevel and not self.compile_state:
5215 self.compile_state = compile_state
5217 is_embedded_select = compound_index is not None or insert_into
5219 # translate step for Oracle, SQL Server which often need to
5220 # restructure the SELECT to allow for LIMIT/OFFSET and possibly
5221 # other conditions
5222 if self.translate_select_structure:
5223 new_select_stmt = self.translate_select_structure(
5224 select_stmt, asfrom=asfrom, **kwargs
5225 )
5227 # if SELECT was restructured, maintain a link to the originals
5228 # and assemble a new compile state
5229 if new_select_stmt is not select_stmt:
5230 compile_state_wraps_for = compile_state
5231 select_wraps_for = select_stmt
5232 select_stmt = new_select_stmt
5234 compile_state = select_stmt._compile_state_factory(
5235 select_stmt,
5236 self,
5237 asfrom=asfrom,
5238 lateral=lateral,
5239 **kwargs,
5240 )
5241 select_stmt = compile_state.statement
5243 entry = self._default_stack_entry if toplevel else self.stack[-1]
5245 populate_result_map = need_column_expressions = (
5246 toplevel
5247 or entry.get("need_result_map_for_compound", False)
5248 or entry.get("need_result_map_for_nested", False)
5249 )
5251 # indicates there is a CompoundSelect in play and we are not the
5252 # first select
5253 if compound_index:
5254 populate_result_map = False
5256 # this was first proposed as part of #3372; however, it is not
5257 # reached in current tests and could possibly be an assertion
5258 # instead.
5259 if not populate_result_map and "add_to_result_map" in kwargs:
5260 del kwargs["add_to_result_map"]
5262 froms = self._setup_select_stack(
5263 select_stmt, compile_state, entry, asfrom, lateral, compound_index
5264 )
5266 column_clause_args = kwargs.copy()
5267 column_clause_args.update(
5268 {"within_label_clause": False, "within_columns_clause": False}
5269 )
5271 text = "SELECT " # we're off to a good start !
5273 if select_stmt._post_select_clause is not None:
5274 psc = self.process(select_stmt._post_select_clause, **kwargs)
5275 if psc is not None:
5276 text += psc + " "
5278 if select_stmt._hints:
5279 hint_text, byfrom = self._setup_select_hints(select_stmt)
5280 if hint_text:
5281 text += hint_text + " "
5282 else:
5283 byfrom = None
5285 if select_stmt._independent_ctes:
5286 self._dispatch_independent_ctes(select_stmt, kwargs)
5288 if select_stmt._prefixes:
5289 text += self._generate_prefixes(
5290 select_stmt, select_stmt._prefixes, **kwargs
5291 )
5293 text += self.get_select_precolumns(select_stmt, **kwargs)
5295 if select_stmt._pre_columns_clause is not None:
5296 pcc = self.process(select_stmt._pre_columns_clause, **kwargs)
5297 if pcc is not None:
5298 text += pcc + " "
5300 # the actual list of columns to print in the SELECT column list.
5301 inner_columns = [
5302 c
5303 for c in [
5304 self._label_select_column(
5305 select_stmt,
5306 column,
5307 populate_result_map,
5308 asfrom,
5309 column_clause_args,
5310 name=name,
5311 proxy_name=proxy_name,
5312 fallback_label_name=fallback_label_name,
5313 column_is_repeated=repeated,
5314 need_column_expressions=need_column_expressions,
5315 )
5316 for (
5317 name,
5318 proxy_name,
5319 fallback_label_name,
5320 column,
5321 repeated,
5322 ) in compile_state.columns_plus_names
5323 ]
5324 if c is not None
5325 ]
5327 if populate_result_map and select_wraps_for is not None:
5328 # if this select was generated from translate_select,
5329 # rewrite the targeted columns in the result map
5331 translate = dict(
5332 zip(
5333 [
5334 name
5335 for (
5336 key,
5337 proxy_name,
5338 fallback_label_name,
5339 name,
5340 repeated,
5341 ) in compile_state.columns_plus_names
5342 ],
5343 [
5344 name
5345 for (
5346 key,
5347 proxy_name,
5348 fallback_label_name,
5349 name,
5350 repeated,
5351 ) in compile_state_wraps_for.columns_plus_names
5352 ],
5353 )
5354 )
5356 self._result_columns = [
5357 ResultColumnsEntry(
5358 key, name, tuple(translate.get(o, o) for o in obj), type_
5359 )
5360 for key, name, obj, type_ in self._result_columns
5361 ]
5363 text = self._compose_select_body(
5364 text,
5365 select_stmt,
5366 compile_state,
5367 inner_columns,
5368 froms,
5369 byfrom,
5370 toplevel,
5371 kwargs,
5372 )
5374 if select_stmt._post_body_clause is not None:
5375 pbc = self.process(select_stmt._post_body_clause, **kwargs)
5376 if pbc:
5377 text += " " + pbc
5379 if select_stmt._statement_hints:
5380 per_dialect = [
5381 ht
5382 for (dialect_name, ht) in select_stmt._statement_hints
5383 if dialect_name in ("*", self.dialect.name)
5384 ]
5385 if per_dialect:
5386 text += " " + self.get_statement_hint_text(per_dialect)
5388 # In compound query, CTEs are shared at the compound level
5389 if self.ctes and (not is_embedded_select or toplevel):
5390 nesting_level = len(self.stack) if not toplevel else None
5391 text = self._render_cte_clause(nesting_level=nesting_level) + text
5393 if select_stmt._suffixes:
5394 text += " " + self._generate_prefixes(
5395 select_stmt, select_stmt._suffixes, **kwargs
5396 )
5398 self.stack.pop(-1)
5400 return text
5402 def _setup_select_hints(
5403 self, select: Select[Unpack[TupleAny]]
5404 ) -> Tuple[str, _FromHintsType]:
5405 byfrom = {
5406 from_: hinttext
5407 % {"name": from_._compiler_dispatch(self, ashint=True)}
5408 for (from_, dialect), hinttext in select._hints.items()
5409 if dialect in ("*", self.dialect.name)
5410 }
5411 hint_text = self.get_select_hint_text(byfrom)
5412 return hint_text, byfrom
5414 def _setup_select_stack(
5415 self, select, compile_state, entry, asfrom, lateral, compound_index
5416 ):
5417 if compound_index == 0:
5418 entry["select_0"] = select
5419 elif compound_index:
5420 select_0 = entry["select_0"]
5421 numcols = len(select_0._all_selected_columns)
5423 if len(compile_state.columns_plus_names) != numcols:
5424 raise exc.CompileError(
5425 "All selectables passed to "
5426 "CompoundSelect must have identical numbers of "
5427 "columns; select #%d has %d columns, select "
5428 "#%d has %d"
5429 % (
5430 1,
5431 numcols,
5432 compound_index + 1,
5433 len(select._all_selected_columns),
5434 )
5435 )
5437 explicit, implicit = self._enclosing_correlate_froms(
5438 entry, asfrom, lateral
5439 )
5440 froms = compile_state._get_display_froms(
5441 explicit_correlate_froms=explicit,
5442 implicit_correlate_froms=implicit,
5443 )
5445 new_correlate_froms = set(_from_objects(*froms))
5446 all_correlate_froms = new_correlate_froms.union(
5447 entry["correlate_froms"]
5448 )
5450 new_entry: _CompilerStackEntry = {
5451 "asfrom_froms": new_correlate_froms,
5452 "correlate_froms": all_correlate_froms,
5453 "selectable": select,
5454 "compile_state": compile_state,
5455 }
5456 self.stack.append(new_entry)
5458 return froms
5460 def _compose_select_body(
5461 self,
5462 text,
5463 select,
5464 compile_state,
5465 inner_columns,
5466 froms,
5467 byfrom,
5468 toplevel,
5469 kwargs,
5470 ):
5471 text += ", ".join(inner_columns)
5473 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
5474 from_linter = FromLinter({}, set())
5475 warn_linting = self.linting & WARN_LINTING
5476 if toplevel:
5477 self.from_linter = from_linter
5478 else:
5479 from_linter = None
5480 warn_linting = False
5482 # adjust the whitespace for no inner columns, part of #9440,
5483 # so that a no-col SELECT comes out as "SELECT WHERE..." or
5484 # "SELECT FROM ...".
5485 # while it would be better to have built the SELECT starting string
5486 # without trailing whitespace first, then add whitespace only if inner
5487 # cols were present, this breaks compatibility with various custom
5488 # compilation schemes that are currently being tested.
5489 if not inner_columns:
5490 text = text.rstrip()
5492 if froms:
5493 text += " \nFROM "
5495 if select._hints:
5496 text += ", ".join(
5497 [
5498 f._compiler_dispatch(
5499 self,
5500 asfrom=True,
5501 fromhints=byfrom,
5502 from_linter=from_linter,
5503 **kwargs,
5504 )
5505 for f in froms
5506 ]
5507 )
5508 else:
5509 text += ", ".join(
5510 [
5511 f._compiler_dispatch(
5512 self,
5513 asfrom=True,
5514 from_linter=from_linter,
5515 **kwargs,
5516 )
5517 for f in froms
5518 ]
5519 )
5520 else:
5521 text += self.default_from()
5523 if select._where_criteria:
5524 t = self._generate_delimited_and_list(
5525 select._where_criteria, from_linter=from_linter, **kwargs
5526 )
5527 if t:
5528 text += " \nWHERE " + t
5530 if warn_linting:
5531 assert from_linter is not None
5532 from_linter.warn()
5534 if select._group_by_clauses:
5535 text += self.group_by_clause(select, **kwargs)
5537 if select._having_criteria:
5538 t = self._generate_delimited_and_list(
5539 select._having_criteria, **kwargs
5540 )
5541 if t:
5542 text += " \nHAVING " + t
5544 if select._post_criteria_clause is not None:
5545 pcc = self.process(select._post_criteria_clause, **kwargs)
5546 if pcc is not None:
5547 text += " \n" + pcc
5549 if select._order_by_clauses:
5550 text += self.order_by_clause(select, **kwargs)
5552 if select._has_row_limiting_clause:
5553 text += self._row_limit_clause(select, **kwargs)
5555 if select._for_update_arg is not None:
5556 text += self.for_update_clause(select, **kwargs)
5558 return text
5560 def _generate_prefixes(self, stmt, prefixes, **kw):
5561 clause = " ".join(
5562 prefix._compiler_dispatch(self, **kw)
5563 for prefix, dialect_name in prefixes
5564 if dialect_name in (None, "*") or dialect_name == self.dialect.name
5565 )
5566 if clause:
5567 clause += " "
5568 return clause
5570 def _render_cte_clause(
5571 self,
5572 nesting_level=None,
5573 include_following_stack=False,
5574 ):
5575 """
5576 include_following_stack
5577 Also render the nesting CTEs on the next stack. Useful for
5578 SQL structures like UNION or INSERT that can wrap SELECT
5579 statements containing nesting CTEs.
5580 """
5581 if not self.ctes:
5582 return ""
5584 ctes: MutableMapping[CTE, str]
5586 if nesting_level and nesting_level > 1:
5587 ctes = util.OrderedDict()
5588 for cte in list(self.ctes.keys()):
5589 cte_level, cte_name, cte_opts = self.level_name_by_cte[
5590 cte._get_reference_cte()
5591 ]
5592 nesting = cte.nesting or cte_opts.nesting
5593 is_rendered_level = cte_level == nesting_level or (
5594 include_following_stack and cte_level == nesting_level + 1
5595 )
5596 if not (nesting and is_rendered_level):
5597 continue
5599 ctes[cte] = self.ctes[cte]
5601 else:
5602 ctes = self.ctes
5604 if not ctes:
5605 return ""
5606 ctes_recursive = any([cte.recursive for cte in ctes])
5608 cte_text = self.get_cte_preamble(ctes_recursive) + " "
5609 cte_text += ", \n".join([txt for txt in ctes.values()])
5610 cte_text += "\n "
5612 if nesting_level and nesting_level > 1:
5613 for cte in list(ctes.keys()):
5614 cte_level, cte_name, cte_opts = self.level_name_by_cte[
5615 cte._get_reference_cte()
5616 ]
5617 del self.ctes[cte]
5618 del self.ctes_by_level_name[(cte_level, cte_name)]
5619 del self.level_name_by_cte[cte._get_reference_cte()]
5621 return cte_text
5623 def get_cte_preamble(self, recursive):
5624 if recursive:
5625 return "WITH RECURSIVE"
5626 else:
5627 return "WITH"
5629 def get_select_precolumns(self, select: Select[Any], **kw: Any) -> str:
5630 """Called when building a ``SELECT`` statement, position is just
5631 before column list.
5633 """
5634 if select._distinct_on:
5635 util.warn_deprecated(
5636 "DISTINCT ON is currently supported only by the PostgreSQL "
5637 "dialect. Use of DISTINCT ON for other backends is currently "
5638 "silently ignored, however this usage is deprecated, and will "
5639 "raise CompileError in a future release for all backends "
5640 "that do not support this syntax.",
5641 version="1.4",
5642 )
5643 return "DISTINCT " if select._distinct else ""
5645 def group_by_clause(self, select, **kw):
5646 """allow dialects to customize how GROUP BY is rendered."""
5648 group_by = self._generate_delimited_list(
5649 select._group_by_clauses, OPERATORS[operators.comma_op], **kw
5650 )
5651 if group_by:
5652 return " GROUP BY " + group_by
5653 else:
5654 return ""
5656 def order_by_clause(self, select, **kw):
5657 """allow dialects to customize how ORDER BY is rendered."""
5659 order_by = self._generate_delimited_list(
5660 select._order_by_clauses, OPERATORS[operators.comma_op], **kw
5661 )
5663 if order_by:
5664 return " ORDER BY " + order_by
5665 else:
5666 return ""
5668 def for_update_clause(self, select, **kw):
5669 return " FOR UPDATE"
5671 def returning_clause(
5672 self,
5673 stmt: UpdateBase,
5674 returning_cols: Sequence[_ColumnsClauseElement],
5675 *,
5676 populate_result_map: bool,
5677 **kw: Any,
5678 ) -> str:
5679 columns = [
5680 self._label_returning_column(
5681 stmt,
5682 column,
5683 populate_result_map,
5684 fallback_label_name=fallback_label_name,
5685 column_is_repeated=repeated,
5686 name=name,
5687 proxy_name=proxy_name,
5688 **kw,
5689 )
5690 for (
5691 name,
5692 proxy_name,
5693 fallback_label_name,
5694 column,
5695 repeated,
5696 ) in stmt._generate_columns_plus_names(
5697 True, cols=base._select_iterables(returning_cols)
5698 )
5699 ]
5701 return "RETURNING " + ", ".join(columns)
5703 def limit_clause(self, select, **kw):
5704 text = ""
5705 if select._limit_clause is not None:
5706 text += "\n LIMIT " + self.process(select._limit_clause, **kw)
5707 if select._offset_clause is not None:
5708 if select._limit_clause is None:
5709 text += "\n LIMIT -1"
5710 text += " OFFSET " + self.process(select._offset_clause, **kw)
5711 return text
5713 def fetch_clause(
5714 self,
5715 select,
5716 fetch_clause=None,
5717 require_offset=False,
5718 use_literal_execute_for_simple_int=False,
5719 **kw,
5720 ):
5721 if fetch_clause is None:
5722 fetch_clause = select._fetch_clause
5723 fetch_clause_options = select._fetch_clause_options
5724 else:
5725 fetch_clause_options = {"percent": False, "with_ties": False}
5727 text = ""
5729 if select._offset_clause is not None:
5730 offset_clause = select._offset_clause
5731 if (
5732 use_literal_execute_for_simple_int
5733 and select._simple_int_clause(offset_clause)
5734 ):
5735 offset_clause = offset_clause.render_literal_execute()
5736 offset_str = self.process(offset_clause, **kw)
5737 text += "\n OFFSET %s ROWS" % offset_str
5738 elif require_offset:
5739 text += "\n OFFSET 0 ROWS"
5741 if fetch_clause is not None:
5742 if (
5743 use_literal_execute_for_simple_int
5744 and select._simple_int_clause(fetch_clause)
5745 ):
5746 fetch_clause = fetch_clause.render_literal_execute()
5747 text += "\n FETCH FIRST %s%s ROWS %s" % (
5748 self.process(fetch_clause, **kw),
5749 " PERCENT" if fetch_clause_options["percent"] else "",
5750 "WITH TIES" if fetch_clause_options["with_ties"] else "ONLY",
5751 )
5752 return text
5754 def visit_table(
5755 self,
5756 table,
5757 asfrom=False,
5758 iscrud=False,
5759 ashint=False,
5760 fromhints=None,
5761 use_schema=True,
5762 from_linter=None,
5763 ambiguous_table_name_map=None,
5764 enclosing_alias=None,
5765 within_tstring=False,
5766 **kwargs,
5767 ):
5768 if from_linter:
5769 from_linter.froms[table] = table.fullname
5771 if asfrom or ashint or within_tstring:
5772 effective_schema = self.preparer.schema_for_object(table)
5774 if use_schema and effective_schema:
5775 ret = (
5776 self.preparer.quote_schema(effective_schema)
5777 + "."
5778 + self.preparer.quote(table.name)
5779 )
5780 else:
5781 ret = self.preparer.quote(table.name)
5783 if (
5784 (
5785 enclosing_alias is None
5786 or enclosing_alias.element is not table
5787 )
5788 and not effective_schema
5789 and ambiguous_table_name_map
5790 and table.name in ambiguous_table_name_map
5791 ):
5792 anon_name = self._truncated_identifier(
5793 "alias", ambiguous_table_name_map[table.name]
5794 )
5796 ret = ret + self.get_render_as_alias_suffix(
5797 self.preparer.format_alias(None, anon_name)
5798 )
5800 if fromhints and table in fromhints:
5801 ret = self.format_from_hint_text(
5802 ret, table, fromhints[table], iscrud
5803 )
5804 return ret
5805 else:
5806 return ""
5808 def visit_join(self, join, asfrom=False, from_linter=None, **kwargs):
5809 if from_linter:
5810 from_linter.edges.update(
5811 itertools.product(
5812 _de_clone(join.left._from_objects),
5813 _de_clone(join.right._from_objects),
5814 )
5815 )
5817 if join.full:
5818 join_type = " FULL OUTER JOIN "
5819 elif join.isouter:
5820 join_type = " LEFT OUTER JOIN "
5821 else:
5822 join_type = " JOIN "
5823 return (
5824 join.left._compiler_dispatch(
5825 self, asfrom=True, from_linter=from_linter, **kwargs
5826 )
5827 + join_type
5828 + join.right._compiler_dispatch(
5829 self, asfrom=True, from_linter=from_linter, **kwargs
5830 )
5831 + " ON "
5832 # TODO: likely need asfrom=True here?
5833 + join.onclause._compiler_dispatch(
5834 self, from_linter=from_linter, **kwargs
5835 )
5836 )
5838 def _setup_crud_hints(self, stmt, table_text):
5839 dialect_hints = {
5840 table: hint_text
5841 for (table, dialect), hint_text in stmt._hints.items()
5842 if dialect in ("*", self.dialect.name)
5843 }
5844 if stmt.table in dialect_hints:
5845 table_text = self.format_from_hint_text(
5846 table_text, stmt.table, dialect_hints[stmt.table], True
5847 )
5848 return dialect_hints, table_text
5850 # within the realm of "insertmanyvalues sentinel columns",
5851 # these lookups match different kinds of Column() configurations
5852 # to specific backend capabilities. they are broken into two
5853 # lookups, one for autoincrement columns and the other for non
5854 # autoincrement columns
5855 _sentinel_col_non_autoinc_lookup = util.immutabledict(
5856 {
5857 _SentinelDefaultCharacterization.CLIENTSIDE: (
5858 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5859 ),
5860 _SentinelDefaultCharacterization.SENTINEL_DEFAULT: (
5861 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5862 ),
5863 _SentinelDefaultCharacterization.NONE: (
5864 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5865 ),
5866 _SentinelDefaultCharacterization.IDENTITY: (
5867 InsertmanyvaluesSentinelOpts.IDENTITY
5868 ),
5869 _SentinelDefaultCharacterization.SEQUENCE: (
5870 InsertmanyvaluesSentinelOpts.SEQUENCE
5871 ),
5872 _SentinelDefaultCharacterization.MONOTONIC_FUNCTION: (
5873 InsertmanyvaluesSentinelOpts.MONOTONIC_FUNCTION
5874 ),
5875 }
5876 )
5877 _sentinel_col_autoinc_lookup = _sentinel_col_non_autoinc_lookup.union(
5878 {
5879 _SentinelDefaultCharacterization.NONE: (
5880 InsertmanyvaluesSentinelOpts.AUTOINCREMENT
5881 ),
5882 }
5883 )
5885 def _get_sentinel_column_for_table(
5886 self, table: Table
5887 ) -> Optional[Sequence[Column[Any]]]:
5888 """given a :class:`.Table`, return a usable sentinel column or
5889 columns for this dialect if any.
5891 Return None if no sentinel columns could be identified, or raise an
5892 error if a column was marked as a sentinel explicitly but isn't
5893 compatible with this dialect.
5895 """
5897 sentinel_opts = self.dialect.insertmanyvalues_implicit_sentinel
5898 sentinel_characteristics = table._sentinel_column_characteristics
5900 sent_cols = sentinel_characteristics.columns
5902 if sent_cols is None:
5903 return None
5905 if sentinel_characteristics.is_autoinc:
5906 bitmask = self._sentinel_col_autoinc_lookup.get(
5907 sentinel_characteristics.default_characterization, 0
5908 )
5909 else:
5910 bitmask = self._sentinel_col_non_autoinc_lookup.get(
5911 sentinel_characteristics.default_characterization, 0
5912 )
5914 if sentinel_opts & bitmask:
5915 return sent_cols
5917 if sentinel_characteristics.is_explicit:
5918 # a column was explicitly marked as insert_sentinel=True,
5919 # however it is not compatible with this dialect. they should
5920 # not indicate this column as a sentinel if they need to include
5921 # this dialect.
5923 # TODO: do we want non-primary key explicit sentinel cols
5924 # that can gracefully degrade for some backends?
5925 # insert_sentinel="degrade" perhaps. not for the initial release.
5926 # I am hoping people are generally not dealing with this sentinel
5927 # business at all.
5929 # if is_explicit is True, there will be only one sentinel column.
5931 raise exc.InvalidRequestError(
5932 f"Column {sent_cols[0]} can't be explicitly "
5933 "marked as a sentinel column when using the "
5934 f"{self.dialect.name} dialect, as the "
5935 "particular type of default generation on this column is "
5936 "not currently compatible with this dialect's specific "
5937 f"INSERT..RETURNING syntax which can receive the "
5938 "server-generated value in "
5939 "a deterministic way. To remove this error, remove "
5940 "insert_sentinel=True from primary key autoincrement "
5941 "columns; these columns are automatically used as "
5942 "sentinels for supported dialects in any case."
5943 )
5945 return None
5947 def _deliver_insertmanyvalues_batches(
5948 self,
5949 statement: str,
5950 parameters: _DBAPIMultiExecuteParams,
5951 compiled_parameters: List[_MutableCoreSingleExecuteParams],
5952 generic_setinputsizes: Optional[_GenericSetInputSizesType],
5953 batch_size: int,
5954 sort_by_parameter_order: bool,
5955 schema_translate_map: Optional[SchemaTranslateMapType],
5956 ) -> Iterator[_InsertManyValuesBatch]:
5957 imv = self._insertmanyvalues
5958 assert imv is not None
5960 if not imv.sentinel_param_keys:
5961 _sentinel_from_params = None
5962 else:
5963 _sentinel_from_params = operator.itemgetter(
5964 *imv.sentinel_param_keys
5965 )
5967 lenparams = len(parameters)
5968 if imv.is_default_expr and not self.dialect.supports_default_metavalue:
5969 # backend doesn't support
5970 # INSERT INTO table (pk_col) VALUES (DEFAULT), (DEFAULT), ...
5971 # at the moment this is basically SQL Server due to
5972 # not being able to use DEFAULT for identity column
5973 # just yield out that many single statements! still
5974 # faster than a whole connection.execute() call ;)
5975 #
5976 # note we still are taking advantage of the fact that we know
5977 # we are using RETURNING. The generalized approach of fetching
5978 # cursor.lastrowid etc. still goes through the more heavyweight
5979 # "ExecutionContext per statement" system as it isn't usable
5980 # as a generic "RETURNING" approach
5981 use_row_at_a_time = True
5982 downgraded = False
5983 elif not self.dialect.supports_multivalues_insert or (
5984 sort_by_parameter_order
5985 and self._result_columns
5986 and (
5987 imv.sentinel_columns is None
5988 or (
5989 imv.includes_upsert_behaviors
5990 and not imv.embed_values_counter
5991 )
5992 )
5993 ):
5994 # deterministic order was requested and the compiler could
5995 # not organize sentinel columns for this dialect/statement.
5996 # use row at a time. Note: if embed_values_counter is True,
5997 # the counter itself provides the ordering capability we need,
5998 # so we can use batch mode even with upsert behaviors.
5999 use_row_at_a_time = True
6000 downgraded = True
6001 elif (
6002 imv.has_upsert_bound_parameters
6003 and not imv.embed_values_counter
6004 and self._result_columns
6005 ):
6006 # For upsert behaviors (ON CONFLICT DO UPDATE, etc.) with RETURNING
6007 # and parametrized bindparams in the SET clause, we must use
6008 # row-at-a-time. Batching multiple rows in a single statement
6009 # doesn't work when the SET clause contains bound parameters that
6010 # will receive different values per row, as there's only one SET
6011 # clause per statement. See issue #13130.
6012 use_row_at_a_time = True
6013 downgraded = True
6014 else:
6015 use_row_at_a_time = False
6016 downgraded = False
6018 if use_row_at_a_time:
6019 for batchnum, (param, compiled_param) in enumerate(
6020 cast(
6021 "Sequence[Tuple[_DBAPISingleExecuteParams, _MutableCoreSingleExecuteParams]]", # noqa: E501
6022 zip(parameters, compiled_parameters),
6023 ),
6024 1,
6025 ):
6026 yield _InsertManyValuesBatch(
6027 statement,
6028 param,
6029 generic_setinputsizes,
6030 [param],
6031 (
6032 [_sentinel_from_params(compiled_param)]
6033 if _sentinel_from_params
6034 else []
6035 ),
6036 1,
6037 batchnum,
6038 lenparams,
6039 sort_by_parameter_order,
6040 downgraded,
6041 )
6042 return
6044 if schema_translate_map:
6045 rst = functools.partial(
6046 self.preparer._render_schema_translates,
6047 schema_translate_map=schema_translate_map,
6048 )
6049 else:
6050 rst = None
6052 imv_single_values_expr = imv.single_values_expr
6053 if rst:
6054 imv_single_values_expr = rst(imv_single_values_expr)
6056 executemany_values = f"({imv_single_values_expr})"
6057 statement = statement.replace(executemany_values, "__EXECMANY_TOKEN__")
6059 # Use optional insertmanyvalues_max_parameters
6060 # to further shrink the batch size so that there are no more than
6061 # insertmanyvalues_max_parameters params.
6062 # Currently used by SQL Server, which limits statements to 2100 bound
6063 # parameters (actually 2099).
6064 max_params = self.dialect.insertmanyvalues_max_parameters
6065 if max_params:
6066 total_num_of_params = len(self.bind_names)
6067 num_params_per_batch = len(imv.insert_crud_params)
6068 num_params_outside_of_batch = (
6069 total_num_of_params - num_params_per_batch
6070 )
6071 batch_size = min(
6072 batch_size,
6073 (
6074 (max_params - num_params_outside_of_batch)
6075 // num_params_per_batch
6076 ),
6077 )
6079 batches = cast("List[Sequence[Any]]", list(parameters))
6080 compiled_batches = cast(
6081 "List[Sequence[Any]]", list(compiled_parameters)
6082 )
6084 processed_setinputsizes: Optional[_GenericSetInputSizesType] = None
6085 batchnum = 1
6086 total_batches = lenparams // batch_size + (
6087 1 if lenparams % batch_size else 0
6088 )
6090 insert_crud_params = imv.insert_crud_params
6091 assert insert_crud_params is not None
6093 if rst:
6094 insert_crud_params = [
6095 (col, key, rst(expr), st)
6096 for col, key, expr, st in insert_crud_params
6097 ]
6099 escaped_bind_names: Mapping[str, str]
6100 expand_pos_lower_index = expand_pos_upper_index = 0
6102 if not self.positional:
6103 if self.escaped_bind_names:
6104 escaped_bind_names = self.escaped_bind_names
6105 else:
6106 escaped_bind_names = {}
6108 all_keys = set(parameters[0])
6110 def apply_placeholders(keys, formatted):
6111 for key in keys:
6112 key = escaped_bind_names.get(key, key)
6113 formatted = formatted.replace(
6114 self.bindtemplate % {"name": key},
6115 self.bindtemplate
6116 % {"name": f"{key}__EXECMANY_INDEX__"},
6117 )
6118 return formatted
6120 if imv.embed_values_counter:
6121 imv_values_counter = ", _IMV_VALUES_COUNTER"
6122 else:
6123 imv_values_counter = ""
6124 formatted_values_clause = f"""({', '.join(
6125 apply_placeholders(bind_keys, formatted)
6126 for _, _, formatted, bind_keys in insert_crud_params
6127 )}{imv_values_counter})"""
6129 keys_to_replace = all_keys.intersection(
6130 escaped_bind_names.get(key, key)
6131 for _, _, _, bind_keys in insert_crud_params
6132 for key in bind_keys
6133 )
6134 base_parameters = {
6135 key: parameters[0][key]
6136 for key in all_keys.difference(keys_to_replace)
6137 }
6139 executemany_values_w_comma = ""
6140 else:
6141 formatted_values_clause = ""
6142 keys_to_replace = set()
6143 base_parameters = {}
6145 if imv.embed_values_counter:
6146 executemany_values_w_comma = (
6147 f"({imv_single_values_expr}, _IMV_VALUES_COUNTER), "
6148 )
6149 else:
6150 executemany_values_w_comma = f"({imv_single_values_expr}), "
6152 all_names_we_will_expand: Set[str] = set()
6153 for elem in imv.insert_crud_params:
6154 all_names_we_will_expand.update(elem[3])
6156 # get the start and end position in a particular list
6157 # of parameters where we will be doing the "expanding".
6158 # statements can have params on either side or both sides,
6159 # given RETURNING and CTEs
6160 if all_names_we_will_expand:
6161 positiontup = self.positiontup
6162 assert positiontup is not None
6164 all_expand_positions = {
6165 idx
6166 for idx, name in enumerate(positiontup)
6167 if name in all_names_we_will_expand
6168 }
6169 expand_pos_lower_index = min(all_expand_positions)
6170 expand_pos_upper_index = max(all_expand_positions) + 1
6171 assert (
6172 len(all_expand_positions)
6173 == expand_pos_upper_index - expand_pos_lower_index
6174 )
6176 if self._numeric_binds:
6177 escaped = re.escape(self._numeric_binds_identifier_char)
6178 executemany_values_w_comma = re.sub(
6179 rf"{escaped}\d+", "%s", executemany_values_w_comma
6180 )
6182 while batches:
6183 batch = batches[0:batch_size]
6184 compiled_batch = compiled_batches[0:batch_size]
6186 batches[0:batch_size] = []
6187 compiled_batches[0:batch_size] = []
6189 if batches:
6190 current_batch_size = batch_size
6191 else:
6192 current_batch_size = len(batch)
6194 if generic_setinputsizes:
6195 # if setinputsizes is present, expand this collection to
6196 # suit the batch length as well
6197 # currently this will be mssql+pyodbc for internal dialects
6198 processed_setinputsizes = [
6199 (new_key, len_, typ)
6200 for new_key, len_, typ in (
6201 (f"{key}_{index}", len_, typ)
6202 for index in range(current_batch_size)
6203 for key, len_, typ in generic_setinputsizes
6204 )
6205 ]
6207 replaced_parameters: Any
6208 if self.positional:
6209 num_ins_params = imv.num_positional_params_counted
6211 batch_iterator: Iterable[Sequence[Any]]
6212 extra_params_left: Sequence[Any]
6213 extra_params_right: Sequence[Any]
6215 if num_ins_params == len(batch[0]):
6216 extra_params_left = extra_params_right = ()
6217 batch_iterator = batch
6218 else:
6219 extra_params_left = batch[0][:expand_pos_lower_index]
6220 extra_params_right = batch[0][expand_pos_upper_index:]
6221 batch_iterator = (
6222 b[expand_pos_lower_index:expand_pos_upper_index]
6223 for b in batch
6224 )
6226 if imv.embed_values_counter:
6227 expanded_values_string = (
6228 "".join(
6229 executemany_values_w_comma.replace(
6230 "_IMV_VALUES_COUNTER", str(i)
6231 )
6232 for i, _ in enumerate(batch)
6233 )
6234 )[:-2]
6235 else:
6236 expanded_values_string = (
6237 (executemany_values_w_comma * current_batch_size)
6238 )[:-2]
6240 if self._numeric_binds and num_ins_params > 0:
6241 # numeric will always number the parameters inside of
6242 # VALUES (and thus order self.positiontup) to be higher
6243 # than non-VALUES parameters, no matter where in the
6244 # statement those non-VALUES parameters appear (this is
6245 # ensured in _process_numeric by numbering first all
6246 # params that are not in _values_bindparam)
6247 # therefore all extra params are always
6248 # on the left side and numbered lower than the VALUES
6249 # parameters
6250 assert not extra_params_right
6252 start = expand_pos_lower_index + 1
6253 end = num_ins_params * (current_batch_size) + start
6255 # need to format here, since statement may contain
6256 # unescaped %, while values_string contains just (%s, %s)
6257 positions = tuple(
6258 f"{self._numeric_binds_identifier_char}{i}"
6259 for i in range(start, end)
6260 )
6261 expanded_values_string = expanded_values_string % positions
6263 replaced_statement = statement.replace(
6264 "__EXECMANY_TOKEN__", expanded_values_string
6265 )
6267 replaced_parameters = tuple(
6268 itertools.chain.from_iterable(batch_iterator)
6269 )
6271 replaced_parameters = (
6272 extra_params_left
6273 + replaced_parameters
6274 + extra_params_right
6275 )
6277 else:
6278 replaced_values_clauses = []
6279 replaced_parameters = base_parameters.copy()
6281 for i, param in enumerate(batch):
6282 fmv = formatted_values_clause.replace(
6283 "EXECMANY_INDEX__", str(i)
6284 )
6285 if imv.embed_values_counter:
6286 fmv = fmv.replace("_IMV_VALUES_COUNTER", str(i))
6288 replaced_values_clauses.append(fmv)
6289 replaced_parameters.update(
6290 {f"{key}__{i}": param[key] for key in keys_to_replace}
6291 )
6293 replaced_statement = statement.replace(
6294 "__EXECMANY_TOKEN__",
6295 ", ".join(replaced_values_clauses),
6296 )
6298 yield _InsertManyValuesBatch(
6299 replaced_statement,
6300 replaced_parameters,
6301 processed_setinputsizes,
6302 batch,
6303 (
6304 [_sentinel_from_params(cb) for cb in compiled_batch]
6305 if _sentinel_from_params
6306 else []
6307 ),
6308 current_batch_size,
6309 batchnum,
6310 total_batches,
6311 sort_by_parameter_order,
6312 False,
6313 )
6314 batchnum += 1
6316 def visit_insert(
6317 self, insert_stmt, visited_bindparam=None, visiting_cte=None, **kw
6318 ):
6319 compile_state = insert_stmt._compile_state_factory(
6320 insert_stmt, self, **kw
6321 )
6322 insert_stmt = compile_state.statement
6324 if visiting_cte is not None:
6325 kw["visiting_cte"] = visiting_cte
6326 toplevel = False
6327 else:
6328 toplevel = not self.stack
6330 if toplevel:
6331 self.isinsert = True
6332 if not self.dml_compile_state:
6333 self.dml_compile_state = compile_state
6334 if not self.compile_state:
6335 self.compile_state = compile_state
6337 self.stack.append(
6338 {
6339 "correlate_froms": set(),
6340 "asfrom_froms": set(),
6341 "selectable": insert_stmt,
6342 }
6343 )
6345 counted_bindparam = 0
6347 # reset any incoming "visited_bindparam" collection
6348 visited_bindparam = None
6350 # for positional, insertmanyvalues needs to know how many
6351 # bound parameters are in the VALUES sequence; there's no simple
6352 # rule because default expressions etc. can have zero or more
6353 # params inside them. After multiple attempts to figure this out,
6354 # this very simplistic "count after" works and is
6355 # likely the least amount of callcounts, though looks clumsy
6356 if self.positional and visiting_cte is None:
6357 # if we are inside a CTE, don't count parameters
6358 # here since they won't be for insertmanyvalues. keep
6359 # visited_bindparam at None so no counting happens.
6360 # see #9173
6361 visited_bindparam = []
6363 crud_params_struct = crud._get_crud_params(
6364 self,
6365 insert_stmt,
6366 compile_state,
6367 toplevel,
6368 visited_bindparam=visited_bindparam,
6369 **kw,
6370 )
6372 if self.positional and visited_bindparam is not None:
6373 counted_bindparam = len(visited_bindparam)
6374 if self._numeric_binds:
6375 if self._values_bindparam is not None:
6376 self._values_bindparam += visited_bindparam
6377 else:
6378 self._values_bindparam = visited_bindparam
6380 crud_params_single = crud_params_struct.single_params
6382 if (
6383 not crud_params_single
6384 and not self.dialect.supports_default_values
6385 and not self.dialect.supports_default_metavalue
6386 and not self.dialect.supports_empty_insert
6387 ):
6388 raise exc.CompileError(
6389 "The '%s' dialect with current database "
6390 "version settings does not support empty "
6391 "inserts." % self.dialect.name
6392 )
6394 if compile_state._has_multi_parameters:
6395 if not self.dialect.supports_multivalues_insert:
6396 raise exc.CompileError(
6397 "The '%s' dialect with current database "
6398 "version settings does not support "
6399 "in-place multirow inserts." % self.dialect.name
6400 )
6401 elif (
6402 self.implicit_returning or insert_stmt._returning
6403 ) and insert_stmt._sort_by_parameter_order:
6404 raise exc.CompileError(
6405 "RETURNING cannot be deterministically sorted when "
6406 "using an INSERT which includes multi-row values()."
6407 )
6408 crud_params_single = crud_params_struct.single_params
6409 else:
6410 crud_params_single = crud_params_struct.single_params
6412 preparer = self.preparer
6413 supports_default_values = self.dialect.supports_default_values
6415 text = "INSERT "
6417 if insert_stmt._prefixes:
6418 text += self._generate_prefixes(
6419 insert_stmt, insert_stmt._prefixes, **kw
6420 )
6422 text += "INTO "
6423 table_text = preparer.format_table(insert_stmt.table)
6425 if insert_stmt._hints:
6426 _, table_text = self._setup_crud_hints(insert_stmt, table_text)
6428 if insert_stmt._independent_ctes:
6429 self._dispatch_independent_ctes(insert_stmt, kw)
6431 text += table_text
6433 if crud_params_single or not supports_default_values:
6434 text += " (%s)" % ", ".join(
6435 [expr for _, expr, _, _ in crud_params_single]
6436 )
6438 # look for insertmanyvalues attributes that would have been configured
6439 # by crud.py as it scanned through the columns to be part of the
6440 # INSERT
6441 use_insertmanyvalues = crud_params_struct.use_insertmanyvalues
6442 named_sentinel_params: Optional[Sequence[str]] = None
6443 add_sentinel_cols = None
6444 implicit_sentinel = False
6446 returning_cols = self.implicit_returning or insert_stmt._returning
6447 if returning_cols:
6448 add_sentinel_cols = crud_params_struct.use_sentinel_columns
6449 if add_sentinel_cols is not None:
6450 assert use_insertmanyvalues
6452 # search for the sentinel column explicitly present
6453 # in the INSERT columns list, and additionally check that
6454 # this column has a bound parameter name set up that's in the
6455 # parameter list. If both of these cases are present, it means
6456 # we will have a client side value for the sentinel in each
6457 # parameter set.
6459 _params_by_col = {
6460 col: param_names
6461 for col, _, _, param_names in crud_params_single
6462 }
6463 named_sentinel_params = []
6464 for _add_sentinel_col in add_sentinel_cols:
6465 if _add_sentinel_col not in _params_by_col:
6466 named_sentinel_params = None
6467 break
6468 param_name = self._within_exec_param_key_getter(
6469 _add_sentinel_col
6470 )
6471 if param_name not in _params_by_col[_add_sentinel_col]:
6472 named_sentinel_params = None
6473 break
6474 named_sentinel_params.append(param_name)
6476 if named_sentinel_params is None:
6477 # if we are not going to have a client side value for
6478 # the sentinel in the parameter set, that means it's
6479 # an autoincrement, an IDENTITY, or a server-side SQL
6480 # expression like nextval('seqname'). So this is
6481 # an "implicit" sentinel; we will look for it in
6482 # RETURNING
6483 # only, and then sort on it. For this case on PG,
6484 # SQL Server we have to use a special INSERT form
6485 # that guarantees the server side function lines up with
6486 # the entries in the VALUES.
6487 if (
6488 self.dialect.insertmanyvalues_implicit_sentinel
6489 & InsertmanyvaluesSentinelOpts.ANY_AUTOINCREMENT
6490 ):
6491 implicit_sentinel = True
6492 else:
6493 # here, we are not using a sentinel at all
6494 # and we are likely the SQLite dialect.
6495 # The first add_sentinel_col that we have should not
6496 # be marked as "insert_sentinel=True". if it was,
6497 # an error should have been raised in
6498 # _get_sentinel_column_for_table.
6499 assert not add_sentinel_cols[0]._insert_sentinel, (
6500 "sentinel selection rules should have prevented "
6501 "us from getting here for this dialect"
6502 )
6504 # always put the sentinel columns last. even if they are
6505 # in the returning list already, they will be there twice
6506 # then.
6507 returning_cols = list(returning_cols) + list(add_sentinel_cols)
6509 returning_clause = self.returning_clause(
6510 insert_stmt,
6511 returning_cols,
6512 populate_result_map=toplevel,
6513 )
6515 if self.returning_precedes_values:
6516 text += " " + returning_clause
6518 else:
6519 returning_clause = None
6521 if insert_stmt.select is not None:
6522 # placed here by crud.py
6523 select_text = self.process(
6524 self.stack[-1]["insert_from_select"], insert_into=True, **kw
6525 )
6527 if self.ctes and self.dialect.cte_follows_insert:
6528 nesting_level = len(self.stack) if not toplevel else None
6529 text += " %s%s" % (
6530 self._render_cte_clause(
6531 nesting_level=nesting_level,
6532 include_following_stack=True,
6533 ),
6534 select_text,
6535 )
6536 else:
6537 text += " %s" % select_text
6538 elif not crud_params_single and supports_default_values:
6539 text += " DEFAULT VALUES"
6540 if use_insertmanyvalues:
6541 self._insertmanyvalues = _InsertManyValues(
6542 True,
6543 self.dialect.default_metavalue_token,
6544 crud_params_single,
6545 counted_bindparam,
6546 sort_by_parameter_order=(
6547 insert_stmt._sort_by_parameter_order
6548 ),
6549 includes_upsert_behaviors=(
6550 insert_stmt._post_values_clause is not None
6551 ),
6552 sentinel_columns=add_sentinel_cols,
6553 num_sentinel_columns=(
6554 len(add_sentinel_cols) if add_sentinel_cols else 0
6555 ),
6556 implicit_sentinel=implicit_sentinel,
6557 )
6558 elif compile_state._has_multi_parameters:
6559 text += " VALUES %s" % (
6560 ", ".join(
6561 "(%s)"
6562 % (", ".join(value for _, _, value, _ in crud_param_set))
6563 for crud_param_set in crud_params_struct.all_multi_params
6564 ),
6565 )
6566 elif use_insertmanyvalues:
6567 if (
6568 implicit_sentinel
6569 and (
6570 self.dialect.insertmanyvalues_implicit_sentinel
6571 & InsertmanyvaluesSentinelOpts.USE_INSERT_FROM_SELECT
6572 )
6573 # this is checking if we have
6574 # INSERT INTO table (id) VALUES (DEFAULT).
6575 and not (crud_params_struct.is_default_metavalue_only)
6576 ):
6577 # if we have a sentinel column that is server generated,
6578 # then for selected backends render the VALUES list as a
6579 # subquery. This is the orderable form supported by
6580 # PostgreSQL and in fewer cases SQL Server
6581 embed_sentinel_value = True
6583 render_bind_casts = (
6584 self.dialect.insertmanyvalues_implicit_sentinel
6585 & InsertmanyvaluesSentinelOpts.RENDER_SELECT_COL_CASTS
6586 )
6588 add_sentinel_set = add_sentinel_cols or ()
6590 insert_single_values_expr = ", ".join(
6591 [
6592 value
6593 for col, _, value, _ in crud_params_single
6594 if col not in add_sentinel_set
6595 ]
6596 )
6598 colnames = ", ".join(
6599 f"p{i}"
6600 for i, cp in enumerate(crud_params_single)
6601 if cp[0] not in add_sentinel_set
6602 )
6604 if render_bind_casts:
6605 # render casts for the SELECT list. For PG, we are
6606 # already rendering bind casts in the parameter list,
6607 # selectively for the more "tricky" types like ARRAY.
6608 # however, even for the "easy" types, if the parameter
6609 # is NULL for every entry, PG gives up and says
6610 # "it must be TEXT", which fails for other easy types
6611 # like ints. So we cast on this side too.
6612 colnames_w_cast = ", ".join(
6613 (
6614 self.render_bind_cast(
6615 col.type,
6616 col.type._unwrapped_dialect_impl(self.dialect),
6617 f"p{i}",
6618 )
6619 if col not in add_sentinel_set
6620 else expr
6621 )
6622 for i, (col, _, expr, _) in enumerate(
6623 crud_params_single
6624 )
6625 )
6626 else:
6627 colnames_w_cast = ", ".join(
6628 (f"p{i}" if col not in add_sentinel_set else expr)
6629 for i, (col, _, expr, _) in enumerate(
6630 crud_params_single
6631 )
6632 )
6634 insert_crud_params = [
6635 elem
6636 for elem in crud_params_single
6637 if elem[0] not in add_sentinel_set
6638 ]
6640 text += (
6641 f" SELECT {colnames_w_cast} FROM "
6642 f"(VALUES ({insert_single_values_expr})) "
6643 f"AS imp_sen({colnames}, sen_counter) "
6644 "ORDER BY sen_counter"
6645 )
6647 else:
6648 # otherwise, if no sentinel or backend doesn't support
6649 # orderable subquery form, use a plain VALUES list
6650 embed_sentinel_value = False
6651 insert_crud_params = crud_params_single
6652 insert_single_values_expr = ", ".join(
6653 [value for _, _, value, _ in crud_params_single]
6654 )
6656 text += f" VALUES ({insert_single_values_expr})"
6658 self._insertmanyvalues = _InsertManyValues(
6659 is_default_expr=False,
6660 single_values_expr=insert_single_values_expr,
6661 insert_crud_params=insert_crud_params,
6662 num_positional_params_counted=counted_bindparam,
6663 sort_by_parameter_order=(insert_stmt._sort_by_parameter_order),
6664 includes_upsert_behaviors=(
6665 insert_stmt._post_values_clause is not None
6666 ),
6667 sentinel_columns=add_sentinel_cols,
6668 num_sentinel_columns=(
6669 len(add_sentinel_cols) if add_sentinel_cols else 0
6670 ),
6671 sentinel_param_keys=named_sentinel_params,
6672 implicit_sentinel=implicit_sentinel,
6673 embed_values_counter=embed_sentinel_value,
6674 )
6676 else:
6677 insert_single_values_expr = ", ".join(
6678 [value for _, _, value, _ in crud_params_single]
6679 )
6681 text += f" VALUES ({insert_single_values_expr})"
6683 if insert_stmt._post_values_clause is not None:
6684 post_values_clause = self.process(
6685 insert_stmt._post_values_clause, **kw
6686 )
6687 if post_values_clause:
6688 text += " " + post_values_clause
6690 if returning_clause and not self.returning_precedes_values:
6691 text += " " + returning_clause
6693 if self.ctes and not self.dialect.cte_follows_insert:
6694 nesting_level = len(self.stack) if not toplevel else None
6695 text = (
6696 self._render_cte_clause(
6697 nesting_level=nesting_level,
6698 include_following_stack=True,
6699 )
6700 + text
6701 )
6703 self.stack.pop(-1)
6705 return text
6707 def update_tables_clause(self, update_stmt, from_table, extra_froms, **kw):
6708 """Provide a hook to override the initial table clause
6709 in an UPDATE statement.
6711 MySQL overrides this.
6713 """
6714 kw["asfrom"] = True
6715 return from_table._compiler_dispatch(self, iscrud=True, **kw)
6717 def update_from_clause(
6718 self, update_stmt, from_table, extra_froms, from_hints, **kw
6719 ):
6720 """Provide a hook to override the generation of an
6721 UPDATE..FROM clause.
6722 MySQL and MSSQL override this.
6723 """
6724 raise NotImplementedError(
6725 "This backend does not support multiple-table "
6726 "criteria within UPDATE"
6727 )
6729 def update_post_criteria_clause(
6730 self, update_stmt: Update, **kw: Any
6731 ) -> Optional[str]:
6732 """provide a hook to override generation after the WHERE criteria
6733 in an UPDATE statement
6735 .. versionadded:: 2.1
6737 """
6738 if update_stmt._post_criteria_clause is not None:
6739 return self.process(
6740 update_stmt._post_criteria_clause,
6741 **kw,
6742 )
6743 else:
6744 return None
6746 def delete_post_criteria_clause(
6747 self, delete_stmt: Delete, **kw: Any
6748 ) -> Optional[str]:
6749 """provide a hook to override generation after the WHERE criteria
6750 in a DELETE statement
6752 .. versionadded:: 2.1
6754 """
6755 if delete_stmt._post_criteria_clause is not None:
6756 return self.process(
6757 delete_stmt._post_criteria_clause,
6758 **kw,
6759 )
6760 else:
6761 return None
6763 def visit_update(
6764 self,
6765 update_stmt: Update,
6766 visiting_cte: Optional[CTE] = None,
6767 **kw: Any,
6768 ) -> str:
6769 compile_state = update_stmt._compile_state_factory(
6770 update_stmt, self, **kw
6771 )
6772 if TYPE_CHECKING:
6773 assert isinstance(compile_state, UpdateDMLState)
6774 update_stmt = compile_state.statement # type: ignore[assignment]
6776 if visiting_cte is not None:
6777 kw["visiting_cte"] = visiting_cte
6778 toplevel = False
6779 else:
6780 toplevel = not self.stack
6782 if toplevel:
6783 self.isupdate = True
6784 if not self.dml_compile_state:
6785 self.dml_compile_state = compile_state
6786 if not self.compile_state:
6787 self.compile_state = compile_state
6789 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
6790 from_linter = FromLinter({}, set())
6791 warn_linting = self.linting & WARN_LINTING
6792 if toplevel:
6793 self.from_linter = from_linter
6794 else:
6795 from_linter = None
6796 warn_linting = False
6798 extra_froms = compile_state._extra_froms
6799 is_multitable = bool(extra_froms)
6801 if is_multitable:
6802 # main table might be a JOIN
6803 main_froms = set(_from_objects(update_stmt.table))
6804 render_extra_froms = [
6805 f for f in extra_froms if f not in main_froms
6806 ]
6807 correlate_froms = main_froms.union(extra_froms)
6808 else:
6809 render_extra_froms = []
6810 correlate_froms = {update_stmt.table}
6812 self.stack.append(
6813 {
6814 "correlate_froms": correlate_froms,
6815 "asfrom_froms": correlate_froms,
6816 "selectable": update_stmt,
6817 }
6818 )
6820 text = "UPDATE "
6822 if update_stmt._prefixes:
6823 text += self._generate_prefixes(
6824 update_stmt, update_stmt._prefixes, **kw
6825 )
6827 table_text = self.update_tables_clause(
6828 update_stmt,
6829 update_stmt.table,
6830 render_extra_froms,
6831 from_linter=from_linter,
6832 **kw,
6833 )
6834 crud_params_struct = crud._get_crud_params(
6835 self, update_stmt, compile_state, toplevel, **kw
6836 )
6837 crud_params = crud_params_struct.single_params
6839 if update_stmt._hints:
6840 dialect_hints, table_text = self._setup_crud_hints(
6841 update_stmt, table_text
6842 )
6843 else:
6844 dialect_hints = None
6846 if update_stmt._independent_ctes:
6847 self._dispatch_independent_ctes(update_stmt, kw)
6849 text += table_text
6851 text += " SET "
6852 text += ", ".join(
6853 expr + "=" + value
6854 for _, expr, value, _ in cast(
6855 "List[Tuple[Any, str, str, Any]]", crud_params
6856 )
6857 )
6859 if self.implicit_returning or update_stmt._returning:
6860 if self.returning_precedes_values:
6861 text += " " + self.returning_clause(
6862 update_stmt,
6863 self.implicit_returning or update_stmt._returning,
6864 populate_result_map=toplevel,
6865 )
6867 if extra_froms:
6868 extra_from_text = self.update_from_clause(
6869 update_stmt,
6870 update_stmt.table,
6871 render_extra_froms,
6872 dialect_hints,
6873 from_linter=from_linter,
6874 **kw,
6875 )
6876 if extra_from_text:
6877 text += " " + extra_from_text
6879 if update_stmt._where_criteria:
6880 t = self._generate_delimited_and_list(
6881 update_stmt._where_criteria, from_linter=from_linter, **kw
6882 )
6883 if t:
6884 text += " WHERE " + t
6886 ulc = self.update_post_criteria_clause(
6887 update_stmt, from_linter=from_linter, **kw
6888 )
6889 if ulc:
6890 text += " " + ulc
6892 if (
6893 self.implicit_returning or update_stmt._returning
6894 ) and not self.returning_precedes_values:
6895 text += " " + self.returning_clause(
6896 update_stmt,
6897 self.implicit_returning or update_stmt._returning,
6898 populate_result_map=toplevel,
6899 )
6901 if self.ctes:
6902 nesting_level = len(self.stack) if not toplevel else None
6903 text = self._render_cte_clause(nesting_level=nesting_level) + text
6905 if warn_linting:
6906 assert from_linter is not None
6907 from_linter.warn(stmt_type="UPDATE")
6909 self.stack.pop(-1)
6911 return text # type: ignore[no-any-return]
6913 def delete_extra_from_clause(
6914 self, delete_stmt, from_table, extra_froms, from_hints, **kw
6915 ):
6916 """Provide a hook to override the generation of an
6917 DELETE..FROM clause.
6919 This can be used to implement DELETE..USING for example.
6921 MySQL and MSSQL override this.
6923 """
6924 raise NotImplementedError(
6925 "This backend does not support multiple-table "
6926 "criteria within DELETE"
6927 )
6929 def delete_table_clause(self, delete_stmt, from_table, extra_froms, **kw):
6930 return from_table._compiler_dispatch(
6931 self, asfrom=True, iscrud=True, **kw
6932 )
6934 def visit_delete(self, delete_stmt, visiting_cte=None, **kw):
6935 compile_state = delete_stmt._compile_state_factory(
6936 delete_stmt, self, **kw
6937 )
6938 delete_stmt = compile_state.statement
6940 if visiting_cte is not None:
6941 kw["visiting_cte"] = visiting_cte
6942 toplevel = False
6943 else:
6944 toplevel = not self.stack
6946 if toplevel:
6947 self.isdelete = True
6948 if not self.dml_compile_state:
6949 self.dml_compile_state = compile_state
6950 if not self.compile_state:
6951 self.compile_state = compile_state
6953 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
6954 from_linter = FromLinter({}, set())
6955 warn_linting = self.linting & WARN_LINTING
6956 if toplevel:
6957 self.from_linter = from_linter
6958 else:
6959 from_linter = None
6960 warn_linting = False
6962 extra_froms = compile_state._extra_froms
6964 correlate_froms = {delete_stmt.table}.union(extra_froms)
6965 self.stack.append(
6966 {
6967 "correlate_froms": correlate_froms,
6968 "asfrom_froms": correlate_froms,
6969 "selectable": delete_stmt,
6970 }
6971 )
6973 text = "DELETE "
6975 if delete_stmt._prefixes:
6976 text += self._generate_prefixes(
6977 delete_stmt, delete_stmt._prefixes, **kw
6978 )
6980 text += "FROM "
6982 try:
6983 table_text = self.delete_table_clause(
6984 delete_stmt,
6985 delete_stmt.table,
6986 extra_froms,
6987 from_linter=from_linter,
6988 )
6989 except TypeError:
6990 # anticipate 3rd party dialects that don't include **kw
6991 # TODO: remove in 2.1
6992 table_text = self.delete_table_clause(
6993 delete_stmt, delete_stmt.table, extra_froms
6994 )
6995 if from_linter:
6996 _ = self.process(delete_stmt.table, from_linter=from_linter)
6998 crud._get_crud_params(self, delete_stmt, compile_state, toplevel, **kw)
7000 if delete_stmt._hints:
7001 dialect_hints, table_text = self._setup_crud_hints(
7002 delete_stmt, table_text
7003 )
7004 else:
7005 dialect_hints = None
7007 if delete_stmt._independent_ctes:
7008 self._dispatch_independent_ctes(delete_stmt, kw)
7010 text += table_text
7012 if (
7013 self.implicit_returning or delete_stmt._returning
7014 ) and self.returning_precedes_values:
7015 text += " " + self.returning_clause(
7016 delete_stmt,
7017 self.implicit_returning or delete_stmt._returning,
7018 populate_result_map=toplevel,
7019 )
7021 if extra_froms:
7022 extra_from_text = self.delete_extra_from_clause(
7023 delete_stmt,
7024 delete_stmt.table,
7025 extra_froms,
7026 dialect_hints,
7027 from_linter=from_linter,
7028 **kw,
7029 )
7030 if extra_from_text:
7031 text += " " + extra_from_text
7033 if delete_stmt._where_criteria:
7034 t = self._generate_delimited_and_list(
7035 delete_stmt._where_criteria, from_linter=from_linter, **kw
7036 )
7037 if t:
7038 text += " WHERE " + t
7040 dlc = self.delete_post_criteria_clause(
7041 delete_stmt, from_linter=from_linter, **kw
7042 )
7043 if dlc:
7044 text += " " + dlc
7046 if (
7047 self.implicit_returning or delete_stmt._returning
7048 ) and not self.returning_precedes_values:
7049 text += " " + self.returning_clause(
7050 delete_stmt,
7051 self.implicit_returning or delete_stmt._returning,
7052 populate_result_map=toplevel,
7053 )
7055 if self.ctes:
7056 nesting_level = len(self.stack) if not toplevel else None
7057 text = self._render_cte_clause(nesting_level=nesting_level) + text
7059 if warn_linting:
7060 assert from_linter is not None
7061 from_linter.warn(stmt_type="DELETE")
7063 self.stack.pop(-1)
7065 return text
7067 def visit_savepoint(self, savepoint_stmt, **kw):
7068 return "SAVEPOINT %s" % self.preparer.format_savepoint(savepoint_stmt)
7070 def visit_rollback_to_savepoint(self, savepoint_stmt, **kw):
7071 return "ROLLBACK TO SAVEPOINT %s" % self.preparer.format_savepoint(
7072 savepoint_stmt
7073 )
7075 def visit_release_savepoint(self, savepoint_stmt, **kw):
7076 return "RELEASE SAVEPOINT %s" % self.preparer.format_savepoint(
7077 savepoint_stmt
7078 )
7081class StrSQLCompiler(SQLCompiler):
7082 """A :class:`.SQLCompiler` subclass which allows a small selection
7083 of non-standard SQL features to render into a string value.
7085 The :class:`.StrSQLCompiler` is invoked whenever a Core expression
7086 element is directly stringified without calling upon the
7087 :meth:`_expression.ClauseElement.compile` method.
7088 It can render a limited set
7089 of non-standard SQL constructs to assist in basic stringification,
7090 however for more substantial custom or dialect-specific SQL constructs,
7091 it will be necessary to make use of
7092 :meth:`_expression.ClauseElement.compile`
7093 directly.
7095 .. seealso::
7097 :ref:`faq_sql_expression_string`
7099 """
7101 def get_select_precolumns(self, select: Select[Any], **kw: Any) -> str:
7102 return "DISTINCT " if select._distinct else ""
7104 def _fallback_column_name(self, column):
7105 return "<name unknown>"
7107 @util.preload_module("sqlalchemy.engine.url")
7108 def visit_unsupported_compilation(self, element, err, **kw):
7109 if element.stringify_dialect != "default":
7110 url = util.preloaded.engine_url
7111 dialect = url.URL.create(element.stringify_dialect).get_dialect()()
7113 compiler = dialect.statement_compiler(
7114 dialect, None, _supporting_against=self
7115 )
7116 if not isinstance(compiler, StrSQLCompiler):
7117 return compiler.process(element, **kw)
7119 return super().visit_unsupported_compilation(element, err)
7121 def visit_getitem_binary(self, binary, operator, **kw):
7122 return "%s[%s]" % (
7123 self.process(binary.left, **kw),
7124 self.process(binary.right, **kw),
7125 )
7127 def visit_json_getitem_op_binary(self, binary, operator, **kw):
7128 return self.visit_getitem_binary(binary, operator, **kw)
7130 def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
7131 return self.visit_getitem_binary(binary, operator, **kw)
7133 def visit_sequence(self, sequence, **kw):
7134 return (
7135 f"<next sequence value: {self.preparer.format_sequence(sequence)}>"
7136 )
7138 def returning_clause(
7139 self,
7140 stmt: UpdateBase,
7141 returning_cols: Sequence[_ColumnsClauseElement],
7142 *,
7143 populate_result_map: bool,
7144 **kw: Any,
7145 ) -> str:
7146 columns = [
7147 self._label_select_column(None, c, True, False, {})
7148 for c in base._select_iterables(returning_cols)
7149 ]
7150 return "RETURNING " + ", ".join(columns)
7152 def update_from_clause(
7153 self, update_stmt, from_table, extra_froms, from_hints, **kw
7154 ):
7155 kw["asfrom"] = True
7156 return "FROM " + ", ".join(
7157 t._compiler_dispatch(self, fromhints=from_hints, **kw)
7158 for t in extra_froms
7159 )
7161 def delete_extra_from_clause(
7162 self, delete_stmt, from_table, extra_froms, from_hints, **kw
7163 ):
7164 kw["asfrom"] = True
7165 return ", " + ", ".join(
7166 t._compiler_dispatch(self, fromhints=from_hints, **kw)
7167 for t in extra_froms
7168 )
7170 def visit_empty_set_expr(self, element_types, **kw):
7171 return "SELECT 1 WHERE 1!=1"
7173 def get_from_hint_text(self, table, text):
7174 return "[%s]" % text
7176 def visit_aggregate_strings_func(self, fn, **kw):
7177 return super().visit_aggregate_strings_func(
7178 fn, use_function_name=fn.name, **kw
7179 )
7181 def visit_regexp_match_op_binary(self, binary, operator, **kw):
7182 return self._generate_generic_binary(binary, " <regexp> ", **kw)
7184 def visit_not_regexp_match_op_binary(self, binary, operator, **kw):
7185 return self._generate_generic_binary(binary, " <not regexp> ", **kw)
7187 def visit_regexp_replace_op_binary(self, binary, operator, **kw):
7188 return "<regexp replace>(%s, %s)" % (
7189 binary.left._compiler_dispatch(self, **kw),
7190 binary.right._compiler_dispatch(self, **kw),
7191 )
7193 def visit_try_cast(self, cast, **kwargs):
7194 return "TRY_CAST(%s AS %s)" % (
7195 cast.clause._compiler_dispatch(self, **kwargs),
7196 cast.typeclause._compiler_dispatch(self, **kwargs),
7197 )
7200class DDLCompiler(Compiled):
7201 is_ddl = True
7203 if TYPE_CHECKING:
7205 def __init__(
7206 self,
7207 dialect: Dialect,
7208 statement: ExecutableDDLElement,
7209 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
7210 render_schema_translate: bool = ...,
7211 compile_kwargs: Mapping[str, Any] = ...,
7212 ): ...
7214 @util.ro_memoized_property
7215 def sql_compiler(self) -> SQLCompiler:
7216 return self.dialect.statement_compiler(
7217 self.dialect, None, schema_translate_map=self.schema_translate_map
7218 )
7220 @util.memoized_property
7221 def type_compiler(self):
7222 return self.dialect.type_compiler_instance
7224 def construct_params(
7225 self,
7226 params: Optional[_CoreSingleExecuteParams] = None,
7227 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
7228 escape_names: bool = True,
7229 ) -> Optional[_MutableCoreSingleExecuteParams]:
7230 return None
7232 def visit_ddl(self, ddl, **kwargs):
7233 # table events can substitute table and schema name
7234 context = ddl.context
7235 if isinstance(ddl.target, schema.Table):
7236 context = context.copy()
7238 preparer = self.preparer
7239 path = preparer.format_table_seq(ddl.target)
7240 if len(path) == 1:
7241 table, sch = path[0], ""
7242 else:
7243 table, sch = path[-1], path[0]
7245 context.setdefault("table", table)
7246 context.setdefault("schema", sch)
7247 context.setdefault("fullname", preparer.format_table(ddl.target))
7249 return self.sql_compiler.post_process_text(ddl.statement % context)
7251 def visit_create_schema(self, create, **kw):
7252 text = "CREATE SCHEMA "
7253 if create.if_not_exists:
7254 text += "IF NOT EXISTS "
7255 return text + self.preparer.format_schema(create.element)
7257 def visit_drop_schema(self, drop, **kw):
7258 text = "DROP SCHEMA "
7259 if drop.if_exists:
7260 text += "IF EXISTS "
7261 text += self.preparer.format_schema(drop.element)
7262 if drop.cascade:
7263 text += " CASCADE"
7264 return text
7266 def visit_create_table(self, create, **kw):
7267 table = create.element
7268 preparer = self.preparer
7270 text = "\nCREATE "
7271 if table._prefixes:
7272 text += " ".join(table._prefixes) + " "
7274 text += "TABLE "
7275 if create.if_not_exists:
7276 text += "IF NOT EXISTS "
7278 text += preparer.format_table(table) + " "
7280 create_table_suffix = self.create_table_suffix(table)
7281 if create_table_suffix:
7282 text += create_table_suffix + " "
7284 text += "("
7286 separator = "\n"
7288 # if only one primary key, specify it along with the column
7289 first_pk = False
7290 for create_column in create.columns:
7291 column = create_column.element
7292 try:
7293 processed = self.process(
7294 create_column, first_pk=column.primary_key and not first_pk
7295 )
7296 if processed is not None:
7297 text += separator
7298 separator = ", \n"
7299 text += "\t" + processed
7300 if column.primary_key:
7301 first_pk = True
7302 except exc.CompileError as ce:
7303 raise exc.CompileError(
7304 "(in table '%s', column '%s'): %s"
7305 % (table.description, column.name, ce.args[0])
7306 ) from ce
7308 const = self.create_table_constraints(
7309 table,
7310 _include_foreign_key_constraints=create.include_foreign_key_constraints, # noqa
7311 )
7312 if const:
7313 text += separator + "\t" + const
7315 text += "\n)%s\n\n" % self.post_create_table(table)
7316 return text
7318 def visit_create_view(self, element: CreateView, **kw: Any) -> str:
7319 return self._generate_table_select(element, "view", **kw)
7321 def visit_create_table_as(self, element: CreateTableAs, **kw: Any) -> str:
7322 return self._generate_table_select(element, "create_table_as", **kw)
7324 def create_table_select_suffixes(
7325 self,
7326 element: _TableViaSelect,
7327 type_: str,
7328 **kw: Any,
7329 ) -> str:
7330 return ""
7332 def _generate_table_select(
7333 self,
7334 element: _TableViaSelect,
7335 type_: str,
7336 if_not_exists: Optional[bool] = None,
7337 **kw: Any,
7338 ) -> str:
7339 prep = self.preparer
7341 inner_kw = dict(kw)
7342 inner_kw["literal_binds"] = True
7343 select_sql = self.sql_compiler.process(element.selectable, **inner_kw)
7345 # Use if_not_exists parameter if provided, otherwise use element's
7346 use_if_not_exists = (
7347 if_not_exists
7348 if if_not_exists is not None
7349 else element.if_not_exists
7350 )
7352 parts: List[Optional[str]] = [
7353 "CREATE",
7354 "OR REPLACE" if getattr(element, "or_replace", False) else None,
7355 "TEMPORARY" if element.temporary else None,
7356 (
7357 "MATERIALIZED VIEW"
7358 if type_ == "view" and getattr(element, "materialized", False)
7359 else "TABLE" if type_ == "create_table_as" else "VIEW"
7360 ),
7361 "IF NOT EXISTS" if use_if_not_exists else None,
7362 prep.format_table(element.table),
7363 ]
7364 suffixes = self.create_table_select_suffixes(element, type_, **kw)
7365 if suffixes:
7366 parts.append(suffixes)
7367 parts += ["AS", select_sql]
7368 return " ".join(p for p in parts if p)
7370 def visit_create_column(self, create, first_pk=False, **kw):
7371 column = create.element
7373 if column.system:
7374 return None
7376 text = self.get_column_specification(column, first_pk=first_pk)
7377 const = " ".join(
7378 self.process(constraint) for constraint in column.constraints
7379 )
7380 if const:
7381 text += " " + const
7383 return text
7385 def create_table_constraints(
7386 self, table, _include_foreign_key_constraints=None, **kw
7387 ):
7388 # On some DB order is significant: visit PK first, then the
7389 # other constraints (engine.ReflectionTest.testbasic failed on FB2)
7390 constraints = []
7391 if table.primary_key:
7392 constraints.append(table.primary_key)
7394 all_fkcs = table.foreign_key_constraints
7395 if _include_foreign_key_constraints is not None:
7396 omit_fkcs = all_fkcs.difference(_include_foreign_key_constraints)
7397 else:
7398 omit_fkcs = set()
7400 constraints.extend(
7401 [
7402 c
7403 for c in table._sorted_constraints
7404 if c is not table.primary_key and c not in omit_fkcs
7405 ]
7406 )
7408 return ", \n\t".join(
7409 p
7410 for p in (
7411 self.process(constraint)
7412 for constraint in constraints
7413 if (constraint._should_create_for_compiler(self))
7414 and (
7415 not self.dialect.supports_alter
7416 or not getattr(constraint, "use_alter", False)
7417 )
7418 )
7419 if p is not None
7420 )
7422 def visit_drop_table(self, drop, **kw):
7423 text = "\nDROP TABLE "
7424 if drop.if_exists:
7425 text += "IF EXISTS "
7426 return text + self.preparer.format_table(drop.element)
7428 def visit_drop_view(self, drop, **kw):
7429 text = "\nDROP "
7430 if drop.materialized:
7431 text += "MATERIALIZED VIEW "
7432 else:
7433 text += "VIEW "
7434 if drop.if_exists:
7435 text += "IF EXISTS "
7436 return text + self.preparer.format_table(drop.element)
7438 def _verify_index_table(self, index: Index) -> None:
7439 if index.table is None:
7440 raise exc.CompileError(
7441 "Index '%s' is not associated with any table." % index.name
7442 )
7444 def visit_create_index(
7445 self, create, include_schema=False, include_table_schema=True, **kw
7446 ):
7447 index = create.element
7448 self._verify_index_table(index)
7449 preparer = self.preparer
7450 text = "CREATE "
7451 if index.unique:
7452 text += "UNIQUE "
7453 if index.name is None:
7454 raise exc.CompileError(
7455 "CREATE INDEX requires that the index have a name"
7456 )
7458 text += "INDEX "
7459 if create.if_not_exists:
7460 text += "IF NOT EXISTS "
7462 text += "%s ON %s (%s)" % (
7463 self._prepared_index_name(index, include_schema=include_schema),
7464 preparer.format_table(
7465 index.table, use_schema=include_table_schema
7466 ),
7467 ", ".join(
7468 self.sql_compiler.process(
7469 expr, include_table=False, literal_binds=True
7470 )
7471 for expr in index.expressions
7472 ),
7473 )
7474 return text
7476 def visit_drop_index(self, drop, **kw):
7477 index = drop.element
7479 if index.name is None:
7480 raise exc.CompileError(
7481 "DROP INDEX requires that the index have a name"
7482 )
7483 text = "\nDROP INDEX "
7484 if drop.if_exists:
7485 text += "IF EXISTS "
7487 return text + self._prepared_index_name(index, include_schema=True)
7489 def _prepared_index_name(
7490 self, index: Index, include_schema: bool = False
7491 ) -> str:
7492 if index.table is not None:
7493 effective_schema = self.preparer.schema_for_object(index.table)
7494 else:
7495 effective_schema = None
7496 if include_schema and effective_schema:
7497 schema_name = self.preparer.quote_schema(effective_schema)
7498 else:
7499 schema_name = None
7501 index_name: str = self.preparer.format_index(index)
7503 if schema_name:
7504 index_name = schema_name + "." + index_name
7505 return index_name
7507 def visit_add_constraint(self, create, **kw):
7508 return "ALTER TABLE %s ADD %s" % (
7509 self.preparer.format_table(create.element.table),
7510 self.process(create.element),
7511 )
7513 def visit_set_table_comment(self, create, **kw):
7514 return "COMMENT ON TABLE %s IS %s" % (
7515 self.preparer.format_table(create.element),
7516 self.sql_compiler.render_literal_value(
7517 create.element.comment, sqltypes.String()
7518 ),
7519 )
7521 def visit_drop_table_comment(self, drop, **kw):
7522 return "COMMENT ON TABLE %s IS NULL" % self.preparer.format_table(
7523 drop.element
7524 )
7526 def visit_set_column_comment(self, create, **kw):
7527 return "COMMENT ON COLUMN %s IS %s" % (
7528 self.preparer.format_column(
7529 create.element, use_table=True, use_schema=True
7530 ),
7531 self.sql_compiler.render_literal_value(
7532 create.element.comment, sqltypes.String()
7533 ),
7534 )
7536 def visit_drop_column_comment(self, drop, **kw):
7537 return "COMMENT ON COLUMN %s IS NULL" % self.preparer.format_column(
7538 drop.element, use_table=True
7539 )
7541 def visit_set_constraint_comment(self, create, **kw):
7542 raise exc.UnsupportedCompilationError(self, type(create))
7544 def visit_drop_constraint_comment(self, drop, **kw):
7545 raise exc.UnsupportedCompilationError(self, type(drop))
7547 def get_identity_options(self, identity_options: IdentityOptions) -> str:
7548 text = []
7549 if identity_options.increment is not None:
7550 text.append("INCREMENT BY %d" % identity_options.increment)
7551 if identity_options.start is not None:
7552 text.append("START WITH %d" % identity_options.start)
7553 if identity_options.minvalue is not None:
7554 text.append("MINVALUE %d" % identity_options.minvalue)
7555 if identity_options.maxvalue is not None:
7556 text.append("MAXVALUE %d" % identity_options.maxvalue)
7557 if identity_options.nominvalue is not None:
7558 text.append("NO MINVALUE")
7559 if identity_options.nomaxvalue is not None:
7560 text.append("NO MAXVALUE")
7561 if identity_options.cache is not None:
7562 text.append("CACHE %d" % identity_options.cache)
7563 if identity_options.cycle is not None:
7564 text.append("CYCLE" if identity_options.cycle else "NO CYCLE")
7565 return " ".join(text)
7567 def visit_create_sequence(self, create, prefix=None, **kw):
7568 text = "CREATE SEQUENCE "
7569 if create.if_not_exists:
7570 text += "IF NOT EXISTS "
7571 text += self.preparer.format_sequence(create.element)
7573 if prefix:
7574 text += prefix
7575 options = self.get_identity_options(create.element)
7576 if options:
7577 text += " " + options
7578 return text
7580 def visit_drop_sequence(self, drop, **kw):
7581 text = "DROP SEQUENCE "
7582 if drop.if_exists:
7583 text += "IF EXISTS "
7584 return text + self.preparer.format_sequence(drop.element)
7586 def visit_drop_constraint(self, drop, **kw):
7587 constraint = drop.element
7588 if constraint.name is not None:
7589 formatted_name = self.preparer.format_constraint(constraint)
7590 else:
7591 formatted_name = None
7593 if formatted_name is None:
7594 raise exc.CompileError(
7595 "Can't emit DROP CONSTRAINT for constraint %r; "
7596 "it has no name" % drop.element
7597 )
7598 return "ALTER TABLE %s DROP CONSTRAINT %s%s%s" % (
7599 self.preparer.format_table(drop.element.table),
7600 "IF EXISTS " if drop.if_exists else "",
7601 formatted_name,
7602 " CASCADE" if drop.cascade else "",
7603 )
7605 def get_column_specification(
7606 self, column: Column[Any], **kwargs: Any
7607 ) -> str:
7608 colspec = (
7609 self.preparer.format_column(column)
7610 + " "
7611 + self.dialect.type_compiler_instance.process(
7612 column.type, type_expression=column
7613 )
7614 )
7615 default = self.get_column_default_string(column)
7616 if default is not None:
7617 colspec += " DEFAULT " + default
7619 if column.computed is not None:
7620 colspec += " " + self.process(column.computed)
7622 if (
7623 column.identity is not None
7624 and self.dialect.supports_identity_columns
7625 ):
7626 colspec += " " + self.process(column.identity)
7628 if not column.nullable and (
7629 not column.identity or not self.dialect.supports_identity_columns
7630 ):
7631 colspec += " NOT NULL"
7632 return colspec
7634 def create_table_suffix(self, table: Table) -> str:
7635 return ""
7637 def post_create_table(self, table: Table) -> str:
7638 return ""
7640 def get_column_default_string(self, column: Column[Any]) -> Optional[str]:
7641 if isinstance(column.server_default, schema.DefaultClause):
7642 return self.render_default_string(column.server_default.arg)
7643 else:
7644 return None
7646 def render_default_string(self, default: Union[Visitable, str]) -> str:
7647 if isinstance(default, str):
7648 return self.sql_compiler.render_literal_value(
7649 default, sqltypes.STRINGTYPE
7650 )
7651 else:
7652 return self.sql_compiler.process(default, literal_binds=True)
7654 def visit_table_or_column_check_constraint(self, constraint, **kw):
7655 if constraint.is_column_level:
7656 return self.visit_column_check_constraint(constraint)
7657 else:
7658 return self.visit_check_constraint(constraint)
7660 def visit_check_constraint(self, constraint, **kw):
7661 text = self.define_constraint_preamble(constraint, **kw)
7662 text += self.define_check_body(constraint, **kw)
7663 text += self.define_constraint_deferrability(constraint)
7664 return text
7666 def visit_column_check_constraint(self, constraint, **kw):
7667 text = self.define_constraint_preamble(constraint, **kw)
7668 text += self.define_check_body(constraint, **kw)
7669 text += self.define_constraint_deferrability(constraint)
7670 return text
7672 def visit_primary_key_constraint(
7673 self, constraint: PrimaryKeyConstraint, **kw: Any
7674 ) -> str:
7675 if len(constraint) == 0:
7676 return ""
7677 text = self.define_constraint_preamble(constraint, **kw)
7678 text += self.define_primary_key_body(constraint, **kw)
7679 text += self.define_constraint_deferrability(constraint)
7680 return text
7682 def visit_foreign_key_constraint(
7683 self, constraint: ForeignKeyConstraint, **kw: Any
7684 ) -> str:
7685 text = self.define_constraint_preamble(constraint, **kw)
7686 text += self.define_foreign_key_body(constraint, **kw)
7687 text += self.define_constraint_match(constraint)
7688 text += self.define_constraint_cascades(constraint)
7689 text += self.define_constraint_deferrability(constraint)
7690 return text
7692 def define_constraint_remote_table(self, constraint, table, preparer):
7693 """Format the remote table clause of a CREATE CONSTRAINT clause."""
7695 return preparer.format_table(table)
7697 def visit_unique_constraint(
7698 self, constraint: UniqueConstraint, **kw: Any
7699 ) -> str:
7700 if len(constraint) == 0:
7701 return ""
7702 text = self.define_constraint_preamble(constraint, **kw)
7703 text += self.define_unique_body(constraint, **kw)
7704 text += self.define_constraint_deferrability(constraint)
7705 return text
7707 def define_constraint_preamble(
7708 self, constraint: Constraint, **kw: Any
7709 ) -> str:
7710 text = ""
7711 if constraint.name is not None:
7712 formatted_name = self.preparer.format_constraint(constraint)
7713 if formatted_name is not None:
7714 text += "CONSTRAINT %s " % formatted_name
7715 return text
7717 def define_primary_key_body(
7718 self, constraint: PrimaryKeyConstraint, **kw: Any
7719 ) -> str:
7720 text = ""
7721 text += "PRIMARY KEY "
7722 text += "(%s)" % ", ".join(
7723 self.preparer.quote(c.name)
7724 for c in (
7725 constraint.columns_autoinc_first
7726 if constraint._implicit_generated
7727 else constraint.columns
7728 )
7729 )
7730 return text
7732 def define_foreign_key_body(
7733 self, constraint: ForeignKeyConstraint, **kw: Any
7734 ) -> str:
7735 preparer = self.preparer
7736 remote_table = list(constraint.elements)[0].column.table
7737 text = "FOREIGN KEY(%s) REFERENCES %s (%s)" % (
7738 ", ".join(
7739 preparer.quote(f.parent.name) for f in constraint.elements
7740 ),
7741 self.define_constraint_remote_table(
7742 constraint, remote_table, preparer
7743 ),
7744 ", ".join(
7745 preparer.quote(f.column.name) for f in constraint.elements
7746 ),
7747 )
7748 return text
7750 def define_unique_body(
7751 self, constraint: UniqueConstraint, **kw: Any
7752 ) -> str:
7753 text = "UNIQUE %s(%s)" % (
7754 self.define_unique_constraint_distinct(constraint, **kw),
7755 ", ".join(self.preparer.quote(c.name) for c in constraint),
7756 )
7757 return text
7759 def define_check_body(self, constraint: CheckConstraint, **kw: Any) -> str:
7760 text = "CHECK (%s)" % self.sql_compiler.process(
7761 constraint.sqltext, include_table=False, literal_binds=True
7762 )
7763 return text
7765 def define_unique_constraint_distinct(
7766 self, constraint: UniqueConstraint, **kw: Any
7767 ) -> str:
7768 return ""
7770 def define_constraint_cascades(
7771 self, constraint: ForeignKeyConstraint
7772 ) -> str:
7773 text = ""
7774 if constraint.ondelete is not None:
7775 text += self.define_constraint_ondelete_cascade(constraint)
7777 if constraint.onupdate is not None:
7778 text += self.define_constraint_onupdate_cascade(constraint)
7779 return text
7781 def define_constraint_ondelete_cascade(
7782 self, constraint: ForeignKeyConstraint
7783 ) -> str:
7784 return " ON DELETE %s" % self.preparer.validate_sql_phrase(
7785 constraint.ondelete, FK_ON_DELETE
7786 )
7788 def define_constraint_onupdate_cascade(
7789 self, constraint: ForeignKeyConstraint
7790 ) -> str:
7791 return " ON UPDATE %s" % self.preparer.validate_sql_phrase(
7792 constraint.onupdate, FK_ON_UPDATE
7793 )
7795 def define_constraint_deferrability(self, constraint: Constraint) -> str:
7796 text = ""
7797 if constraint.deferrable is not None:
7798 if constraint.deferrable:
7799 text += " DEFERRABLE"
7800 else:
7801 text += " NOT DEFERRABLE"
7802 if constraint.initially is not None:
7803 text += " INITIALLY %s" % self.preparer.validate_sql_phrase(
7804 constraint.initially, FK_INITIALLY
7805 )
7806 return text
7808 def define_constraint_match(self, constraint: ForeignKeyConstraint) -> str:
7809 text = ""
7810 if constraint.match is not None:
7811 text += " MATCH %s" % constraint.match
7812 return text
7814 def visit_computed_column(self, generated, **kw):
7815 text = "GENERATED ALWAYS AS (%s)" % self.sql_compiler.process(
7816 generated.sqltext, include_table=False, literal_binds=True
7817 )
7818 if generated.persisted is True:
7819 text += " STORED"
7820 elif generated.persisted is False:
7821 text += " VIRTUAL"
7822 return text
7824 def visit_identity_column(self, identity, **kw):
7825 text = "GENERATED %s AS IDENTITY" % (
7826 "ALWAYS" if identity.always else "BY DEFAULT",
7827 )
7828 options = self.get_identity_options(identity)
7829 if options:
7830 text += " (%s)" % options
7831 return text
7834class GenericTypeCompiler(TypeCompiler):
7835 def visit_FLOAT(self, type_: sqltypes.Float[Any], **kw: Any) -> str:
7836 return "FLOAT"
7838 def visit_DOUBLE(self, type_: sqltypes.Double[Any], **kw: Any) -> str:
7839 return "DOUBLE"
7841 def visit_DOUBLE_PRECISION(
7842 self, type_: sqltypes.DOUBLE_PRECISION[Any], **kw: Any
7843 ) -> str:
7844 return "DOUBLE PRECISION"
7846 def visit_REAL(self, type_: sqltypes.REAL[Any], **kw: Any) -> str:
7847 return "REAL"
7849 def visit_NUMERIC(self, type_: sqltypes.Numeric[Any], **kw: Any) -> str:
7850 if type_.precision is None:
7851 return "NUMERIC"
7852 elif type_.scale is None:
7853 return "NUMERIC(%(precision)s)" % {"precision": type_.precision}
7854 else:
7855 return "NUMERIC(%(precision)s, %(scale)s)" % {
7856 "precision": type_.precision,
7857 "scale": type_.scale,
7858 }
7860 def visit_DECIMAL(self, type_: sqltypes.DECIMAL[Any], **kw: Any) -> str:
7861 if type_.precision is None:
7862 return "DECIMAL"
7863 elif type_.scale is None:
7864 return "DECIMAL(%(precision)s)" % {"precision": type_.precision}
7865 else:
7866 return "DECIMAL(%(precision)s, %(scale)s)" % {
7867 "precision": type_.precision,
7868 "scale": type_.scale,
7869 }
7871 def visit_INTEGER(self, type_: sqltypes.Integer, **kw: Any) -> str:
7872 return "INTEGER"
7874 def visit_SMALLINT(self, type_: sqltypes.SmallInteger, **kw: Any) -> str:
7875 return "SMALLINT"
7877 def visit_BIGINT(self, type_: sqltypes.BigInteger, **kw: Any) -> str:
7878 return "BIGINT"
7880 def visit_TIMESTAMP(self, type_: sqltypes.TIMESTAMP, **kw: Any) -> str:
7881 return "TIMESTAMP"
7883 def visit_DATETIME(self, type_: sqltypes.DateTime, **kw: Any) -> str:
7884 return "DATETIME"
7886 def visit_DATE(self, type_: sqltypes.Date, **kw: Any) -> str:
7887 return "DATE"
7889 def visit_TIME(self, type_: sqltypes.Time, **kw: Any) -> str:
7890 return "TIME"
7892 def visit_CLOB(self, type_: sqltypes.CLOB, **kw: Any) -> str:
7893 return "CLOB"
7895 def visit_NCLOB(self, type_: sqltypes.Text, **kw: Any) -> str:
7896 return "NCLOB"
7898 def _render_string_type(
7899 self,
7900 name: str,
7901 length: Optional[int],
7902 collation: Optional[str],
7903 collation_schema: Optional[str] = None,
7904 identifier_preparer: Optional[IdentifierPreparer] = None,
7905 **kw: Any,
7906 ) -> str:
7907 text = name
7908 if length:
7909 text += f"({length})"
7910 if collation:
7911 if identifier_preparer is None:
7912 identifier_preparer = self.dialect.identifier_preparer
7913 text += " COLLATE " + identifier_preparer.format_collation(
7914 collation, collation_schema
7915 )
7916 return text
7918 def visit_CHAR(self, type_: sqltypes.CHAR, **kw: Any) -> str:
7919 return self._render_string_type(
7920 "CHAR",
7921 type_.length,
7922 type_.collation,
7923 type_.collation_schema,
7924 **kw,
7925 )
7927 def visit_NCHAR(self, type_: sqltypes.NCHAR, **kw: Any) -> str:
7928 return self._render_string_type(
7929 "NCHAR",
7930 type_.length,
7931 type_.collation,
7932 type_.collation_schema,
7933 **kw,
7934 )
7936 def visit_VARCHAR(self, type_: sqltypes.String, **kw: Any) -> str:
7937 return self._render_string_type(
7938 "VARCHAR",
7939 type_.length,
7940 type_.collation,
7941 type_.collation_schema,
7942 **kw,
7943 )
7945 def visit_NVARCHAR(self, type_: sqltypes.NVARCHAR, **kw: Any) -> str:
7946 return self._render_string_type(
7947 "NVARCHAR",
7948 type_.length,
7949 type_.collation,
7950 type_.collation_schema,
7951 **kw,
7952 )
7954 def visit_TEXT(self, type_: sqltypes.Text, **kw: Any) -> str:
7955 return self._render_string_type(
7956 "TEXT",
7957 type_.length,
7958 type_.collation,
7959 type_.collation_schema,
7960 **kw,
7961 )
7963 def visit_UUID(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
7964 return "UUID"
7966 def visit_BLOB(self, type_: sqltypes.LargeBinary, **kw: Any) -> str:
7967 return "BLOB"
7969 def visit_BINARY(self, type_: sqltypes.BINARY, **kw: Any) -> str:
7970 return "BINARY" + (type_.length and "(%d)" % type_.length or "")
7972 def visit_VARBINARY(self, type_: sqltypes.VARBINARY, **kw: Any) -> str:
7973 return "VARBINARY" + (type_.length and "(%d)" % type_.length or "")
7975 def visit_BOOLEAN(self, type_: sqltypes.Boolean, **kw: Any) -> str:
7976 return "BOOLEAN"
7978 def visit_uuid(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
7979 if not type_.native_uuid or not self.dialect.supports_native_uuid:
7980 return self._render_string_type(
7981 "CHAR",
7982 length=32,
7983 collation=None,
7984 collation_schema=None,
7985 **kw,
7986 )
7987 else:
7988 return self.visit_UUID(type_, **kw)
7990 def visit_large_binary(
7991 self, type_: sqltypes.LargeBinary, **kw: Any
7992 ) -> str:
7993 return self.visit_BLOB(type_, **kw)
7995 def visit_boolean(self, type_: sqltypes.Boolean, **kw: Any) -> str:
7996 return self.visit_BOOLEAN(type_, **kw)
7998 def visit_time(self, type_: sqltypes.Time, **kw: Any) -> str:
7999 return self.visit_TIME(type_, **kw)
8001 def visit_datetime(self, type_: sqltypes.DateTime, **kw: Any) -> str:
8002 return self.visit_DATETIME(type_, **kw)
8004 def visit_date(self, type_: sqltypes.Date, **kw: Any) -> str:
8005 return self.visit_DATE(type_, **kw)
8007 def visit_big_integer(self, type_: sqltypes.BigInteger, **kw: Any) -> str:
8008 return self.visit_BIGINT(type_, **kw)
8010 def visit_small_integer(
8011 self, type_: sqltypes.SmallInteger, **kw: Any
8012 ) -> str:
8013 return self.visit_SMALLINT(type_, **kw)
8015 def visit_integer(self, type_: sqltypes.Integer, **kw: Any) -> str:
8016 return self.visit_INTEGER(type_, **kw)
8018 def visit_real(self, type_: sqltypes.REAL[Any], **kw: Any) -> str:
8019 return self.visit_REAL(type_, **kw)
8021 def visit_float(self, type_: sqltypes.Float[Any], **kw: Any) -> str:
8022 return self.visit_FLOAT(type_, **kw)
8024 def visit_double(self, type_: sqltypes.Double[Any], **kw: Any) -> str:
8025 return self.visit_DOUBLE(type_, **kw)
8027 def visit_numeric(self, type_: sqltypes.Numeric[Any], **kw: Any) -> str:
8028 return self.visit_NUMERIC(type_, **kw)
8030 def visit_string(self, type_: sqltypes.String, **kw: Any) -> str:
8031 return self.visit_VARCHAR(type_, **kw)
8033 def visit_unicode(self, type_: sqltypes.Unicode, **kw: Any) -> str:
8034 return self.visit_VARCHAR(type_, **kw)
8036 def visit_text(self, type_: sqltypes.Text, **kw: Any) -> str:
8037 return self.visit_TEXT(type_, **kw)
8039 def visit_unicode_text(
8040 self, type_: sqltypes.UnicodeText, **kw: Any
8041 ) -> str:
8042 return self.visit_TEXT(type_, **kw)
8044 def visit_enum(self, type_: sqltypes.Enum, **kw: Any) -> str:
8045 return self.visit_VARCHAR(type_, **kw)
8047 def visit_null(self, type_, **kw):
8048 raise exc.CompileError(
8049 "Can't generate DDL for %r; "
8050 "did you forget to specify a "
8051 "type on this Column?" % type_
8052 )
8054 def visit_type_decorator(
8055 self, type_: TypeDecorator[Any], **kw: Any
8056 ) -> str:
8057 return self.process(type_.type_engine(self.dialect), **kw)
8059 def visit_user_defined(
8060 self, type_: UserDefinedType[Any], **kw: Any
8061 ) -> str:
8062 return type_.get_col_spec(**kw)
8065class StrSQLTypeCompiler(GenericTypeCompiler):
8066 def process(self, type_, **kw):
8067 try:
8068 _compiler_dispatch = type_._compiler_dispatch
8069 except AttributeError:
8070 return self._visit_unknown(type_, **kw)
8071 else:
8072 return _compiler_dispatch(self, **kw)
8074 def __getattr__(self, key):
8075 if key.startswith("visit_"):
8076 return self._visit_unknown
8077 else:
8078 raise AttributeError(key)
8080 def _visit_unknown(self, type_, **kw):
8081 if type_.__class__.__name__ == type_.__class__.__name__.upper():
8082 return type_.__class__.__name__
8083 else:
8084 return repr(type_)
8086 def visit_null(self, type_, **kw):
8087 return "NULL"
8089 def visit_user_defined(self, type_, **kw):
8090 try:
8091 get_col_spec = type_.get_col_spec
8092 except AttributeError:
8093 return repr(type_)
8094 else:
8095 return get_col_spec(**kw)
8098class _SchemaForObjectCallable(Protocol):
8099 def __call__(self, obj: Any, /) -> str: ...
8102class _BindNameForColProtocol(Protocol):
8103 def __call__(self, col: ColumnClause[Any]) -> str: ...
8106class IdentifierPreparer:
8107 """Handle quoting and case-folding of identifiers based on options."""
8109 reserved_words = RESERVED_WORDS
8111 legal_characters = LEGAL_CHARACTERS
8113 illegal_initial_characters = ILLEGAL_INITIAL_CHARACTERS
8115 initial_quote: str
8117 final_quote: str
8119 _strings: MutableMapping[str, str]
8121 schema_for_object: _SchemaForObjectCallable = operator.attrgetter("schema")
8122 """Return the .schema attribute for an object.
8124 For the default IdentifierPreparer, the schema for an object is always
8125 the value of the ".schema" attribute. if the preparer is replaced
8126 with one that has a non-empty schema_translate_map, the value of the
8127 ".schema" attribute is rendered a symbol that will be converted to a
8128 real schema name from the mapping post-compile.
8130 """
8132 _includes_none_schema_translate: bool = False
8134 def __init__(
8135 self,
8136 dialect: Dialect,
8137 initial_quote: str = '"',
8138 final_quote: Optional[str] = None,
8139 escape_quote: str = '"',
8140 quote_case_sensitive_collations: bool = True,
8141 omit_schema: bool = False,
8142 ):
8143 """Construct a new ``IdentifierPreparer`` object.
8145 initial_quote
8146 Character that begins a delimited identifier.
8148 final_quote
8149 Character that ends a delimited identifier. Defaults to
8150 `initial_quote`.
8152 omit_schema
8153 Prevent prepending schema name. Useful for databases that do
8154 not support schemae.
8155 """
8157 self.dialect = dialect
8158 self.initial_quote = initial_quote
8159 self.final_quote = final_quote or self.initial_quote
8160 self.escape_quote = escape_quote
8161 self.escape_to_quote = self.escape_quote * 2
8162 self.omit_schema = omit_schema
8163 self.quote_case_sensitive_collations = quote_case_sensitive_collations
8164 self._strings = {}
8165 self._double_percents = self.dialect.paramstyle in (
8166 "format",
8167 "pyformat",
8168 )
8170 def _with_schema_translate(self, schema_translate_map):
8171 prep = self.__class__.__new__(self.__class__)
8172 prep.__dict__.update(self.__dict__)
8174 includes_none = None in schema_translate_map
8176 def symbol_getter(obj):
8177 name = obj.schema
8178 if obj._use_schema_map and (name is not None or includes_none):
8179 if name is not None and ("[" in name or "]" in name):
8180 raise exc.CompileError(
8181 "Square bracket characters ([]) not supported "
8182 "in schema translate name '%s'" % name
8183 )
8184 return quoted_name(
8185 "__[SCHEMA_%s]" % (name or "_none"), quote=False
8186 )
8187 else:
8188 return obj.schema
8190 prep.schema_for_object = symbol_getter
8191 prep._includes_none_schema_translate = includes_none
8192 return prep
8194 def _render_schema_translates(
8195 self, statement: str, schema_translate_map: SchemaTranslateMapType
8196 ) -> str:
8197 d = schema_translate_map
8198 if None in d:
8199 if not self._includes_none_schema_translate:
8200 raise exc.InvalidRequestError(
8201 "schema translate map which previously did not have "
8202 "`None` present as a key now has `None` present; compiled "
8203 "statement may lack adequate placeholders. Please use "
8204 "consistent keys in successive "
8205 "schema_translate_map dictionaries."
8206 )
8208 d["_none"] = d[None] # type: ignore[index]
8210 def replace(m):
8211 name = m.group(2)
8212 if name in d:
8213 effective_schema = d[name]
8214 else:
8215 if name in (None, "_none"):
8216 raise exc.InvalidRequestError(
8217 "schema translate map which previously had `None` "
8218 "present as a key now no longer has it present; don't "
8219 "know how to apply schema for compiled statement. "
8220 "Please use consistent keys in successive "
8221 "schema_translate_map dictionaries."
8222 )
8223 effective_schema = name
8225 if not effective_schema:
8226 effective_schema = self.dialect.default_schema_name
8227 if not effective_schema:
8228 # TODO: no coverage here
8229 raise exc.CompileError(
8230 "Dialect has no default schema name; can't "
8231 "use None as dynamic schema target."
8232 )
8233 return self.quote_schema(effective_schema)
8235 return re.sub(r"(__\[SCHEMA_([^\]]+)\])", replace, statement)
8237 def _escape_identifier(self, value: str) -> str:
8238 """Escape an identifier.
8240 Subclasses should override this to provide database-dependent
8241 escaping behavior.
8242 """
8244 value = value.replace(self.escape_quote, self.escape_to_quote)
8245 if self._double_percents:
8246 value = value.replace("%", "%%")
8247 return value
8249 def _unescape_identifier(self, value: str) -> str:
8250 """Canonicalize an escaped identifier.
8252 Subclasses should override this to provide database-dependent
8253 unescaping behavior that reverses _escape_identifier.
8254 """
8256 return value.replace(self.escape_to_quote, self.escape_quote)
8258 def validate_sql_phrase(self, element, reg):
8259 """keyword sequence filter.
8261 a filter for elements that are intended to represent keyword sequences,
8262 such as "INITIALLY", "INITIALLY DEFERRED", etc. no special characters
8263 should be present.
8265 """
8267 if element is not None and not reg.match(element):
8268 raise exc.CompileError(
8269 "Unexpected SQL phrase: %r (matching against %r)"
8270 % (element, reg.pattern)
8271 )
8272 return element
8274 def quote_identifier(self, value: str) -> str:
8275 """Quote an identifier.
8277 Subclasses should override this to provide database-dependent
8278 quoting behavior.
8279 """
8281 return (
8282 self.initial_quote
8283 + self._escape_identifier(value)
8284 + self.final_quote
8285 )
8287 def _requires_quotes(self, value: str) -> bool:
8288 """Return True if the given identifier requires quoting."""
8289 if not value:
8290 # a blank name is legal on SQLite only, where it can be
8291 # delivered by reflection; quote it so that it renders as the
8292 # database has it, rather than indexing into an empty string
8293 # below
8294 return True
8295 lc_value = value.lower()
8296 return (
8297 lc_value in self.reserved_words
8298 or value[0] in self.illegal_initial_characters
8299 or not self.legal_characters.match(str(value))
8300 or (lc_value != value)
8301 )
8303 def _requires_quotes_illegal_chars(self, value):
8304 """Return True if the given identifier requires quoting, but
8305 not taking case convention into account."""
8306 return not self.legal_characters.match(str(value))
8308 def quote_schema(self, schema: str) -> str:
8309 """Conditionally quote a schema name.
8312 The name is quoted if it is a reserved word, contains quote-necessary
8313 characters, or is an instance of :class:`.quoted_name` which includes
8314 ``quote`` set to ``True``.
8316 Subclasses can override this to provide database-dependent
8317 quoting behavior for schema names.
8319 :param schema: string schema name
8320 """
8321 return self.quote(schema)
8323 def quote(self, ident: str) -> str:
8324 """Conditionally quote an identifier.
8326 The identifier is quoted if it is a reserved word, contains
8327 quote-necessary characters, or is an instance of
8328 :class:`.quoted_name` which includes ``quote`` set to ``True``.
8330 Subclasses can override this to provide database-dependent
8331 quoting behavior for identifier names.
8333 :param ident: string identifier
8334 """
8335 force = getattr(ident, "quote", None)
8337 if force is None:
8338 if ident in self._strings:
8339 return self._strings[ident]
8340 else:
8341 if self._requires_quotes(ident):
8342 self._strings[ident] = self.quote_identifier(ident)
8343 else:
8344 self._strings[ident] = ident
8345 return self._strings[ident]
8346 elif force:
8347 return self.quote_identifier(ident)
8348 else:
8349 return ident
8351 def format_collation(self, collation_name, collation_schema=None):
8352 if self.quote_case_sensitive_collations:
8353 name = self.quote(collation_name)
8354 else:
8355 name = collation_name
8357 if collation_schema is not None:
8358 return f"{self.quote_schema(collation_schema)}.{name}"
8359 else:
8360 return name
8362 def format_sequence(
8363 self, sequence: schema.Sequence, use_schema: bool = True
8364 ) -> str:
8365 name = self.quote(sequence.name)
8367 effective_schema = self.schema_for_object(sequence)
8369 if (
8370 not self.omit_schema
8371 and use_schema
8372 and effective_schema is not None
8373 ):
8374 name = self.quote_schema(effective_schema) + "." + name
8375 return name
8377 def format_label(
8378 self, label: Label[Any], name: Optional[str] = None
8379 ) -> str:
8380 return self.quote(name or label.name)
8382 def format_alias(
8383 self, alias: Optional[AliasedReturnsRows], name: Optional[str] = None
8384 ) -> str:
8385 if name is None:
8386 assert alias is not None
8387 return self.quote(alias.name)
8388 else:
8389 return self.quote(name)
8391 def format_savepoint(self, savepoint, name=None):
8392 # Running the savepoint name through quoting is unnecessary
8393 # for all known dialects. This is here to support potential
8394 # third party use cases
8395 ident = name or savepoint.ident
8396 if self._requires_quotes(ident):
8397 ident = self.quote_identifier(ident)
8398 return ident
8400 @util.preload_module("sqlalchemy.sql.naming")
8401 def format_constraint(
8402 self, constraint: Union[Constraint, Index], _alembic_quote: bool = True
8403 ) -> Optional[str]:
8404 naming = util.preloaded.sql_naming
8406 if constraint.name is _NONE_NAME:
8407 name = naming._constraint_name_for_table(
8408 constraint, constraint.table
8409 )
8411 if name is None:
8412 return None
8413 else:
8414 name = constraint.name
8416 assert name is not None
8417 if constraint.__visit_name__ == "index":
8418 return self.truncate_and_render_index_name(
8419 name, _alembic_quote=_alembic_quote
8420 )
8421 else:
8422 return self.truncate_and_render_constraint_name(
8423 name, _alembic_quote=_alembic_quote
8424 )
8426 def truncate_and_render_index_name(
8427 self, name: str, _alembic_quote: bool = True
8428 ) -> str:
8429 # calculate these at format time so that ad-hoc changes
8430 # to dialect.max_identifier_length etc. can be reflected
8431 # as IdentifierPreparer is long lived
8432 max_ = (
8433 self.dialect.max_index_name_length
8434 or self.dialect.max_identifier_length
8435 )
8436 return self._truncate_and_render_maxlen_name(
8437 name, max_, _alembic_quote
8438 )
8440 def truncate_and_render_constraint_name(
8441 self, name: str, _alembic_quote: bool = True
8442 ) -> str:
8443 # calculate these at format time so that ad-hoc changes
8444 # to dialect.max_identifier_length etc. can be reflected
8445 # as IdentifierPreparer is long lived
8446 max_ = (
8447 self.dialect.max_constraint_name_length
8448 or self.dialect.max_identifier_length
8449 )
8450 return self._truncate_and_render_maxlen_name(
8451 name, max_, _alembic_quote
8452 )
8454 def _truncate_and_render_maxlen_name(
8455 self, name: str, max_: int, _alembic_quote: bool
8456 ) -> str:
8457 if isinstance(name, elements._truncated_label):
8458 if len(name) > max_:
8459 name = name[0 : max_ - 8] + "_" + util.md5_hex(name)[-4:]
8460 else:
8461 self.dialect.validate_identifier(name)
8463 if not _alembic_quote:
8464 return name
8465 else:
8466 return self.quote(name)
8468 def format_index(self, index: Index) -> str:
8469 name = self.format_constraint(index)
8470 assert name is not None
8471 return name
8473 def format_table(
8474 self,
8475 table: FromClause,
8476 use_schema: bool = True,
8477 name: Optional[str] = None,
8478 ) -> str:
8479 """Prepare a quoted table and schema name."""
8480 if name is None:
8481 if TYPE_CHECKING:
8482 assert isinstance(table, NamedFromClause)
8483 name = table.name
8485 result = self.quote(name)
8487 effective_schema = self.schema_for_object(table)
8489 if not self.omit_schema and use_schema and effective_schema:
8490 result = self.quote_schema(effective_schema) + "." + result
8491 return result
8493 def format_schema(self, name):
8494 """Prepare a quoted schema name."""
8496 return self.quote(name)
8498 def format_label_name(
8499 self,
8500 name,
8501 anon_map=None,
8502 ):
8503 """Prepare a quoted column name."""
8505 if anon_map is not None and isinstance(
8506 name, elements._truncated_label
8507 ):
8508 name = name.apply_map(anon_map)
8510 return self.quote(name)
8512 def format_column(
8513 self,
8514 column: ColumnElement[Any],
8515 use_table: bool = False,
8516 name: Optional[str] = None,
8517 table_name: Optional[str] = None,
8518 use_schema: bool = False,
8519 anon_map: Optional[Mapping[str, Any]] = None,
8520 ) -> str:
8521 """Prepare a quoted column name."""
8523 if name is None:
8524 name = column.name
8525 assert name is not None
8527 if anon_map is not None and isinstance(
8528 name, elements._truncated_label
8529 ):
8530 name = name.apply_map(anon_map)
8532 if not getattr(column, "is_literal", False):
8533 if use_table:
8534 return (
8535 self.format_table(
8536 column.table, use_schema=use_schema, name=table_name
8537 )
8538 + "."
8539 + self.quote(name)
8540 )
8541 else:
8542 return self.quote(name)
8543 else:
8544 # literal textual elements get stuck into ColumnClause a lot,
8545 # which shouldn't get quoted
8547 if use_table:
8548 return (
8549 self.format_table(
8550 column.table, use_schema=use_schema, name=table_name
8551 )
8552 + "."
8553 + name
8554 )
8555 else:
8556 return name
8558 def format_table_seq(self, table, use_schema=True):
8559 """Format table name and schema as a tuple."""
8561 # Dialects with more levels in their fully qualified references
8562 # ('database', 'owner', etc.) could override this and return
8563 # a longer sequence.
8565 effective_schema = self.schema_for_object(table)
8567 if not self.omit_schema and use_schema and effective_schema:
8568 return (
8569 self.quote_schema(effective_schema),
8570 self.format_table(table, use_schema=False),
8571 )
8572 else:
8573 return (self.format_table(table, use_schema=False),)
8575 @util.memoized_property
8576 def _r_identifiers(self):
8577 initial, final, escaped_final = (
8578 re.escape(s)
8579 for s in (
8580 self.initial_quote,
8581 self.final_quote,
8582 self._escape_identifier(self.final_quote),
8583 )
8584 )
8585 r = re.compile(
8586 r"(?:"
8587 r"(?:%(initial)s((?:%(escaped)s|[^%(final)s])+)%(final)s"
8588 r"|([^\.]+))(?=\.|$))+"
8589 % {"initial": initial, "final": final, "escaped": escaped_final}
8590 )
8591 return r
8593 def unformat_identifiers(self, identifiers: str) -> Sequence[str]:
8594 """Unpack 'schema.table.column'-like strings into components."""
8596 r = self._r_identifiers
8597 return [
8598 self._unescape_identifier(i)
8599 for i in [a or b for a, b in r.findall(identifiers)]
8600 ]