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 Any
40from typing import Callable
41from typing import cast
42from typing import ClassVar
43from typing import Dict
44from typing import Final
45from typing import FrozenSet
46from typing import Iterable
47from typing import Iterator
48from typing import List
49from typing import Literal
50from typing import Mapping
51from typing import MutableMapping
52from typing import NamedTuple
53from typing import NoReturn
54from typing import Optional
55from typing import Pattern
56from typing import Protocol
57from typing import Sequence
58from typing import Set
59from typing import Tuple
60from typing import Type
61from typing import TYPE_CHECKING
62from typing import TypedDict
63from typing import Union
65from . import base
66from . import coercions
67from . import crud
68from . import elements
69from . import functions
70from . import operators
71from . import roles
72from . import schema
73from . import selectable
74from . import sqltypes
75from . import util as sql_util
76from ._typing import is_column_element
77from ._typing import is_dml
78from .base import _de_clone
79from .base import _from_objects
80from .base import _NONE_NAME
81from .base import _SentinelDefaultCharacterization
82from .base import NO_ARG
83from .elements import quoted_name
84from .sqltypes import TupleType
85from .visitors import prefix_anon_map
86from .. import exc
87from .. import util
88from ..util import FastIntFlag
89from ..util.typing import Self
90from ..util.typing import TupleAny
91from ..util.typing import Unpack
93if typing.TYPE_CHECKING:
94 from .annotation import _AnnotationDict
95 from .base import _AmbiguousTableNameMap
96 from .base import CompileState
97 from .base import Executable
98 from .base import ExecutableStatement
99 from .cache_key import CacheKey
100 from .ddl import _TableViaSelect
101 from .ddl import CreateTableAs
102 from .ddl import CreateView
103 from .ddl import ExecutableDDLElement
104 from .dml import Delete
105 from .dml import Insert
106 from .dml import Update
107 from .dml import UpdateBase
108 from .dml import UpdateDMLState
109 from .dml import ValuesBase
110 from .elements import _truncated_label
111 from .elements import BinaryExpression
112 from .elements import BindParameter
113 from .elements import ClauseElement
114 from .elements import ColumnClause
115 from .elements import ColumnElement
116 from .elements import False_
117 from .elements import Label
118 from .elements import Null
119 from .elements import True_
120 from .functions import Function
121 from .schema import CheckConstraint
122 from .schema import Column
123 from .schema import Constraint
124 from .schema import ForeignKeyConstraint
125 from .schema import IdentityOptions
126 from .schema import Index
127 from .schema import PrimaryKeyConstraint
128 from .schema import Table
129 from .schema import UniqueConstraint
130 from .selectable import _ColumnsClauseElement
131 from .selectable import AliasedReturnsRows
132 from .selectable import CompoundSelectState
133 from .selectable import CTE
134 from .selectable import FromClause
135 from .selectable import NamedFromClause
136 from .selectable import ReturnsRows
137 from .selectable import Select
138 from .selectable import SelectState
139 from .type_api import _BindProcessorType
140 from .type_api import TypeDecorator
141 from .type_api import TypeEngine
142 from .type_api import UserDefinedType
143 from .visitors import Visitable
144 from ..engine.cursor import CursorResultMetaData
145 from ..engine.interfaces import _CoreSingleExecuteParams
146 from ..engine.interfaces import _DBAPIAnyExecuteParams
147 from ..engine.interfaces import _DBAPIMultiExecuteParams
148 from ..engine.interfaces import _DBAPISingleExecuteParams
149 from ..engine.interfaces import _ExecuteOptions
150 from ..engine.interfaces import _GenericSetInputSizesType
151 from ..engine.interfaces import _MutableCoreSingleExecuteParams
152 from ..engine.interfaces import Dialect
153 from ..engine.interfaces import SchemaTranslateMapType
156_FromHintsType = Dict["FromClause", str]
158RESERVED_WORDS = {
159 "all",
160 "analyse",
161 "analyze",
162 "and",
163 "any",
164 "array",
165 "as",
166 "asc",
167 "asymmetric",
168 "authorization",
169 "between",
170 "binary",
171 "both",
172 "case",
173 "cast",
174 "check",
175 "collate",
176 "column",
177 "constraint",
178 "create",
179 "cross",
180 "current_date",
181 "current_role",
182 "current_time",
183 "current_timestamp",
184 "current_user",
185 "default",
186 "deferrable",
187 "desc",
188 "distinct",
189 "do",
190 "else",
191 "end",
192 "except",
193 "false",
194 "for",
195 "foreign",
196 "freeze",
197 "from",
198 "full",
199 "grant",
200 "group",
201 "having",
202 "ilike",
203 "in",
204 "initially",
205 "inner",
206 "intersect",
207 "into",
208 "is",
209 "isnull",
210 "join",
211 "leading",
212 "left",
213 "like",
214 "limit",
215 "localtime",
216 "localtimestamp",
217 "natural",
218 "new",
219 "not",
220 "notnull",
221 "null",
222 "off",
223 "offset",
224 "old",
225 "on",
226 "only",
227 "or",
228 "order",
229 "outer",
230 "overlaps",
231 "placing",
232 "primary",
233 "references",
234 "right",
235 "select",
236 "session_user",
237 "set",
238 "similar",
239 "some",
240 "symmetric",
241 "table",
242 "then",
243 "to",
244 "trailing",
245 "true",
246 "union",
247 "unique",
248 "user",
249 "using",
250 "verbose",
251 "when",
252 "where",
253}
255LEGAL_CHARACTERS = re.compile(r"^[A-Z0-9_$]+$", re.I)
256LEGAL_CHARACTERS_PLUS_SPACE = re.compile(r"^[A-Z0-9_ $]+$", re.I)
257ILLEGAL_INITIAL_CHARACTERS = {str(x) for x in range(0, 10)}.union(["$"])
259FK_ON_DELETE = re.compile(
260 r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I
261)
262FK_ON_UPDATE = re.compile(
263 r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I
264)
265FK_INITIALLY = re.compile(r"^(?:DEFERRED|IMMEDIATE)$", re.I)
266_WINDOW_EXCLUDE_RE = re.compile(
267 r"^(?:CURRENT ROW|GROUP|TIES|NO OTHERS)$", re.I
268)
269BIND_PARAMS = re.compile(r"(?<![:\w\$\x5c]):([\w\$]+)(?![:\w\$])", re.UNICODE)
270BIND_PARAMS_ESC = re.compile(r"\x5c(:[\w\$]*)(?![:\w\$])", re.UNICODE)
272_pyformat_template = "%%(%(name)s)s"
273BIND_TEMPLATES = {
274 "pyformat": _pyformat_template,
275 "qmark": "?",
276 "format": "%%s",
277 "numeric": ":[_POSITION]",
278 "numeric_dollar": "$[_POSITION]",
279 "named": ":%(name)s",
280}
283OPERATORS = {
284 # binary
285 operators.and_: " AND ",
286 operators.or_: " OR ",
287 operators.add: " + ",
288 operators.mul: " * ",
289 operators.sub: " - ",
290 operators.mod: " % ",
291 operators.neg: "-",
292 operators.lt: " < ",
293 operators.le: " <= ",
294 operators.ne: " != ",
295 operators.gt: " > ",
296 operators.ge: " >= ",
297 operators.eq: " = ",
298 operators.is_distinct_from: " IS DISTINCT FROM ",
299 operators.is_not_distinct_from: " IS NOT DISTINCT FROM ",
300 operators.concat_op: " || ",
301 operators.match_op: " MATCH ",
302 operators.not_match_op: " NOT MATCH ",
303 operators.in_op: " IN ",
304 operators.not_in_op: " NOT IN ",
305 operators.comma_op: ", ",
306 operators.from_: " FROM ",
307 operators.as_: " AS ",
308 operators.is_: " IS ",
309 operators.is_not: " IS NOT ",
310 operators.collate: " COLLATE ",
311 # unary
312 operators.exists: "EXISTS ",
313 operators.distinct_op: "DISTINCT ",
314 operators.inv: "NOT ",
315 operators.any_op: "ANY ",
316 operators.all_op: "ALL ",
317 # modifiers
318 operators.desc_op: " DESC",
319 operators.asc_op: " ASC",
320 operators.nulls_first_op: " NULLS FIRST",
321 operators.nulls_last_op: " NULLS LAST",
322 # bitwise
323 operators.bitwise_xor_op: " ^ ",
324 operators.bitwise_or_op: " | ",
325 operators.bitwise_and_op: " & ",
326 operators.bitwise_not_op: "~",
327 operators.bitwise_lshift_op: " << ",
328 operators.bitwise_rshift_op: " >> ",
329}
331FUNCTIONS: Dict[Type[Function[Any]], str] = {
332 functions.coalesce: "coalesce",
333 functions.current_date: "CURRENT_DATE",
334 functions.current_time: "CURRENT_TIME",
335 functions.current_timestamp: "CURRENT_TIMESTAMP",
336 functions.current_user: "CURRENT_USER",
337 functions.localtime: "LOCALTIME",
338 functions.localtimestamp: "LOCALTIMESTAMP",
339 functions.random: "random",
340 functions.sysdate: "sysdate",
341 functions.session_user: "SESSION_USER",
342 functions.user: "USER",
343 functions.cube: "CUBE",
344 functions.rollup: "ROLLUP",
345 functions.grouping_sets: "GROUPING SETS",
346}
349EXTRACT_MAP = {
350 "month": "month",
351 "day": "day",
352 "year": "year",
353 "second": "second",
354 "hour": "hour",
355 "doy": "doy",
356 "minute": "minute",
357 "quarter": "quarter",
358 "dow": "dow",
359 "week": "week",
360 "epoch": "epoch",
361 "milliseconds": "milliseconds",
362 "microseconds": "microseconds",
363 "timezone_hour": "timezone_hour",
364 "timezone_minute": "timezone_minute",
365}
367COMPOUND_KEYWORDS = {
368 selectable._CompoundSelectKeyword.UNION: "UNION",
369 selectable._CompoundSelectKeyword.UNION_ALL: "UNION ALL",
370 selectable._CompoundSelectKeyword.EXCEPT: "EXCEPT",
371 selectable._CompoundSelectKeyword.EXCEPT_ALL: "EXCEPT ALL",
372 selectable._CompoundSelectKeyword.INTERSECT: "INTERSECT",
373 selectable._CompoundSelectKeyword.INTERSECT_ALL: "INTERSECT ALL",
374}
377class ResultColumnsEntry(NamedTuple):
378 """Tracks a column expression that is expected to be represented
379 in the result rows for this statement.
381 This normally refers to the columns clause of a SELECT statement
382 but may also refer to a RETURNING clause, as well as for dialect-specific
383 emulations.
385 """
387 keyname: str
388 """string name that's expected in cursor.description"""
390 name: str
391 """column name, may be labeled"""
393 objects: Tuple[Any, ...]
394 """sequence of objects that should be able to locate this column
395 in a RowMapping. This is typically string names and aliases
396 as well as Column objects.
398 """
400 type: TypeEngine[Any]
401 """Datatype to be associated with this column. This is where
402 the "result processing" logic directly links the compiled statement
403 to the rows that come back from the cursor.
405 """
408class _ResultMapAppender(Protocol):
409 def __call__(
410 self,
411 keyname: str,
412 name: str,
413 objects: Sequence[Any],
414 type_: TypeEngine[Any],
415 ) -> None: ...
418# integer indexes into ResultColumnsEntry used by cursor.py.
419# some profiling showed integer access faster than named tuple
420RM_RENDERED_NAME: Literal[0] = 0
421RM_NAME: Literal[1] = 1
422RM_OBJECTS: Literal[2] = 2
423RM_TYPE: Literal[3] = 3
426class _BaseCompilerStackEntry(TypedDict):
427 asfrom_froms: Set[FromClause]
428 correlate_froms: Set[FromClause]
429 selectable: ReturnsRows
432class _CompilerStackEntry(_BaseCompilerStackEntry, total=False):
433 compile_state: CompileState
434 need_result_map_for_nested: bool
435 need_result_map_for_compound: bool
436 select_0: ReturnsRows
437 insert_from_select: Select[Unpack[TupleAny]]
440class ExpandedState(NamedTuple):
441 """represents state to use when producing "expanded" and
442 "post compile" bound parameters for a statement.
444 "expanded" parameters are parameters that are generated at
445 statement execution time to suit a number of parameters passed, the most
446 prominent example being the individual elements inside of an IN expression.
448 "post compile" parameters are parameters where the SQL literal value
449 will be rendered into the SQL statement at execution time, rather than
450 being passed as separate parameters to the driver.
452 To create an :class:`.ExpandedState` instance, use the
453 :meth:`.SQLCompiler.construct_expanded_state` method on any
454 :class:`.SQLCompiler` instance.
456 """
458 statement: str
459 """String SQL statement with parameters fully expanded"""
461 parameters: _CoreSingleExecuteParams
462 """Parameter dictionary with parameters fully expanded.
464 For a statement that uses named parameters, this dictionary will map
465 exactly to the names in the statement. For a statement that uses
466 positional parameters, the :attr:`.ExpandedState.positional_parameters`
467 will yield a tuple with the positional parameter set.
469 """
471 processors: Mapping[str, _BindProcessorType[Any]]
472 """mapping of bound value processors"""
474 positiontup: Optional[Sequence[str]]
475 """Sequence of string names indicating the order of positional
476 parameters"""
478 parameter_expansion: Mapping[str, List[str]]
479 """Mapping representing the intermediary link from original parameter
480 name to list of "expanded" parameter names, for those parameters that
481 were expanded."""
483 @property
484 def positional_parameters(self) -> Tuple[Any, ...]:
485 """Tuple of positional parameters, for statements that were compiled
486 using a positional paramstyle.
488 """
489 if self.positiontup is None:
490 raise exc.InvalidRequestError(
491 "statement does not use a positional paramstyle"
492 )
493 return tuple(self.parameters[key] for key in self.positiontup)
495 @property
496 def additional_parameters(self) -> _CoreSingleExecuteParams:
497 """synonym for :attr:`.ExpandedState.parameters`."""
498 return self.parameters
501class _InsertManyValues(NamedTuple):
502 """represents state to use for executing an "insertmanyvalues" statement.
504 The primary consumers of this object are the
505 :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and
506 :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods.
508 .. versionadded:: 2.0
510 """
512 is_default_expr: bool
513 """if True, the statement is of the form
514 ``INSERT INTO TABLE DEFAULT VALUES``, and can't be rewritten as a "batch"
516 """
518 single_values_expr: str
519 """The rendered "values" clause of the INSERT statement.
521 This is typically the parenthesized section e.g. "(?, ?, ?)" or similar.
522 The insertmanyvalues logic uses this string as a search and replace
523 target.
525 """
527 insert_crud_params: List[crud._CrudParamElementStr]
528 """List of Column / bind names etc. used while rewriting the statement"""
530 num_positional_params_counted: int
531 """the number of bound parameters in a single-row statement.
533 This count may be larger or smaller than the actual number of columns
534 targeted in the INSERT, as it accommodates for SQL expressions
535 in the values list that may have zero or more parameters embedded
536 within them.
538 This count is part of what's used to organize rewritten parameter lists
539 when batching.
541 """
543 sort_by_parameter_order: bool = False
544 """if the deterministic_returnined_order parameter were used on the
545 insert.
547 All of the attributes following this will only be used if this is True.
549 """
551 includes_upsert_behaviors: bool = False
552 """if True, we have to accommodate for upsert behaviors.
554 This will in some cases downgrade "insertmanyvalues" that requests
555 deterministic ordering.
557 """
559 sentinel_columns: Optional[Sequence[Column[Any]]] = None
560 """List of sentinel columns that were located.
562 This list is only here if the INSERT asked for
563 sort_by_parameter_order=True,
564 and dialect-appropriate sentinel columns were located.
566 .. versionadded:: 2.0.10
568 """
570 num_sentinel_columns: int = 0
571 """how many sentinel columns are in the above list, if any.
573 This is the same as
574 ``len(sentinel_columns) if sentinel_columns is not None else 0``
576 """
578 sentinel_param_keys: Optional[Sequence[str]] = None
579 """parameter str keys in each param dictionary / tuple
580 that would link to the client side "sentinel" values for that row, which
581 we can use to match up parameter sets to result rows.
583 This is only present if sentinel_columns is present and the INSERT
584 statement actually refers to client side values for these sentinel
585 columns.
587 .. versionadded:: 2.0.10
589 .. versionchanged:: 2.0.29 - the sequence is now string dictionary keys
590 only, used against the "compiled parameteters" collection before
591 the parameters were converted by bound parameter processors
593 """
595 implicit_sentinel: bool = False
596 """if True, we have exactly one sentinel column and it uses a server side
597 value, currently has to generate an incrementing integer value.
599 The dialect in question would have asserted that it supports receiving
600 these values back and sorting on that value as a means of guaranteeing
601 correlation with the incoming parameter list.
603 .. versionadded:: 2.0.10
605 """
607 has_upsert_bound_parameters: bool = False
608 """if True, the upsert SET clause contains bound parameters that will
609 receive their values from the parameters dict (i.e., parametrized
610 bindparams where value is None and callable is None).
612 This means we can't batch multiple rows in a single statement, since
613 each row would need different values in the SET clause but there's only
614 one SET clause per statement. See issue #13130.
616 .. versionadded:: 2.0.37
618 """
620 embed_values_counter: bool = False
621 """Whether to embed an incrementing integer counter in each parameter
622 set within the VALUES clause as parameters are batched over.
624 This is only used for a specific INSERT..SELECT..VALUES..RETURNING syntax
625 where a subquery is used to produce value tuples. Current support
626 includes PostgreSQL, Microsoft SQL Server.
628 .. versionadded:: 2.0.10
630 """
633class _InsertManyValuesBatch(NamedTuple):
634 """represents an individual batch SQL statement for insertmanyvalues.
636 This is passed through the
637 :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and
638 :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods out
639 to the :class:`.Connection` within the
640 :meth:`.Connection._exec_insertmany_context` method.
642 .. versionadded:: 2.0.10
644 """
646 replaced_statement: str
647 replaced_parameters: _DBAPIAnyExecuteParams
648 processed_setinputsizes: Optional[_GenericSetInputSizesType]
649 batch: Sequence[_DBAPISingleExecuteParams]
650 sentinel_values: Sequence[Tuple[Any, ...]]
651 current_batch_size: int
652 batchnum: int
653 total_batches: int
654 rows_sorted: bool
655 is_downgraded: bool
658class InsertmanyvaluesSentinelOpts(FastIntFlag):
659 """bitflag enum indicating styles of PK defaults
660 which can work as implicit sentinel columns
662 """
664 NOT_SUPPORTED = 1
665 AUTOINCREMENT = 2
666 IDENTITY = 4
667 SEQUENCE = 8
668 MONOTONIC_FUNCTION = 16
670 ANY_AUTOINCREMENT = (
671 AUTOINCREMENT | IDENTITY | SEQUENCE | MONOTONIC_FUNCTION
672 )
673 _SUPPORTED_OR_NOT = NOT_SUPPORTED | ANY_AUTOINCREMENT
675 USE_INSERT_FROM_SELECT = 32
676 RENDER_SELECT_COL_CASTS = 64
679class AggregateOrderByStyle(IntEnum):
680 """Describes backend database's capabilities with ORDER BY for aggregate
681 functions
683 .. versionadded:: 2.1
685 """
687 NONE = 0
688 """database has no ORDER BY for aggregate functions"""
690 INLINE = 1
691 """ORDER BY is rendered inside the function's argument list, typically as
692 the last element"""
694 WITHIN_GROUP = 2
695 """the WITHIN GROUP (ORDER BY ...) phrase is used for all aggregate
696 functions (not just the ordered set ones)"""
699class CompilerState(IntEnum):
700 COMPILING = 0
701 """statement is present, compilation phase in progress"""
703 STRING_APPLIED = 1
704 """statement is present, string form of the statement has been applied.
706 Additional processors by subclasses may still be pending.
708 """
710 NO_STATEMENT = 2
711 """compiler does not have a statement to compile, is used
712 for method access"""
715class Linting(IntEnum):
716 """represent preferences for the 'SQL linting' feature.
718 this feature currently includes support for flagging cartesian products
719 in SQL statements.
721 """
723 NO_LINTING = 0
724 "Disable all linting."
726 COLLECT_CARTESIAN_PRODUCTS = 1
727 """Collect data on FROMs and cartesian products and gather into
728 'self.from_linter'"""
730 WARN_LINTING = 2
731 "Emit warnings for linters that find problems"
733 FROM_LINTING = COLLECT_CARTESIAN_PRODUCTS | WARN_LINTING
734 """Warn for cartesian products; combines COLLECT_CARTESIAN_PRODUCTS
735 and WARN_LINTING"""
738NO_LINTING, COLLECT_CARTESIAN_PRODUCTS, WARN_LINTING, FROM_LINTING = tuple(
739 Linting
740)
743class FromLinter(collections.namedtuple("FromLinter", ["froms", "edges"])):
744 """represents current state for the "cartesian product" detection
745 feature."""
747 def lint(self, start=None):
748 froms = self.froms
749 if not froms:
750 return None, None
752 edges = set(self.edges)
753 the_rest = set(froms)
755 if start is not None:
756 start_with = start
757 the_rest.remove(start_with)
758 else:
759 start_with = the_rest.pop()
761 stack = collections.deque([start_with])
763 while stack and the_rest:
764 node = stack.popleft()
765 the_rest.discard(node)
767 # comparison of nodes in edges here is based on hash equality, as
768 # there are "annotated" elements that match the non-annotated ones.
769 # to remove the need for in-python hash() calls, use native
770 # containment routines (e.g. "node in edge", "edge.index(node)")
771 to_remove = {edge for edge in edges if node in edge}
773 # appendleft the node in each edge that is not
774 # the one that matched.
775 stack.extendleft(edge[not edge.index(node)] for edge in to_remove)
776 edges.difference_update(to_remove)
778 # FROMS left over? boom
779 if the_rest:
780 return the_rest, start_with
781 else:
782 return None, None
784 def warn(self, stmt_type="SELECT"):
785 the_rest, start_with = self.lint()
787 # FROMS left over? boom
788 if the_rest:
789 froms = the_rest
790 if froms:
791 template = (
792 "{stmt_type} statement has a cartesian product between "
793 "FROM element(s) {froms} and "
794 'FROM element "{start}". Apply join condition(s) '
795 "between each element to resolve."
796 )
797 froms_str = ", ".join(
798 f'"{self.froms[from_]}"' for from_ in froms
799 )
800 message = template.format(
801 stmt_type=stmt_type,
802 froms=froms_str,
803 start=self.froms[start_with],
804 )
806 util.warn(message)
809class Compiled:
810 """Represent a compiled SQL or DDL expression.
812 The ``__str__`` method of the ``Compiled`` object should produce
813 the actual text of the statement. ``Compiled`` objects are
814 specific to their underlying database dialect, and also may
815 or may not be specific to the columns referenced within a
816 particular set of bind parameters. In no case should the
817 ``Compiled`` object be dependent on the actual values of those
818 bind parameters, even though it may reference those values as
819 defaults.
820 """
822 statement: Optional[ClauseElement] = None
823 "The statement to compile."
824 string: str = ""
825 "The string representation of the ``statement``"
827 state: CompilerState
828 """description of the compiler's state"""
830 is_sql = False
831 is_ddl = False
833 _cached_metadata: Optional[CursorResultMetaData] = None
835 _result_columns: Optional[List[ResultColumnsEntry]] = None
837 schema_translate_map: Optional[SchemaTranslateMapType] = None
839 execution_options: _ExecuteOptions = util.EMPTY_DICT
840 """
841 Execution options propagated from the statement. In some cases,
842 sub-elements of the statement can modify these.
843 """
845 preparer: IdentifierPreparer
847 _annotations: _AnnotationDict = util.EMPTY_DICT
849 compile_state: Optional[CompileState] = None
850 """Optional :class:`.CompileState` object that maintains additional
851 state used by the compiler.
853 Major executable objects such as :class:`_expression.Insert`,
854 :class:`_expression.Update`, :class:`_expression.Delete`,
855 :class:`_expression.Select` will generate this
856 state when compiled in order to calculate additional information about the
857 object. For the top level object that is to be executed, the state can be
858 stored here where it can also have applicability towards result set
859 processing.
861 .. versionadded:: 1.4
863 """
865 dml_compile_state: Optional[CompileState] = None
866 """Optional :class:`.CompileState` assigned at the same point that
867 .isinsert, .isupdate, or .isdelete is assigned.
869 This will normally be the same object as .compile_state, with the
870 exception of cases like the :class:`.ORMFromStatementCompileState`
871 object.
873 .. versionadded:: 1.4.40
875 """
877 cache_key: Optional[CacheKey] = None
878 """The :class:`.CacheKey` that was generated ahead of creating this
879 :class:`.Compiled` object.
881 This is used for routines that need access to the original
882 :class:`.CacheKey` instance generated when the :class:`.Compiled`
883 instance was first cached, typically in order to reconcile
884 the original list of :class:`.BindParameter` objects with a
885 per-statement list that's generated on each call.
887 """
889 _gen_time: float
890 """Generation time of this :class:`.Compiled`, used for reporting
891 cache stats."""
893 def __init__(
894 self,
895 dialect: Dialect,
896 statement: Optional[ClauseElement],
897 schema_translate_map: Optional[SchemaTranslateMapType] = None,
898 render_schema_translate: bool = False,
899 compile_kwargs: Mapping[str, Any] = util.immutabledict(),
900 ):
901 """Construct a new :class:`.Compiled` object.
903 :param dialect: :class:`.Dialect` to compile against.
905 :param statement: :class:`_expression.ClauseElement` to be compiled.
907 :param schema_translate_map: dictionary of schema names to be
908 translated when forming the resultant SQL
910 .. seealso::
912 :ref:`schema_translating`
914 :param compile_kwargs: additional kwargs that will be
915 passed to the initial call to :meth:`.Compiled.process`.
918 """
919 self.dialect = dialect
920 self.preparer = self.dialect.identifier_preparer
921 if schema_translate_map:
922 self.schema_translate_map = schema_translate_map
923 self.preparer = self.preparer._with_schema_translate(
924 schema_translate_map
925 )
927 if statement is not None:
928 self.state = CompilerState.COMPILING
929 self.statement = statement
930 self.can_execute = statement.supports_execution
931 self._annotations = statement._annotations
932 if self.can_execute:
933 if TYPE_CHECKING:
934 assert isinstance(statement, Executable)
935 self.execution_options = statement._execution_options
936 self.string = self.process(self.statement, **compile_kwargs)
938 if render_schema_translate:
939 assert schema_translate_map is not None
940 self.string = self.preparer._render_schema_translates(
941 self.string, schema_translate_map
942 )
944 self.state = CompilerState.STRING_APPLIED
945 else:
946 self.state = CompilerState.NO_STATEMENT
948 self._gen_time = perf_counter()
950 def __init_subclass__(cls) -> None:
951 cls._init_compiler_cls()
952 return super().__init_subclass__()
954 @classmethod
955 def _init_compiler_cls(cls):
956 pass
958 def visit_unsupported_compilation(self, element, err, **kw):
959 raise exc.UnsupportedCompilationError(self, type(element)) from err
961 @property
962 def sql_compiler(self) -> SQLCompiler:
963 """Return a Compiled that is capable of processing SQL expressions.
965 If this compiler is one, it would likely just return 'self'.
967 """
969 raise NotImplementedError()
971 def process(self, obj: Visitable, **kwargs: Any) -> str:
972 return obj._compiler_dispatch(self, **kwargs)
974 def __str__(self) -> str:
975 """Return the string text of the generated SQL or DDL."""
977 if self.state is CompilerState.STRING_APPLIED:
978 return self.string
979 else:
980 return ""
982 def construct_params(
983 self,
984 params: Optional[_CoreSingleExecuteParams] = None,
985 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
986 escape_names: bool = True,
987 ) -> Optional[_MutableCoreSingleExecuteParams]:
988 """Return the bind params for this compiled object.
990 :param params: a dict of string/object pairs whose values will
991 override bind values compiled in to the
992 statement.
993 """
995 raise NotImplementedError()
997 @property
998 def params(self):
999 """Return the bind params for this compiled object."""
1000 return self.construct_params()
1003class TypeCompiler(util.EnsureKWArg):
1004 """Produces DDL specification for TypeEngine objects."""
1006 ensure_kwarg = r"visit_\w+"
1008 def __init__(self, dialect: Dialect):
1009 self.dialect = dialect
1011 def process(self, type_: TypeEngine[Any], **kw: Any) -> str:
1012 if (
1013 type_._variant_mapping
1014 and self.dialect.name in type_._variant_mapping
1015 ):
1016 type_ = type_._variant_mapping[self.dialect.name]
1017 return type_._compiler_dispatch(self, **kw)
1019 def visit_unsupported_compilation(
1020 self, element: Any, err: Exception, **kw: Any
1021 ) -> NoReturn:
1022 raise exc.UnsupportedCompilationError(self, element) from err
1025# this was a Visitable, but to allow accurate detection of
1026# column elements this is actually a column element
1027class _CompileLabel(
1028 roles.BinaryElementRole[Any], elements.CompilerColumnElement
1029):
1030 """lightweight label object which acts as an expression.Label."""
1032 __visit_name__ = "label"
1033 __slots__ = "element", "name", "_alt_names"
1035 def __init__(self, col, name, alt_names=()):
1036 self.element = col
1037 self.name = name
1038 self._alt_names = (col,) + alt_names
1040 @property
1041 def proxy_set(self):
1042 return self.element.proxy_set
1044 @property
1045 def type(self):
1046 return self.element.type
1048 def self_group(self, **kw):
1049 return self
1052class aggregate_orderby_inline(
1053 roles.BinaryElementRole[Any], elements.CompilerColumnElement
1054):
1055 """produce ORDER BY inside of function argument lists"""
1057 __visit_name__ = "aggregate_orderby_inline"
1058 __slots__ = "element", "aggregate_order_by"
1060 def __init__(self, element, orderby):
1061 self.element = element
1062 self.aggregate_order_by = orderby
1064 def __iter__(self):
1065 return iter(self.element)
1067 @property
1068 def proxy_set(self):
1069 return self.element.proxy_set
1071 @property
1072 def type(self):
1073 return self.element.type
1075 def self_group(self, **kw):
1076 return self
1078 def _with_binary_element_type(self, type_):
1079 return aggregate_orderby_inline(
1080 self.element._with_binary_element_type(type_),
1081 self.aggregate_order_by,
1082 )
1085class ilike_case_insensitive(
1086 roles.BinaryElementRole[Any], elements.CompilerColumnElement
1087):
1088 """produce a wrapping element for a case-insensitive portion of
1089 an ILIKE construct.
1091 The construct usually renders the ``lower()`` function, but on
1092 PostgreSQL will pass silently with the assumption that "ILIKE"
1093 is being used.
1095 .. versionadded:: 2.0
1097 """
1099 __visit_name__ = "ilike_case_insensitive_operand"
1100 __slots__ = "element", "comparator"
1102 def __init__(self, element):
1103 self.element = element
1104 self.comparator = element.comparator
1106 @property
1107 def proxy_set(self):
1108 return self.element.proxy_set
1110 @property
1111 def type(self):
1112 return self.element.type
1114 def self_group(self, **kw):
1115 return self
1117 def _with_binary_element_type(self, type_):
1118 return ilike_case_insensitive(
1119 self.element._with_binary_element_type(type_)
1120 )
1123class SQLCompiler(Compiled):
1124 """Default implementation of :class:`.Compiled`.
1126 Compiles :class:`_expression.ClauseElement` objects into SQL strings.
1128 """
1130 extract_map = EXTRACT_MAP
1132 bindname_escape_characters: ClassVar[Mapping[str, str]] = (
1133 util.immutabledict(
1134 {
1135 "%": "P",
1136 "(": "A",
1137 ")": "Z",
1138 ":": "C",
1139 ".": "_",
1140 "[": "_",
1141 "]": "_",
1142 " ": "_",
1143 }
1144 )
1145 )
1146 """A mapping (e.g. dict or similar) containing a lookup of
1147 characters keyed to replacement characters which will be applied to all
1148 'bind names' used in SQL statements as a form of 'escaping'; the given
1149 characters are replaced entirely with the 'replacement' character when
1150 rendered in the SQL statement, and a similar translation is performed
1151 on the incoming names used in parameter dictionaries passed to methods
1152 like :meth:`_engine.Connection.execute`.
1154 This allows bound parameter names used in :func:`_sql.bindparam` and
1155 other constructs to have any arbitrary characters present without any
1156 concern for characters that aren't allowed at all on the target database.
1158 Third party dialects can establish their own dictionary here to replace the
1159 default mapping, which will ensure that the particular characters in the
1160 mapping will never appear in a bound parameter name.
1162 The dictionary is evaluated at **class creation time**, so cannot be
1163 modified at runtime; it must be present on the class when the class
1164 is first declared.
1166 Note that for dialects that have additional bound parameter rules such
1167 as additional restrictions on leading characters, the
1168 :meth:`_sql.SQLCompiler.bindparam_string` method may need to be augmented.
1169 See the cx_Oracle compiler for an example of this.
1171 .. versionadded:: 2.0.0rc1
1173 """
1175 _bind_translate_re: ClassVar[Pattern[str]]
1176 _bind_translate_chars: ClassVar[Mapping[str, str]]
1178 is_sql = True
1180 compound_keywords = COMPOUND_KEYWORDS
1182 isdelete: bool = False
1183 isinsert: bool = False
1184 isupdate: bool = False
1185 """class-level defaults which can be set at the instance
1186 level to define if this Compiled instance represents
1187 INSERT/UPDATE/DELETE
1188 """
1190 postfetch: Optional[List[Column[Any]]]
1191 """list of columns that can be post-fetched after INSERT or UPDATE to
1192 receive server-updated values"""
1194 insert_prefetch: Sequence[Column[Any]] = ()
1195 """list of columns for which default values should be evaluated before
1196 an INSERT takes place"""
1198 update_prefetch: Sequence[Column[Any]] = ()
1199 """list of columns for which onupdate default values should be evaluated
1200 before an UPDATE takes place"""
1202 implicit_returning: Optional[Sequence[ColumnElement[Any]]] = None
1203 """list of "implicit" returning columns for a toplevel INSERT or UPDATE
1204 statement, used to receive newly generated values of columns.
1206 .. versionadded:: 2.0 ``implicit_returning`` replaces the previous
1207 ``returning`` collection, which was not a generalized RETURNING
1208 collection and instead was in fact specific to the "implicit returning"
1209 feature.
1211 """
1213 isplaintext: bool = False
1215 binds: Dict[str, BindParameter[Any]]
1216 """a dictionary of bind parameter keys to BindParameter instances."""
1218 bind_names: Dict[BindParameter[Any], str]
1219 """a dictionary of BindParameter instances to "compiled" names
1220 that are actually present in the generated SQL"""
1222 stack: List[_CompilerStackEntry]
1223 """major statements such as SELECT, INSERT, UPDATE, DELETE are
1224 tracked in this stack using an entry format."""
1226 returning_precedes_values: bool = False
1227 """set to True classwide to generate RETURNING
1228 clauses before the VALUES or WHERE clause (i.e. MSSQL)
1229 """
1231 render_table_with_column_in_update_from: bool = False
1232 """set to True classwide to indicate the SET clause
1233 in a multi-table UPDATE statement should qualify
1234 columns with the table name (i.e. MySQL only)
1235 """
1237 ansi_bind_rules: bool = False
1238 """SQL 92 doesn't allow bind parameters to be used
1239 in the columns clause of a SELECT, nor does it allow
1240 ambiguous expressions like "? = ?". A compiler
1241 subclass can set this flag to False if the target
1242 driver/DB enforces this
1243 """
1245 bindtemplate: str
1246 """template to render bound parameters based on paramstyle."""
1248 compilation_bindtemplate: str
1249 """template used by compiler to render parameters before positional
1250 paramstyle application"""
1252 _numeric_binds_identifier_char: str
1253 """Character that's used to as the identifier of a numerical bind param.
1254 For example if this char is set to ``$``, numerical binds will be rendered
1255 in the form ``$1, $2, $3``.
1256 """
1258 _result_columns: List[ResultColumnsEntry]
1259 """relates label names in the final SQL to a tuple of local
1260 column/label name, ColumnElement object (if any) and
1261 TypeEngine. CursorResult uses this for type processing and
1262 column targeting"""
1264 _textual_ordered_columns: bool = False
1265 """tell the result object that the column names as rendered are important,
1266 but they are also "ordered" vs. what is in the compiled object here.
1268 As of 1.4.42 this condition is only present when the statement is a
1269 TextualSelect, e.g. text("....").columns(...), where it is required
1270 that the columns are considered positionally and not by name.
1272 """
1274 _ad_hoc_textual: bool = False
1275 """tell the result that we encountered text() or '*' constructs in the
1276 middle of the result columns, but we also have compiled columns, so
1277 if the number of columns in cursor.description does not match how many
1278 expressions we have, that means we can't rely on positional at all and
1279 should match on name.
1281 """
1283 _ordered_columns: bool = True
1284 """
1285 if False, means we can't be sure the list of entries
1286 in _result_columns is actually the rendered order. Usually
1287 True unless using an unordered TextualSelect.
1288 """
1290 _loose_column_name_matching: bool = False
1291 """tell the result object that the SQL statement is textual, wants to match
1292 up to Column objects, and may be using the ._tq_label in the SELECT rather
1293 than the base name.
1295 """
1297 _numeric_binds: bool = False
1298 """
1299 True if paramstyle is "numeric". This paramstyle is trickier than
1300 all the others.
1302 """
1304 _render_postcompile: bool = False
1305 """
1306 whether to render out POSTCOMPILE params during the compile phase.
1308 This attribute is used only for end-user invocation of stmt.compile();
1309 it's never used for actual statement execution, where instead the
1310 dialect internals access and render the internal postcompile structure
1311 directly.
1313 """
1315 _post_compile_expanded_state: Optional[ExpandedState] = None
1316 """When render_postcompile is used, the ``ExpandedState`` used to create
1317 the "expanded" SQL is assigned here, and then used by the ``.params``
1318 accessor and ``.construct_params()`` methods for their return values.
1320 .. versionadded:: 2.0.0rc1
1322 """
1324 _pre_expanded_string: Optional[str] = None
1325 """Stores the original string SQL before 'post_compile' is applied,
1326 for cases where 'post_compile' were used.
1328 """
1330 _pre_expanded_positiontup: Optional[List[str]] = None
1332 _insertmanyvalues: Optional[_InsertManyValues] = None
1334 _insert_crud_params: Optional[crud._CrudParamSequence] = None
1336 literal_execute_params: FrozenSet[BindParameter[Any]] = frozenset()
1337 """bindparameter objects that are rendered as literal values at statement
1338 execution time.
1340 """
1342 post_compile_params: FrozenSet[BindParameter[Any]] = frozenset()
1343 """bindparameter objects that are rendered as bound parameter placeholders
1344 at statement execution time.
1346 """
1348 escaped_bind_names: util.immutabledict[str, str] = util.EMPTY_DICT
1350 # the names ``escaped_bind_names`` maps *to*, maintained alongside it so
1351 # that collision checks don't rebuild the value collection each time.
1352 # only assigned once a name actually needs escaping, which is rare;
1353 # ``escaped_bind_names`` being empty means this was never set.
1354 _escaped_bind_names_used: Set[str]
1355 """Late escaping of bound parameter names that has to be converted
1356 to the original name when looking in the parameter dictionary.
1358 """
1360 has_out_parameters = False
1361 """if True, there are bindparam() objects that have the isoutparam
1362 flag set."""
1364 postfetch_lastrowid = False
1365 """if True, and this in insert, use cursor.lastrowid to populate
1366 result.inserted_primary_key. """
1368 _cache_key_bind_match: Optional[
1369 Tuple[
1370 Dict[
1371 BindParameter[Any],
1372 List[BindParameter[Any]],
1373 ],
1374 Dict[
1375 str,
1376 BindParameter[Any],
1377 ],
1378 ]
1379 ] = None
1380 """a mapping that will relate the BindParameter object we compile
1381 to those that are part of the extracted collection of parameters
1382 in the cache key, if we were given a cache key.
1384 """
1386 positiontup: Optional[List[str]] = None
1387 """for a compiled construct that uses a positional paramstyle, will be
1388 a sequence of strings, indicating the names of bound parameters in order.
1390 This is used in order to render bound parameters in their correct order,
1391 and is combined with the :attr:`_sql.Compiled.params` dictionary to
1392 render parameters.
1394 This sequence always contains the unescaped name of the parameters.
1396 .. seealso::
1398 :ref:`faq_sql_expression_string` - includes a usage example for
1399 debugging use cases.
1401 """
1402 _values_bindparam: Optional[List[str]] = None
1404 _visited_bindparam: Optional[List[str]] = None
1406 inline: bool = False
1408 ctes: Optional[MutableMapping[CTE, str]]
1410 # Detect same CTE references - Dict[(level, name), cte]
1411 # Level is required for supporting nesting
1412 ctes_by_level_name: Dict[Tuple[int, str], CTE]
1414 # To retrieve key/level in ctes_by_level_name -
1415 # Dict[cte_reference, (level, cte_name, cte_opts)]
1416 level_name_by_cte: Dict[CTE, Tuple[int, str, selectable._CTEOpts]]
1418 ctes_recursive: bool
1420 _post_compile_pattern = re.compile(r"__\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]")
1421 _pyformat_pattern = re.compile(r"%\(([^)]+?)\)s")
1422 _positional_pattern = re.compile(
1423 f"{_pyformat_pattern.pattern}|{_post_compile_pattern.pattern}"
1424 )
1425 _collect_params: Final[bool]
1426 _collected_params: util.immutabledict[str, Any]
1428 @classmethod
1429 def _init_compiler_cls(cls):
1430 cls._init_bind_translate()
1432 @classmethod
1433 def _init_bind_translate(cls):
1434 reg = re.escape("".join(cls.bindname_escape_characters))
1435 cls._bind_translate_re = re.compile(f"[{reg}]")
1436 cls._bind_translate_chars = cls.bindname_escape_characters
1438 def __init__(
1439 self,
1440 dialect: Dialect,
1441 statement: Optional[ClauseElement],
1442 cache_key: Optional[CacheKey] = None,
1443 column_keys: Optional[Sequence[str]] = None,
1444 for_executemany: bool = False,
1445 linting: Linting = NO_LINTING,
1446 _supporting_against: Optional[SQLCompiler] = None,
1447 **kwargs: Any,
1448 ):
1449 """Construct a new :class:`.SQLCompiler` object.
1451 :param dialect: :class:`.Dialect` to be used
1453 :param statement: :class:`_expression.ClauseElement` to be compiled
1455 :param column_keys: a list of column names to be compiled into an
1456 INSERT or UPDATE statement.
1458 :param for_executemany: whether INSERT / UPDATE statements should
1459 expect that they are to be invoked in an "executemany" style,
1460 which may impact how the statement will be expected to return the
1461 values of defaults and autoincrement / sequences and similar.
1462 Depending on the backend and driver in use, support for retrieving
1463 these values may be disabled which means SQL expressions may
1464 be rendered inline, RETURNING may not be rendered, etc.
1466 :param kwargs: additional keyword arguments to be consumed by the
1467 superclass.
1469 """
1470 self.column_keys = column_keys
1472 self.cache_key = cache_key
1474 if cache_key:
1475 cksm = {b.key: b for b in cache_key[1]}
1476 ckbm = {b: [b] for b in cache_key[1]}
1477 self._cache_key_bind_match = (ckbm, cksm)
1479 # compile INSERT/UPDATE defaults/sequences to expect executemany
1480 # style execution, which may mean no pre-execute of defaults,
1481 # or no RETURNING
1482 self.for_executemany = for_executemany
1484 self.linting = linting
1486 # a dictionary of bind parameter keys to BindParameter
1487 # instances.
1488 self.binds = {}
1490 # a dictionary of BindParameter instances to "compiled" names
1491 # that are actually present in the generated SQL
1492 self.bind_names = util.column_dict()
1494 # stack which keeps track of nested SELECT statements
1495 self.stack = []
1497 self._result_columns = []
1499 # true if the paramstyle is positional
1500 self.positional = dialect.positional
1501 if self.positional:
1502 self._numeric_binds = nb = dialect.paramstyle.startswith("numeric")
1503 if nb:
1504 self._numeric_binds_identifier_char = (
1505 "$" if dialect.paramstyle == "numeric_dollar" else ":"
1506 )
1508 self.compilation_bindtemplate = _pyformat_template
1509 else:
1510 self.compilation_bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
1512 self.ctes = None
1514 self.label_length = (
1515 dialect.label_length or dialect.max_identifier_length
1516 )
1518 # a map which tracks "anonymous" identifiers that are created on
1519 # the fly here
1520 self.anon_map = prefix_anon_map()
1522 # a map which tracks "truncated" names based on
1523 # dialect.label_length or dialect.max_identifier_length
1524 self.truncated_names: Dict[Tuple[str, str], str] = {}
1525 self._truncated_counters: Dict[str, int] = {}
1526 if not cache_key:
1527 self._collect_params = True
1528 self._collected_params = util.EMPTY_DICT
1529 else:
1530 self._collect_params = False # type: ignore[misc]
1532 Compiled.__init__(self, dialect, statement, **kwargs)
1534 if self.isinsert or self.isupdate or self.isdelete:
1535 if TYPE_CHECKING:
1536 assert isinstance(statement, UpdateBase)
1538 if self.isinsert or self.isupdate:
1539 if TYPE_CHECKING:
1540 assert isinstance(statement, ValuesBase)
1541 if statement._inline:
1542 self.inline = True
1543 elif self.for_executemany and (
1544 not self.isinsert
1545 or (
1546 self.dialect.insert_executemany_returning
1547 and statement._return_defaults
1548 )
1549 ):
1550 self.inline = True
1552 self.bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
1554 if _supporting_against:
1555 self.__dict__.update(
1556 {
1557 k: v
1558 for k, v in _supporting_against.__dict__.items()
1559 if k
1560 not in {
1561 "state",
1562 "dialect",
1563 "preparer",
1564 "positional",
1565 "_numeric_binds",
1566 "compilation_bindtemplate",
1567 "bindtemplate",
1568 }
1569 }
1570 )
1572 if self.state is CompilerState.STRING_APPLIED:
1573 if self.positional:
1574 if self._numeric_binds:
1575 self._process_numeric()
1576 else:
1577 self._process_positional()
1579 if self._render_postcompile:
1580 parameters = self.construct_params(
1581 escape_names=False,
1582 _no_postcompile=True,
1583 )
1585 self._process_parameters_for_postcompile(
1586 parameters, _populate_self=True
1587 )
1589 @property
1590 def insert_single_values_expr(self) -> Optional[str]:
1591 """When an INSERT is compiled with a single set of parameters inside
1592 a VALUES expression, the string is assigned here, where it can be
1593 used for insert batching schemes to rewrite the VALUES expression.
1595 .. versionchanged:: 2.0 This collection is no longer used by
1596 SQLAlchemy's built-in dialects, in favor of the currently
1597 internal ``_insertmanyvalues`` collection that is used only by
1598 :class:`.SQLCompiler`.
1600 """
1601 if self._insertmanyvalues is None:
1602 return None
1603 else:
1604 return self._insertmanyvalues.single_values_expr
1606 @util.ro_memoized_property
1607 def effective_returning(self) -> Optional[Sequence[ColumnElement[Any]]]:
1608 """The effective "returning" columns for INSERT, UPDATE or DELETE.
1610 This is either the so-called "implicit returning" columns which are
1611 calculated by the compiler on the fly, or those present based on what's
1612 present in ``self.statement._returning`` (expanded into individual
1613 columns using the ``._all_selected_columns`` attribute) i.e. those set
1614 explicitly using the :meth:`.UpdateBase.returning` method.
1616 .. versionadded:: 2.0
1618 """
1619 if self.implicit_returning:
1620 return self.implicit_returning
1621 elif self.statement is not None and is_dml(self.statement):
1622 return [
1623 c
1624 for c in self.statement._all_selected_columns
1625 if is_column_element(c)
1626 ]
1628 else:
1629 return None
1631 @property
1632 def returning(self):
1633 """backwards compatibility; returns the
1634 effective_returning collection.
1636 """
1637 return self.effective_returning
1639 @property
1640 def current_executable(self):
1641 """Return the current 'executable' that is being compiled.
1643 This is currently the :class:`_sql.Select`, :class:`_sql.Insert`,
1644 :class:`_sql.Update`, :class:`_sql.Delete`,
1645 :class:`_sql.CompoundSelect` object that is being compiled.
1646 Specifically it's assigned to the ``self.stack`` list of elements.
1648 When a statement like the above is being compiled, it normally
1649 is also assigned to the ``.statement`` attribute of the
1650 :class:`_sql.Compiler` object. However, all SQL constructs are
1651 ultimately nestable, and this attribute should never be consulted
1652 by a ``visit_`` method, as it is not guaranteed to be assigned
1653 nor guaranteed to correspond to the current statement being compiled.
1655 """
1656 try:
1657 return self.stack[-1]["selectable"]
1658 except IndexError as ie:
1659 raise IndexError("Compiler does not have a stack entry") from ie
1661 @property
1662 def prefetch(self):
1663 return list(self.insert_prefetch) + list(self.update_prefetch)
1665 @util.memoized_property
1666 def _global_attributes(self) -> Dict[Any, Any]:
1667 return {}
1669 def _add_to_params(self, item: ExecutableStatement) -> None:
1670 # assumes that this is called before traversing the statement
1671 # so the call happens outer to inner, meaning that existing params
1672 # take precedence
1673 if item._params:
1674 self._collected_params = item._params | self._collected_params
1676 @util.memoized_instancemethod
1677 def _init_cte_state(self) -> MutableMapping[CTE, str]:
1678 """Initialize collections related to CTEs only if
1679 a CTE is located, to save on the overhead of
1680 these collections otherwise.
1682 """
1683 # collect CTEs to tack on top of a SELECT
1684 # To store the query to print - Dict[cte, text_query]
1685 ctes: MutableMapping[CTE, str] = util.OrderedDict()
1686 self.ctes = ctes
1688 # Detect same CTE references - Dict[(level, name), cte]
1689 # Level is required for supporting nesting
1690 self.ctes_by_level_name = {}
1692 # To retrieve key/level in ctes_by_level_name -
1693 # Dict[cte_reference, (level, cte_name, cte_opts)]
1694 self.level_name_by_cte = {}
1696 self.ctes_recursive = False
1698 return ctes
1700 @contextlib.contextmanager
1701 def _nested_result(self):
1702 """special API to support the use case of 'nested result sets'"""
1703 result_columns, ordered_columns = (
1704 self._result_columns,
1705 self._ordered_columns,
1706 )
1707 self._result_columns, self._ordered_columns = [], False
1709 try:
1710 if self.stack:
1711 entry = self.stack[-1]
1712 entry["need_result_map_for_nested"] = True
1713 else:
1714 entry = None
1715 yield self._result_columns, self._ordered_columns
1716 finally:
1717 if entry:
1718 entry.pop("need_result_map_for_nested")
1719 self._result_columns, self._ordered_columns = (
1720 result_columns,
1721 ordered_columns,
1722 )
1724 def _process_positional(self):
1725 assert not self.positiontup
1726 assert self.state is CompilerState.STRING_APPLIED
1727 assert not self._numeric_binds
1729 if self.dialect.paramstyle == "format":
1730 placeholder = "%s"
1731 else:
1732 assert self.dialect.paramstyle == "qmark"
1733 placeholder = "?"
1735 positions = []
1737 def find_position(m: re.Match[str]) -> str:
1738 normal_bind = m.group(1)
1739 if normal_bind:
1740 positions.append(normal_bind)
1741 return placeholder
1742 else:
1743 # this a post-compile bind
1744 positions.append(m.group(2))
1745 return m.group(0)
1747 self.string = re.sub(
1748 self._positional_pattern, find_position, self.string
1749 )
1751 if self.escaped_bind_names:
1752 reverse_escape = {v: k for k, v in self.escaped_bind_names.items()}
1753 assert len(self.escaped_bind_names) == len(reverse_escape)
1754 self.positiontup = [
1755 reverse_escape.get(name, name) for name in positions
1756 ]
1757 else:
1758 self.positiontup = positions
1760 if self._insertmanyvalues:
1761 positions = []
1763 single_values_expr = re.sub(
1764 self._positional_pattern,
1765 find_position,
1766 self._insertmanyvalues.single_values_expr,
1767 )
1768 insert_crud_params = [
1769 (
1770 v[0],
1771 v[1],
1772 re.sub(self._positional_pattern, find_position, v[2]),
1773 v[3],
1774 )
1775 for v in self._insertmanyvalues.insert_crud_params
1776 ]
1778 self._insertmanyvalues = self._insertmanyvalues._replace(
1779 single_values_expr=single_values_expr,
1780 insert_crud_params=insert_crud_params,
1781 )
1783 def _process_numeric(self):
1784 assert self._numeric_binds
1785 assert self.state is CompilerState.STRING_APPLIED
1787 num = 1
1788 param_pos: Dict[str, str] = {}
1789 order: Iterable[str]
1790 if self._insertmanyvalues and self._values_bindparam is not None:
1791 # bindparams that are not in values are always placed first.
1792 # this avoids the need of changing them when using executemany
1793 # values () ()
1794 order = itertools.chain(
1795 (
1796 name
1797 for name in self.bind_names.values()
1798 if name not in self._values_bindparam
1799 ),
1800 self.bind_names.values(),
1801 )
1802 else:
1803 order = self.bind_names.values()
1805 for bind_name in order:
1806 if bind_name in param_pos:
1807 continue
1808 bind = self.binds[bind_name]
1809 if (
1810 bind in self.post_compile_params
1811 or bind in self.literal_execute_params
1812 ):
1813 # set to None to just mark the in positiontup, it will not
1814 # be replaced below.
1815 param_pos[bind_name] = None # type: ignore[assignment]
1816 else:
1817 ph = f"{self._numeric_binds_identifier_char}{num}"
1818 num += 1
1819 param_pos[bind_name] = ph
1821 self.next_numeric_pos = num
1823 self.positiontup = list(param_pos)
1824 if self.escaped_bind_names:
1825 len_before = len(param_pos)
1826 param_pos = {
1827 self.escaped_bind_names.get(name, name): pos
1828 for name, pos in param_pos.items()
1829 }
1830 assert len(param_pos) == len_before
1832 # Can't use format here since % chars are not escaped.
1833 self.string = self._pyformat_pattern.sub(
1834 lambda m: param_pos[m.group(1)], self.string
1835 )
1837 if self._insertmanyvalues:
1838 single_values_expr = (
1839 # format is ok here since single_values_expr includes only
1840 # place-holders
1841 self._insertmanyvalues.single_values_expr
1842 % param_pos
1843 )
1844 insert_crud_params = [
1845 (v[0], v[1], "%s", v[3])
1846 for v in self._insertmanyvalues.insert_crud_params
1847 ]
1849 self._insertmanyvalues = self._insertmanyvalues._replace(
1850 # This has the numbers (:1, :2)
1851 single_values_expr=single_values_expr,
1852 # The single binds are instead %s so they can be formatted
1853 insert_crud_params=insert_crud_params,
1854 )
1856 @util.memoized_property
1857 def _bind_processors(
1858 self,
1859 ) -> MutableMapping[
1860 str, Union[_BindProcessorType[Any], Sequence[_BindProcessorType[Any]]]
1861 ]:
1862 # mypy is not able to see the two value types as the above Union,
1863 # it just sees "object". don't know how to resolve
1864 return {
1865 key: value # type: ignore[misc]
1866 for key, value in (
1867 (
1868 self.bind_names[bindparam],
1869 (
1870 bindparam.type._cached_bind_processor(self.dialect)
1871 if not bindparam.type._is_tuple_type
1872 else tuple(
1873 elem_type._cached_bind_processor(self.dialect)
1874 for elem_type in cast(
1875 TupleType, bindparam.type
1876 ).types
1877 )
1878 ),
1879 )
1880 for bindparam in self.bind_names
1881 # literal_execute parameters are rendered into the SQL string
1882 # via their literal_processor and never bound as values, so
1883 # they do not need a bind processor. Skipping them also avoids
1884 # invoking bind-processor construction that may require the
1885 # DBAPI to be present (see asyncpg, psycopgcffi cases),
1886 # facilitating testing.
1887 if bindparam not in self.literal_execute_params
1888 )
1889 if value is not None
1890 }
1892 def is_subquery(self):
1893 return len(self.stack) > 1
1895 @property
1896 def sql_compiler(self) -> Self:
1897 return self
1899 def construct_expanded_state(
1900 self,
1901 params: Optional[_CoreSingleExecuteParams] = None,
1902 escape_names: bool = True,
1903 ) -> ExpandedState:
1904 """Return a new :class:`.ExpandedState` for a given parameter set.
1906 For queries that use "expanding" or other late-rendered parameters,
1907 this method will provide for both the finalized SQL string as well
1908 as the parameters that would be used for a particular parameter set.
1910 .. versionadded:: 2.0.0rc1
1912 """
1913 parameters = self.construct_params(
1914 params,
1915 escape_names=escape_names,
1916 _no_postcompile=True,
1917 )
1918 return self._process_parameters_for_postcompile(
1919 parameters,
1920 )
1922 def construct_params(
1923 self,
1924 params: Optional[_CoreSingleExecuteParams] = None,
1925 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
1926 escape_names: bool = True,
1927 _group_number: Optional[int] = None,
1928 _check: bool = True,
1929 _no_postcompile: bool = False,
1930 _collected_params: _CoreSingleExecuteParams | None = None,
1931 ) -> _MutableCoreSingleExecuteParams:
1932 """return a dictionary of bind parameter keys and values"""
1933 if _collected_params is not None:
1934 assert not self._collect_params
1935 elif self._collect_params:
1936 _collected_params = self._collected_params
1938 if _collected_params:
1939 if not params:
1940 params = _collected_params
1941 else:
1942 params = {**_collected_params, **params}
1944 if self._render_postcompile and not _no_postcompile:
1945 assert self._post_compile_expanded_state is not None
1946 if not params:
1947 return dict(self._post_compile_expanded_state.parameters)
1948 else:
1949 raise exc.InvalidRequestError(
1950 "can't construct new parameters when render_postcompile "
1951 "is used; the statement is hard-linked to the original "
1952 "parameters. Use construct_expanded_state to generate a "
1953 "new statement and parameters."
1954 )
1956 has_escaped_names = escape_names and bool(self.escaped_bind_names)
1958 if extracted_parameters:
1959 # related the bound parameters collected in the original cache key
1960 # to those collected in the incoming cache key. They will not have
1961 # matching names but they will line up positionally in the same
1962 # way. The parameters present in self.bind_names may be clones of
1963 # these original cache key params in the case of DML but the .key
1964 # will be guaranteed to match.
1965 if self.cache_key is None:
1966 raise exc.CompileError(
1967 "This compiled object has no original cache key; "
1968 "can't pass extracted_parameters to construct_params"
1969 )
1970 else:
1971 orig_extracted = self.cache_key[1]
1973 ckbm_tuple = self._cache_key_bind_match
1974 assert ckbm_tuple is not None
1975 ckbm, _ = ckbm_tuple
1976 resolved_extracted = {
1977 bind: extracted
1978 for b, extracted in zip(orig_extracted, extracted_parameters)
1979 for bind in ckbm[b]
1980 }
1981 else:
1982 resolved_extracted = None
1984 if params:
1985 pd = {}
1986 for bindparam, name in self.bind_names.items():
1987 escaped_name = (
1988 self.escaped_bind_names.get(name, name)
1989 if has_escaped_names
1990 else name
1991 )
1993 if bindparam.key in params:
1994 pd[escaped_name] = params[bindparam.key]
1995 elif name in params:
1996 pd[escaped_name] = params[name]
1998 elif _check and bindparam.required:
1999 if _group_number:
2000 raise exc.InvalidRequestError(
2001 "A value is required for bind parameter %r, "
2002 "in parameter group %d"
2003 % (bindparam.key, _group_number),
2004 code="cd3x",
2005 )
2006 else:
2007 raise exc.InvalidRequestError(
2008 "A value is required for bind parameter %r"
2009 % bindparam.key,
2010 code="cd3x",
2011 )
2012 else:
2013 if resolved_extracted:
2014 value_param = resolved_extracted.get(
2015 bindparam, bindparam
2016 )
2017 else:
2018 value_param = bindparam
2020 if bindparam.callable:
2021 pd[escaped_name] = value_param.effective_value
2022 else:
2023 pd[escaped_name] = value_param.value
2024 return pd
2025 else:
2026 pd = {}
2027 for bindparam, name in self.bind_names.items():
2028 escaped_name = (
2029 self.escaped_bind_names.get(name, name)
2030 if has_escaped_names
2031 else name
2032 )
2034 if _check and bindparam.required:
2035 if _group_number:
2036 raise exc.InvalidRequestError(
2037 "A value is required for bind parameter %r, "
2038 "in parameter group %d"
2039 % (bindparam.key, _group_number),
2040 code="cd3x",
2041 )
2042 else:
2043 raise exc.InvalidRequestError(
2044 "A value is required for bind parameter %r"
2045 % bindparam.key,
2046 code="cd3x",
2047 )
2049 if resolved_extracted:
2050 value_param = resolved_extracted.get(bindparam, bindparam)
2051 else:
2052 value_param = bindparam
2054 if bindparam.callable:
2055 pd[escaped_name] = value_param.effective_value
2056 else:
2057 pd[escaped_name] = value_param.value
2059 return pd
2061 @util.memoized_instancemethod
2062 def _get_set_input_sizes_lookup(self):
2063 dialect = self.dialect
2065 include_types = dialect.include_set_input_sizes
2066 exclude_types = dialect.exclude_set_input_sizes
2068 dbapi = dialect.dbapi
2070 def lookup_type(typ):
2071 dbtype = typ._unwrapped_dialect_impl(dialect).get_dbapi_type(dbapi)
2073 if (
2074 dbtype is not None
2075 and (exclude_types is None or dbtype not in exclude_types)
2076 and (include_types is None or dbtype in include_types)
2077 ):
2078 return dbtype
2079 else:
2080 return None
2082 inputsizes = {}
2084 literal_execute_params = self.literal_execute_params
2086 for bindparam in self.bind_names:
2087 if bindparam in literal_execute_params:
2088 continue
2090 if bindparam.type._is_tuple_type:
2091 inputsizes[bindparam] = [
2092 lookup_type(typ)
2093 for typ in cast(TupleType, bindparam.type).types
2094 ]
2095 else:
2096 inputsizes[bindparam] = lookup_type(bindparam.type)
2098 return inputsizes
2100 @property
2101 def params(self):
2102 """Return the bind param dictionary embedded into this
2103 compiled object, for those values that are present.
2105 .. seealso::
2107 :ref:`faq_sql_expression_string` - includes a usage example for
2108 debugging use cases.
2110 """
2111 return self.construct_params(_check=False)
2113 def _process_parameters_for_postcompile(
2114 self,
2115 parameters: _MutableCoreSingleExecuteParams,
2116 _populate_self: bool = False,
2117 ) -> ExpandedState:
2118 """handle special post compile parameters.
2120 These include:
2122 * "expanding" parameters -typically IN tuples that are rendered
2123 on a per-parameter basis for an otherwise fixed SQL statement string.
2125 * literal_binds compiled with the literal_execute flag. Used for
2126 things like SQL Server "TOP N" where the driver does not accommodate
2127 N as a bound parameter.
2129 """
2131 expanded_parameters = {}
2132 new_positiontup: Optional[List[str]]
2134 pre_expanded_string = self._pre_expanded_string
2135 if pre_expanded_string is None:
2136 pre_expanded_string = self.string
2138 if self.positional:
2139 new_positiontup = []
2141 pre_expanded_positiontup = self._pre_expanded_positiontup
2142 if pre_expanded_positiontup is None:
2143 pre_expanded_positiontup = self.positiontup
2145 else:
2146 new_positiontup = pre_expanded_positiontup = None
2148 processors = self._bind_processors
2149 single_processors = cast(
2150 "Mapping[str, _BindProcessorType[Any]]", processors
2151 )
2152 tuple_processors = cast(
2153 "Mapping[str, Sequence[_BindProcessorType[Any]]]", processors
2154 )
2156 new_processors: Dict[str, _BindProcessorType[Any]] = {}
2158 replacement_expressions: Dict[str, Any] = {}
2159 to_update_sets: Dict[str, Any] = {}
2161 # notes:
2162 # *unescaped* parameter names in:
2163 # self.bind_names, self.binds, self._bind_processors, self.positiontup
2164 #
2165 # *escaped* parameter names in:
2166 # construct_params(), replacement_expressions
2168 numeric_positiontup: Optional[List[str]] = None
2170 if self.positional and pre_expanded_positiontup is not None:
2171 names: Iterable[str] = pre_expanded_positiontup
2172 if self._numeric_binds:
2173 numeric_positiontup = []
2174 else:
2175 names = self.bind_names.values()
2177 ebn = self.escaped_bind_names
2178 for name in names:
2179 escaped_name = ebn.get(name, name) if ebn else name
2180 parameter = self.binds[name]
2182 if parameter in self.literal_execute_params:
2183 if escaped_name not in replacement_expressions:
2184 replacement_expressions[escaped_name] = (
2185 self.render_literal_bindparam(
2186 parameter,
2187 render_literal_value=parameters.pop(escaped_name),
2188 )
2189 )
2190 continue
2192 if parameter in self.post_compile_params:
2193 if escaped_name in replacement_expressions:
2194 to_update = to_update_sets[escaped_name]
2195 values = None
2196 else:
2197 # we are removing the parameter from parameters
2198 # because it is a list value, which is not expected by
2199 # TypeEngine objects that would otherwise be asked to
2200 # process it. the single name is being replaced with
2201 # individual numbered parameters for each value in the
2202 # param.
2203 #
2204 # note we are also inserting *escaped* parameter names
2205 # into the given dictionary. default dialect will
2206 # use these param names directly as they will not be
2207 # in the escaped_bind_names dictionary.
2208 values = parameters.pop(name)
2210 leep_res = self._literal_execute_expanding_parameter(
2211 escaped_name, parameter, values
2212 )
2213 to_update, replacement_expr = leep_res
2215 to_update_sets[escaped_name] = to_update
2216 replacement_expressions[escaped_name] = replacement_expr
2218 if not parameter.literal_execute:
2219 parameters.update(to_update)
2220 if parameter.type._is_tuple_type:
2221 assert values is not None
2222 new_processors.update(
2223 (
2224 "%s_%s_%s" % (name, i, j),
2225 tuple_processors[name][j - 1],
2226 )
2227 for i, tuple_element in enumerate(values, 1)
2228 for j, _ in enumerate(tuple_element, 1)
2229 if name in tuple_processors
2230 and tuple_processors[name][j - 1] is not None
2231 )
2232 else:
2233 new_processors.update(
2234 (key, single_processors[name])
2235 for key, _ in to_update
2236 if name in single_processors
2237 )
2238 if numeric_positiontup is not None:
2239 numeric_positiontup.extend(
2240 name for name, _ in to_update
2241 )
2242 elif new_positiontup is not None:
2243 # to_update has escaped names, but that's ok since
2244 # these are new names, that aren't in the
2245 # escaped_bind_names dict.
2246 new_positiontup.extend(name for name, _ in to_update)
2247 expanded_parameters[name] = [
2248 expand_key for expand_key, _ in to_update
2249 ]
2250 elif new_positiontup is not None:
2251 new_positiontup.append(name)
2253 def process_expanding(m):
2254 key = m.group(1)
2255 expr = replacement_expressions[key]
2257 # if POSTCOMPILE included a bind_expression, render that
2258 # around each element
2259 if m.group(2):
2260 tok = m.group(2).split("~~")
2261 be_left, be_right = tok[1], tok[3]
2262 expr = ", ".join(
2263 "%s%s%s" % (be_left, exp, be_right)
2264 for exp in expr.split(", ")
2265 )
2266 return expr
2268 statement = re.sub(
2269 self._post_compile_pattern, process_expanding, pre_expanded_string
2270 )
2272 if numeric_positiontup is not None:
2273 assert new_positiontup is not None
2274 param_pos = {
2275 key: f"{self._numeric_binds_identifier_char}{num}"
2276 for num, key in enumerate(
2277 numeric_positiontup, self.next_numeric_pos
2278 )
2279 }
2280 # Can't use format here since % chars are not escaped.
2281 statement = self._pyformat_pattern.sub(
2282 lambda m: param_pos[m.group(1)], statement
2283 )
2284 new_positiontup.extend(numeric_positiontup)
2286 expanded_state = ExpandedState(
2287 statement,
2288 parameters,
2289 new_processors,
2290 new_positiontup,
2291 expanded_parameters,
2292 )
2294 if _populate_self:
2295 # this is for the "render_postcompile" flag, which is not
2296 # otherwise used internally and is for end-user debugging and
2297 # special use cases.
2298 self._pre_expanded_string = pre_expanded_string
2299 self._pre_expanded_positiontup = pre_expanded_positiontup
2300 self.string = expanded_state.statement
2301 self.positiontup = (
2302 list(expanded_state.positiontup or ())
2303 if self.positional
2304 else None
2305 )
2306 self._post_compile_expanded_state = expanded_state
2308 return expanded_state
2310 @util.preload_module("sqlalchemy.engine.cursor")
2311 def _create_result_map(self):
2312 """utility method used for unit tests only."""
2313 cursor = util.preloaded.engine_cursor
2314 return cursor.CursorResultMetaData._create_description_match_map(
2315 self._result_columns
2316 )
2318 # assigned by crud.py for insert/update statements
2319 _get_bind_name_for_col: _BindNameForColProtocol
2321 @util.memoized_property
2322 def _within_exec_param_key_getter(self) -> Callable[[Any], str]:
2323 getter = self._get_bind_name_for_col
2324 return getter
2326 @util.memoized_property
2327 @util.preload_module("sqlalchemy.engine.result")
2328 def _inserted_primary_key_from_lastrowid_getter(self):
2329 result = util.preloaded.engine_result
2331 param_key_getter = self._within_exec_param_key_getter
2333 assert self.compile_state is not None
2334 statement = self.compile_state.statement
2336 if TYPE_CHECKING:
2337 assert isinstance(statement, Insert)
2339 table = statement.table
2341 getters = [
2342 (operator.methodcaller("get", param_key_getter(col), None), col)
2343 for col in table.primary_key
2344 ]
2346 autoinc_getter = None
2347 autoinc_col = table._autoincrement_column
2348 if autoinc_col is not None:
2349 # apply type post processors to the lastrowid
2350 lastrowid_processor = autoinc_col.type._cached_result_processor(
2351 self.dialect, None
2352 )
2353 autoinc_key = param_key_getter(autoinc_col)
2355 # if a bind value is present for the autoincrement column
2356 # in the parameters, we need to do the logic dictated by
2357 # #7998; honor a non-None user-passed parameter over lastrowid.
2358 # previously in the 1.4 series we weren't fetching lastrowid
2359 # at all if the key were present in the parameters
2360 if autoinc_key in self.binds:
2362 def _autoinc_getter(lastrowid, parameters):
2363 param_value = parameters.get(autoinc_key, lastrowid)
2364 if param_value is not None:
2365 # they supplied non-None parameter, use that.
2366 # SQLite at least is observed to return the wrong
2367 # cursor.lastrowid for INSERT..ON CONFLICT so it
2368 # can't be used in all cases
2369 return param_value
2370 else:
2371 # use lastrowid
2372 return lastrowid
2374 # work around mypy https://github.com/python/mypy/issues/14027
2375 autoinc_getter = _autoinc_getter
2377 else:
2378 lastrowid_processor = None
2380 row_fn = result.result_tuple([col.key for col in table.primary_key])
2382 def get(lastrowid, parameters):
2383 """given cursor.lastrowid value and the parameters used for INSERT,
2384 return a "row" that represents the primary key, either by
2385 using the "lastrowid" or by extracting values from the parameters
2386 that were sent along with the INSERT.
2388 """
2389 if lastrowid_processor is not None:
2390 lastrowid = lastrowid_processor(lastrowid)
2392 if lastrowid is None:
2393 return row_fn(getter(parameters) for getter, col in getters)
2394 else:
2395 return row_fn(
2396 (
2397 (
2398 autoinc_getter(lastrowid, parameters)
2399 if autoinc_getter is not None
2400 else lastrowid
2401 )
2402 if col is autoinc_col
2403 else getter(parameters)
2404 )
2405 for getter, col in getters
2406 )
2408 return get
2410 @util.memoized_property
2411 @util.preload_module("sqlalchemy.engine.result")
2412 def _inserted_primary_key_from_returning_getter(self):
2413 result = util.preloaded.engine_result
2415 assert self.compile_state is not None
2416 statement = self.compile_state.statement
2418 if TYPE_CHECKING:
2419 assert isinstance(statement, Insert)
2421 param_key_getter = self._within_exec_param_key_getter
2422 table = statement.table
2424 returning = self.implicit_returning
2425 assert returning is not None
2426 ret = {col: idx for idx, col in enumerate(returning)}
2428 getters = cast(
2429 "List[Tuple[Callable[[Any], Any], bool]]",
2430 [
2431 (
2432 (operator.itemgetter(ret[col]), True)
2433 if col in ret
2434 else (
2435 operator.methodcaller(
2436 "get", param_key_getter(col), None
2437 ),
2438 False,
2439 )
2440 )
2441 for col in table.primary_key
2442 ],
2443 )
2445 row_fn = result.result_tuple([col.key for col in table.primary_key])
2447 def get(row, parameters):
2448 return row_fn(
2449 getter(row) if use_row else getter(parameters)
2450 for getter, use_row in getters
2451 )
2453 return get
2455 def default_from(self) -> str:
2456 """Called when a SELECT statement has no froms, and no FROM clause is
2457 to be appended.
2459 Gives Oracle Database a chance to tack on a ``FROM DUAL`` to the string
2460 output.
2462 """
2463 return ""
2465 def visit_override_binds(
2466 self,
2467 override_binds,
2468 add_to_result_map=None,
2469 result_map_targets=(),
2470 **kw,
2471 ):
2472 """SQL compile the nested element of an _OverrideBinds with
2473 bindparams swapped out.
2475 The _OverrideBinds is not normally expected to be compiled; it
2476 is meant to be used when an already cached statement is to be used,
2477 the compilation was already performed, and only the bound params should
2478 be swapped in at execution time.
2480 However, there are test cases that exericise this object, and
2481 additionally the ORM subquery loader is known to feed in expressions
2482 which include this construct into new queries (discovered in #11173),
2483 so it has to do the right thing at compile time as well.
2485 """
2487 if add_to_result_map is not None:
2488 # the ORM looks up result columns using the objects it placed
2489 # into the loader option, which for a cached statement is the
2490 # _OverrideBinds wrapper rather than the element inside of it.
2491 # make sure the wrapper is present in the result map so that
2492 # lookup succeeds. See #13560
2493 result_map_targets += (override_binds,)
2494 kw["add_to_result_map"] = add_to_result_map
2495 kw["result_map_targets"] = result_map_targets
2497 # get SQL text first
2498 sqltext = override_binds.element._compiler_dispatch(self, **kw)
2500 # for a test compile that is not for caching, change binds after the
2501 # fact. note that we don't try to
2502 # swap the bindparam as we compile, because our element may be
2503 # elsewhere in the statement already (e.g. a subquery or perhaps a
2504 # CTE) and was already visited / compiled. See
2505 # test_relationship_criteria.py ->
2506 # test_selectinload_local_criteria_subquery
2507 for k in override_binds.translate:
2508 if k not in self.binds:
2509 continue
2510 bp = self.binds[k]
2512 # so this would work, just change the value of bp in place.
2513 # but we dont want to mutate things outside.
2514 # bp.value = override_binds.translate[bp.key]
2515 # continue
2517 # instead, need to replace bp with new_bp or otherwise accommodate
2518 # in all internal collections
2519 new_bp = bp._with_value(
2520 override_binds.translate[bp.key],
2521 maintain_key=True,
2522 required=False,
2523 )
2525 name = self.bind_names[bp]
2526 self.binds[k] = self.binds[name] = new_bp
2527 self.bind_names[new_bp] = name
2528 self.bind_names.pop(bp, None)
2530 if bp in self.post_compile_params:
2531 self.post_compile_params |= {new_bp}
2532 if bp in self.literal_execute_params:
2533 self.literal_execute_params |= {new_bp}
2535 ckbm_tuple = self._cache_key_bind_match
2536 if ckbm_tuple:
2537 ckbm, cksm = ckbm_tuple
2538 for bp in bp._cloned_set:
2539 if bp.key in cksm:
2540 cb = cksm[bp.key]
2541 ckbm[cb].append(new_bp)
2543 return sqltext
2545 def visit_grouping(self, grouping, asfrom=False, **kwargs):
2546 return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")"
2548 def visit_select_statement_grouping(self, grouping, **kwargs):
2549 return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")"
2551 def visit_label_reference(
2552 self, element, within_columns_clause=False, **kwargs
2553 ):
2554 if self.stack and self.dialect.supports_simple_order_by_label:
2555 try:
2556 compile_state = cast(
2557 "Union[SelectState, CompoundSelectState]",
2558 self.stack[-1]["compile_state"],
2559 )
2560 except KeyError as ke:
2561 raise exc.CompileError(
2562 "Can't resolve label reference for ORDER BY / "
2563 "GROUP BY / DISTINCT etc."
2564 ) from ke
2566 (
2567 with_cols,
2568 only_froms,
2569 only_cols,
2570 ) = compile_state._label_resolve_dict
2571 if within_columns_clause:
2572 resolve_dict = only_froms
2573 else:
2574 resolve_dict = only_cols
2576 # this can be None in the case that a _label_reference()
2577 # were subject to a replacement operation, in which case
2578 # the replacement of the Label element may have changed
2579 # to something else like a ColumnClause expression.
2580 order_by_elem = element.element._order_by_label_element
2582 if (
2583 order_by_elem is not None
2584 and order_by_elem.name in resolve_dict
2585 and order_by_elem.shares_lineage(
2586 resolve_dict[order_by_elem.name]
2587 )
2588 ):
2589 kwargs["render_label_as_label"] = (
2590 element.element._order_by_label_element
2591 )
2592 return self.process(
2593 element.element,
2594 within_columns_clause=within_columns_clause,
2595 **kwargs,
2596 )
2598 def visit_textual_label_reference(
2599 self, element, within_columns_clause=False, **kwargs
2600 ):
2601 if not self.stack:
2602 # compiling the element outside of the context of a SELECT
2603 return self.process(element._text_clause)
2605 try:
2606 compile_state = cast(
2607 "Union[SelectState, CompoundSelectState]",
2608 self.stack[-1]["compile_state"],
2609 )
2610 except KeyError as ke:
2611 coercions._no_text_coercion(
2612 element.element,
2613 extra=(
2614 "Can't resolve label reference for ORDER BY / "
2615 "GROUP BY / DISTINCT etc."
2616 ),
2617 exc_cls=exc.CompileError,
2618 err=ke,
2619 )
2621 with_cols, only_froms, only_cols = compile_state._label_resolve_dict
2622 try:
2623 if within_columns_clause:
2624 col = only_froms[element.element]
2625 else:
2626 col = with_cols[element.element]
2627 except KeyError as err:
2628 coercions._no_text_coercion(
2629 element.element,
2630 extra=(
2631 "Can't resolve label reference for ORDER BY / "
2632 "GROUP BY / DISTINCT etc."
2633 ),
2634 exc_cls=exc.CompileError,
2635 err=err,
2636 )
2637 else:
2638 kwargs["render_label_as_label"] = col
2639 return self.process(
2640 col, within_columns_clause=within_columns_clause, **kwargs
2641 )
2643 def visit_label(
2644 self,
2645 label,
2646 add_to_result_map=None,
2647 within_label_clause=False,
2648 within_columns_clause=False,
2649 render_label_as_label=None,
2650 result_map_targets=(),
2651 within_tstring=False,
2652 **kw,
2653 ):
2654 if within_tstring:
2655 raise exc.CompileError(
2656 "Using label() directly inside tstring is not supported "
2657 "as it is ambiguous how the label expression should be "
2658 "rendered without knowledge of how it's being used in SQL"
2659 )
2660 # only render labels within the columns clause
2661 # or ORDER BY clause of a select. dialect-specific compilers
2662 # can modify this behavior.
2663 render_label_with_as = (
2664 within_columns_clause and not within_label_clause
2665 )
2666 render_label_only = render_label_as_label is label
2668 if render_label_only or render_label_with_as:
2669 if isinstance(label.name, elements._truncated_label):
2670 labelname = self._truncated_identifier("colident", label.name)
2671 else:
2672 labelname = label.name
2674 if render_label_with_as:
2675 if add_to_result_map is not None:
2676 add_to_result_map(
2677 labelname,
2678 label.name,
2679 (label, labelname) + label._alt_names + result_map_targets,
2680 label.type,
2681 )
2682 return (
2683 label.element._compiler_dispatch(
2684 self,
2685 within_columns_clause=True,
2686 within_label_clause=True,
2687 **kw,
2688 )
2689 + OPERATORS[operators.as_]
2690 + self.preparer.format_label(label, labelname)
2691 )
2692 elif render_label_only:
2693 return self.preparer.format_label(label, labelname)
2694 else:
2695 return label.element._compiler_dispatch(
2696 self, within_columns_clause=False, **kw
2697 )
2699 def _fallback_column_name(self, column):
2700 raise exc.CompileError(
2701 "Cannot compile Column object until its 'name' is assigned."
2702 )
2704 def visit_lambda_element(self, element, **kw):
2705 sql_element = element._resolved
2706 return self.process(sql_element, **kw)
2708 def visit_column(
2709 self,
2710 column: ColumnClause[Any],
2711 add_to_result_map: Optional[_ResultMapAppender] = None,
2712 include_table: bool = True,
2713 result_map_targets: Tuple[Any, ...] = (),
2714 ambiguous_table_name_map: Optional[_AmbiguousTableNameMap] = None,
2715 **kwargs: Any,
2716 ) -> str:
2717 name = orig_name = column.name
2718 if name is None:
2719 name = self._fallback_column_name(column)
2721 is_literal = column.is_literal
2722 if not is_literal and isinstance(name, elements._truncated_label):
2723 name = self._truncated_identifier("colident", name)
2725 if add_to_result_map is not None:
2726 targets = (column, name, column.key) + result_map_targets
2727 if column._tq_label:
2728 targets += (column._tq_label,)
2730 add_to_result_map(name, orig_name, targets, column.type)
2732 if is_literal:
2733 # note we are not currently accommodating for
2734 # literal_column(quoted_name('ident', True)) here
2735 name = self.escape_literal_column(name)
2736 else:
2737 name = self.preparer.quote(name)
2738 table = column.table
2739 if table is None or not include_table or not table.named_with_column:
2740 return name
2741 else:
2742 effective_schema = self.preparer.schema_for_object(table)
2744 if effective_schema:
2745 schema_prefix = (
2746 self.preparer.quote_schema(effective_schema) + "."
2747 )
2748 else:
2749 schema_prefix = ""
2751 if TYPE_CHECKING:
2752 assert isinstance(table, NamedFromClause)
2753 tablename = table.name
2755 if (
2756 not effective_schema
2757 and ambiguous_table_name_map
2758 and tablename in ambiguous_table_name_map
2759 ):
2760 tablename = ambiguous_table_name_map[tablename]
2762 if isinstance(tablename, elements._truncated_label):
2763 tablename = self._truncated_identifier("alias", tablename)
2765 return schema_prefix + self.preparer.quote(tablename) + "." + name
2767 def visit_collation(self, element, **kw):
2768 return self.preparer.format_collation(
2769 element.collation, element.collation_schema
2770 )
2772 def visit_fromclause(self, fromclause, **kwargs):
2773 return fromclause.name
2775 def visit_index(self, index, **kwargs):
2776 return index.name
2778 def visit_typeclause(self, typeclause, **kw):
2779 kw["type_expression"] = typeclause
2780 kw["identifier_preparer"] = self.preparer
2781 return self.dialect.type_compiler_instance.process(
2782 typeclause.type, **kw
2783 )
2785 def post_process_text(self, text):
2786 if self.preparer._double_percents:
2787 text = text.replace("%", "%%")
2788 return text
2790 def escape_literal_column(self, text):
2791 if self.preparer._double_percents:
2792 text = text.replace("%", "%%")
2793 return text
2795 def visit_textclause(self, textclause, add_to_result_map=None, **kw):
2796 if self._collect_params:
2797 self._add_to_params(textclause)
2799 def do_bindparam(m):
2800 name = m.group(1)
2801 if name in textclause._bindparams:
2802 return self.process(textclause._bindparams[name], **kw)
2803 else:
2804 return self.bindparam_string(name, **kw)
2806 if not self.stack:
2807 self.isplaintext = True
2809 if add_to_result_map:
2810 # text() object is present in the columns clause of a
2811 # select(). Add a no-name entry to the result map so that
2812 # row[text()] produces a result
2813 add_to_result_map(None, None, (textclause,), sqltypes.NULLTYPE)
2815 # un-escape any \:params
2816 return BIND_PARAMS_ESC.sub(
2817 lambda m: m.group(1),
2818 BIND_PARAMS.sub(
2819 do_bindparam, self.post_process_text(textclause.text)
2820 ),
2821 )
2823 def visit_tstring(self, tstring, add_to_result_map=None, **kw):
2824 if self._collect_params:
2825 self._add_to_params(tstring)
2827 if not self.stack:
2828 self.isplaintext = True
2830 if add_to_result_map:
2831 # tstring() object is present in the columns clause of a
2832 # select(). Add a no-name entry to the result map so that
2833 # row[tstring()] produces a result
2834 add_to_result_map(None, None, (tstring,), sqltypes.NULLTYPE)
2836 # Process each part and concatenate
2837 kw["within_tstring"] = True
2838 return "".join(self.process(part, **kw) for part in tstring.parts)
2840 def visit_textual_select(
2841 self, taf, compound_index=None, asfrom=False, **kw
2842 ):
2843 if self._collect_params:
2844 self._add_to_params(taf)
2845 toplevel = not self.stack
2846 entry = self._default_stack_entry if toplevel else self.stack[-1]
2848 new_entry: _CompilerStackEntry = {
2849 "correlate_froms": set(),
2850 "asfrom_froms": set(),
2851 "selectable": taf,
2852 }
2853 self.stack.append(new_entry)
2855 if taf._independent_ctes:
2856 self._dispatch_independent_ctes(taf, kw)
2858 populate_result_map = (
2859 toplevel
2860 or (
2861 compound_index == 0
2862 and entry.get("need_result_map_for_compound", False)
2863 )
2864 or entry.get("need_result_map_for_nested", False)
2865 )
2867 if populate_result_map:
2868 self._ordered_columns = self._textual_ordered_columns = (
2869 taf.positional
2870 )
2872 # enable looser result column matching when the SQL text links to
2873 # Column objects by name only
2874 self._loose_column_name_matching = not taf.positional and bool(
2875 taf.column_args
2876 )
2878 for c in taf.column_args:
2879 self.process(
2880 c,
2881 within_columns_clause=True,
2882 add_to_result_map=self._add_to_result_map,
2883 )
2885 text = self.process(taf.element, **kw)
2886 if self.ctes:
2887 nesting_level = len(self.stack) if not toplevel else None
2888 text = self._render_cte_clause(nesting_level=nesting_level) + text
2890 self.stack.pop(-1)
2892 return text
2894 def visit_null(self, expr: Null, **kw: Any) -> str:
2895 return "NULL"
2897 def visit_true(self, expr: True_, **kw: Any) -> str:
2898 if self.dialect.supports_native_boolean:
2899 return "true"
2900 else:
2901 return "1"
2903 def visit_false(self, expr: False_, **kw: Any) -> str:
2904 if self.dialect.supports_native_boolean:
2905 return "false"
2906 else:
2907 return "0"
2909 def _generate_delimited_list(self, elements, separator, **kw):
2910 return separator.join(
2911 s
2912 for s in (c._compiler_dispatch(self, **kw) for c in elements)
2913 if s
2914 )
2916 def _generate_delimited_and_list(self, clauses, **kw):
2917 lcc, clauses = elements.BooleanClauseList._process_clauses_for_boolean(
2918 operators.and_,
2919 elements.True_._singleton,
2920 elements.False_._singleton,
2921 clauses,
2922 )
2923 if lcc == 1:
2924 return clauses[0]._compiler_dispatch(self, **kw)
2925 else:
2926 separator = OPERATORS[operators.and_]
2927 return separator.join(
2928 s
2929 for s in (c._compiler_dispatch(self, **kw) for c in clauses)
2930 if s
2931 )
2933 def visit_tuple(self, clauselist, **kw):
2934 return "(%s)" % self.visit_clauselist(clauselist, **kw)
2936 def visit_element_list(self, element, **kw):
2937 return self._generate_delimited_list(element.clauses, " ", **kw)
2939 def visit_order_by_list(self, element, **kw):
2940 return self._generate_delimited_list(element.clauses, ", ", **kw)
2942 def visit_clauselist(self, clauselist, **kw):
2943 sep = clauselist.operator
2944 if sep is None:
2945 sep = " "
2946 else:
2947 sep = OPERATORS[clauselist.operator]
2949 return self._generate_delimited_list(clauselist.clauses, sep, **kw)
2951 def visit_expression_clauselist(self, clauselist, **kw):
2952 operator_ = clauselist.operator
2954 disp = self._get_operator_dispatch(
2955 operator_, "expression_clauselist", None
2956 )
2957 if disp:
2958 return disp(clauselist, operator_, **kw)
2960 try:
2961 opstring = OPERATORS[operator_]
2962 except KeyError as err:
2963 raise exc.UnsupportedCompilationError(self, operator_) from err
2964 else:
2965 kw["_in_operator_expression"] = True
2966 return self._generate_delimited_list(
2967 clauselist.clauses, opstring, **kw
2968 )
2970 def visit_case(self, clause, **kwargs):
2971 x = "CASE "
2972 if clause.value is not None:
2973 x += clause.value._compiler_dispatch(self, **kwargs) + " "
2974 for cond, result in clause.whens:
2975 x += (
2976 "WHEN "
2977 + cond._compiler_dispatch(self, **kwargs)
2978 + " THEN "
2979 + result._compiler_dispatch(self, **kwargs)
2980 + " "
2981 )
2982 if clause.else_ is not None:
2983 x += (
2984 "ELSE " + clause.else_._compiler_dispatch(self, **kwargs) + " "
2985 )
2986 x += "END"
2987 return x
2989 def visit_type_coerce(self, type_coerce, **kw):
2990 return type_coerce.typed_expression._compiler_dispatch(self, **kw)
2992 def visit_cast(self, cast, **kwargs):
2993 type_clause = cast.typeclause._compiler_dispatch(self, **kwargs)
2994 match = re.match("(.*)( COLLATE .*)", type_clause)
2995 return "CAST(%s AS %s)%s" % (
2996 cast.clause._compiler_dispatch(self, **kwargs),
2997 match.group(1) if match else type_clause,
2998 match.group(2) if match else "",
2999 )
3001 def visit_frame_clause(self, frameclause, **kw):
3003 if frameclause.lower_type is elements.FrameClauseType.UNBOUNDED:
3004 left = "UNBOUNDED PRECEDING"
3005 elif frameclause.lower_type is elements.FrameClauseType.CURRENT:
3006 left = "CURRENT ROW"
3007 else:
3008 val = self.process(frameclause.lower_bind, **kw)
3009 if frameclause.lower_type is elements.FrameClauseType.PRECEDING:
3010 left = f"{val} PRECEDING"
3011 else:
3012 left = f"{val} FOLLOWING"
3014 if frameclause.upper_type is elements.FrameClauseType.UNBOUNDED:
3015 right = "UNBOUNDED FOLLOWING"
3016 elif frameclause.upper_type is elements.FrameClauseType.CURRENT:
3017 right = "CURRENT ROW"
3018 else:
3019 val = self.process(frameclause.upper_bind, **kw)
3020 if frameclause.upper_type is elements.FrameClauseType.PRECEDING:
3021 right = f"{val} PRECEDING"
3022 else:
3023 right = f"{val} FOLLOWING"
3025 return f"{left} AND {right}"
3027 def visit_over(self, over, **kwargs):
3028 text = over.element._compiler_dispatch(self, **kwargs)
3029 if over.range_ is not None:
3030 range_ = f"RANGE BETWEEN {self.process(over.range_, **kwargs)}"
3031 elif over.rows is not None:
3032 range_ = f"ROWS BETWEEN {self.process(over.rows, **kwargs)}"
3033 elif over.groups is not None:
3034 range_ = f"GROUPS BETWEEN {self.process(over.groups, **kwargs)}"
3035 else:
3036 range_ = None
3038 if range_ is not None and over.exclude is not None:
3039 range_ += " EXCLUDE " + self.preparer.validate_sql_phrase(
3040 over.exclude, _WINDOW_EXCLUDE_RE
3041 )
3043 return "%s OVER (%s)" % (
3044 text,
3045 " ".join(
3046 [
3047 "%s BY %s"
3048 % (word, clause._compiler_dispatch(self, **kwargs))
3049 for word, clause in (
3050 ("PARTITION", over.partition_by),
3051 ("ORDER", over.order_by),
3052 )
3053 if clause is not None and len(clause)
3054 ]
3055 + ([range_] if range_ else [])
3056 ),
3057 )
3059 def visit_withingroup(self, withingroup, **kwargs):
3060 return "%s WITHIN GROUP (ORDER BY %s)" % (
3061 withingroup.element._compiler_dispatch(self, **kwargs),
3062 withingroup.order_by._compiler_dispatch(self, **kwargs),
3063 )
3065 def visit_funcfilter(self, funcfilter, **kwargs):
3066 return "%s FILTER (WHERE %s)" % (
3067 funcfilter.func._compiler_dispatch(self, **kwargs),
3068 funcfilter.criterion._compiler_dispatch(self, **kwargs),
3069 )
3071 def visit_aggregateorderby(self, aggregateorderby, **kwargs):
3072 if self.dialect.aggregate_order_by_style is AggregateOrderByStyle.NONE:
3073 raise exc.CompileError(
3074 "this dialect does not support "
3075 "ORDER BY within an aggregate function"
3076 )
3077 elif (
3078 self.dialect.aggregate_order_by_style
3079 is AggregateOrderByStyle.INLINE
3080 ):
3081 new_fn = aggregateorderby.element._clone()
3082 new_fn.clause_expr = elements.Grouping(
3083 aggregate_orderby_inline(
3084 new_fn.clause_expr.element, aggregateorderby.order_by
3085 )
3086 )
3088 return new_fn._compiler_dispatch(self, **kwargs)
3089 else:
3090 return self.visit_withingroup(aggregateorderby, **kwargs)
3092 def visit_aggregate_orderby_inline(self, element, **kw):
3093 return "%s ORDER BY %s" % (
3094 self.process(element.element, **kw),
3095 self.process(element.aggregate_order_by, **kw),
3096 )
3098 def visit_aggregate_strings_func(self, fn, *, use_function_name, **kw):
3099 # aggreagate_order_by attribute is present if visit_function
3100 # gave us a Function with aggregate_orderby_inline() as the inner
3101 # contents
3102 order_by = getattr(fn.clauses, "aggregate_order_by", None)
3104 literal_exec = dict(kw)
3105 literal_exec["literal_execute"] = True
3107 # break up the function into its components so we can apply
3108 # literal_execute to the second argument (the delimiter)
3109 cl = list(fn.clauses)
3110 expr, delimiter = cl[0:2]
3111 if (
3112 order_by is not None
3113 and self.dialect.aggregate_order_by_style
3114 is AggregateOrderByStyle.INLINE
3115 ):
3116 return (
3117 f"{use_function_name}({expr._compiler_dispatch(self, **kw)}, "
3118 f"{delimiter._compiler_dispatch(self, **literal_exec)} "
3119 f"ORDER BY {order_by._compiler_dispatch(self, **kw)})"
3120 )
3121 else:
3122 return (
3123 f"{use_function_name}({expr._compiler_dispatch(self, **kw)}, "
3124 f"{delimiter._compiler_dispatch(self, **literal_exec)})"
3125 )
3127 def visit_extract(self, extract, **kwargs):
3128 field = self.extract_map.get(extract.field, extract.field)
3129 return "EXTRACT(%s FROM %s)" % (
3130 field,
3131 extract.expr._compiler_dispatch(self, **kwargs),
3132 )
3134 def visit_scalar_function_column(self, element, **kw):
3135 compiled_fn = self.visit_function(element.fn, **kw)
3136 compiled_col = self.visit_column(element, **kw)
3137 return "(%s).%s" % (compiled_fn, compiled_col)
3139 def visit_function(
3140 self,
3141 func: Function[Any],
3142 add_to_result_map: Optional[_ResultMapAppender] = None,
3143 **kwargs: Any,
3144 ) -> str:
3145 if self._collect_params:
3146 self._add_to_params(func)
3147 if add_to_result_map is not None:
3148 add_to_result_map(func.name, func.name, (func.name,), func.type)
3150 disp = getattr(self, "visit_%s_func" % func.name.lower(), None)
3152 text: str
3154 kwargs["within_aggregate_function"] = True
3156 if disp:
3157 text = disp(func, **kwargs)
3158 else:
3159 name = FUNCTIONS.get(func._deannotate().__class__, None)
3160 if name:
3161 if func._has_args:
3162 name += "%(expr)s"
3163 else:
3164 name = func.name
3165 name = (
3166 self.preparer.quote(name)
3167 if self.preparer._requires_quotes_illegal_chars(name)
3168 or isinstance(name, elements.quoted_name)
3169 else name
3170 )
3171 name = name + "%(expr)s"
3172 text = ".".join(
3173 [
3174 (
3175 self.preparer.quote(tok)
3176 if self.preparer._requires_quotes_illegal_chars(tok)
3177 or isinstance(name, elements.quoted_name)
3178 else tok
3179 )
3180 for tok in func.packagenames
3181 ]
3182 + [name]
3183 ) % {"expr": self.function_argspec(func, **kwargs)}
3185 if func._with_ordinality:
3186 text += " WITH ORDINALITY"
3187 return text
3189 def visit_next_value_func(self, next_value, **kw):
3190 return self.visit_sequence(next_value.sequence)
3192 def visit_sequence(self, sequence, **kw):
3193 raise NotImplementedError(
3194 "Dialect '%s' does not support sequence increments."
3195 % self.dialect.name
3196 )
3198 def function_argspec(self, func: Function[Any], **kwargs: Any) -> str:
3199 return func.clause_expr._compiler_dispatch(self, **kwargs)
3201 def visit_compound_select(
3202 self, cs, asfrom=False, compound_index=None, **kwargs
3203 ):
3204 if self._collect_params:
3205 self._add_to_params(cs)
3206 toplevel = not self.stack
3208 compile_state = cs._compile_state_factory(cs, self, **kwargs)
3210 if toplevel and not self.compile_state:
3211 self.compile_state = compile_state
3213 compound_stmt = compile_state.statement
3215 entry = self._default_stack_entry if toplevel else self.stack[-1]
3216 need_result_map = toplevel or (
3217 not compound_index
3218 and entry.get("need_result_map_for_compound", False)
3219 )
3221 # indicates there is already a CompoundSelect in play
3222 if compound_index == 0:
3223 entry["select_0"] = cs
3225 self.stack.append(
3226 {
3227 "correlate_froms": entry["correlate_froms"],
3228 "asfrom_froms": entry["asfrom_froms"],
3229 "selectable": cs,
3230 "compile_state": compile_state,
3231 "need_result_map_for_compound": need_result_map,
3232 }
3233 )
3235 if compound_stmt._independent_ctes:
3236 self._dispatch_independent_ctes(compound_stmt, kwargs)
3238 keyword = self.compound_keywords[cs.keyword]
3240 text = (" " + keyword + " ").join(
3241 (
3242 c._compiler_dispatch(
3243 self, asfrom=asfrom, compound_index=i, **kwargs
3244 )
3245 for i, c in enumerate(cs.selects)
3246 )
3247 )
3249 kwargs["include_table"] = False
3250 text += self.group_by_clause(cs, **dict(asfrom=asfrom, **kwargs))
3251 text += self.order_by_clause(cs, **kwargs)
3252 if cs._has_row_limiting_clause:
3253 text += self._row_limit_clause(cs, **kwargs)
3255 if self.ctes:
3256 nesting_level = len(self.stack) if not toplevel else None
3257 text = (
3258 self._render_cte_clause(
3259 nesting_level=nesting_level,
3260 include_following_stack=True,
3261 )
3262 + text
3263 )
3265 self.stack.pop(-1)
3266 return text
3268 def _row_limit_clause(self, cs, **kwargs):
3269 if cs._fetch_clause is not None:
3270 return self.fetch_clause(cs, **kwargs)
3271 else:
3272 return self.limit_clause(cs, **kwargs)
3274 def _get_operator_dispatch(self, operator_, qualifier1, qualifier2):
3275 attrname = "visit_%s_%s%s" % (
3276 operator_.__name__,
3277 qualifier1,
3278 "_" + qualifier2 if qualifier2 else "",
3279 )
3280 return getattr(self, attrname, None)
3282 def _get_custom_operator_dispatch(self, operator_, qualifier1):
3283 attrname = "visit_%s_op_%s" % (operator_.visit_name, qualifier1)
3284 return getattr(self, attrname, None)
3286 def visit_unary(
3287 self, unary, add_to_result_map=None, result_map_targets=(), **kw
3288 ):
3289 if add_to_result_map is not None:
3290 result_map_targets += (unary,)
3291 kw["add_to_result_map"] = add_to_result_map
3292 kw["result_map_targets"] = result_map_targets
3294 if unary.operator is operators.distinct_op and not kw.get(
3295 "within_aggregate_function", False
3296 ):
3297 util.warn(
3298 "Column-expression-level unary distinct() "
3299 "should not be used outside of an aggregate "
3300 "function. For general 'SELECT DISTINCT' support"
3301 "use select().distinct()."
3302 )
3304 if unary.operator:
3305 if unary.modifier:
3306 raise exc.CompileError(
3307 "Unary expression does not support operator "
3308 "and modifier simultaneously"
3309 )
3310 disp = self._get_operator_dispatch(
3311 unary.operator, "unary", "operator"
3312 )
3313 if disp:
3314 return disp(unary, unary.operator, **kw)
3315 else:
3316 return self._generate_generic_unary_operator(
3317 unary, OPERATORS[unary.operator], **kw
3318 )
3319 elif unary.modifier:
3320 disp = self._get_operator_dispatch(
3321 unary.modifier, "unary", "modifier"
3322 )
3323 if disp:
3324 return disp(unary, unary.modifier, **kw)
3325 else:
3326 return self._generate_generic_unary_modifier(
3327 unary, OPERATORS[unary.modifier], **kw
3328 )
3329 else:
3330 raise exc.CompileError(
3331 "Unary expression has no operator or modifier"
3332 )
3334 def visit_truediv_binary(self, binary, operator, **kw):
3335 if self.dialect.div_is_floordiv:
3336 return (
3337 self.process(binary.left, **kw)
3338 + " / "
3339 # TODO: would need a fast cast again here,
3340 # unless we want to use an implicit cast like "+ 0.0"
3341 + self.process(
3342 elements.Cast(
3343 binary.right,
3344 (
3345 binary.right.type
3346 if binary.right.type._type_affinity
3347 in (sqltypes.Numeric, sqltypes.Float)
3348 else sqltypes.Numeric()
3349 ),
3350 ),
3351 **kw,
3352 )
3353 )
3354 else:
3355 return (
3356 self.process(binary.left, **kw)
3357 + " / "
3358 + self.process(binary.right, **kw)
3359 )
3361 def visit_floordiv_binary(self, binary, operator, **kw):
3362 if (
3363 self.dialect.div_is_floordiv
3364 and binary.right.type._type_affinity is sqltypes.Integer
3365 and binary.left.type._type_affinity is sqltypes.Integer
3366 ):
3367 return (
3368 self.process(binary.left, **kw)
3369 + " / "
3370 + self.process(binary.right, **kw)
3371 )
3372 else:
3373 return "FLOOR(%s)" % (
3374 self.process(binary.left, **kw)
3375 + " / "
3376 + self.process(binary.right, **kw)
3377 )
3379 def visit_is_true_unary_operator(self, element, operator, **kw):
3380 if (
3381 element._is_implicitly_boolean
3382 or self.dialect.supports_native_boolean
3383 ):
3384 return self.process(element.element, **kw)
3385 else:
3386 return "%s = 1" % self.process(element.element, **kw)
3388 def visit_is_false_unary_operator(self, element, operator, **kw):
3389 if (
3390 element._is_implicitly_boolean
3391 or self.dialect.supports_native_boolean
3392 ):
3393 return "NOT %s" % self.process(element.element, **kw)
3394 else:
3395 return "%s = 0" % self.process(element.element, **kw)
3397 def visit_not_match_op_binary(self, binary, operator, **kw):
3398 return "NOT %s" % self.visit_binary(
3399 binary, override_operator=operators.match_op
3400 )
3402 def visit_not_in_op_binary(self, binary, operator, **kw):
3403 # The brackets are required in the NOT IN operation because the empty
3404 # case is handled using the form "(col NOT IN (null) OR 1 = 1)".
3405 # The presence of the OR makes the brackets required.
3406 return "(%s)" % self._generate_generic_binary(
3407 binary, OPERATORS[operator], **kw
3408 )
3410 def visit_empty_set_op_expr(self, type_, expand_op, **kw):
3411 if expand_op is operators.not_in_op:
3412 if len(type_) > 1:
3413 return "(%s)) OR (1 = 1" % (
3414 ", ".join("NULL" for element in type_)
3415 )
3416 else:
3417 return "NULL) OR (1 = 1"
3418 elif expand_op is operators.in_op:
3419 if len(type_) > 1:
3420 return "(%s)) AND (1 != 1" % (
3421 ", ".join("NULL" for element in type_)
3422 )
3423 else:
3424 return "NULL) AND (1 != 1"
3425 else:
3426 return self.visit_empty_set_expr(type_)
3428 def visit_empty_set_expr(self, element_types, **kw):
3429 raise NotImplementedError(
3430 "Dialect '%s' does not support empty set expression."
3431 % self.dialect.name
3432 )
3434 def _literal_execute_expanding_parameter_literal_binds(
3435 self, parameter, values, bind_expression_template=None
3436 ):
3437 typ_dialect_impl = parameter.type._unwrapped_dialect_impl(self.dialect)
3439 if not values:
3440 # empty IN expression. note we don't need to use
3441 # bind_expression_template here because there are no
3442 # expressions to render.
3444 if typ_dialect_impl._is_tuple_type:
3445 replacement_expression = (
3446 "VALUES " if self.dialect.tuple_in_values else ""
3447 ) + self.visit_empty_set_op_expr(
3448 parameter.type.types, parameter.expand_op
3449 )
3451 else:
3452 replacement_expression = self.visit_empty_set_op_expr(
3453 [parameter.type], parameter.expand_op
3454 )
3456 elif typ_dialect_impl._is_tuple_type or (
3457 typ_dialect_impl._isnull
3458 and isinstance(values[0], collections_abc.Sequence)
3459 and not isinstance(values[0], (str, bytes))
3460 ):
3461 if typ_dialect_impl._has_bind_expression:
3462 raise NotImplementedError(
3463 "bind_expression() on TupleType not supported with "
3464 "literal_binds"
3465 )
3467 replacement_expression = (
3468 "VALUES " if self.dialect.tuple_in_values else ""
3469 ) + ", ".join(
3470 "(%s)"
3471 % (
3472 ", ".join(
3473 self.render_literal_value(value, param_type)
3474 for value, param_type in zip(
3475 tuple_element, parameter.type.types
3476 )
3477 )
3478 )
3479 for i, tuple_element in enumerate(values)
3480 )
3481 else:
3482 if bind_expression_template:
3483 post_compile_pattern = self._post_compile_pattern
3484 m = post_compile_pattern.search(bind_expression_template)
3485 assert m and m.group(
3486 2
3487 ), "unexpected format for expanding parameter"
3489 tok = m.group(2).split("~~")
3490 be_left, be_right = tok[1], tok[3]
3491 replacement_expression = ", ".join(
3492 "%s%s%s"
3493 % (
3494 be_left,
3495 self.render_literal_value(value, parameter.type),
3496 be_right,
3497 )
3498 for value in values
3499 )
3500 else:
3501 replacement_expression = ", ".join(
3502 self.render_literal_value(value, parameter.type)
3503 for value in values
3504 )
3506 return (), replacement_expression
3508 def _literal_execute_expanding_parameter(self, name, parameter, values):
3509 if parameter.literal_execute:
3510 return self._literal_execute_expanding_parameter_literal_binds(
3511 parameter, values
3512 )
3514 dialect = self.dialect
3515 typ_dialect_impl = parameter.type._unwrapped_dialect_impl(dialect)
3517 if self._numeric_binds:
3518 bind_template = self.compilation_bindtemplate
3519 else:
3520 bind_template = self.bindtemplate
3522 if (
3523 self.dialect._bind_typing_render_casts
3524 and typ_dialect_impl.render_bind_cast
3525 ):
3527 def _render_bindtemplate(name):
3528 return self.render_bind_cast(
3529 parameter.type,
3530 typ_dialect_impl,
3531 bind_template % {"name": name},
3532 )
3534 else:
3536 def _render_bindtemplate(name):
3537 return bind_template % {"name": name}
3539 if not values:
3540 to_update = []
3541 if typ_dialect_impl._is_tuple_type:
3542 replacement_expression = self.visit_empty_set_op_expr(
3543 parameter.type.types, parameter.expand_op
3544 )
3545 else:
3546 replacement_expression = self.visit_empty_set_op_expr(
3547 [parameter.type], parameter.expand_op
3548 )
3550 elif typ_dialect_impl._is_tuple_type or (
3551 typ_dialect_impl._isnull
3552 and isinstance(values[0], collections_abc.Sequence)
3553 and not isinstance(values[0], (str, bytes))
3554 ):
3555 assert not typ_dialect_impl._is_array
3556 to_update = [
3557 ("%s_%s_%s" % (name, i, j), value)
3558 for i, tuple_element in enumerate(values, 1)
3559 for j, value in enumerate(tuple_element, 1)
3560 ]
3562 replacement_expression = (
3563 "VALUES " if dialect.tuple_in_values else ""
3564 ) + ", ".join(
3565 "(%s)"
3566 % (
3567 ", ".join(
3568 _render_bindtemplate(
3569 to_update[i * len(tuple_element) + j][0]
3570 )
3571 for j, value in enumerate(tuple_element)
3572 )
3573 )
3574 for i, tuple_element in enumerate(values)
3575 )
3576 else:
3577 to_update = [
3578 ("%s_%s" % (name, i), value)
3579 for i, value in enumerate(values, 1)
3580 ]
3581 replacement_expression = ", ".join(
3582 _render_bindtemplate(key) for key, value in to_update
3583 )
3585 return to_update, replacement_expression
3587 def visit_binary(
3588 self,
3589 binary,
3590 override_operator=None,
3591 eager_grouping=False,
3592 from_linter=None,
3593 lateral_from_linter=None,
3594 **kw,
3595 ):
3596 if from_linter and operators.is_comparison(binary.operator):
3597 if lateral_from_linter is not None:
3598 enclosing_lateral = kw["enclosing_lateral"]
3599 lateral_from_linter.edges.update(
3600 itertools.product(
3601 _de_clone(
3602 binary.left._from_objects + [enclosing_lateral]
3603 ),
3604 _de_clone(
3605 binary.right._from_objects + [enclosing_lateral]
3606 ),
3607 )
3608 )
3609 else:
3610 from_linter.edges.update(
3611 itertools.product(
3612 _de_clone(binary.left._from_objects),
3613 _de_clone(binary.right._from_objects),
3614 )
3615 )
3617 # don't allow "? = ?" to render
3618 if (
3619 self.ansi_bind_rules
3620 and isinstance(binary.left, elements.BindParameter)
3621 and isinstance(binary.right, elements.BindParameter)
3622 ):
3623 kw["literal_execute"] = True
3625 operator_ = override_operator or binary.operator
3626 disp = self._get_operator_dispatch(operator_, "binary", None)
3627 if disp:
3628 return disp(binary, operator_, **kw)
3629 else:
3630 try:
3631 opstring = OPERATORS[operator_]
3632 except KeyError as err:
3633 raise exc.UnsupportedCompilationError(self, operator_) from err
3634 else:
3635 return self._generate_generic_binary(
3636 binary,
3637 opstring,
3638 from_linter=from_linter,
3639 lateral_from_linter=lateral_from_linter,
3640 **kw,
3641 )
3643 def visit_function_as_comparison_op_binary(self, element, operator, **kw):
3644 return self.process(element.sql_function, **kw)
3646 def visit_mod_binary(self, binary, operator, **kw):
3647 if self.preparer._double_percents:
3648 return (
3649 self.process(binary.left, **kw)
3650 + " %% "
3651 + self.process(binary.right, **kw)
3652 )
3653 else:
3654 return (
3655 self.process(binary.left, **kw)
3656 + " % "
3657 + self.process(binary.right, **kw)
3658 )
3660 def visit_custom_op_binary(self, element, operator, **kw):
3661 if operator.visit_name:
3662 disp = self._get_custom_operator_dispatch(operator, "binary")
3663 if disp:
3664 return disp(element, operator, **kw)
3666 kw["eager_grouping"] = operator.eager_grouping
3667 return self._generate_generic_binary(
3668 element,
3669 " " + self.escape_literal_column(operator.opstring) + " ",
3670 **kw,
3671 )
3673 def visit_custom_op_unary_operator(self, element, operator, **kw):
3674 if operator.visit_name:
3675 disp = self._get_custom_operator_dispatch(operator, "unary")
3676 if disp:
3677 return disp(element, operator, **kw)
3679 return self._generate_generic_unary_operator(
3680 element, self.escape_literal_column(operator.opstring) + " ", **kw
3681 )
3683 def visit_custom_op_unary_modifier(self, element, operator, **kw):
3684 if operator.visit_name:
3685 disp = self._get_custom_operator_dispatch(operator, "unary")
3686 if disp:
3687 return disp(element, operator, **kw)
3689 return self._generate_generic_unary_modifier(
3690 element, " " + self.escape_literal_column(operator.opstring), **kw
3691 )
3693 def _generate_generic_binary(
3694 self,
3695 binary: BinaryExpression[Any],
3696 opstring: str,
3697 eager_grouping: bool = False,
3698 **kw: Any,
3699 ) -> str:
3700 _in_operator_expression = kw.get("_in_operator_expression", False)
3702 kw["_in_operator_expression"] = True
3703 kw["_binary_op"] = binary.operator
3704 text = (
3705 binary.left._compiler_dispatch(
3706 self, eager_grouping=eager_grouping, **kw
3707 )
3708 + opstring
3709 + binary.right._compiler_dispatch(
3710 self, eager_grouping=eager_grouping, **kw
3711 )
3712 )
3714 if _in_operator_expression and eager_grouping:
3715 text = "(%s)" % text
3716 return text
3718 def _generate_generic_unary_operator(self, unary, opstring, **kw):
3719 return opstring + unary.element._compiler_dispatch(self, **kw)
3721 def _generate_generic_unary_modifier(self, unary, opstring, **kw):
3722 return unary.element._compiler_dispatch(self, **kw) + opstring
3724 @util.memoized_property
3725 def _like_percent_literal(self):
3726 return elements.literal_column("'%'", type_=sqltypes.STRINGTYPE)
3728 def visit_ilike_case_insensitive_operand(self, element, **kw):
3729 return f"lower({element.element._compiler_dispatch(self, **kw)})"
3731 def visit_contains_op_binary(self, binary, operator, **kw):
3732 binary = binary._clone()
3733 percent = self._like_percent_literal
3734 binary.right = percent.concat(binary.right).concat(percent)
3735 return self.visit_like_op_binary(binary, operator, **kw)
3737 def visit_not_contains_op_binary(self, binary, operator, **kw):
3738 binary = binary._clone()
3739 percent = self._like_percent_literal
3740 binary.right = percent.concat(binary.right).concat(percent)
3741 return self.visit_not_like_op_binary(binary, operator, **kw)
3743 def visit_icontains_op_binary(self, binary, operator, **kw):
3744 binary = binary._clone()
3745 percent = self._like_percent_literal
3746 binary.left = ilike_case_insensitive(binary.left)
3747 binary.right = percent.concat(
3748 ilike_case_insensitive(binary.right)
3749 ).concat(percent)
3750 return self.visit_ilike_op_binary(binary, operator, **kw)
3752 def visit_not_icontains_op_binary(self, binary, operator, **kw):
3753 binary = binary._clone()
3754 percent = self._like_percent_literal
3755 binary.left = ilike_case_insensitive(binary.left)
3756 binary.right = percent.concat(
3757 ilike_case_insensitive(binary.right)
3758 ).concat(percent)
3759 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3761 def visit_startswith_op_binary(self, binary, operator, **kw):
3762 binary = binary._clone()
3763 percent = self._like_percent_literal
3764 binary.right = percent._rconcat(binary.right)
3765 return self.visit_like_op_binary(binary, operator, **kw)
3767 def visit_not_startswith_op_binary(self, binary, operator, **kw):
3768 binary = binary._clone()
3769 percent = self._like_percent_literal
3770 binary.right = percent._rconcat(binary.right)
3771 return self.visit_not_like_op_binary(binary, operator, **kw)
3773 def visit_istartswith_op_binary(self, binary, operator, **kw):
3774 binary = binary._clone()
3775 percent = self._like_percent_literal
3776 binary.left = ilike_case_insensitive(binary.left)
3777 binary.right = percent._rconcat(ilike_case_insensitive(binary.right))
3778 return self.visit_ilike_op_binary(binary, operator, **kw)
3780 def visit_not_istartswith_op_binary(self, binary, operator, **kw):
3781 binary = binary._clone()
3782 percent = self._like_percent_literal
3783 binary.left = ilike_case_insensitive(binary.left)
3784 binary.right = percent._rconcat(ilike_case_insensitive(binary.right))
3785 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3787 def visit_endswith_op_binary(self, binary, operator, **kw):
3788 binary = binary._clone()
3789 percent = self._like_percent_literal
3790 binary.right = percent.concat(binary.right)
3791 return self.visit_like_op_binary(binary, operator, **kw)
3793 def visit_not_endswith_op_binary(self, binary, operator, **kw):
3794 binary = binary._clone()
3795 percent = self._like_percent_literal
3796 binary.right = percent.concat(binary.right)
3797 return self.visit_not_like_op_binary(binary, operator, **kw)
3799 def visit_iendswith_op_binary(self, binary, operator, **kw):
3800 binary = binary._clone()
3801 percent = self._like_percent_literal
3802 binary.left = ilike_case_insensitive(binary.left)
3803 binary.right = percent.concat(ilike_case_insensitive(binary.right))
3804 return self.visit_ilike_op_binary(binary, operator, **kw)
3806 def visit_not_iendswith_op_binary(self, binary, operator, **kw):
3807 binary = binary._clone()
3808 percent = self._like_percent_literal
3809 binary.left = ilike_case_insensitive(binary.left)
3810 binary.right = percent.concat(ilike_case_insensitive(binary.right))
3811 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3813 def visit_like_op_binary(self, binary, operator, **kw):
3814 escape = binary.modifiers.get("escape", None)
3816 return "%s LIKE %s" % (
3817 binary.left._compiler_dispatch(self, **kw),
3818 binary.right._compiler_dispatch(self, **kw),
3819 ) + (
3820 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
3821 if escape is not None
3822 else ""
3823 )
3825 def visit_not_like_op_binary(self, binary, operator, **kw):
3826 escape = binary.modifiers.get("escape", None)
3827 return "%s NOT LIKE %s" % (
3828 binary.left._compiler_dispatch(self, **kw),
3829 binary.right._compiler_dispatch(self, **kw),
3830 ) + (
3831 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
3832 if escape is not None
3833 else ""
3834 )
3836 def visit_ilike_op_binary(self, binary, operator, **kw):
3837 if operator is operators.ilike_op:
3838 binary = binary._clone()
3839 binary.left = ilike_case_insensitive(binary.left)
3840 binary.right = ilike_case_insensitive(binary.right)
3841 # else we assume ilower() has been applied
3843 return self.visit_like_op_binary(binary, operator, **kw)
3845 def visit_not_ilike_op_binary(self, binary, operator, **kw):
3846 if operator is operators.not_ilike_op:
3847 binary = binary._clone()
3848 binary.left = ilike_case_insensitive(binary.left)
3849 binary.right = ilike_case_insensitive(binary.right)
3850 # else we assume ilower() has been applied
3852 return self.visit_not_like_op_binary(binary, operator, **kw)
3854 def visit_between_op_binary(self, binary, operator, **kw):
3855 symmetric = binary.modifiers.get("symmetric", False)
3856 return self._generate_generic_binary(
3857 binary, " BETWEEN SYMMETRIC " if symmetric else " BETWEEN ", **kw
3858 )
3860 def visit_not_between_op_binary(self, binary, operator, **kw):
3861 symmetric = binary.modifiers.get("symmetric", False)
3862 return self._generate_generic_binary(
3863 binary,
3864 " NOT BETWEEN SYMMETRIC " if symmetric else " NOT BETWEEN ",
3865 **kw,
3866 )
3868 def visit_regexp_match_op_binary(
3869 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3870 ) -> str:
3871 raise exc.CompileError(
3872 "%s dialect does not support regular expressions"
3873 % self.dialect.name
3874 )
3876 def visit_not_regexp_match_op_binary(
3877 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3878 ) -> str:
3879 raise exc.CompileError(
3880 "%s dialect does not support regular expressions"
3881 % self.dialect.name
3882 )
3884 def visit_regexp_replace_op_binary(
3885 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3886 ) -> str:
3887 raise exc.CompileError(
3888 "%s dialect does not support regular expression replacements"
3889 % self.dialect.name
3890 )
3892 def visit_dmltargetcopy(self, element, *, bindmarkers=None, **kw):
3893 if bindmarkers is None:
3894 raise exc.CompileError(
3895 "DML target objects may only be used with "
3896 "compiled INSERT or UPDATE statements"
3897 )
3899 bindmarkers[element.column.key] = element
3900 return f"__BINDMARKER_~~{element.column.key}~~"
3902 def visit_bindparam(
3903 self,
3904 bindparam,
3905 within_columns_clause=False,
3906 literal_binds=False,
3907 skip_bind_expression=False,
3908 literal_execute=False,
3909 render_postcompile=False,
3910 is_upsert_set=False,
3911 **kwargs,
3912 ):
3913 # Detect parametrized bindparams in upsert SET clause for issue #13130
3914 if (
3915 is_upsert_set
3916 and bindparam.value is None
3917 and bindparam.callable is None
3918 and self._insertmanyvalues is not None
3919 ):
3920 self._insertmanyvalues = self._insertmanyvalues._replace(
3921 has_upsert_bound_parameters=True
3922 )
3924 if not skip_bind_expression:
3925 impl = bindparam.type.dialect_impl(self.dialect)
3926 if impl._has_bind_expression:
3927 bind_expression = impl.bind_expression(bindparam)
3928 wrapped = self.process(
3929 bind_expression,
3930 skip_bind_expression=True,
3931 within_columns_clause=within_columns_clause,
3932 literal_binds=literal_binds and not bindparam.expanding,
3933 literal_execute=literal_execute,
3934 render_postcompile=render_postcompile,
3935 **kwargs,
3936 )
3937 if bindparam.expanding:
3938 # for postcompile w/ expanding, move the "wrapped" part
3939 # of this into the inside
3941 m = re.match(
3942 r"^(.*)\(__\[POSTCOMPILE_(\S+?)\]\)(.*)$", wrapped
3943 )
3944 assert m, "unexpected format for expanding parameter"
3945 wrapped = "(__[POSTCOMPILE_%s~~%s~~REPL~~%s~~])" % (
3946 m.group(2),
3947 m.group(1),
3948 m.group(3),
3949 )
3951 if literal_binds:
3952 ret = self.render_literal_bindparam(
3953 bindparam,
3954 within_columns_clause=True,
3955 bind_expression_template=wrapped,
3956 **kwargs,
3957 )
3958 return f"({ret})"
3960 return wrapped
3962 if not literal_binds:
3963 literal_execute = (
3964 literal_execute
3965 or bindparam.literal_execute
3966 or (within_columns_clause and self.ansi_bind_rules)
3967 )
3968 post_compile = literal_execute or bindparam.expanding
3969 else:
3970 post_compile = False
3972 if literal_binds:
3973 ret = self.render_literal_bindparam(
3974 bindparam, within_columns_clause=True, **kwargs
3975 )
3976 if bindparam.expanding:
3977 ret = f"({ret})"
3978 return ret
3980 name = self._truncate_bindparam(bindparam)
3982 if name in self.binds:
3983 existing = self.binds[name]
3984 if existing is not bindparam:
3985 if (
3986 (existing.unique or bindparam.unique)
3987 and not existing.proxy_set.intersection(
3988 bindparam.proxy_set
3989 )
3990 and not existing._cloned_set.intersection(
3991 bindparam._cloned_set
3992 )
3993 ):
3994 raise exc.CompileError(
3995 "Bind parameter '%s' conflicts with "
3996 "unique bind parameter of the same name" % name
3997 )
3998 elif existing.expanding != bindparam.expanding:
3999 raise exc.CompileError(
4000 "Can't reuse bound parameter name '%s' in both "
4001 "'expanding' (e.g. within an IN expression) and "
4002 "non-expanding contexts. If this parameter is to "
4003 "receive a list/array value, set 'expanding=True' on "
4004 "it for expressions that aren't IN, otherwise use "
4005 "a different parameter name." % (name,)
4006 )
4007 elif existing._is_crud or bindparam._is_crud:
4008 if existing._is_crud and bindparam._is_crud:
4009 # TODO: this condition is not well understood.
4010 # see tests in test/sql/test_update.py
4011 raise exc.CompileError(
4012 "Encountered unsupported case when compiling an "
4013 "INSERT or UPDATE statement. If this is a "
4014 "multi-table "
4015 "UPDATE statement, please provide string-named "
4016 "arguments to the "
4017 "values() method with distinct names; support for "
4018 "multi-table UPDATE statements that "
4019 "target multiple tables for UPDATE is very "
4020 "limited",
4021 )
4022 else:
4023 raise exc.CompileError(
4024 f"bindparam() name '{bindparam.key}' is reserved "
4025 "for automatic usage in the VALUES or SET "
4026 "clause of this "
4027 "insert/update statement. Please use a "
4028 "name other than column name when using "
4029 "bindparam() "
4030 "with insert() or update() (for example, "
4031 f"'b_{bindparam.key}')."
4032 )
4034 self.binds[bindparam.key] = self.binds[name] = bindparam
4036 # if we are given a cache key that we're going to match against,
4037 # relate the bindparam here to one that is most likely present
4038 # in the "extracted params" portion of the cache key. this is used
4039 # to set up a positional mapping that is used to determine the
4040 # correct parameters for a subsequent use of this compiled with
4041 # a different set of parameter values. here, we accommodate for
4042 # parameters that may have been cloned both before and after the cache
4043 # key was been generated.
4044 ckbm_tuple = self._cache_key_bind_match
4046 if ckbm_tuple:
4047 ckbm, cksm = ckbm_tuple
4048 for bp in bindparam._cloned_set:
4049 if bp.key in cksm:
4050 cb = cksm[bp.key]
4051 ckbm[cb].append(bindparam)
4053 if bindparam.isoutparam:
4054 self.has_out_parameters = True
4056 if post_compile:
4057 if render_postcompile:
4058 self._render_postcompile = True
4060 if literal_execute:
4061 self.literal_execute_params |= {bindparam}
4062 else:
4063 self.post_compile_params |= {bindparam}
4065 ret = self.bindparam_string(
4066 name,
4067 post_compile=post_compile,
4068 expanding=bindparam.expanding,
4069 bindparam_type=bindparam.type,
4070 **kwargs,
4071 )
4073 if bindparam.expanding:
4074 ret = f"({ret})"
4076 return ret
4078 def render_bind_cast(self, type_, dbapi_type, sqltext):
4079 raise NotImplementedError()
4081 def render_literal_bindparam(
4082 self,
4083 bindparam,
4084 render_literal_value=NO_ARG,
4085 bind_expression_template=None,
4086 **kw,
4087 ):
4088 if render_literal_value is not NO_ARG:
4089 value = render_literal_value
4090 else:
4091 if bindparam.value is None and bindparam.callable is None:
4092 op = kw.get("_binary_op", None)
4093 if op and op not in (operators.is_, operators.is_not):
4094 util.warn_limited(
4095 "Bound parameter '%s' rendering literal NULL in a SQL "
4096 "expression; comparisons to NULL should not use "
4097 "operators outside of 'is' or 'is not'",
4098 (bindparam.key,),
4099 )
4100 return self.process(sqltypes.NULLTYPE, **kw)
4101 value = bindparam.effective_value
4103 if bindparam.expanding:
4104 leep = self._literal_execute_expanding_parameter_literal_binds
4105 to_update, replacement_expr = leep(
4106 bindparam,
4107 value,
4108 bind_expression_template=bind_expression_template,
4109 )
4110 return replacement_expr
4111 else:
4112 return self.render_literal_value(value, bindparam.type)
4114 def render_literal_value(
4115 self, value: Any, type_: sqltypes.TypeEngine[Any]
4116 ) -> str:
4117 """Render the value of a bind parameter as a quoted literal.
4119 This is used for statement sections that do not accept bind parameters
4120 on the target driver/database.
4122 This should be implemented by subclasses using the quoting services
4123 of the DBAPI.
4125 """
4127 if value is None and not type_.should_evaluate_none:
4128 # issue #10535 - handle NULL in the compiler without placing
4129 # this onto each type, except for "evaluate None" types
4130 # (e.g. JSON)
4131 return self.process(elements.Null._instance())
4133 processor = type_._cached_literal_processor(self.dialect)
4134 if processor:
4135 try:
4136 return processor(value)
4137 except Exception as e:
4138 raise exc.CompileError(
4139 f"Could not render literal value "
4140 f'"{sql_util._repr_single_value(value)}" '
4141 f"with datatype "
4142 f"{type_}; see parent stack trace for "
4143 "more detail."
4144 ) from e
4146 else:
4147 raise exc.CompileError(
4148 f"No literal value renderer is available for literal value "
4149 f'"{sql_util._repr_single_value(value)}" '
4150 f"with datatype {type_}"
4151 )
4153 def _truncate_bindparam(self, bindparam):
4154 if bindparam in self.bind_names:
4155 return self.bind_names[bindparam]
4157 bind_name = bindparam.key
4158 if isinstance(bind_name, elements._truncated_label):
4159 bind_name = self._truncated_identifier("bindparam", bind_name)
4161 # add to bind_names for translation
4162 self.bind_names[bindparam] = bind_name
4164 return bind_name
4166 def _truncated_identifier(
4167 self, ident_class: str, name: _truncated_label
4168 ) -> str:
4169 if (ident_class, name) in self.truncated_names:
4170 return self.truncated_names[(ident_class, name)]
4172 anonname = name.apply_map(self.anon_map)
4174 if len(anonname) > self.label_length - 6:
4175 counter = self._truncated_counters.get(ident_class, 1)
4176 truncname = (
4177 anonname[0 : max(self.label_length - 6, 0)]
4178 + "_"
4179 + hex(counter)[2:]
4180 )
4181 self._truncated_counters[ident_class] = counter + 1
4182 else:
4183 truncname = anonname
4184 self.truncated_names[(ident_class, name)] = truncname
4185 return truncname
4187 def _anonymize(self, name: str) -> str:
4188 return name % self.anon_map
4190 def _escaped_bind_name(self, escaped_from: str, name: str) -> str:
4191 """Record the escaped form of a bind name, uniquifying it against
4192 the names already in use.
4194 Distinct bind names can escape to the same string, e.g. parameters
4195 named ``"a.b"`` and ``"a_b"`` both escape to ``a_b``. As the escaped
4196 name is what keys the parameter dictionary handed to the driver, the
4197 second parameter would otherwise overwrite the first and its value be
4198 silently discarded. A counter is appended so that the two remain
4199 distinct. The ``.key`` of each :class:`.BindParameter` is untouched,
4200 so parameter dictionaries passed by the caller are unaffected.
4202 See :ticket:`13534`.
4204 """
4205 try:
4206 return self.escaped_bind_names[escaped_from]
4207 except KeyError:
4208 pass
4210 if self.escaped_bind_names:
4211 used = self._escaped_bind_names_used
4212 else:
4213 self._escaped_bind_names_used = used = set()
4215 if name in used or (name != escaped_from and name in self.binds):
4216 base = name
4217 counter = 0
4218 while name in used or name in self.binds:
4219 counter += 1
4220 name = "%s__%d" % (base, counter)
4222 used.add(name)
4223 self.escaped_bind_names = self.escaped_bind_names.union(
4224 {escaped_from: name}
4225 )
4226 return name
4228 def bindparam_string(
4229 self,
4230 name: str,
4231 post_compile: bool = False,
4232 expanding: bool = False,
4233 escaped_from: Optional[str] = None,
4234 bindparam_type: Optional[TypeEngine[Any]] = None,
4235 accumulate_bind_names: Optional[Set[str]] = None,
4236 visited_bindparam: Optional[List[str]] = None,
4237 **kw: Any,
4238 ) -> str:
4239 # TODO: accumulate_bind_names is passed by crud.py to gather
4240 # names on a per-value basis, visited_bindparam is passed by
4241 # visit_insert() to collect all parameters in the statement.
4242 # see if this gathering can be simplified somehow
4243 if accumulate_bind_names is not None:
4244 accumulate_bind_names.add(name)
4245 if visited_bindparam is not None:
4246 visited_bindparam.append(name)
4248 if not escaped_from:
4249 if self._bind_translate_re.search(name):
4250 # not quite the translate use case as we want to
4251 # also get a quick boolean if we even found
4252 # unusual characters in the name
4253 new_name = self._bind_translate_re.sub(
4254 lambda m: self._bind_translate_chars[m.group(0)],
4255 name,
4256 )
4257 escaped_from = name
4258 name = new_name
4259 elif (
4260 self.escaped_bind_names
4261 and name in self._escaped_bind_names_used
4262 ):
4263 # this name needs no escaping of its own, but another
4264 # parameter has already escaped to it; #13534
4265 escaped_from = name
4267 if escaped_from:
4268 name = self._escaped_bind_name(escaped_from, name)
4269 if post_compile:
4270 ret = "__[POSTCOMPILE_%s]" % name
4271 if expanding:
4272 # for expanding, bound parameters or literal values will be
4273 # rendered per item
4274 return ret
4276 # otherwise, for non-expanding "literal execute", apply
4277 # bind casts as determined by the datatype
4278 if bindparam_type is not None:
4279 type_impl = bindparam_type._unwrapped_dialect_impl(
4280 self.dialect
4281 )
4282 if type_impl.render_literal_cast:
4283 ret = self.render_bind_cast(bindparam_type, type_impl, ret)
4284 return ret
4285 elif self.state is CompilerState.COMPILING:
4286 ret = self.compilation_bindtemplate % {"name": name}
4287 else:
4288 ret = self.bindtemplate % {"name": name}
4290 if (
4291 bindparam_type is not None
4292 and self.dialect._bind_typing_render_casts
4293 ):
4294 type_impl = bindparam_type._unwrapped_dialect_impl(self.dialect)
4295 if type_impl.render_bind_cast:
4296 ret = self.render_bind_cast(bindparam_type, type_impl, ret)
4298 return ret
4300 def _dispatch_independent_ctes(self, stmt, kw):
4301 local_kw = kw.copy()
4302 local_kw.pop("cte_opts", None)
4303 for cte, opt in zip(
4304 stmt._independent_ctes, stmt._independent_ctes_opts
4305 ):
4306 cte._compiler_dispatch(self, cte_opts=opt, **local_kw)
4308 def visit_cte(
4309 self,
4310 cte: CTE,
4311 asfrom: bool = False,
4312 ashint: bool = False,
4313 fromhints: Optional[_FromHintsType] = None,
4314 visiting_cte: Optional[CTE] = None,
4315 from_linter: Optional[FromLinter] = None,
4316 cte_opts: selectable._CTEOpts = selectable._CTEOpts(False),
4317 **kwargs: Any,
4318 ) -> Optional[str]:
4319 self_ctes = self._init_cte_state()
4320 assert self_ctes is self.ctes
4322 kwargs["visiting_cte"] = cte
4324 cte_name = cte.name
4326 if isinstance(cte_name, elements._truncated_label):
4327 cte_name = self._truncated_identifier("alias", cte_name)
4329 is_new_cte = True
4330 embedded_in_current_named_cte = False
4332 _reference_cte = cte._get_reference_cte()
4334 nesting = cte.nesting or cte_opts.nesting
4336 # check for CTE already encountered
4337 if _reference_cte in self.level_name_by_cte:
4338 cte_level, _, existing_cte_opts = self.level_name_by_cte[
4339 _reference_cte
4340 ]
4341 assert _ == cte_name
4343 cte_level_name = (cte_level, cte_name)
4344 existing_cte = self.ctes_by_level_name[cte_level_name]
4346 # check if we are receiving it here with a specific
4347 # "nest_here" location; if so, move it to this location
4349 if cte_opts.nesting:
4350 if existing_cte_opts.nesting:
4351 raise exc.CompileError(
4352 "CTE is stated as 'nest_here' in "
4353 "more than one location"
4354 )
4356 old_level_name = (cte_level, cte_name)
4357 cte_level = len(self.stack) if nesting else 1
4358 cte_level_name = new_level_name = (cte_level, cte_name)
4360 del self.ctes_by_level_name[old_level_name]
4361 self.ctes_by_level_name[new_level_name] = existing_cte
4362 self.level_name_by_cte[_reference_cte] = new_level_name + (
4363 cte_opts,
4364 )
4366 else:
4367 cte_level = len(self.stack) if nesting else 1
4368 cte_level_name = (cte_level, cte_name)
4370 if cte_level_name in self.ctes_by_level_name:
4371 existing_cte = self.ctes_by_level_name[cte_level_name]
4372 else:
4373 existing_cte = None
4375 if existing_cte is not None:
4376 embedded_in_current_named_cte = visiting_cte is existing_cte
4378 # we've generated a same-named CTE that we are enclosed in,
4379 # or this is the same CTE. just return the name.
4380 if cte is existing_cte._restates or cte is existing_cte:
4381 is_new_cte = False
4382 elif existing_cte is cte._restates:
4383 # we've generated a same-named CTE that is
4384 # enclosed in us - we take precedence, so
4385 # discard the text for the "inner".
4386 del self_ctes[existing_cte]
4388 existing_cte_reference_cte = existing_cte._get_reference_cte()
4390 assert existing_cte_reference_cte is _reference_cte
4391 assert existing_cte_reference_cte is existing_cte
4393 del self.level_name_by_cte[existing_cte_reference_cte]
4394 else:
4395 if (
4396 # if the two CTEs have the same hash, which we expect
4397 # here means that one/both is an annotated of the other
4398 (hash(cte) == hash(existing_cte))
4399 # or...
4400 or (
4401 (
4402 # if they are clones, i.e. they came from the ORM
4403 # or some other visit method
4404 cte._is_clone_of is not None
4405 or existing_cte._is_clone_of is not None
4406 )
4407 # and are deep-copy identical
4408 and cte.compare(existing_cte)
4409 )
4410 ):
4411 # then consider these two CTEs the same
4412 is_new_cte = False
4413 else:
4414 # otherwise these are two CTEs that either will render
4415 # differently, or were indicated separately by the user,
4416 # with the same name
4417 raise exc.CompileError(
4418 "Multiple, unrelated CTEs found with "
4419 "the same name: %r" % cte_name
4420 )
4422 if not asfrom and not is_new_cte:
4423 return None
4425 if cte._cte_alias is not None:
4426 pre_alias_cte = cte._cte_alias
4427 cte_pre_alias_name = cte._cte_alias.name
4428 if isinstance(cte_pre_alias_name, elements._truncated_label):
4429 cte_pre_alias_name = self._truncated_identifier(
4430 "alias", cte_pre_alias_name
4431 )
4432 else:
4433 pre_alias_cte = cte
4434 cte_pre_alias_name = None
4436 if is_new_cte:
4437 self.ctes_by_level_name[cte_level_name] = cte
4438 self.level_name_by_cte[_reference_cte] = cte_level_name + (
4439 cte_opts,
4440 )
4442 if pre_alias_cte not in self.ctes:
4443 self.visit_cte(pre_alias_cte, **kwargs)
4445 if not cte_pre_alias_name and cte not in self_ctes:
4446 if cte.recursive:
4447 self.ctes_recursive = True
4448 text = self.preparer.format_alias(cte, cte_name)
4449 if cte.recursive or cte.element.name_cte_columns:
4450 col_source = cte.element
4452 # TODO: can we get at the .columns_plus_names collection
4453 # that is already (or will be?) generated for the SELECT
4454 # rather than calling twice?
4455 recur_cols = [
4456 # TODO: proxy_name is not technically safe,
4457 # see test_cte->
4458 # test_with_recursive_no_name_currently_buggy. not
4459 # clear what should be done with such a case
4460 fallback_label_name or proxy_name
4461 for (
4462 _,
4463 proxy_name,
4464 fallback_label_name,
4465 c,
4466 repeated,
4467 ) in (col_source._generate_columns_plus_names(True))
4468 if not repeated
4469 ]
4471 text += "(%s)" % (
4472 ", ".join(
4473 self.preparer.format_label_name(
4474 ident, anon_map=self.anon_map
4475 )
4476 for ident in recur_cols
4477 )
4478 )
4480 assert kwargs.get("subquery", False) is False
4482 if not self.stack:
4483 # toplevel, this is a stringify of the
4484 # cte directly. just compile the inner
4485 # the way alias() does.
4486 return cte.element._compiler_dispatch(
4487 self, asfrom=asfrom, **kwargs
4488 )
4489 else:
4490 prefixes = self._generate_prefixes(
4491 cte, cte._prefixes, **kwargs
4492 )
4493 inner = cte.element._compiler_dispatch(
4494 self, asfrom=True, **kwargs
4495 )
4497 text += " AS %s\n(%s)" % (prefixes, inner)
4499 if cte._suffixes:
4500 text += " " + self._generate_prefixes(
4501 cte, cte._suffixes, **kwargs
4502 )
4504 self_ctes[cte] = text
4506 if asfrom:
4507 if from_linter:
4508 from_linter.froms[cte._de_clone()] = cte_name
4510 if not is_new_cte and embedded_in_current_named_cte:
4511 return self.preparer.format_alias(cte, cte_name)
4513 if cte_pre_alias_name:
4514 text = self.preparer.format_alias(cte, cte_pre_alias_name)
4515 if self.preparer._requires_quotes(cte_name):
4516 cte_name = self.preparer.quote(cte_name)
4517 text += self.get_render_as_alias_suffix(cte_name)
4518 return text # type: ignore[no-any-return]
4519 else:
4520 return self.preparer.format_alias(cte, cte_name)
4522 return None
4524 def visit_table_valued_alias(self, element, **kw):
4525 if element.joins_implicitly:
4526 kw["from_linter"] = None
4527 if element._is_lateral:
4528 return self.visit_lateral(element, **kw)
4529 else:
4530 return self.visit_alias(element, **kw)
4532 def visit_table_valued_column(self, element, **kw):
4533 return self.visit_column(element, **kw)
4535 def visit_alias(
4536 self,
4537 alias,
4538 asfrom=False,
4539 ashint=False,
4540 iscrud=False,
4541 fromhints=None,
4542 subquery=False,
4543 lateral=False,
4544 enclosing_alias=None,
4545 from_linter=None,
4546 within_tstring=False,
4547 **kwargs,
4548 ):
4549 if lateral:
4550 if "enclosing_lateral" not in kwargs:
4551 # if lateral is set and enclosing_lateral is not
4552 # present, we assume we are being called directly
4553 # from visit_lateral() and we need to set enclosing_lateral.
4554 assert alias._is_lateral
4555 kwargs["enclosing_lateral"] = alias
4557 # for lateral objects, we track a second from_linter that is...
4558 # lateral! to the level above us.
4559 if (
4560 from_linter
4561 and "lateral_from_linter" not in kwargs
4562 and "enclosing_lateral" in kwargs
4563 ):
4564 kwargs["lateral_from_linter"] = from_linter
4566 if enclosing_alias is not None and enclosing_alias.element is alias:
4567 inner = alias.element._compiler_dispatch(
4568 self,
4569 asfrom=asfrom,
4570 ashint=ashint,
4571 iscrud=iscrud,
4572 fromhints=fromhints,
4573 lateral=lateral,
4574 enclosing_alias=alias,
4575 **kwargs,
4576 )
4577 if subquery and (asfrom or lateral):
4578 inner = "(%s)" % (inner,)
4579 return inner
4580 else:
4581 kwargs["enclosing_alias"] = alias
4583 if asfrom or ashint or within_tstring:
4584 if isinstance(alias.name, elements._truncated_label):
4585 alias_name = self._truncated_identifier("alias", alias.name)
4586 else:
4587 alias_name = alias.name
4589 if ashint:
4590 return self.preparer.format_alias(alias, alias_name)
4591 elif asfrom or within_tstring:
4592 if from_linter:
4593 from_linter.froms[alias._de_clone()] = alias_name
4595 inner = alias.element._compiler_dispatch(
4596 self, asfrom=True, lateral=lateral, **kwargs
4597 )
4598 if subquery:
4599 inner = "(%s)" % (inner,)
4601 ret = inner + self.get_render_as_alias_suffix(
4602 self.preparer.format_alias(alias, alias_name)
4603 )
4605 if alias._supports_derived_columns and alias._render_derived:
4606 ret += "(%s)" % (
4607 ", ".join(
4608 "%s%s"
4609 % (
4610 self.preparer.quote(col.name),
4611 (
4612 " %s"
4613 % self.dialect.type_compiler_instance.process(
4614 col.type, **kwargs
4615 )
4616 if alias._render_derived_w_types
4617 else ""
4618 ),
4619 )
4620 for col in alias.c
4621 )
4622 )
4624 if fromhints and alias in fromhints:
4625 ret = self.format_from_hint_text(
4626 ret, alias, fromhints[alias], iscrud
4627 )
4629 return ret
4630 else:
4631 # note we cancel the "subquery" flag here as well
4632 return alias.element._compiler_dispatch(
4633 self, lateral=lateral, **kwargs
4634 )
4636 def visit_subquery(self, subquery, **kw):
4637 kw["subquery"] = True
4638 return self.visit_alias(subquery, **kw)
4640 def visit_lateral(self, lateral_, **kw):
4641 kw["lateral"] = True
4642 return "LATERAL %s" % self.visit_alias(lateral_, **kw)
4644 def visit_tablesample(self, tablesample, asfrom=False, **kw):
4645 text = "%s TABLESAMPLE %s" % (
4646 self.visit_alias(tablesample, asfrom=True, **kw),
4647 tablesample._get_method()._compiler_dispatch(self, **kw),
4648 )
4650 if tablesample.seed is not None:
4651 text += " REPEATABLE (%s)" % (
4652 tablesample.seed._compiler_dispatch(self, **kw)
4653 )
4655 return text
4657 def _render_values(self, element, **kw):
4658 kw.setdefault("literal_binds", element.literal_binds)
4659 tuples = ", ".join(
4660 self.process(
4661 elements.Tuple(
4662 types=element._column_types, *elem
4663 ).self_group(),
4664 **kw,
4665 )
4666 for chunk in element._data
4667 for elem in chunk
4668 )
4669 return f"VALUES {tuples}"
4671 def visit_values(
4672 self, element, asfrom=False, from_linter=None, visiting_cte=None, **kw
4673 ):
4675 if element._independent_ctes:
4676 self._dispatch_independent_ctes(element, kw)
4678 v = self._render_values(element, **kw)
4680 if element._unnamed:
4681 name = None
4682 elif isinstance(element.name, elements._truncated_label):
4683 name = self._truncated_identifier("values", element.name)
4684 else:
4685 name = element.name
4687 if element._is_lateral:
4688 lateral = "LATERAL "
4689 else:
4690 lateral = ""
4692 if asfrom:
4693 if from_linter:
4694 from_linter.froms[element._de_clone()] = (
4695 name if name is not None else "(unnamed VALUES element)"
4696 )
4698 if visiting_cte is not None and visiting_cte.element is element:
4699 if element._is_lateral:
4700 raise exc.CompileError(
4701 "Can't use a LATERAL VALUES expression inside of a CTE"
4702 )
4703 elif name:
4704 kw["include_table"] = False
4705 v = "%s(%s)%s (%s)" % (
4706 lateral,
4707 v,
4708 self.get_render_as_alias_suffix(self.preparer.quote(name)),
4709 (
4710 ", ".join(
4711 c._compiler_dispatch(self, **kw)
4712 for c in element.columns
4713 )
4714 ),
4715 )
4716 else:
4717 v = "%s(%s)" % (lateral, v)
4718 return v
4720 def visit_scalar_values(self, element, **kw):
4721 return f"({self._render_values(element, **kw)})"
4723 def get_render_as_alias_suffix(self, alias_name_text):
4724 return " AS " + alias_name_text
4726 def _add_to_result_map(
4727 self,
4728 keyname: str,
4729 name: str,
4730 objects: Tuple[Any, ...],
4731 type_: TypeEngine[Any],
4732 ) -> None:
4734 # note objects must be non-empty for cursor.py to handle the
4735 # collection properly
4736 assert objects
4738 if keyname is None or keyname == "*":
4739 self._ordered_columns = False
4740 self._ad_hoc_textual = True
4741 if type_._is_tuple_type:
4742 raise exc.CompileError(
4743 "Most backends don't support SELECTing "
4744 "from a tuple() object. If this is an ORM query, "
4745 "consider using the Bundle object."
4746 )
4747 self._result_columns.append(
4748 ResultColumnsEntry(keyname, name, objects, type_)
4749 )
4751 def _label_returning_column(
4752 self, stmt, column, populate_result_map, column_clause_args=None, **kw
4753 ):
4754 """Render a column with necessary labels inside of a RETURNING clause.
4756 This method is provided for individual dialects in place of calling
4757 the _label_select_column method directly, so that the two use cases
4758 of RETURNING vs. SELECT can be disambiguated going forward.
4760 .. versionadded:: 1.4.21
4762 """
4763 return self._label_select_column(
4764 None,
4765 column,
4766 populate_result_map,
4767 False,
4768 {} if column_clause_args is None else column_clause_args,
4769 **kw,
4770 )
4772 def _label_select_column(
4773 self,
4774 select,
4775 column,
4776 populate_result_map,
4777 asfrom,
4778 column_clause_args,
4779 name=None,
4780 proxy_name=None,
4781 fallback_label_name=None,
4782 within_columns_clause=True,
4783 column_is_repeated=False,
4784 need_column_expressions=False,
4785 include_table=True,
4786 ):
4787 """produce labeled columns present in a select()."""
4788 impl = column.type.dialect_impl(self.dialect)
4790 if impl._has_column_expression and (
4791 need_column_expressions or populate_result_map
4792 ):
4793 col_expr = impl.column_expression(column)
4794 else:
4795 col_expr = column
4797 if populate_result_map:
4798 # pass an "add_to_result_map" callable into the compilation
4799 # of embedded columns. this collects information about the
4800 # column as it will be fetched in the result and is coordinated
4801 # with cursor.description when the query is executed.
4802 add_to_result_map = self._add_to_result_map
4804 # if the SELECT statement told us this column is a repeat,
4805 # wrap the callable with one that prevents the addition of the
4806 # targets
4807 if column_is_repeated:
4808 _add_to_result_map = add_to_result_map
4810 def add_to_result_map(keyname, name, objects, type_):
4811 _add_to_result_map(keyname, name, (keyname,), type_)
4813 # if we redefined col_expr for type expressions, wrap the
4814 # callable with one that adds the original column to the targets
4815 elif col_expr is not column:
4816 _add_to_result_map = add_to_result_map
4818 def add_to_result_map(keyname, name, objects, type_):
4819 _add_to_result_map(
4820 keyname, name, (column,) + objects, type_
4821 )
4823 else:
4824 add_to_result_map = None
4826 # this method is used by some of the dialects for RETURNING,
4827 # which has different inputs. _label_returning_column was added
4828 # as the better target for this now however for 1.4 we will keep
4829 # _label_select_column directly compatible with this use case.
4830 # these assertions right now set up the current expected inputs
4831 assert within_columns_clause, (
4832 "_label_select_column is only relevant within "
4833 "the columns clause of a SELECT or RETURNING"
4834 )
4835 result_expr: elements.Label[Any] | _CompileLabel
4837 if isinstance(column, elements.Label):
4838 if col_expr is not column:
4839 result_expr = _CompileLabel(
4840 col_expr, column.name, alt_names=(column.element,)
4841 )
4842 else:
4843 result_expr = col_expr
4845 elif name:
4846 # here, _columns_plus_names has determined there's an explicit
4847 # label name we need to use. this is the default for
4848 # tablenames_plus_columnnames as well as when columns are being
4849 # deduplicated on name
4851 assert (
4852 proxy_name is not None
4853 ), "proxy_name is required if 'name' is passed"
4855 result_expr = _CompileLabel(
4856 col_expr,
4857 name,
4858 alt_names=(
4859 proxy_name,
4860 # this is a hack to allow legacy result column lookups
4861 # to work as they did before; this goes away in 2.0.
4862 # TODO: this only seems to be tested indirectly
4863 # via test/orm/test_deprecations.py. should be a
4864 # resultset test for this
4865 column._tq_label,
4866 ),
4867 )
4868 else:
4869 # determine here whether this column should be rendered in
4870 # a labelled context or not, as we were given no required label
4871 # name from the caller. Here we apply heuristics based on the kind
4872 # of SQL expression involved.
4874 if col_expr is not column:
4875 # type-specific expression wrapping the given column,
4876 # so we render a label
4877 render_with_label = True
4878 elif isinstance(column, elements.ColumnClause):
4879 # table-bound column, we render its name as a label if we are
4880 # inside of a subquery only
4881 render_with_label = (
4882 asfrom
4883 and not column.is_literal
4884 and column.table is not None
4885 )
4886 elif isinstance(column, elements.TextClause):
4887 render_with_label = False
4888 elif isinstance(column, elements.UnaryExpression):
4889 # unary expression. notes added as of #12681
4890 #
4891 # By convention, the visit_unary() method
4892 # itself does not add an entry to the result map, and relies
4893 # upon either the inner expression creating a result map
4894 # entry, or if not, by creating a label here that produces
4895 # the result map entry. Where that happens is based on whether
4896 # or not the element immediately inside the unary is a
4897 # NamedColumn subclass or not.
4898 #
4899 # Now, this also impacts how the SELECT is written; if
4900 # we decide to generate a label here, we get the usual
4901 # "~(x+y) AS anon_1" thing in the columns clause. If we
4902 # don't, we don't get an AS at all, we get like
4903 # "~table.column".
4904 #
4905 # But here is the important thing as of modernish (like 1.4)
4906 # versions of SQLAlchemy - **whether or not the AS <label>
4907 # is present in the statement is not actually important**.
4908 # We target result columns **positionally** for a fully
4909 # compiled ``Select()`` object; before 1.4 we needed those
4910 # labels to match in cursor.description etc etc but now it
4911 # really doesn't matter.
4912 # So really, we could set render_with_label True in all cases.
4913 # Or we could just have visit_unary() populate the result map
4914 # in all cases.
4915 #
4916 # What we're doing here is strictly trying to not rock the
4917 # boat too much with when we do/don't render "AS label";
4918 # labels being present helps in the edge cases that we
4919 # "fall back" to named cursor.description matching, labels
4920 # not being present for columns keeps us from having awkward
4921 # phrases like "SELECT DISTINCT table.x AS x".
4922 render_with_label = (
4923 (
4924 # exception case to detect if we render "not boolean"
4925 # as "not <col>" for native boolean or "<col> = 1"
4926 # for non-native boolean. this is controlled by
4927 # visit_is_<true|false>_unary_operator
4928 column.operator
4929 in (operators.is_false, operators.is_true)
4930 and not self.dialect.supports_native_boolean
4931 )
4932 or column._wraps_unnamed_column()
4933 or asfrom
4934 )
4935 elif (
4936 # general class of expressions that don't have a SQL-column
4937 # addressable name. includes scalar selects, bind parameters,
4938 # SQL functions, others
4939 not isinstance(column, elements.NamedColumn)
4940 # deeper check that indicates there's no natural "name" to
4941 # this element, which accommodates for custom SQL constructs
4942 # that might have a ".name" attribute (but aren't SQL
4943 # functions) but are not implementing this more recently added
4944 # base class. in theory the "NamedColumn" check should be
4945 # enough, however here we seek to maintain legacy behaviors
4946 # as well.
4947 and column._non_anon_label is None
4948 ):
4949 render_with_label = True
4950 else:
4951 render_with_label = False
4953 if render_with_label:
4954 if not fallback_label_name:
4955 # used by the RETURNING case right now. we generate it
4956 # here as 3rd party dialects may be referring to
4957 # _label_select_column method directly instead of the
4958 # just-added _label_returning_column method
4959 assert not column_is_repeated
4960 fallback_label_name = column._anon_name_label
4962 fallback_label_name = (
4963 elements._truncated_label(fallback_label_name)
4964 if not isinstance(
4965 fallback_label_name, elements._truncated_label
4966 )
4967 else fallback_label_name
4968 )
4970 result_expr = _CompileLabel(
4971 col_expr, fallback_label_name, alt_names=(proxy_name,)
4972 )
4973 else:
4974 result_expr = col_expr
4976 column_clause_args.update(
4977 within_columns_clause=within_columns_clause,
4978 add_to_result_map=add_to_result_map,
4979 include_table=include_table,
4980 within_tstring=False,
4981 )
4982 return result_expr._compiler_dispatch(self, **column_clause_args)
4984 def format_from_hint_text(self, sqltext, table, hint, iscrud):
4985 hinttext = self.get_from_hint_text(table, hint)
4986 if hinttext:
4987 sqltext += " " + hinttext
4988 return sqltext
4990 def get_select_hint_text(self, byfroms):
4991 return None
4993 def get_from_hint_text(
4994 self, table: FromClause, text: Optional[str]
4995 ) -> Optional[str]:
4996 return None
4998 def get_crud_hint_text(self, table, text):
4999 return None
5001 def get_statement_hint_text(self, hint_texts):
5002 return " ".join(hint_texts)
5004 _default_stack_entry: _CompilerStackEntry
5006 if not typing.TYPE_CHECKING:
5007 _default_stack_entry = util.immutabledict(
5008 [("correlate_froms", frozenset()), ("asfrom_froms", frozenset())]
5009 )
5011 def _display_froms_for_select(
5012 self, select_stmt, asfrom, lateral=False, **kw
5013 ):
5014 # utility method to help external dialects
5015 # get the correct from list for a select.
5016 # specifically the oracle dialect needs this feature
5017 # right now.
5018 toplevel = not self.stack
5019 entry = self._default_stack_entry if toplevel else self.stack[-1]
5021 compile_state = select_stmt._compile_state_factory(select_stmt, self)
5023 correlate_froms = entry["correlate_froms"]
5024 asfrom_froms = entry["asfrom_froms"]
5026 if asfrom and not lateral:
5027 froms = compile_state._get_display_froms(
5028 explicit_correlate_froms=correlate_froms.difference(
5029 asfrom_froms
5030 ),
5031 implicit_correlate_froms=(),
5032 )
5033 else:
5034 froms = compile_state._get_display_froms(
5035 explicit_correlate_froms=correlate_froms,
5036 implicit_correlate_froms=asfrom_froms,
5037 )
5038 return froms
5040 translate_select_structure: Any = None
5041 """if not ``None``, should be a callable which accepts ``(select_stmt,
5042 **kw)`` and returns a select object. this is used for structural changes
5043 mostly to accommodate for LIMIT/OFFSET schemes
5045 """
5047 def visit_select(
5048 self,
5049 select_stmt,
5050 asfrom=False,
5051 insert_into=False,
5052 fromhints=None,
5053 compound_index=None,
5054 select_wraps_for=None,
5055 lateral=False,
5056 from_linter=None,
5057 **kwargs,
5058 ):
5059 assert select_wraps_for is None, (
5060 "SQLAlchemy 1.4 requires use of "
5061 "the translate_select_structure hook for structural "
5062 "translations of SELECT objects"
5063 )
5064 if self._collect_params:
5065 self._add_to_params(select_stmt)
5067 # initial setup of SELECT. the compile_state_factory may now
5068 # be creating a totally different SELECT from the one that was
5069 # passed in. for ORM use this will convert from an ORM-state
5070 # SELECT to a regular "Core" SELECT. other composed operations
5071 # such as computation of joins will be performed.
5073 kwargs["within_columns_clause"] = False
5075 compile_state = select_stmt._compile_state_factory(
5076 select_stmt, self, **kwargs
5077 )
5078 kwargs["ambiguous_table_name_map"] = (
5079 compile_state._ambiguous_table_name_map
5080 )
5082 select_stmt = compile_state.statement
5084 toplevel = not self.stack
5086 if toplevel and not self.compile_state:
5087 self.compile_state = compile_state
5089 is_embedded_select = compound_index is not None or insert_into
5091 # translate step for Oracle, SQL Server which often need to
5092 # restructure the SELECT to allow for LIMIT/OFFSET and possibly
5093 # other conditions
5094 if self.translate_select_structure:
5095 new_select_stmt = self.translate_select_structure(
5096 select_stmt, asfrom=asfrom, **kwargs
5097 )
5099 # if SELECT was restructured, maintain a link to the originals
5100 # and assemble a new compile state
5101 if new_select_stmt is not select_stmt:
5102 compile_state_wraps_for = compile_state
5103 select_wraps_for = select_stmt
5104 select_stmt = new_select_stmt
5106 compile_state = select_stmt._compile_state_factory(
5107 select_stmt, self, **kwargs
5108 )
5109 select_stmt = compile_state.statement
5111 entry = self._default_stack_entry if toplevel else self.stack[-1]
5113 populate_result_map = need_column_expressions = (
5114 toplevel
5115 or entry.get("need_result_map_for_compound", False)
5116 or entry.get("need_result_map_for_nested", False)
5117 )
5119 # indicates there is a CompoundSelect in play and we are not the
5120 # first select
5121 if compound_index:
5122 populate_result_map = False
5124 # this was first proposed as part of #3372; however, it is not
5125 # reached in current tests and could possibly be an assertion
5126 # instead.
5127 if not populate_result_map and "add_to_result_map" in kwargs:
5128 del kwargs["add_to_result_map"]
5130 froms = self._setup_select_stack(
5131 select_stmt, compile_state, entry, asfrom, lateral, compound_index
5132 )
5134 column_clause_args = kwargs.copy()
5135 column_clause_args.update(
5136 {"within_label_clause": False, "within_columns_clause": False}
5137 )
5139 text = "SELECT " # we're off to a good start !
5141 if select_stmt._post_select_clause is not None:
5142 psc = self.process(select_stmt._post_select_clause, **kwargs)
5143 if psc is not None:
5144 text += psc + " "
5146 if select_stmt._hints:
5147 hint_text, byfrom = self._setup_select_hints(select_stmt)
5148 if hint_text:
5149 text += hint_text + " "
5150 else:
5151 byfrom = None
5153 if select_stmt._independent_ctes:
5154 self._dispatch_independent_ctes(select_stmt, kwargs)
5156 if select_stmt._prefixes:
5157 text += self._generate_prefixes(
5158 select_stmt, select_stmt._prefixes, **kwargs
5159 )
5161 text += self.get_select_precolumns(select_stmt, **kwargs)
5163 if select_stmt._pre_columns_clause is not None:
5164 pcc = self.process(select_stmt._pre_columns_clause, **kwargs)
5165 if pcc is not None:
5166 text += pcc + " "
5168 # the actual list of columns to print in the SELECT column list.
5169 inner_columns = [
5170 c
5171 for c in [
5172 self._label_select_column(
5173 select_stmt,
5174 column,
5175 populate_result_map,
5176 asfrom,
5177 column_clause_args,
5178 name=name,
5179 proxy_name=proxy_name,
5180 fallback_label_name=fallback_label_name,
5181 column_is_repeated=repeated,
5182 need_column_expressions=need_column_expressions,
5183 )
5184 for (
5185 name,
5186 proxy_name,
5187 fallback_label_name,
5188 column,
5189 repeated,
5190 ) in compile_state.columns_plus_names
5191 ]
5192 if c is not None
5193 ]
5195 if populate_result_map and select_wraps_for is not None:
5196 # if this select was generated from translate_select,
5197 # rewrite the targeted columns in the result map
5199 translate = dict(
5200 zip(
5201 [
5202 name
5203 for (
5204 key,
5205 proxy_name,
5206 fallback_label_name,
5207 name,
5208 repeated,
5209 ) in compile_state.columns_plus_names
5210 ],
5211 [
5212 name
5213 for (
5214 key,
5215 proxy_name,
5216 fallback_label_name,
5217 name,
5218 repeated,
5219 ) in compile_state_wraps_for.columns_plus_names
5220 ],
5221 )
5222 )
5224 self._result_columns = [
5225 ResultColumnsEntry(
5226 key, name, tuple(translate.get(o, o) for o in obj), type_
5227 )
5228 for key, name, obj, type_ in self._result_columns
5229 ]
5231 text = self._compose_select_body(
5232 text,
5233 select_stmt,
5234 compile_state,
5235 inner_columns,
5236 froms,
5237 byfrom,
5238 toplevel,
5239 kwargs,
5240 )
5242 if select_stmt._post_body_clause is not None:
5243 pbc = self.process(select_stmt._post_body_clause, **kwargs)
5244 if pbc:
5245 text += " " + pbc
5247 if select_stmt._statement_hints:
5248 per_dialect = [
5249 ht
5250 for (dialect_name, ht) in select_stmt._statement_hints
5251 if dialect_name in ("*", self.dialect.name)
5252 ]
5253 if per_dialect:
5254 text += " " + self.get_statement_hint_text(per_dialect)
5256 # In compound query, CTEs are shared at the compound level
5257 if self.ctes and (not is_embedded_select or toplevel):
5258 nesting_level = len(self.stack) if not toplevel else None
5259 text = self._render_cte_clause(nesting_level=nesting_level) + text
5261 if select_stmt._suffixes:
5262 text += " " + self._generate_prefixes(
5263 select_stmt, select_stmt._suffixes, **kwargs
5264 )
5266 self.stack.pop(-1)
5268 return text
5270 def _setup_select_hints(
5271 self, select: Select[Unpack[TupleAny]]
5272 ) -> Tuple[str, _FromHintsType]:
5273 byfrom = {
5274 from_: hinttext
5275 % {"name": from_._compiler_dispatch(self, ashint=True)}
5276 for (from_, dialect), hinttext in select._hints.items()
5277 if dialect in ("*", self.dialect.name)
5278 }
5279 hint_text = self.get_select_hint_text(byfrom)
5280 return hint_text, byfrom
5282 def _setup_select_stack(
5283 self, select, compile_state, entry, asfrom, lateral, compound_index
5284 ):
5285 correlate_froms = entry["correlate_froms"]
5286 asfrom_froms = entry["asfrom_froms"]
5288 if compound_index == 0:
5289 entry["select_0"] = select
5290 elif compound_index:
5291 select_0 = entry["select_0"]
5292 numcols = len(select_0._all_selected_columns)
5294 if len(compile_state.columns_plus_names) != numcols:
5295 raise exc.CompileError(
5296 "All selectables passed to "
5297 "CompoundSelect must have identical numbers of "
5298 "columns; select #%d has %d columns, select "
5299 "#%d has %d"
5300 % (
5301 1,
5302 numcols,
5303 compound_index + 1,
5304 len(select._all_selected_columns),
5305 )
5306 )
5308 if asfrom and not lateral:
5309 froms = compile_state._get_display_froms(
5310 explicit_correlate_froms=correlate_froms.difference(
5311 asfrom_froms
5312 ),
5313 implicit_correlate_froms=(),
5314 )
5315 else:
5316 froms = compile_state._get_display_froms(
5317 explicit_correlate_froms=correlate_froms,
5318 implicit_correlate_froms=asfrom_froms,
5319 )
5321 new_correlate_froms = set(_from_objects(*froms))
5322 all_correlate_froms = new_correlate_froms.union(correlate_froms)
5324 new_entry: _CompilerStackEntry = {
5325 "asfrom_froms": new_correlate_froms,
5326 "correlate_froms": all_correlate_froms,
5327 "selectable": select,
5328 "compile_state": compile_state,
5329 }
5330 self.stack.append(new_entry)
5332 return froms
5334 def _compose_select_body(
5335 self,
5336 text,
5337 select,
5338 compile_state,
5339 inner_columns,
5340 froms,
5341 byfrom,
5342 toplevel,
5343 kwargs,
5344 ):
5345 text += ", ".join(inner_columns)
5347 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
5348 from_linter = FromLinter({}, set())
5349 warn_linting = self.linting & WARN_LINTING
5350 if toplevel:
5351 self.from_linter = from_linter
5352 else:
5353 from_linter = None
5354 warn_linting = False
5356 # adjust the whitespace for no inner columns, part of #9440,
5357 # so that a no-col SELECT comes out as "SELECT WHERE..." or
5358 # "SELECT FROM ...".
5359 # while it would be better to have built the SELECT starting string
5360 # without trailing whitespace first, then add whitespace only if inner
5361 # cols were present, this breaks compatibility with various custom
5362 # compilation schemes that are currently being tested.
5363 if not inner_columns:
5364 text = text.rstrip()
5366 if froms:
5367 text += " \nFROM "
5369 if select._hints:
5370 text += ", ".join(
5371 [
5372 f._compiler_dispatch(
5373 self,
5374 asfrom=True,
5375 fromhints=byfrom,
5376 from_linter=from_linter,
5377 **kwargs,
5378 )
5379 for f in froms
5380 ]
5381 )
5382 else:
5383 text += ", ".join(
5384 [
5385 f._compiler_dispatch(
5386 self,
5387 asfrom=True,
5388 from_linter=from_linter,
5389 **kwargs,
5390 )
5391 for f in froms
5392 ]
5393 )
5394 else:
5395 text += self.default_from()
5397 if select._where_criteria:
5398 t = self._generate_delimited_and_list(
5399 select._where_criteria, from_linter=from_linter, **kwargs
5400 )
5401 if t:
5402 text += " \nWHERE " + t
5404 if warn_linting:
5405 assert from_linter is not None
5406 from_linter.warn()
5408 if select._group_by_clauses:
5409 text += self.group_by_clause(select, **kwargs)
5411 if select._having_criteria:
5412 t = self._generate_delimited_and_list(
5413 select._having_criteria, **kwargs
5414 )
5415 if t:
5416 text += " \nHAVING " + t
5418 if select._post_criteria_clause is not None:
5419 pcc = self.process(select._post_criteria_clause, **kwargs)
5420 if pcc is not None:
5421 text += " \n" + pcc
5423 if select._order_by_clauses:
5424 text += self.order_by_clause(select, **kwargs)
5426 if select._has_row_limiting_clause:
5427 text += self._row_limit_clause(select, **kwargs)
5429 if select._for_update_arg is not None:
5430 text += self.for_update_clause(select, **kwargs)
5432 return text
5434 def _generate_prefixes(self, stmt, prefixes, **kw):
5435 clause = " ".join(
5436 prefix._compiler_dispatch(self, **kw)
5437 for prefix, dialect_name in prefixes
5438 if dialect_name in (None, "*") or dialect_name == self.dialect.name
5439 )
5440 if clause:
5441 clause += " "
5442 return clause
5444 def _render_cte_clause(
5445 self,
5446 nesting_level=None,
5447 include_following_stack=False,
5448 ):
5449 """
5450 include_following_stack
5451 Also render the nesting CTEs on the next stack. Useful for
5452 SQL structures like UNION or INSERT that can wrap SELECT
5453 statements containing nesting CTEs.
5454 """
5455 if not self.ctes:
5456 return ""
5458 ctes: MutableMapping[CTE, str]
5460 if nesting_level and nesting_level > 1:
5461 ctes = util.OrderedDict()
5462 for cte in list(self.ctes.keys()):
5463 cte_level, cte_name, cte_opts = self.level_name_by_cte[
5464 cte._get_reference_cte()
5465 ]
5466 nesting = cte.nesting or cte_opts.nesting
5467 is_rendered_level = cte_level == nesting_level or (
5468 include_following_stack and cte_level == nesting_level + 1
5469 )
5470 if not (nesting and is_rendered_level):
5471 continue
5473 ctes[cte] = self.ctes[cte]
5475 else:
5476 ctes = self.ctes
5478 if not ctes:
5479 return ""
5480 ctes_recursive = any([cte.recursive for cte in ctes])
5482 cte_text = self.get_cte_preamble(ctes_recursive) + " "
5483 cte_text += ", \n".join([txt for txt in ctes.values()])
5484 cte_text += "\n "
5486 if nesting_level and nesting_level > 1:
5487 for cte in list(ctes.keys()):
5488 cte_level, cte_name, cte_opts = self.level_name_by_cte[
5489 cte._get_reference_cte()
5490 ]
5491 del self.ctes[cte]
5492 del self.ctes_by_level_name[(cte_level, cte_name)]
5493 del self.level_name_by_cte[cte._get_reference_cte()]
5495 return cte_text
5497 def get_cte_preamble(self, recursive):
5498 if recursive:
5499 return "WITH RECURSIVE"
5500 else:
5501 return "WITH"
5503 def get_select_precolumns(self, select: Select[Any], **kw: Any) -> str:
5504 """Called when building a ``SELECT`` statement, position is just
5505 before column list.
5507 """
5508 if select._distinct_on:
5509 util.warn_deprecated(
5510 "DISTINCT ON is currently supported only by the PostgreSQL "
5511 "dialect. Use of DISTINCT ON for other backends is currently "
5512 "silently ignored, however this usage is deprecated, and will "
5513 "raise CompileError in a future release for all backends "
5514 "that do not support this syntax.",
5515 version="1.4",
5516 )
5517 return "DISTINCT " if select._distinct else ""
5519 def group_by_clause(self, select, **kw):
5520 """allow dialects to customize how GROUP BY is rendered."""
5522 group_by = self._generate_delimited_list(
5523 select._group_by_clauses, OPERATORS[operators.comma_op], **kw
5524 )
5525 if group_by:
5526 return " GROUP BY " + group_by
5527 else:
5528 return ""
5530 def order_by_clause(self, select, **kw):
5531 """allow dialects to customize how ORDER BY is rendered."""
5533 order_by = self._generate_delimited_list(
5534 select._order_by_clauses, OPERATORS[operators.comma_op], **kw
5535 )
5537 if order_by:
5538 return " ORDER BY " + order_by
5539 else:
5540 return ""
5542 def for_update_clause(self, select, **kw):
5543 return " FOR UPDATE"
5545 def returning_clause(
5546 self,
5547 stmt: UpdateBase,
5548 returning_cols: Sequence[_ColumnsClauseElement],
5549 *,
5550 populate_result_map: bool,
5551 **kw: Any,
5552 ) -> str:
5553 columns = [
5554 self._label_returning_column(
5555 stmt,
5556 column,
5557 populate_result_map,
5558 fallback_label_name=fallback_label_name,
5559 column_is_repeated=repeated,
5560 name=name,
5561 proxy_name=proxy_name,
5562 **kw,
5563 )
5564 for (
5565 name,
5566 proxy_name,
5567 fallback_label_name,
5568 column,
5569 repeated,
5570 ) in stmt._generate_columns_plus_names(
5571 True, cols=base._select_iterables(returning_cols)
5572 )
5573 ]
5575 return "RETURNING " + ", ".join(columns)
5577 def limit_clause(self, select, **kw):
5578 text = ""
5579 if select._limit_clause is not None:
5580 text += "\n LIMIT " + self.process(select._limit_clause, **kw)
5581 if select._offset_clause is not None:
5582 if select._limit_clause is None:
5583 text += "\n LIMIT -1"
5584 text += " OFFSET " + self.process(select._offset_clause, **kw)
5585 return text
5587 def fetch_clause(
5588 self,
5589 select,
5590 fetch_clause=None,
5591 require_offset=False,
5592 use_literal_execute_for_simple_int=False,
5593 **kw,
5594 ):
5595 if fetch_clause is None:
5596 fetch_clause = select._fetch_clause
5597 fetch_clause_options = select._fetch_clause_options
5598 else:
5599 fetch_clause_options = {"percent": False, "with_ties": False}
5601 text = ""
5603 if select._offset_clause is not None:
5604 offset_clause = select._offset_clause
5605 if (
5606 use_literal_execute_for_simple_int
5607 and select._simple_int_clause(offset_clause)
5608 ):
5609 offset_clause = offset_clause.render_literal_execute()
5610 offset_str = self.process(offset_clause, **kw)
5611 text += "\n OFFSET %s ROWS" % offset_str
5612 elif require_offset:
5613 text += "\n OFFSET 0 ROWS"
5615 if fetch_clause is not None:
5616 if (
5617 use_literal_execute_for_simple_int
5618 and select._simple_int_clause(fetch_clause)
5619 ):
5620 fetch_clause = fetch_clause.render_literal_execute()
5621 text += "\n FETCH FIRST %s%s ROWS %s" % (
5622 self.process(fetch_clause, **kw),
5623 " PERCENT" if fetch_clause_options["percent"] else "",
5624 "WITH TIES" if fetch_clause_options["with_ties"] else "ONLY",
5625 )
5626 return text
5628 def visit_table(
5629 self,
5630 table,
5631 asfrom=False,
5632 iscrud=False,
5633 ashint=False,
5634 fromhints=None,
5635 use_schema=True,
5636 from_linter=None,
5637 ambiguous_table_name_map=None,
5638 enclosing_alias=None,
5639 within_tstring=False,
5640 **kwargs,
5641 ):
5642 if from_linter:
5643 from_linter.froms[table] = table.fullname
5645 if asfrom or ashint or within_tstring:
5646 effective_schema = self.preparer.schema_for_object(table)
5648 if use_schema and effective_schema:
5649 ret = (
5650 self.preparer.quote_schema(effective_schema)
5651 + "."
5652 + self.preparer.quote(table.name)
5653 )
5654 else:
5655 ret = self.preparer.quote(table.name)
5657 if (
5658 (
5659 enclosing_alias is None
5660 or enclosing_alias.element is not table
5661 )
5662 and not effective_schema
5663 and ambiguous_table_name_map
5664 and table.name in ambiguous_table_name_map
5665 ):
5666 anon_name = self._truncated_identifier(
5667 "alias", ambiguous_table_name_map[table.name]
5668 )
5670 ret = ret + self.get_render_as_alias_suffix(
5671 self.preparer.format_alias(None, anon_name)
5672 )
5674 if fromhints and table in fromhints:
5675 ret = self.format_from_hint_text(
5676 ret, table, fromhints[table], iscrud
5677 )
5678 return ret
5679 else:
5680 return ""
5682 def visit_join(self, join, asfrom=False, from_linter=None, **kwargs):
5683 if from_linter:
5684 from_linter.edges.update(
5685 itertools.product(
5686 _de_clone(join.left._from_objects),
5687 _de_clone(join.right._from_objects),
5688 )
5689 )
5691 if join.full:
5692 join_type = " FULL OUTER JOIN "
5693 elif join.isouter:
5694 join_type = " LEFT OUTER JOIN "
5695 else:
5696 join_type = " JOIN "
5697 return (
5698 join.left._compiler_dispatch(
5699 self, asfrom=True, from_linter=from_linter, **kwargs
5700 )
5701 + join_type
5702 + join.right._compiler_dispatch(
5703 self, asfrom=True, from_linter=from_linter, **kwargs
5704 )
5705 + " ON "
5706 # TODO: likely need asfrom=True here?
5707 + join.onclause._compiler_dispatch(
5708 self, from_linter=from_linter, **kwargs
5709 )
5710 )
5712 def _setup_crud_hints(self, stmt, table_text):
5713 dialect_hints = {
5714 table: hint_text
5715 for (table, dialect), hint_text in stmt._hints.items()
5716 if dialect in ("*", self.dialect.name)
5717 }
5718 if stmt.table in dialect_hints:
5719 table_text = self.format_from_hint_text(
5720 table_text, stmt.table, dialect_hints[stmt.table], True
5721 )
5722 return dialect_hints, table_text
5724 # within the realm of "insertmanyvalues sentinel columns",
5725 # these lookups match different kinds of Column() configurations
5726 # to specific backend capabilities. they are broken into two
5727 # lookups, one for autoincrement columns and the other for non
5728 # autoincrement columns
5729 _sentinel_col_non_autoinc_lookup = util.immutabledict(
5730 {
5731 _SentinelDefaultCharacterization.CLIENTSIDE: (
5732 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5733 ),
5734 _SentinelDefaultCharacterization.SENTINEL_DEFAULT: (
5735 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5736 ),
5737 _SentinelDefaultCharacterization.NONE: (
5738 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5739 ),
5740 _SentinelDefaultCharacterization.IDENTITY: (
5741 InsertmanyvaluesSentinelOpts.IDENTITY
5742 ),
5743 _SentinelDefaultCharacterization.SEQUENCE: (
5744 InsertmanyvaluesSentinelOpts.SEQUENCE
5745 ),
5746 _SentinelDefaultCharacterization.MONOTONIC_FUNCTION: (
5747 InsertmanyvaluesSentinelOpts.MONOTONIC_FUNCTION
5748 ),
5749 }
5750 )
5751 _sentinel_col_autoinc_lookup = _sentinel_col_non_autoinc_lookup.union(
5752 {
5753 _SentinelDefaultCharacterization.NONE: (
5754 InsertmanyvaluesSentinelOpts.AUTOINCREMENT
5755 ),
5756 }
5757 )
5759 def _get_sentinel_column_for_table(
5760 self, table: Table
5761 ) -> Optional[Sequence[Column[Any]]]:
5762 """given a :class:`.Table`, return a usable sentinel column or
5763 columns for this dialect if any.
5765 Return None if no sentinel columns could be identified, or raise an
5766 error if a column was marked as a sentinel explicitly but isn't
5767 compatible with this dialect.
5769 """
5771 sentinel_opts = self.dialect.insertmanyvalues_implicit_sentinel
5772 sentinel_characteristics = table._sentinel_column_characteristics
5774 sent_cols = sentinel_characteristics.columns
5776 if sent_cols is None:
5777 return None
5779 if sentinel_characteristics.is_autoinc:
5780 bitmask = self._sentinel_col_autoinc_lookup.get(
5781 sentinel_characteristics.default_characterization, 0
5782 )
5783 else:
5784 bitmask = self._sentinel_col_non_autoinc_lookup.get(
5785 sentinel_characteristics.default_characterization, 0
5786 )
5788 if sentinel_opts & bitmask:
5789 return sent_cols
5791 if sentinel_characteristics.is_explicit:
5792 # a column was explicitly marked as insert_sentinel=True,
5793 # however it is not compatible with this dialect. they should
5794 # not indicate this column as a sentinel if they need to include
5795 # this dialect.
5797 # TODO: do we want non-primary key explicit sentinel cols
5798 # that can gracefully degrade for some backends?
5799 # insert_sentinel="degrade" perhaps. not for the initial release.
5800 # I am hoping people are generally not dealing with this sentinel
5801 # business at all.
5803 # if is_explicit is True, there will be only one sentinel column.
5805 raise exc.InvalidRequestError(
5806 f"Column {sent_cols[0]} can't be explicitly "
5807 "marked as a sentinel column when using the "
5808 f"{self.dialect.name} dialect, as the "
5809 "particular type of default generation on this column is "
5810 "not currently compatible with this dialect's specific "
5811 f"INSERT..RETURNING syntax which can receive the "
5812 "server-generated value in "
5813 "a deterministic way. To remove this error, remove "
5814 "insert_sentinel=True from primary key autoincrement "
5815 "columns; these columns are automatically used as "
5816 "sentinels for supported dialects in any case."
5817 )
5819 return None
5821 def _deliver_insertmanyvalues_batches(
5822 self,
5823 statement: str,
5824 parameters: _DBAPIMultiExecuteParams,
5825 compiled_parameters: List[_MutableCoreSingleExecuteParams],
5826 generic_setinputsizes: Optional[_GenericSetInputSizesType],
5827 batch_size: int,
5828 sort_by_parameter_order: bool,
5829 schema_translate_map: Optional[SchemaTranslateMapType],
5830 ) -> Iterator[_InsertManyValuesBatch]:
5831 imv = self._insertmanyvalues
5832 assert imv is not None
5834 if not imv.sentinel_param_keys:
5835 _sentinel_from_params = None
5836 else:
5837 _sentinel_from_params = operator.itemgetter(
5838 *imv.sentinel_param_keys
5839 )
5841 lenparams = len(parameters)
5842 if imv.is_default_expr and not self.dialect.supports_default_metavalue:
5843 # backend doesn't support
5844 # INSERT INTO table (pk_col) VALUES (DEFAULT), (DEFAULT), ...
5845 # at the moment this is basically SQL Server due to
5846 # not being able to use DEFAULT for identity column
5847 # just yield out that many single statements! still
5848 # faster than a whole connection.execute() call ;)
5849 #
5850 # note we still are taking advantage of the fact that we know
5851 # we are using RETURNING. The generalized approach of fetching
5852 # cursor.lastrowid etc. still goes through the more heavyweight
5853 # "ExecutionContext per statement" system as it isn't usable
5854 # as a generic "RETURNING" approach
5855 use_row_at_a_time = True
5856 downgraded = False
5857 elif not self.dialect.supports_multivalues_insert or (
5858 sort_by_parameter_order
5859 and self._result_columns
5860 and (
5861 imv.sentinel_columns is None
5862 or (
5863 imv.includes_upsert_behaviors
5864 and not imv.embed_values_counter
5865 )
5866 )
5867 ):
5868 # deterministic order was requested and the compiler could
5869 # not organize sentinel columns for this dialect/statement.
5870 # use row at a time. Note: if embed_values_counter is True,
5871 # the counter itself provides the ordering capability we need,
5872 # so we can use batch mode even with upsert behaviors.
5873 use_row_at_a_time = True
5874 downgraded = True
5875 elif (
5876 imv.has_upsert_bound_parameters
5877 and not imv.embed_values_counter
5878 and self._result_columns
5879 ):
5880 # For upsert behaviors (ON CONFLICT DO UPDATE, etc.) with RETURNING
5881 # and parametrized bindparams in the SET clause, we must use
5882 # row-at-a-time. Batching multiple rows in a single statement
5883 # doesn't work when the SET clause contains bound parameters that
5884 # will receive different values per row, as there's only one SET
5885 # clause per statement. See issue #13130.
5886 use_row_at_a_time = True
5887 downgraded = True
5888 else:
5889 use_row_at_a_time = False
5890 downgraded = False
5892 if use_row_at_a_time:
5893 for batchnum, (param, compiled_param) in enumerate(
5894 cast(
5895 "Sequence[Tuple[_DBAPISingleExecuteParams, _MutableCoreSingleExecuteParams]]", # noqa: E501
5896 zip(parameters, compiled_parameters),
5897 ),
5898 1,
5899 ):
5900 yield _InsertManyValuesBatch(
5901 statement,
5902 param,
5903 generic_setinputsizes,
5904 [param],
5905 (
5906 [_sentinel_from_params(compiled_param)]
5907 if _sentinel_from_params
5908 else []
5909 ),
5910 1,
5911 batchnum,
5912 lenparams,
5913 sort_by_parameter_order,
5914 downgraded,
5915 )
5916 return
5918 if schema_translate_map:
5919 rst = functools.partial(
5920 self.preparer._render_schema_translates,
5921 schema_translate_map=schema_translate_map,
5922 )
5923 else:
5924 rst = None
5926 imv_single_values_expr = imv.single_values_expr
5927 if rst:
5928 imv_single_values_expr = rst(imv_single_values_expr)
5930 executemany_values = f"({imv_single_values_expr})"
5931 statement = statement.replace(executemany_values, "__EXECMANY_TOKEN__")
5933 # Use optional insertmanyvalues_max_parameters
5934 # to further shrink the batch size so that there are no more than
5935 # insertmanyvalues_max_parameters params.
5936 # Currently used by SQL Server, which limits statements to 2100 bound
5937 # parameters (actually 2099).
5938 max_params = self.dialect.insertmanyvalues_max_parameters
5939 if max_params:
5940 total_num_of_params = len(self.bind_names)
5941 num_params_per_batch = len(imv.insert_crud_params)
5942 num_params_outside_of_batch = (
5943 total_num_of_params - num_params_per_batch
5944 )
5945 batch_size = min(
5946 batch_size,
5947 (
5948 (max_params - num_params_outside_of_batch)
5949 // num_params_per_batch
5950 ),
5951 )
5953 batches = cast("List[Sequence[Any]]", list(parameters))
5954 compiled_batches = cast(
5955 "List[Sequence[Any]]", list(compiled_parameters)
5956 )
5958 processed_setinputsizes: Optional[_GenericSetInputSizesType] = None
5959 batchnum = 1
5960 total_batches = lenparams // batch_size + (
5961 1 if lenparams % batch_size else 0
5962 )
5964 insert_crud_params = imv.insert_crud_params
5965 assert insert_crud_params is not None
5967 if rst:
5968 insert_crud_params = [
5969 (col, key, rst(expr), st)
5970 for col, key, expr, st in insert_crud_params
5971 ]
5973 escaped_bind_names: Mapping[str, str]
5974 expand_pos_lower_index = expand_pos_upper_index = 0
5976 if not self.positional:
5977 if self.escaped_bind_names:
5978 escaped_bind_names = self.escaped_bind_names
5979 else:
5980 escaped_bind_names = {}
5982 all_keys = set(parameters[0])
5984 def apply_placeholders(keys, formatted):
5985 for key in keys:
5986 key = escaped_bind_names.get(key, key)
5987 formatted = formatted.replace(
5988 self.bindtemplate % {"name": key},
5989 self.bindtemplate
5990 % {"name": f"{key}__EXECMANY_INDEX__"},
5991 )
5992 return formatted
5994 if imv.embed_values_counter:
5995 imv_values_counter = ", _IMV_VALUES_COUNTER"
5996 else:
5997 imv_values_counter = ""
5998 formatted_values_clause = f"""({', '.join(
5999 apply_placeholders(bind_keys, formatted)
6000 for _, _, formatted, bind_keys in insert_crud_params
6001 )}{imv_values_counter})"""
6003 keys_to_replace = all_keys.intersection(
6004 escaped_bind_names.get(key, key)
6005 for _, _, _, bind_keys in insert_crud_params
6006 for key in bind_keys
6007 )
6008 base_parameters = {
6009 key: parameters[0][key]
6010 for key in all_keys.difference(keys_to_replace)
6011 }
6013 executemany_values_w_comma = ""
6014 else:
6015 formatted_values_clause = ""
6016 keys_to_replace = set()
6017 base_parameters = {}
6019 if imv.embed_values_counter:
6020 executemany_values_w_comma = (
6021 f"({imv_single_values_expr}, _IMV_VALUES_COUNTER), "
6022 )
6023 else:
6024 executemany_values_w_comma = f"({imv_single_values_expr}), "
6026 all_names_we_will_expand: Set[str] = set()
6027 for elem in imv.insert_crud_params:
6028 all_names_we_will_expand.update(elem[3])
6030 # get the start and end position in a particular list
6031 # of parameters where we will be doing the "expanding".
6032 # statements can have params on either side or both sides,
6033 # given RETURNING and CTEs
6034 if all_names_we_will_expand:
6035 positiontup = self.positiontup
6036 assert positiontup is not None
6038 all_expand_positions = {
6039 idx
6040 for idx, name in enumerate(positiontup)
6041 if name in all_names_we_will_expand
6042 }
6043 expand_pos_lower_index = min(all_expand_positions)
6044 expand_pos_upper_index = max(all_expand_positions) + 1
6045 assert (
6046 len(all_expand_positions)
6047 == expand_pos_upper_index - expand_pos_lower_index
6048 )
6050 if self._numeric_binds:
6051 escaped = re.escape(self._numeric_binds_identifier_char)
6052 executemany_values_w_comma = re.sub(
6053 rf"{escaped}\d+", "%s", executemany_values_w_comma
6054 )
6056 while batches:
6057 batch = batches[0:batch_size]
6058 compiled_batch = compiled_batches[0:batch_size]
6060 batches[0:batch_size] = []
6061 compiled_batches[0:batch_size] = []
6063 if batches:
6064 current_batch_size = batch_size
6065 else:
6066 current_batch_size = len(batch)
6068 if generic_setinputsizes:
6069 # if setinputsizes is present, expand this collection to
6070 # suit the batch length as well
6071 # currently this will be mssql+pyodbc for internal dialects
6072 processed_setinputsizes = [
6073 (new_key, len_, typ)
6074 for new_key, len_, typ in (
6075 (f"{key}_{index}", len_, typ)
6076 for index in range(current_batch_size)
6077 for key, len_, typ in generic_setinputsizes
6078 )
6079 ]
6081 replaced_parameters: Any
6082 if self.positional:
6083 num_ins_params = imv.num_positional_params_counted
6085 batch_iterator: Iterable[Sequence[Any]]
6086 extra_params_left: Sequence[Any]
6087 extra_params_right: Sequence[Any]
6089 if num_ins_params == len(batch[0]):
6090 extra_params_left = extra_params_right = ()
6091 batch_iterator = batch
6092 else:
6093 extra_params_left = batch[0][:expand_pos_lower_index]
6094 extra_params_right = batch[0][expand_pos_upper_index:]
6095 batch_iterator = (
6096 b[expand_pos_lower_index:expand_pos_upper_index]
6097 for b in batch
6098 )
6100 if imv.embed_values_counter:
6101 expanded_values_string = (
6102 "".join(
6103 executemany_values_w_comma.replace(
6104 "_IMV_VALUES_COUNTER", str(i)
6105 )
6106 for i, _ in enumerate(batch)
6107 )
6108 )[:-2]
6109 else:
6110 expanded_values_string = (
6111 (executemany_values_w_comma * current_batch_size)
6112 )[:-2]
6114 if self._numeric_binds and num_ins_params > 0:
6115 # numeric will always number the parameters inside of
6116 # VALUES (and thus order self.positiontup) to be higher
6117 # than non-VALUES parameters, no matter where in the
6118 # statement those non-VALUES parameters appear (this is
6119 # ensured in _process_numeric by numbering first all
6120 # params that are not in _values_bindparam)
6121 # therefore all extra params are always
6122 # on the left side and numbered lower than the VALUES
6123 # parameters
6124 assert not extra_params_right
6126 start = expand_pos_lower_index + 1
6127 end = num_ins_params * (current_batch_size) + start
6129 # need to format here, since statement may contain
6130 # unescaped %, while values_string contains just (%s, %s)
6131 positions = tuple(
6132 f"{self._numeric_binds_identifier_char}{i}"
6133 for i in range(start, end)
6134 )
6135 expanded_values_string = expanded_values_string % positions
6137 replaced_statement = statement.replace(
6138 "__EXECMANY_TOKEN__", expanded_values_string
6139 )
6141 replaced_parameters = tuple(
6142 itertools.chain.from_iterable(batch_iterator)
6143 )
6145 replaced_parameters = (
6146 extra_params_left
6147 + replaced_parameters
6148 + extra_params_right
6149 )
6151 else:
6152 replaced_values_clauses = []
6153 replaced_parameters = base_parameters.copy()
6155 for i, param in enumerate(batch):
6156 fmv = formatted_values_clause.replace(
6157 "EXECMANY_INDEX__", str(i)
6158 )
6159 if imv.embed_values_counter:
6160 fmv = fmv.replace("_IMV_VALUES_COUNTER", str(i))
6162 replaced_values_clauses.append(fmv)
6163 replaced_parameters.update(
6164 {f"{key}__{i}": param[key] for key in keys_to_replace}
6165 )
6167 replaced_statement = statement.replace(
6168 "__EXECMANY_TOKEN__",
6169 ", ".join(replaced_values_clauses),
6170 )
6172 yield _InsertManyValuesBatch(
6173 replaced_statement,
6174 replaced_parameters,
6175 processed_setinputsizes,
6176 batch,
6177 (
6178 [_sentinel_from_params(cb) for cb in compiled_batch]
6179 if _sentinel_from_params
6180 else []
6181 ),
6182 current_batch_size,
6183 batchnum,
6184 total_batches,
6185 sort_by_parameter_order,
6186 False,
6187 )
6188 batchnum += 1
6190 def visit_insert(
6191 self, insert_stmt, visited_bindparam=None, visiting_cte=None, **kw
6192 ):
6193 compile_state = insert_stmt._compile_state_factory(
6194 insert_stmt, self, **kw
6195 )
6196 insert_stmt = compile_state.statement
6198 if visiting_cte is not None:
6199 kw["visiting_cte"] = visiting_cte
6200 toplevel = False
6201 else:
6202 toplevel = not self.stack
6204 if toplevel:
6205 self.isinsert = True
6206 if not self.dml_compile_state:
6207 self.dml_compile_state = compile_state
6208 if not self.compile_state:
6209 self.compile_state = compile_state
6211 self.stack.append(
6212 {
6213 "correlate_froms": set(),
6214 "asfrom_froms": set(),
6215 "selectable": insert_stmt,
6216 }
6217 )
6219 counted_bindparam = 0
6221 # reset any incoming "visited_bindparam" collection
6222 visited_bindparam = None
6224 # for positional, insertmanyvalues needs to know how many
6225 # bound parameters are in the VALUES sequence; there's no simple
6226 # rule because default expressions etc. can have zero or more
6227 # params inside them. After multiple attempts to figure this out,
6228 # this very simplistic "count after" works and is
6229 # likely the least amount of callcounts, though looks clumsy
6230 if self.positional and visiting_cte is None:
6231 # if we are inside a CTE, don't count parameters
6232 # here since they won't be for insertmanyvalues. keep
6233 # visited_bindparam at None so no counting happens.
6234 # see #9173
6235 visited_bindparam = []
6237 crud_params_struct = crud._get_crud_params(
6238 self,
6239 insert_stmt,
6240 compile_state,
6241 toplevel,
6242 visited_bindparam=visited_bindparam,
6243 **kw,
6244 )
6246 if self.positional and visited_bindparam is not None:
6247 counted_bindparam = len(visited_bindparam)
6248 if self._numeric_binds:
6249 if self._values_bindparam is not None:
6250 self._values_bindparam += visited_bindparam
6251 else:
6252 self._values_bindparam = visited_bindparam
6254 crud_params_single = crud_params_struct.single_params
6256 if (
6257 not crud_params_single
6258 and not self.dialect.supports_default_values
6259 and not self.dialect.supports_default_metavalue
6260 and not self.dialect.supports_empty_insert
6261 ):
6262 raise exc.CompileError(
6263 "The '%s' dialect with current database "
6264 "version settings does not support empty "
6265 "inserts." % self.dialect.name
6266 )
6268 if compile_state._has_multi_parameters:
6269 if not self.dialect.supports_multivalues_insert:
6270 raise exc.CompileError(
6271 "The '%s' dialect with current database "
6272 "version settings does not support "
6273 "in-place multirow inserts." % self.dialect.name
6274 )
6275 elif (
6276 self.implicit_returning or insert_stmt._returning
6277 ) and insert_stmt._sort_by_parameter_order:
6278 raise exc.CompileError(
6279 "RETURNING cannot be deterministically sorted when "
6280 "using an INSERT which includes multi-row values()."
6281 )
6282 crud_params_single = crud_params_struct.single_params
6283 else:
6284 crud_params_single = crud_params_struct.single_params
6286 preparer = self.preparer
6287 supports_default_values = self.dialect.supports_default_values
6289 text = "INSERT "
6291 if insert_stmt._prefixes:
6292 text += self._generate_prefixes(
6293 insert_stmt, insert_stmt._prefixes, **kw
6294 )
6296 text += "INTO "
6297 table_text = preparer.format_table(insert_stmt.table)
6299 if insert_stmt._hints:
6300 _, table_text = self._setup_crud_hints(insert_stmt, table_text)
6302 if insert_stmt._independent_ctes:
6303 self._dispatch_independent_ctes(insert_stmt, kw)
6305 text += table_text
6307 if crud_params_single or not supports_default_values:
6308 text += " (%s)" % ", ".join(
6309 [expr for _, expr, _, _ in crud_params_single]
6310 )
6312 # look for insertmanyvalues attributes that would have been configured
6313 # by crud.py as it scanned through the columns to be part of the
6314 # INSERT
6315 use_insertmanyvalues = crud_params_struct.use_insertmanyvalues
6316 named_sentinel_params: Optional[Sequence[str]] = None
6317 add_sentinel_cols = None
6318 implicit_sentinel = False
6320 returning_cols = self.implicit_returning or insert_stmt._returning
6321 if returning_cols:
6322 add_sentinel_cols = crud_params_struct.use_sentinel_columns
6323 if add_sentinel_cols is not None:
6324 assert use_insertmanyvalues
6326 # search for the sentinel column explicitly present
6327 # in the INSERT columns list, and additionally check that
6328 # this column has a bound parameter name set up that's in the
6329 # parameter list. If both of these cases are present, it means
6330 # we will have a client side value for the sentinel in each
6331 # parameter set.
6333 _params_by_col = {
6334 col: param_names
6335 for col, _, _, param_names in crud_params_single
6336 }
6337 named_sentinel_params = []
6338 for _add_sentinel_col in add_sentinel_cols:
6339 if _add_sentinel_col not in _params_by_col:
6340 named_sentinel_params = None
6341 break
6342 param_name = self._within_exec_param_key_getter(
6343 _add_sentinel_col
6344 )
6345 if param_name not in _params_by_col[_add_sentinel_col]:
6346 named_sentinel_params = None
6347 break
6348 named_sentinel_params.append(param_name)
6350 if named_sentinel_params is None:
6351 # if we are not going to have a client side value for
6352 # the sentinel in the parameter set, that means it's
6353 # an autoincrement, an IDENTITY, or a server-side SQL
6354 # expression like nextval('seqname'). So this is
6355 # an "implicit" sentinel; we will look for it in
6356 # RETURNING
6357 # only, and then sort on it. For this case on PG,
6358 # SQL Server we have to use a special INSERT form
6359 # that guarantees the server side function lines up with
6360 # the entries in the VALUES.
6361 if (
6362 self.dialect.insertmanyvalues_implicit_sentinel
6363 & InsertmanyvaluesSentinelOpts.ANY_AUTOINCREMENT
6364 ):
6365 implicit_sentinel = True
6366 else:
6367 # here, we are not using a sentinel at all
6368 # and we are likely the SQLite dialect.
6369 # The first add_sentinel_col that we have should not
6370 # be marked as "insert_sentinel=True". if it was,
6371 # an error should have been raised in
6372 # _get_sentinel_column_for_table.
6373 assert not add_sentinel_cols[0]._insert_sentinel, (
6374 "sentinel selection rules should have prevented "
6375 "us from getting here for this dialect"
6376 )
6378 # always put the sentinel columns last. even if they are
6379 # in the returning list already, they will be there twice
6380 # then.
6381 returning_cols = list(returning_cols) + list(add_sentinel_cols)
6383 returning_clause = self.returning_clause(
6384 insert_stmt,
6385 returning_cols,
6386 populate_result_map=toplevel,
6387 )
6389 if self.returning_precedes_values:
6390 text += " " + returning_clause
6392 else:
6393 returning_clause = None
6395 if insert_stmt.select is not None:
6396 # placed here by crud.py
6397 select_text = self.process(
6398 self.stack[-1]["insert_from_select"], insert_into=True, **kw
6399 )
6401 if self.ctes and self.dialect.cte_follows_insert:
6402 nesting_level = len(self.stack) if not toplevel else None
6403 text += " %s%s" % (
6404 self._render_cte_clause(
6405 nesting_level=nesting_level,
6406 include_following_stack=True,
6407 ),
6408 select_text,
6409 )
6410 else:
6411 text += " %s" % select_text
6412 elif not crud_params_single and supports_default_values:
6413 text += " DEFAULT VALUES"
6414 if use_insertmanyvalues:
6415 self._insertmanyvalues = _InsertManyValues(
6416 True,
6417 self.dialect.default_metavalue_token,
6418 crud_params_single,
6419 counted_bindparam,
6420 sort_by_parameter_order=(
6421 insert_stmt._sort_by_parameter_order
6422 ),
6423 includes_upsert_behaviors=(
6424 insert_stmt._post_values_clause is not None
6425 ),
6426 sentinel_columns=add_sentinel_cols,
6427 num_sentinel_columns=(
6428 len(add_sentinel_cols) if add_sentinel_cols else 0
6429 ),
6430 implicit_sentinel=implicit_sentinel,
6431 )
6432 elif compile_state._has_multi_parameters:
6433 text += " VALUES %s" % (
6434 ", ".join(
6435 "(%s)"
6436 % (", ".join(value for _, _, value, _ in crud_param_set))
6437 for crud_param_set in crud_params_struct.all_multi_params
6438 ),
6439 )
6440 elif use_insertmanyvalues:
6441 if (
6442 implicit_sentinel
6443 and (
6444 self.dialect.insertmanyvalues_implicit_sentinel
6445 & InsertmanyvaluesSentinelOpts.USE_INSERT_FROM_SELECT
6446 )
6447 # this is checking if we have
6448 # INSERT INTO table (id) VALUES (DEFAULT).
6449 and not (crud_params_struct.is_default_metavalue_only)
6450 ):
6451 # if we have a sentinel column that is server generated,
6452 # then for selected backends render the VALUES list as a
6453 # subquery. This is the orderable form supported by
6454 # PostgreSQL and in fewer cases SQL Server
6455 embed_sentinel_value = True
6457 render_bind_casts = (
6458 self.dialect.insertmanyvalues_implicit_sentinel
6459 & InsertmanyvaluesSentinelOpts.RENDER_SELECT_COL_CASTS
6460 )
6462 add_sentinel_set = add_sentinel_cols or ()
6464 insert_single_values_expr = ", ".join(
6465 [
6466 value
6467 for col, _, value, _ in crud_params_single
6468 if col not in add_sentinel_set
6469 ]
6470 )
6472 colnames = ", ".join(
6473 f"p{i}"
6474 for i, cp in enumerate(crud_params_single)
6475 if cp[0] not in add_sentinel_set
6476 )
6478 if render_bind_casts:
6479 # render casts for the SELECT list. For PG, we are
6480 # already rendering bind casts in the parameter list,
6481 # selectively for the more "tricky" types like ARRAY.
6482 # however, even for the "easy" types, if the parameter
6483 # is NULL for every entry, PG gives up and says
6484 # "it must be TEXT", which fails for other easy types
6485 # like ints. So we cast on this side too.
6486 colnames_w_cast = ", ".join(
6487 (
6488 self.render_bind_cast(
6489 col.type,
6490 col.type._unwrapped_dialect_impl(self.dialect),
6491 f"p{i}",
6492 )
6493 if col not in add_sentinel_set
6494 else expr
6495 )
6496 for i, (col, _, expr, _) in enumerate(
6497 crud_params_single
6498 )
6499 )
6500 else:
6501 colnames_w_cast = ", ".join(
6502 (f"p{i}" if col not in add_sentinel_set else expr)
6503 for i, (col, _, expr, _) in enumerate(
6504 crud_params_single
6505 )
6506 )
6508 insert_crud_params = [
6509 elem
6510 for elem in crud_params_single
6511 if elem[0] not in add_sentinel_set
6512 ]
6514 text += (
6515 f" SELECT {colnames_w_cast} FROM "
6516 f"(VALUES ({insert_single_values_expr})) "
6517 f"AS imp_sen({colnames}, sen_counter) "
6518 "ORDER BY sen_counter"
6519 )
6521 else:
6522 # otherwise, if no sentinel or backend doesn't support
6523 # orderable subquery form, use a plain VALUES list
6524 embed_sentinel_value = False
6525 insert_crud_params = crud_params_single
6526 insert_single_values_expr = ", ".join(
6527 [value for _, _, value, _ in crud_params_single]
6528 )
6530 text += f" VALUES ({insert_single_values_expr})"
6532 self._insertmanyvalues = _InsertManyValues(
6533 is_default_expr=False,
6534 single_values_expr=insert_single_values_expr,
6535 insert_crud_params=insert_crud_params,
6536 num_positional_params_counted=counted_bindparam,
6537 sort_by_parameter_order=(insert_stmt._sort_by_parameter_order),
6538 includes_upsert_behaviors=(
6539 insert_stmt._post_values_clause is not None
6540 ),
6541 sentinel_columns=add_sentinel_cols,
6542 num_sentinel_columns=(
6543 len(add_sentinel_cols) if add_sentinel_cols else 0
6544 ),
6545 sentinel_param_keys=named_sentinel_params,
6546 implicit_sentinel=implicit_sentinel,
6547 embed_values_counter=embed_sentinel_value,
6548 )
6550 else:
6551 insert_single_values_expr = ", ".join(
6552 [value for _, _, value, _ in crud_params_single]
6553 )
6555 text += f" VALUES ({insert_single_values_expr})"
6557 if insert_stmt._post_values_clause is not None:
6558 post_values_clause = self.process(
6559 insert_stmt._post_values_clause, **kw
6560 )
6561 if post_values_clause:
6562 text += " " + post_values_clause
6564 if returning_clause and not self.returning_precedes_values:
6565 text += " " + returning_clause
6567 if self.ctes and not self.dialect.cte_follows_insert:
6568 nesting_level = len(self.stack) if not toplevel else None
6569 text = (
6570 self._render_cte_clause(
6571 nesting_level=nesting_level,
6572 include_following_stack=True,
6573 )
6574 + text
6575 )
6577 self.stack.pop(-1)
6579 return text
6581 def update_tables_clause(self, update_stmt, from_table, extra_froms, **kw):
6582 """Provide a hook to override the initial table clause
6583 in an UPDATE statement.
6585 MySQL overrides this.
6587 """
6588 kw["asfrom"] = True
6589 return from_table._compiler_dispatch(self, iscrud=True, **kw)
6591 def update_from_clause(
6592 self, update_stmt, from_table, extra_froms, from_hints, **kw
6593 ):
6594 """Provide a hook to override the generation of an
6595 UPDATE..FROM clause.
6596 MySQL and MSSQL override this.
6597 """
6598 raise NotImplementedError(
6599 "This backend does not support multiple-table "
6600 "criteria within UPDATE"
6601 )
6603 def update_post_criteria_clause(
6604 self, update_stmt: Update, **kw: Any
6605 ) -> Optional[str]:
6606 """provide a hook to override generation after the WHERE criteria
6607 in an UPDATE statement
6609 .. versionadded:: 2.1
6611 """
6612 if update_stmt._post_criteria_clause is not None:
6613 return self.process(
6614 update_stmt._post_criteria_clause,
6615 **kw,
6616 )
6617 else:
6618 return None
6620 def delete_post_criteria_clause(
6621 self, delete_stmt: Delete, **kw: Any
6622 ) -> Optional[str]:
6623 """provide a hook to override generation after the WHERE criteria
6624 in a DELETE statement
6626 .. versionadded:: 2.1
6628 """
6629 if delete_stmt._post_criteria_clause is not None:
6630 return self.process(
6631 delete_stmt._post_criteria_clause,
6632 **kw,
6633 )
6634 else:
6635 return None
6637 def visit_update(
6638 self,
6639 update_stmt: Update,
6640 visiting_cte: Optional[CTE] = None,
6641 **kw: Any,
6642 ) -> str:
6643 compile_state = update_stmt._compile_state_factory(
6644 update_stmt, self, **kw
6645 )
6646 if TYPE_CHECKING:
6647 assert isinstance(compile_state, UpdateDMLState)
6648 update_stmt = compile_state.statement # type: ignore[assignment]
6650 if visiting_cte is not None:
6651 kw["visiting_cte"] = visiting_cte
6652 toplevel = False
6653 else:
6654 toplevel = not self.stack
6656 if toplevel:
6657 self.isupdate = True
6658 if not self.dml_compile_state:
6659 self.dml_compile_state = compile_state
6660 if not self.compile_state:
6661 self.compile_state = compile_state
6663 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
6664 from_linter = FromLinter({}, set())
6665 warn_linting = self.linting & WARN_LINTING
6666 if toplevel:
6667 self.from_linter = from_linter
6668 else:
6669 from_linter = None
6670 warn_linting = False
6672 extra_froms = compile_state._extra_froms
6673 is_multitable = bool(extra_froms)
6675 if is_multitable:
6676 # main table might be a JOIN
6677 main_froms = set(_from_objects(update_stmt.table))
6678 render_extra_froms = [
6679 f for f in extra_froms if f not in main_froms
6680 ]
6681 correlate_froms = main_froms.union(extra_froms)
6682 else:
6683 render_extra_froms = []
6684 correlate_froms = {update_stmt.table}
6686 self.stack.append(
6687 {
6688 "correlate_froms": correlate_froms,
6689 "asfrom_froms": correlate_froms,
6690 "selectable": update_stmt,
6691 }
6692 )
6694 text = "UPDATE "
6696 if update_stmt._prefixes:
6697 text += self._generate_prefixes(
6698 update_stmt, update_stmt._prefixes, **kw
6699 )
6701 table_text = self.update_tables_clause(
6702 update_stmt,
6703 update_stmt.table,
6704 render_extra_froms,
6705 from_linter=from_linter,
6706 **kw,
6707 )
6708 crud_params_struct = crud._get_crud_params(
6709 self, update_stmt, compile_state, toplevel, **kw
6710 )
6711 crud_params = crud_params_struct.single_params
6713 if update_stmt._hints:
6714 dialect_hints, table_text = self._setup_crud_hints(
6715 update_stmt, table_text
6716 )
6717 else:
6718 dialect_hints = None
6720 if update_stmt._independent_ctes:
6721 self._dispatch_independent_ctes(update_stmt, kw)
6723 text += table_text
6725 text += " SET "
6726 text += ", ".join(
6727 expr + "=" + value
6728 for _, expr, value, _ in cast(
6729 "List[Tuple[Any, str, str, Any]]", crud_params
6730 )
6731 )
6733 if self.implicit_returning or update_stmt._returning:
6734 if self.returning_precedes_values:
6735 text += " " + self.returning_clause(
6736 update_stmt,
6737 self.implicit_returning or update_stmt._returning,
6738 populate_result_map=toplevel,
6739 )
6741 if extra_froms:
6742 extra_from_text = self.update_from_clause(
6743 update_stmt,
6744 update_stmt.table,
6745 render_extra_froms,
6746 dialect_hints,
6747 from_linter=from_linter,
6748 **kw,
6749 )
6750 if extra_from_text:
6751 text += " " + extra_from_text
6753 if update_stmt._where_criteria:
6754 t = self._generate_delimited_and_list(
6755 update_stmt._where_criteria, from_linter=from_linter, **kw
6756 )
6757 if t:
6758 text += " WHERE " + t
6760 ulc = self.update_post_criteria_clause(
6761 update_stmt, from_linter=from_linter, **kw
6762 )
6763 if ulc:
6764 text += " " + ulc
6766 if (
6767 self.implicit_returning or update_stmt._returning
6768 ) and not self.returning_precedes_values:
6769 text += " " + self.returning_clause(
6770 update_stmt,
6771 self.implicit_returning or update_stmt._returning,
6772 populate_result_map=toplevel,
6773 )
6775 if self.ctes:
6776 nesting_level = len(self.stack) if not toplevel else None
6777 text = self._render_cte_clause(nesting_level=nesting_level) + text
6779 if warn_linting:
6780 assert from_linter is not None
6781 from_linter.warn(stmt_type="UPDATE")
6783 self.stack.pop(-1)
6785 return text # type: ignore[no-any-return]
6787 def delete_extra_from_clause(
6788 self, delete_stmt, from_table, extra_froms, from_hints, **kw
6789 ):
6790 """Provide a hook to override the generation of an
6791 DELETE..FROM clause.
6793 This can be used to implement DELETE..USING for example.
6795 MySQL and MSSQL override this.
6797 """
6798 raise NotImplementedError(
6799 "This backend does not support multiple-table "
6800 "criteria within DELETE"
6801 )
6803 def delete_table_clause(self, delete_stmt, from_table, extra_froms, **kw):
6804 return from_table._compiler_dispatch(
6805 self, asfrom=True, iscrud=True, **kw
6806 )
6808 def visit_delete(self, delete_stmt, visiting_cte=None, **kw):
6809 compile_state = delete_stmt._compile_state_factory(
6810 delete_stmt, self, **kw
6811 )
6812 delete_stmt = compile_state.statement
6814 if visiting_cte is not None:
6815 kw["visiting_cte"] = visiting_cte
6816 toplevel = False
6817 else:
6818 toplevel = not self.stack
6820 if toplevel:
6821 self.isdelete = True
6822 if not self.dml_compile_state:
6823 self.dml_compile_state = compile_state
6824 if not self.compile_state:
6825 self.compile_state = compile_state
6827 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
6828 from_linter = FromLinter({}, set())
6829 warn_linting = self.linting & WARN_LINTING
6830 if toplevel:
6831 self.from_linter = from_linter
6832 else:
6833 from_linter = None
6834 warn_linting = False
6836 extra_froms = compile_state._extra_froms
6838 correlate_froms = {delete_stmt.table}.union(extra_froms)
6839 self.stack.append(
6840 {
6841 "correlate_froms": correlate_froms,
6842 "asfrom_froms": correlate_froms,
6843 "selectable": delete_stmt,
6844 }
6845 )
6847 text = "DELETE "
6849 if delete_stmt._prefixes:
6850 text += self._generate_prefixes(
6851 delete_stmt, delete_stmt._prefixes, **kw
6852 )
6854 text += "FROM "
6856 try:
6857 table_text = self.delete_table_clause(
6858 delete_stmt,
6859 delete_stmt.table,
6860 extra_froms,
6861 from_linter=from_linter,
6862 )
6863 except TypeError:
6864 # anticipate 3rd party dialects that don't include **kw
6865 # TODO: remove in 2.1
6866 table_text = self.delete_table_clause(
6867 delete_stmt, delete_stmt.table, extra_froms
6868 )
6869 if from_linter:
6870 _ = self.process(delete_stmt.table, from_linter=from_linter)
6872 crud._get_crud_params(self, delete_stmt, compile_state, toplevel, **kw)
6874 if delete_stmt._hints:
6875 dialect_hints, table_text = self._setup_crud_hints(
6876 delete_stmt, table_text
6877 )
6878 else:
6879 dialect_hints = None
6881 if delete_stmt._independent_ctes:
6882 self._dispatch_independent_ctes(delete_stmt, kw)
6884 text += table_text
6886 if (
6887 self.implicit_returning or delete_stmt._returning
6888 ) and self.returning_precedes_values:
6889 text += " " + self.returning_clause(
6890 delete_stmt,
6891 self.implicit_returning or delete_stmt._returning,
6892 populate_result_map=toplevel,
6893 )
6895 if extra_froms:
6896 extra_from_text = self.delete_extra_from_clause(
6897 delete_stmt,
6898 delete_stmt.table,
6899 extra_froms,
6900 dialect_hints,
6901 from_linter=from_linter,
6902 **kw,
6903 )
6904 if extra_from_text:
6905 text += " " + extra_from_text
6907 if delete_stmt._where_criteria:
6908 t = self._generate_delimited_and_list(
6909 delete_stmt._where_criteria, from_linter=from_linter, **kw
6910 )
6911 if t:
6912 text += " WHERE " + t
6914 dlc = self.delete_post_criteria_clause(
6915 delete_stmt, from_linter=from_linter, **kw
6916 )
6917 if dlc:
6918 text += " " + dlc
6920 if (
6921 self.implicit_returning or delete_stmt._returning
6922 ) and not self.returning_precedes_values:
6923 text += " " + self.returning_clause(
6924 delete_stmt,
6925 self.implicit_returning or delete_stmt._returning,
6926 populate_result_map=toplevel,
6927 )
6929 if self.ctes:
6930 nesting_level = len(self.stack) if not toplevel else None
6931 text = self._render_cte_clause(nesting_level=nesting_level) + text
6933 if warn_linting:
6934 assert from_linter is not None
6935 from_linter.warn(stmt_type="DELETE")
6937 self.stack.pop(-1)
6939 return text
6941 def visit_savepoint(self, savepoint_stmt, **kw):
6942 return "SAVEPOINT %s" % self.preparer.format_savepoint(savepoint_stmt)
6944 def visit_rollback_to_savepoint(self, savepoint_stmt, **kw):
6945 return "ROLLBACK TO SAVEPOINT %s" % self.preparer.format_savepoint(
6946 savepoint_stmt
6947 )
6949 def visit_release_savepoint(self, savepoint_stmt, **kw):
6950 return "RELEASE SAVEPOINT %s" % self.preparer.format_savepoint(
6951 savepoint_stmt
6952 )
6955class StrSQLCompiler(SQLCompiler):
6956 """A :class:`.SQLCompiler` subclass which allows a small selection
6957 of non-standard SQL features to render into a string value.
6959 The :class:`.StrSQLCompiler` is invoked whenever a Core expression
6960 element is directly stringified without calling upon the
6961 :meth:`_expression.ClauseElement.compile` method.
6962 It can render a limited set
6963 of non-standard SQL constructs to assist in basic stringification,
6964 however for more substantial custom or dialect-specific SQL constructs,
6965 it will be necessary to make use of
6966 :meth:`_expression.ClauseElement.compile`
6967 directly.
6969 .. seealso::
6971 :ref:`faq_sql_expression_string`
6973 """
6975 def get_select_precolumns(self, select: Select[Any], **kw: Any) -> str:
6976 return "DISTINCT " if select._distinct else ""
6978 def _fallback_column_name(self, column):
6979 return "<name unknown>"
6981 @util.preload_module("sqlalchemy.engine.url")
6982 def visit_unsupported_compilation(self, element, err, **kw):
6983 if element.stringify_dialect != "default":
6984 url = util.preloaded.engine_url
6985 dialect = url.URL.create(element.stringify_dialect).get_dialect()()
6987 compiler = dialect.statement_compiler(
6988 dialect, None, _supporting_against=self
6989 )
6990 if not isinstance(compiler, StrSQLCompiler):
6991 return compiler.process(element, **kw)
6993 return super().visit_unsupported_compilation(element, err)
6995 def visit_getitem_binary(self, binary, operator, **kw):
6996 return "%s[%s]" % (
6997 self.process(binary.left, **kw),
6998 self.process(binary.right, **kw),
6999 )
7001 def visit_json_getitem_op_binary(self, binary, operator, **kw):
7002 return self.visit_getitem_binary(binary, operator, **kw)
7004 def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
7005 return self.visit_getitem_binary(binary, operator, **kw)
7007 def visit_sequence(self, sequence, **kw):
7008 return (
7009 f"<next sequence value: {self.preparer.format_sequence(sequence)}>"
7010 )
7012 def returning_clause(
7013 self,
7014 stmt: UpdateBase,
7015 returning_cols: Sequence[_ColumnsClauseElement],
7016 *,
7017 populate_result_map: bool,
7018 **kw: Any,
7019 ) -> str:
7020 columns = [
7021 self._label_select_column(None, c, True, False, {})
7022 for c in base._select_iterables(returning_cols)
7023 ]
7024 return "RETURNING " + ", ".join(columns)
7026 def update_from_clause(
7027 self, update_stmt, from_table, extra_froms, from_hints, **kw
7028 ):
7029 kw["asfrom"] = True
7030 return "FROM " + ", ".join(
7031 t._compiler_dispatch(self, fromhints=from_hints, **kw)
7032 for t in extra_froms
7033 )
7035 def delete_extra_from_clause(
7036 self, delete_stmt, from_table, extra_froms, from_hints, **kw
7037 ):
7038 kw["asfrom"] = True
7039 return ", " + ", ".join(
7040 t._compiler_dispatch(self, fromhints=from_hints, **kw)
7041 for t in extra_froms
7042 )
7044 def visit_empty_set_expr(self, element_types, **kw):
7045 return "SELECT 1 WHERE 1!=1"
7047 def get_from_hint_text(self, table, text):
7048 return "[%s]" % text
7050 def visit_regexp_match_op_binary(self, binary, operator, **kw):
7051 return self._generate_generic_binary(binary, " <regexp> ", **kw)
7053 def visit_not_regexp_match_op_binary(self, binary, operator, **kw):
7054 return self._generate_generic_binary(binary, " <not regexp> ", **kw)
7056 def visit_regexp_replace_op_binary(self, binary, operator, **kw):
7057 return "<regexp replace>(%s, %s)" % (
7058 binary.left._compiler_dispatch(self, **kw),
7059 binary.right._compiler_dispatch(self, **kw),
7060 )
7062 def visit_try_cast(self, cast, **kwargs):
7063 return "TRY_CAST(%s AS %s)" % (
7064 cast.clause._compiler_dispatch(self, **kwargs),
7065 cast.typeclause._compiler_dispatch(self, **kwargs),
7066 )
7069class DDLCompiler(Compiled):
7070 is_ddl = True
7072 if TYPE_CHECKING:
7074 def __init__(
7075 self,
7076 dialect: Dialect,
7077 statement: ExecutableDDLElement,
7078 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
7079 render_schema_translate: bool = ...,
7080 compile_kwargs: Mapping[str, Any] = ...,
7081 ): ...
7083 @util.ro_memoized_property
7084 def sql_compiler(self) -> SQLCompiler:
7085 return self.dialect.statement_compiler(
7086 self.dialect, None, schema_translate_map=self.schema_translate_map
7087 )
7089 @util.memoized_property
7090 def type_compiler(self):
7091 return self.dialect.type_compiler_instance
7093 def construct_params(
7094 self,
7095 params: Optional[_CoreSingleExecuteParams] = None,
7096 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
7097 escape_names: bool = True,
7098 ) -> Optional[_MutableCoreSingleExecuteParams]:
7099 return None
7101 def visit_ddl(self, ddl, **kwargs):
7102 # table events can substitute table and schema name
7103 context = ddl.context
7104 if isinstance(ddl.target, schema.Table):
7105 context = context.copy()
7107 preparer = self.preparer
7108 path = preparer.format_table_seq(ddl.target)
7109 if len(path) == 1:
7110 table, sch = path[0], ""
7111 else:
7112 table, sch = path[-1], path[0]
7114 context.setdefault("table", table)
7115 context.setdefault("schema", sch)
7116 context.setdefault("fullname", preparer.format_table(ddl.target))
7118 return self.sql_compiler.post_process_text(ddl.statement % context)
7120 def visit_create_schema(self, create, **kw):
7121 text = "CREATE SCHEMA "
7122 if create.if_not_exists:
7123 text += "IF NOT EXISTS "
7124 return text + self.preparer.format_schema(create.element)
7126 def visit_drop_schema(self, drop, **kw):
7127 text = "DROP SCHEMA "
7128 if drop.if_exists:
7129 text += "IF EXISTS "
7130 text += self.preparer.format_schema(drop.element)
7131 if drop.cascade:
7132 text += " CASCADE"
7133 return text
7135 def visit_create_table(self, create, **kw):
7136 table = create.element
7137 preparer = self.preparer
7139 text = "\nCREATE "
7140 if table._prefixes:
7141 text += " ".join(table._prefixes) + " "
7143 text += "TABLE "
7144 if create.if_not_exists:
7145 text += "IF NOT EXISTS "
7147 text += preparer.format_table(table) + " "
7149 create_table_suffix = self.create_table_suffix(table)
7150 if create_table_suffix:
7151 text += create_table_suffix + " "
7153 text += "("
7155 separator = "\n"
7157 # if only one primary key, specify it along with the column
7158 first_pk = False
7159 for create_column in create.columns:
7160 column = create_column.element
7161 try:
7162 processed = self.process(
7163 create_column, first_pk=column.primary_key and not first_pk
7164 )
7165 if processed is not None:
7166 text += separator
7167 separator = ", \n"
7168 text += "\t" + processed
7169 if column.primary_key:
7170 first_pk = True
7171 except exc.CompileError as ce:
7172 raise exc.CompileError(
7173 "(in table '%s', column '%s'): %s"
7174 % (table.description, column.name, ce.args[0])
7175 ) from ce
7177 const = self.create_table_constraints(
7178 table,
7179 _include_foreign_key_constraints=create.include_foreign_key_constraints, # noqa
7180 )
7181 if const:
7182 text += separator + "\t" + const
7184 text += "\n)%s\n\n" % self.post_create_table(table)
7185 return text
7187 def visit_create_view(self, element: CreateView, **kw: Any) -> str:
7188 return self._generate_table_select(element, "view", **kw)
7190 def visit_create_table_as(self, element: CreateTableAs, **kw: Any) -> str:
7191 return self._generate_table_select(element, "create_table_as", **kw)
7193 def create_table_select_suffixes(
7194 self,
7195 element: _TableViaSelect,
7196 type_: str,
7197 **kw: Any,
7198 ) -> str:
7199 return ""
7201 def _generate_table_select(
7202 self,
7203 element: _TableViaSelect,
7204 type_: str,
7205 if_not_exists: Optional[bool] = None,
7206 **kw: Any,
7207 ) -> str:
7208 prep = self.preparer
7210 inner_kw = dict(kw)
7211 inner_kw["literal_binds"] = True
7212 select_sql = self.sql_compiler.process(element.selectable, **inner_kw)
7214 # Use if_not_exists parameter if provided, otherwise use element's
7215 use_if_not_exists = (
7216 if_not_exists
7217 if if_not_exists is not None
7218 else element.if_not_exists
7219 )
7221 parts: List[Optional[str]] = [
7222 "CREATE",
7223 "OR REPLACE" if getattr(element, "or_replace", False) else None,
7224 "TEMPORARY" if element.temporary else None,
7225 (
7226 "MATERIALIZED VIEW"
7227 if type_ == "view" and getattr(element, "materialized", False)
7228 else "TABLE" if type_ == "create_table_as" else "VIEW"
7229 ),
7230 "IF NOT EXISTS" if use_if_not_exists else None,
7231 prep.format_table(element.table),
7232 ]
7233 suffixes = self.create_table_select_suffixes(element, type_, **kw)
7234 if suffixes:
7235 parts.append(suffixes)
7236 parts += ["AS", select_sql]
7237 return " ".join(p for p in parts if p)
7239 def visit_create_column(self, create, first_pk=False, **kw):
7240 column = create.element
7242 if column.system:
7243 return None
7245 text = self.get_column_specification(column, first_pk=first_pk)
7246 const = " ".join(
7247 self.process(constraint) for constraint in column.constraints
7248 )
7249 if const:
7250 text += " " + const
7252 return text
7254 def create_table_constraints(
7255 self, table, _include_foreign_key_constraints=None, **kw
7256 ):
7257 # On some DB order is significant: visit PK first, then the
7258 # other constraints (engine.ReflectionTest.testbasic failed on FB2)
7259 constraints = []
7260 if table.primary_key:
7261 constraints.append(table.primary_key)
7263 all_fkcs = table.foreign_key_constraints
7264 if _include_foreign_key_constraints is not None:
7265 omit_fkcs = all_fkcs.difference(_include_foreign_key_constraints)
7266 else:
7267 omit_fkcs = set()
7269 constraints.extend(
7270 [
7271 c
7272 for c in table._sorted_constraints
7273 if c is not table.primary_key and c not in omit_fkcs
7274 ]
7275 )
7277 return ", \n\t".join(
7278 p
7279 for p in (
7280 self.process(constraint)
7281 for constraint in constraints
7282 if (constraint._should_create_for_compiler(self))
7283 and (
7284 not self.dialect.supports_alter
7285 or not getattr(constraint, "use_alter", False)
7286 )
7287 )
7288 if p is not None
7289 )
7291 def visit_drop_table(self, drop, **kw):
7292 text = "\nDROP TABLE "
7293 if drop.if_exists:
7294 text += "IF EXISTS "
7295 return text + self.preparer.format_table(drop.element)
7297 def visit_drop_view(self, drop, **kw):
7298 text = "\nDROP "
7299 if drop.materialized:
7300 text += "MATERIALIZED VIEW "
7301 else:
7302 text += "VIEW "
7303 if drop.if_exists:
7304 text += "IF EXISTS "
7305 return text + self.preparer.format_table(drop.element)
7307 def _verify_index_table(self, index: Index) -> None:
7308 if index.table is None:
7309 raise exc.CompileError(
7310 "Index '%s' is not associated with any table." % index.name
7311 )
7313 def visit_create_index(
7314 self, create, include_schema=False, include_table_schema=True, **kw
7315 ):
7316 index = create.element
7317 self._verify_index_table(index)
7318 preparer = self.preparer
7319 text = "CREATE "
7320 if index.unique:
7321 text += "UNIQUE "
7322 if index.name is None:
7323 raise exc.CompileError(
7324 "CREATE INDEX requires that the index have a name"
7325 )
7327 text += "INDEX "
7328 if create.if_not_exists:
7329 text += "IF NOT EXISTS "
7331 text += "%s ON %s (%s)" % (
7332 self._prepared_index_name(index, include_schema=include_schema),
7333 preparer.format_table(
7334 index.table, use_schema=include_table_schema
7335 ),
7336 ", ".join(
7337 self.sql_compiler.process(
7338 expr, include_table=False, literal_binds=True
7339 )
7340 for expr in index.expressions
7341 ),
7342 )
7343 return text
7345 def visit_drop_index(self, drop, **kw):
7346 index = drop.element
7348 if index.name is None:
7349 raise exc.CompileError(
7350 "DROP INDEX requires that the index have a name"
7351 )
7352 text = "\nDROP INDEX "
7353 if drop.if_exists:
7354 text += "IF EXISTS "
7356 return text + self._prepared_index_name(index, include_schema=True)
7358 def _prepared_index_name(
7359 self, index: Index, include_schema: bool = False
7360 ) -> str:
7361 if index.table is not None:
7362 effective_schema = self.preparer.schema_for_object(index.table)
7363 else:
7364 effective_schema = None
7365 if include_schema and effective_schema:
7366 schema_name = self.preparer.quote_schema(effective_schema)
7367 else:
7368 schema_name = None
7370 index_name: str = self.preparer.format_index(index)
7372 if schema_name:
7373 index_name = schema_name + "." + index_name
7374 return index_name
7376 def visit_add_constraint(self, create, **kw):
7377 return "ALTER TABLE %s ADD %s" % (
7378 self.preparer.format_table(create.element.table),
7379 self.process(create.element),
7380 )
7382 def visit_set_table_comment(self, create, **kw):
7383 return "COMMENT ON TABLE %s IS %s" % (
7384 self.preparer.format_table(create.element),
7385 self.sql_compiler.render_literal_value(
7386 create.element.comment, sqltypes.String()
7387 ),
7388 )
7390 def visit_drop_table_comment(self, drop, **kw):
7391 return "COMMENT ON TABLE %s IS NULL" % self.preparer.format_table(
7392 drop.element
7393 )
7395 def visit_set_column_comment(self, create, **kw):
7396 return "COMMENT ON COLUMN %s IS %s" % (
7397 self.preparer.format_column(
7398 create.element, use_table=True, use_schema=True
7399 ),
7400 self.sql_compiler.render_literal_value(
7401 create.element.comment, sqltypes.String()
7402 ),
7403 )
7405 def visit_drop_column_comment(self, drop, **kw):
7406 return "COMMENT ON COLUMN %s IS NULL" % self.preparer.format_column(
7407 drop.element, use_table=True
7408 )
7410 def visit_set_constraint_comment(self, create, **kw):
7411 raise exc.UnsupportedCompilationError(self, type(create))
7413 def visit_drop_constraint_comment(self, drop, **kw):
7414 raise exc.UnsupportedCompilationError(self, type(drop))
7416 def get_identity_options(self, identity_options: IdentityOptions) -> str:
7417 text = []
7418 if identity_options.increment is not None:
7419 text.append("INCREMENT BY %d" % identity_options.increment)
7420 if identity_options.start is not None:
7421 text.append("START WITH %d" % identity_options.start)
7422 if identity_options.minvalue is not None:
7423 text.append("MINVALUE %d" % identity_options.minvalue)
7424 if identity_options.maxvalue is not None:
7425 text.append("MAXVALUE %d" % identity_options.maxvalue)
7426 if identity_options.nominvalue is not None:
7427 text.append("NO MINVALUE")
7428 if identity_options.nomaxvalue is not None:
7429 text.append("NO MAXVALUE")
7430 if identity_options.cache is not None:
7431 text.append("CACHE %d" % identity_options.cache)
7432 if identity_options.cycle is not None:
7433 text.append("CYCLE" if identity_options.cycle else "NO CYCLE")
7434 return " ".join(text)
7436 def visit_create_sequence(self, create, prefix=None, **kw):
7437 text = "CREATE SEQUENCE "
7438 if create.if_not_exists:
7439 text += "IF NOT EXISTS "
7440 text += self.preparer.format_sequence(create.element)
7442 if prefix:
7443 text += prefix
7444 options = self.get_identity_options(create.element)
7445 if options:
7446 text += " " + options
7447 return text
7449 def visit_drop_sequence(self, drop, **kw):
7450 text = "DROP SEQUENCE "
7451 if drop.if_exists:
7452 text += "IF EXISTS "
7453 return text + self.preparer.format_sequence(drop.element)
7455 def visit_drop_constraint(self, drop, **kw):
7456 constraint = drop.element
7457 if constraint.name is not None:
7458 formatted_name = self.preparer.format_constraint(constraint)
7459 else:
7460 formatted_name = None
7462 if formatted_name is None:
7463 raise exc.CompileError(
7464 "Can't emit DROP CONSTRAINT for constraint %r; "
7465 "it has no name" % drop.element
7466 )
7467 return "ALTER TABLE %s DROP CONSTRAINT %s%s%s" % (
7468 self.preparer.format_table(drop.element.table),
7469 "IF EXISTS " if drop.if_exists else "",
7470 formatted_name,
7471 " CASCADE" if drop.cascade else "",
7472 )
7474 def get_column_specification(
7475 self, column: Column[Any], **kwargs: Any
7476 ) -> str:
7477 colspec = (
7478 self.preparer.format_column(column)
7479 + " "
7480 + self.dialect.type_compiler_instance.process(
7481 column.type, type_expression=column
7482 )
7483 )
7484 default = self.get_column_default_string(column)
7485 if default is not None:
7486 colspec += " DEFAULT " + default
7488 if column.computed is not None:
7489 colspec += " " + self.process(column.computed)
7491 if (
7492 column.identity is not None
7493 and self.dialect.supports_identity_columns
7494 ):
7495 colspec += " " + self.process(column.identity)
7497 if not column.nullable and (
7498 not column.identity or not self.dialect.supports_identity_columns
7499 ):
7500 colspec += " NOT NULL"
7501 return colspec
7503 def create_table_suffix(self, table: Table) -> str:
7504 return ""
7506 def post_create_table(self, table: Table) -> str:
7507 return ""
7509 def get_column_default_string(self, column: Column[Any]) -> Optional[str]:
7510 if isinstance(column.server_default, schema.DefaultClause):
7511 return self.render_default_string(column.server_default.arg)
7512 else:
7513 return None
7515 def render_default_string(self, default: Union[Visitable, str]) -> str:
7516 if isinstance(default, str):
7517 return self.sql_compiler.render_literal_value(
7518 default, sqltypes.STRINGTYPE
7519 )
7520 else:
7521 return self.sql_compiler.process(default, literal_binds=True)
7523 def visit_table_or_column_check_constraint(self, constraint, **kw):
7524 if constraint.is_column_level:
7525 return self.visit_column_check_constraint(constraint)
7526 else:
7527 return self.visit_check_constraint(constraint)
7529 def visit_check_constraint(self, constraint, **kw):
7530 text = self.define_constraint_preamble(constraint, **kw)
7531 text += self.define_check_body(constraint, **kw)
7532 text += self.define_constraint_deferrability(constraint)
7533 return text
7535 def visit_column_check_constraint(self, constraint, **kw):
7536 text = self.define_constraint_preamble(constraint, **kw)
7537 text += self.define_check_body(constraint, **kw)
7538 text += self.define_constraint_deferrability(constraint)
7539 return text
7541 def visit_primary_key_constraint(
7542 self, constraint: PrimaryKeyConstraint, **kw: Any
7543 ) -> str:
7544 if len(constraint) == 0:
7545 return ""
7546 text = self.define_constraint_preamble(constraint, **kw)
7547 text += self.define_primary_key_body(constraint, **kw)
7548 text += self.define_constraint_deferrability(constraint)
7549 return text
7551 def visit_foreign_key_constraint(
7552 self, constraint: ForeignKeyConstraint, **kw: Any
7553 ) -> str:
7554 text = self.define_constraint_preamble(constraint, **kw)
7555 text += self.define_foreign_key_body(constraint, **kw)
7556 text += self.define_constraint_match(constraint)
7557 text += self.define_constraint_cascades(constraint)
7558 text += self.define_constraint_deferrability(constraint)
7559 return text
7561 def define_constraint_remote_table(self, constraint, table, preparer):
7562 """Format the remote table clause of a CREATE CONSTRAINT clause."""
7564 return preparer.format_table(table)
7566 def visit_unique_constraint(
7567 self, constraint: UniqueConstraint, **kw: Any
7568 ) -> str:
7569 if len(constraint) == 0:
7570 return ""
7571 text = self.define_constraint_preamble(constraint, **kw)
7572 text += self.define_unique_body(constraint, **kw)
7573 text += self.define_constraint_deferrability(constraint)
7574 return text
7576 def define_constraint_preamble(
7577 self, constraint: Constraint, **kw: Any
7578 ) -> str:
7579 text = ""
7580 if constraint.name is not None:
7581 formatted_name = self.preparer.format_constraint(constraint)
7582 if formatted_name is not None:
7583 text += "CONSTRAINT %s " % formatted_name
7584 return text
7586 def define_primary_key_body(
7587 self, constraint: PrimaryKeyConstraint, **kw: Any
7588 ) -> str:
7589 text = ""
7590 text += "PRIMARY KEY "
7591 text += "(%s)" % ", ".join(
7592 self.preparer.quote(c.name)
7593 for c in (
7594 constraint.columns_autoinc_first
7595 if constraint._implicit_generated
7596 else constraint.columns
7597 )
7598 )
7599 return text
7601 def define_foreign_key_body(
7602 self, constraint: ForeignKeyConstraint, **kw: Any
7603 ) -> str:
7604 preparer = self.preparer
7605 remote_table = list(constraint.elements)[0].column.table
7606 text = "FOREIGN KEY(%s) REFERENCES %s (%s)" % (
7607 ", ".join(
7608 preparer.quote(f.parent.name) for f in constraint.elements
7609 ),
7610 self.define_constraint_remote_table(
7611 constraint, remote_table, preparer
7612 ),
7613 ", ".join(
7614 preparer.quote(f.column.name) for f in constraint.elements
7615 ),
7616 )
7617 return text
7619 def define_unique_body(
7620 self, constraint: UniqueConstraint, **kw: Any
7621 ) -> str:
7622 text = "UNIQUE %s(%s)" % (
7623 self.define_unique_constraint_distinct(constraint, **kw),
7624 ", ".join(self.preparer.quote(c.name) for c in constraint),
7625 )
7626 return text
7628 def define_check_body(self, constraint: CheckConstraint, **kw: Any) -> str:
7629 text = "CHECK (%s)" % self.sql_compiler.process(
7630 constraint.sqltext, include_table=False, literal_binds=True
7631 )
7632 return text
7634 def define_unique_constraint_distinct(
7635 self, constraint: UniqueConstraint, **kw: Any
7636 ) -> str:
7637 return ""
7639 def define_constraint_cascades(
7640 self, constraint: ForeignKeyConstraint
7641 ) -> str:
7642 text = ""
7643 if constraint.ondelete is not None:
7644 text += self.define_constraint_ondelete_cascade(constraint)
7646 if constraint.onupdate is not None:
7647 text += self.define_constraint_onupdate_cascade(constraint)
7648 return text
7650 def define_constraint_ondelete_cascade(
7651 self, constraint: ForeignKeyConstraint
7652 ) -> str:
7653 return " ON DELETE %s" % self.preparer.validate_sql_phrase(
7654 constraint.ondelete, FK_ON_DELETE
7655 )
7657 def define_constraint_onupdate_cascade(
7658 self, constraint: ForeignKeyConstraint
7659 ) -> str:
7660 return " ON UPDATE %s" % self.preparer.validate_sql_phrase(
7661 constraint.onupdate, FK_ON_UPDATE
7662 )
7664 def define_constraint_deferrability(self, constraint: Constraint) -> str:
7665 text = ""
7666 if constraint.deferrable is not None:
7667 if constraint.deferrable:
7668 text += " DEFERRABLE"
7669 else:
7670 text += " NOT DEFERRABLE"
7671 if constraint.initially is not None:
7672 text += " INITIALLY %s" % self.preparer.validate_sql_phrase(
7673 constraint.initially, FK_INITIALLY
7674 )
7675 return text
7677 def define_constraint_match(self, constraint: ForeignKeyConstraint) -> str:
7678 text = ""
7679 if constraint.match is not None:
7680 text += " MATCH %s" % constraint.match
7681 return text
7683 def visit_computed_column(self, generated, **kw):
7684 text = "GENERATED ALWAYS AS (%s)" % self.sql_compiler.process(
7685 generated.sqltext, include_table=False, literal_binds=True
7686 )
7687 if generated.persisted is True:
7688 text += " STORED"
7689 elif generated.persisted is False:
7690 text += " VIRTUAL"
7691 return text
7693 def visit_identity_column(self, identity, **kw):
7694 text = "GENERATED %s AS IDENTITY" % (
7695 "ALWAYS" if identity.always else "BY DEFAULT",
7696 )
7697 options = self.get_identity_options(identity)
7698 if options:
7699 text += " (%s)" % options
7700 return text
7703class GenericTypeCompiler(TypeCompiler):
7704 def visit_FLOAT(self, type_: sqltypes.Float[Any], **kw: Any) -> str:
7705 return "FLOAT"
7707 def visit_DOUBLE(self, type_: sqltypes.Double[Any], **kw: Any) -> str:
7708 return "DOUBLE"
7710 def visit_DOUBLE_PRECISION(
7711 self, type_: sqltypes.DOUBLE_PRECISION[Any], **kw: Any
7712 ) -> str:
7713 return "DOUBLE PRECISION"
7715 def visit_REAL(self, type_: sqltypes.REAL[Any], **kw: Any) -> str:
7716 return "REAL"
7718 def visit_NUMERIC(self, type_: sqltypes.Numeric[Any], **kw: Any) -> str:
7719 if type_.precision is None:
7720 return "NUMERIC"
7721 elif type_.scale is None:
7722 return "NUMERIC(%(precision)s)" % {"precision": type_.precision}
7723 else:
7724 return "NUMERIC(%(precision)s, %(scale)s)" % {
7725 "precision": type_.precision,
7726 "scale": type_.scale,
7727 }
7729 def visit_DECIMAL(self, type_: sqltypes.DECIMAL[Any], **kw: Any) -> str:
7730 if type_.precision is None:
7731 return "DECIMAL"
7732 elif type_.scale is None:
7733 return "DECIMAL(%(precision)s)" % {"precision": type_.precision}
7734 else:
7735 return "DECIMAL(%(precision)s, %(scale)s)" % {
7736 "precision": type_.precision,
7737 "scale": type_.scale,
7738 }
7740 def visit_INTEGER(self, type_: sqltypes.Integer, **kw: Any) -> str:
7741 return "INTEGER"
7743 def visit_SMALLINT(self, type_: sqltypes.SmallInteger, **kw: Any) -> str:
7744 return "SMALLINT"
7746 def visit_BIGINT(self, type_: sqltypes.BigInteger, **kw: Any) -> str:
7747 return "BIGINT"
7749 def visit_TIMESTAMP(self, type_: sqltypes.TIMESTAMP, **kw: Any) -> str:
7750 return "TIMESTAMP"
7752 def visit_DATETIME(self, type_: sqltypes.DateTime, **kw: Any) -> str:
7753 return "DATETIME"
7755 def visit_DATE(self, type_: sqltypes.Date, **kw: Any) -> str:
7756 return "DATE"
7758 def visit_TIME(self, type_: sqltypes.Time, **kw: Any) -> str:
7759 return "TIME"
7761 def visit_CLOB(self, type_: sqltypes.CLOB, **kw: Any) -> str:
7762 return "CLOB"
7764 def visit_NCLOB(self, type_: sqltypes.Text, **kw: Any) -> str:
7765 return "NCLOB"
7767 def _render_string_type(
7768 self,
7769 name: str,
7770 length: Optional[int],
7771 collation: Optional[str],
7772 collation_schema: Optional[str] = None,
7773 identifier_preparer: Optional[IdentifierPreparer] = None,
7774 **kw: Any,
7775 ) -> str:
7776 text = name
7777 if length:
7778 text += f"({length})"
7779 if collation:
7780 if identifier_preparer is None:
7781 identifier_preparer = self.dialect.identifier_preparer
7782 text += " COLLATE " + identifier_preparer.format_collation(
7783 collation, collation_schema
7784 )
7785 return text
7787 def visit_CHAR(self, type_: sqltypes.CHAR, **kw: Any) -> str:
7788 return self._render_string_type(
7789 "CHAR",
7790 type_.length,
7791 type_.collation,
7792 type_.collation_schema,
7793 **kw,
7794 )
7796 def visit_NCHAR(self, type_: sqltypes.NCHAR, **kw: Any) -> str:
7797 return self._render_string_type(
7798 "NCHAR",
7799 type_.length,
7800 type_.collation,
7801 type_.collation_schema,
7802 **kw,
7803 )
7805 def visit_VARCHAR(self, type_: sqltypes.String, **kw: Any) -> str:
7806 return self._render_string_type(
7807 "VARCHAR",
7808 type_.length,
7809 type_.collation,
7810 type_.collation_schema,
7811 **kw,
7812 )
7814 def visit_NVARCHAR(self, type_: sqltypes.NVARCHAR, **kw: Any) -> str:
7815 return self._render_string_type(
7816 "NVARCHAR",
7817 type_.length,
7818 type_.collation,
7819 type_.collation_schema,
7820 **kw,
7821 )
7823 def visit_TEXT(self, type_: sqltypes.Text, **kw: Any) -> str:
7824 return self._render_string_type(
7825 "TEXT",
7826 type_.length,
7827 type_.collation,
7828 type_.collation_schema,
7829 **kw,
7830 )
7832 def visit_UUID(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
7833 return "UUID"
7835 def visit_BLOB(self, type_: sqltypes.LargeBinary, **kw: Any) -> str:
7836 return "BLOB"
7838 def visit_BINARY(self, type_: sqltypes.BINARY, **kw: Any) -> str:
7839 return "BINARY" + (type_.length and "(%d)" % type_.length or "")
7841 def visit_VARBINARY(self, type_: sqltypes.VARBINARY, **kw: Any) -> str:
7842 return "VARBINARY" + (type_.length and "(%d)" % type_.length or "")
7844 def visit_BOOLEAN(self, type_: sqltypes.Boolean, **kw: Any) -> str:
7845 return "BOOLEAN"
7847 def visit_uuid(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
7848 if not type_.native_uuid or not self.dialect.supports_native_uuid:
7849 return self._render_string_type(
7850 "CHAR",
7851 length=32,
7852 collation=None,
7853 collation_schema=None,
7854 **kw,
7855 )
7856 else:
7857 return self.visit_UUID(type_, **kw)
7859 def visit_large_binary(
7860 self, type_: sqltypes.LargeBinary, **kw: Any
7861 ) -> str:
7862 return self.visit_BLOB(type_, **kw)
7864 def visit_boolean(self, type_: sqltypes.Boolean, **kw: Any) -> str:
7865 return self.visit_BOOLEAN(type_, **kw)
7867 def visit_time(self, type_: sqltypes.Time, **kw: Any) -> str:
7868 return self.visit_TIME(type_, **kw)
7870 def visit_datetime(self, type_: sqltypes.DateTime, **kw: Any) -> str:
7871 return self.visit_DATETIME(type_, **kw)
7873 def visit_date(self, type_: sqltypes.Date, **kw: Any) -> str:
7874 return self.visit_DATE(type_, **kw)
7876 def visit_big_integer(self, type_: sqltypes.BigInteger, **kw: Any) -> str:
7877 return self.visit_BIGINT(type_, **kw)
7879 def visit_small_integer(
7880 self, type_: sqltypes.SmallInteger, **kw: Any
7881 ) -> str:
7882 return self.visit_SMALLINT(type_, **kw)
7884 def visit_integer(self, type_: sqltypes.Integer, **kw: Any) -> str:
7885 return self.visit_INTEGER(type_, **kw)
7887 def visit_real(self, type_: sqltypes.REAL[Any], **kw: Any) -> str:
7888 return self.visit_REAL(type_, **kw)
7890 def visit_float(self, type_: sqltypes.Float[Any], **kw: Any) -> str:
7891 return self.visit_FLOAT(type_, **kw)
7893 def visit_double(self, type_: sqltypes.Double[Any], **kw: Any) -> str:
7894 return self.visit_DOUBLE(type_, **kw)
7896 def visit_numeric(self, type_: sqltypes.Numeric[Any], **kw: Any) -> str:
7897 return self.visit_NUMERIC(type_, **kw)
7899 def visit_string(self, type_: sqltypes.String, **kw: Any) -> str:
7900 return self.visit_VARCHAR(type_, **kw)
7902 def visit_unicode(self, type_: sqltypes.Unicode, **kw: Any) -> str:
7903 return self.visit_VARCHAR(type_, **kw)
7905 def visit_text(self, type_: sqltypes.Text, **kw: Any) -> str:
7906 return self.visit_TEXT(type_, **kw)
7908 def visit_unicode_text(
7909 self, type_: sqltypes.UnicodeText, **kw: Any
7910 ) -> str:
7911 return self.visit_TEXT(type_, **kw)
7913 def visit_enum(self, type_: sqltypes.Enum, **kw: Any) -> str:
7914 return self.visit_VARCHAR(type_, **kw)
7916 def visit_null(self, type_, **kw):
7917 raise exc.CompileError(
7918 "Can't generate DDL for %r; "
7919 "did you forget to specify a "
7920 "type on this Column?" % type_
7921 )
7923 def visit_type_decorator(
7924 self, type_: TypeDecorator[Any], **kw: Any
7925 ) -> str:
7926 return self.process(type_.type_engine(self.dialect), **kw)
7928 def visit_user_defined(
7929 self, type_: UserDefinedType[Any], **kw: Any
7930 ) -> str:
7931 return type_.get_col_spec(**kw)
7934class StrSQLTypeCompiler(GenericTypeCompiler):
7935 def process(self, type_, **kw):
7936 try:
7937 _compiler_dispatch = type_._compiler_dispatch
7938 except AttributeError:
7939 return self._visit_unknown(type_, **kw)
7940 else:
7941 return _compiler_dispatch(self, **kw)
7943 def __getattr__(self, key):
7944 if key.startswith("visit_"):
7945 return self._visit_unknown
7946 else:
7947 raise AttributeError(key)
7949 def _visit_unknown(self, type_, **kw):
7950 if type_.__class__.__name__ == type_.__class__.__name__.upper():
7951 return type_.__class__.__name__
7952 else:
7953 return repr(type_)
7955 def visit_null(self, type_, **kw):
7956 return "NULL"
7958 def visit_user_defined(self, type_, **kw):
7959 try:
7960 get_col_spec = type_.get_col_spec
7961 except AttributeError:
7962 return repr(type_)
7963 else:
7964 return get_col_spec(**kw)
7967class _SchemaForObjectCallable(Protocol):
7968 def __call__(self, obj: Any, /) -> str: ...
7971class _BindNameForColProtocol(Protocol):
7972 def __call__(self, col: ColumnClause[Any]) -> str: ...
7975class IdentifierPreparer:
7976 """Handle quoting and case-folding of identifiers based on options."""
7978 reserved_words = RESERVED_WORDS
7980 legal_characters = LEGAL_CHARACTERS
7982 illegal_initial_characters = ILLEGAL_INITIAL_CHARACTERS
7984 initial_quote: str
7986 final_quote: str
7988 _strings: MutableMapping[str, str]
7990 schema_for_object: _SchemaForObjectCallable = operator.attrgetter("schema")
7991 """Return the .schema attribute for an object.
7993 For the default IdentifierPreparer, the schema for an object is always
7994 the value of the ".schema" attribute. if the preparer is replaced
7995 with one that has a non-empty schema_translate_map, the value of the
7996 ".schema" attribute is rendered a symbol that will be converted to a
7997 real schema name from the mapping post-compile.
7999 """
8001 _includes_none_schema_translate: bool = False
8003 def __init__(
8004 self,
8005 dialect: Dialect,
8006 initial_quote: str = '"',
8007 final_quote: Optional[str] = None,
8008 escape_quote: str = '"',
8009 quote_case_sensitive_collations: bool = True,
8010 omit_schema: bool = False,
8011 ):
8012 """Construct a new ``IdentifierPreparer`` object.
8014 initial_quote
8015 Character that begins a delimited identifier.
8017 final_quote
8018 Character that ends a delimited identifier. Defaults to
8019 `initial_quote`.
8021 omit_schema
8022 Prevent prepending schema name. Useful for databases that do
8023 not support schemae.
8024 """
8026 self.dialect = dialect
8027 self.initial_quote = initial_quote
8028 self.final_quote = final_quote or self.initial_quote
8029 self.escape_quote = escape_quote
8030 self.escape_to_quote = self.escape_quote * 2
8031 self.omit_schema = omit_schema
8032 self.quote_case_sensitive_collations = quote_case_sensitive_collations
8033 self._strings = {}
8034 self._double_percents = self.dialect.paramstyle in (
8035 "format",
8036 "pyformat",
8037 )
8039 def _with_schema_translate(self, schema_translate_map):
8040 prep = self.__class__.__new__(self.__class__)
8041 prep.__dict__.update(self.__dict__)
8043 includes_none = None in schema_translate_map
8045 def symbol_getter(obj):
8046 name = obj.schema
8047 if obj._use_schema_map and (name is not None or includes_none):
8048 if name is not None and ("[" in name or "]" in name):
8049 raise exc.CompileError(
8050 "Square bracket characters ([]) not supported "
8051 "in schema translate name '%s'" % name
8052 )
8053 return quoted_name(
8054 "__[SCHEMA_%s]" % (name or "_none"), quote=False
8055 )
8056 else:
8057 return obj.schema
8059 prep.schema_for_object = symbol_getter
8060 prep._includes_none_schema_translate = includes_none
8061 return prep
8063 def _render_schema_translates(
8064 self, statement: str, schema_translate_map: SchemaTranslateMapType
8065 ) -> str:
8066 d = schema_translate_map
8067 if None in d:
8068 if not self._includes_none_schema_translate:
8069 raise exc.InvalidRequestError(
8070 "schema translate map which previously did not have "
8071 "`None` present as a key now has `None` present; compiled "
8072 "statement may lack adequate placeholders. Please use "
8073 "consistent keys in successive "
8074 "schema_translate_map dictionaries."
8075 )
8077 d["_none"] = d[None] # type: ignore[index]
8079 def replace(m):
8080 name = m.group(2)
8081 if name in d:
8082 effective_schema = d[name]
8083 else:
8084 if name in (None, "_none"):
8085 raise exc.InvalidRequestError(
8086 "schema translate map which previously had `None` "
8087 "present as a key now no longer has it present; don't "
8088 "know how to apply schema for compiled statement. "
8089 "Please use consistent keys in successive "
8090 "schema_translate_map dictionaries."
8091 )
8092 effective_schema = name
8094 if not effective_schema:
8095 effective_schema = self.dialect.default_schema_name
8096 if not effective_schema:
8097 # TODO: no coverage here
8098 raise exc.CompileError(
8099 "Dialect has no default schema name; can't "
8100 "use None as dynamic schema target."
8101 )
8102 return self.quote_schema(effective_schema)
8104 return re.sub(r"(__\[SCHEMA_([^\]]+)\])", replace, statement)
8106 def _escape_identifier(self, value: str) -> str:
8107 """Escape an identifier.
8109 Subclasses should override this to provide database-dependent
8110 escaping behavior.
8111 """
8113 value = value.replace(self.escape_quote, self.escape_to_quote)
8114 if self._double_percents:
8115 value = value.replace("%", "%%")
8116 return value
8118 def _unescape_identifier(self, value: str) -> str:
8119 """Canonicalize an escaped identifier.
8121 Subclasses should override this to provide database-dependent
8122 unescaping behavior that reverses _escape_identifier.
8123 """
8125 return value.replace(self.escape_to_quote, self.escape_quote)
8127 def validate_sql_phrase(self, element, reg):
8128 """keyword sequence filter.
8130 a filter for elements that are intended to represent keyword sequences,
8131 such as "INITIALLY", "INITIALLY DEFERRED", etc. no special characters
8132 should be present.
8134 """
8136 if element is not None and not reg.match(element):
8137 raise exc.CompileError(
8138 "Unexpected SQL phrase: %r (matching against %r)"
8139 % (element, reg.pattern)
8140 )
8141 return element
8143 def quote_identifier(self, value: str) -> str:
8144 """Quote an identifier.
8146 Subclasses should override this to provide database-dependent
8147 quoting behavior.
8148 """
8150 return (
8151 self.initial_quote
8152 + self._escape_identifier(value)
8153 + self.final_quote
8154 )
8156 def _requires_quotes(self, value: str) -> bool:
8157 """Return True if the given identifier requires quoting."""
8158 if not value:
8159 # a blank name is legal on SQLite only, where it can be
8160 # delivered by reflection; quote it so that it renders as the
8161 # database has it, rather than indexing into an empty string
8162 # below
8163 return True
8164 lc_value = value.lower()
8165 return (
8166 lc_value in self.reserved_words
8167 or value[0] in self.illegal_initial_characters
8168 or not self.legal_characters.match(str(value))
8169 or (lc_value != value)
8170 )
8172 def _requires_quotes_illegal_chars(self, value):
8173 """Return True if the given identifier requires quoting, but
8174 not taking case convention into account."""
8175 return not self.legal_characters.match(str(value))
8177 def quote_schema(self, schema: str) -> str:
8178 """Conditionally quote a schema name.
8181 The name is quoted if it is a reserved word, contains quote-necessary
8182 characters, or is an instance of :class:`.quoted_name` which includes
8183 ``quote`` set to ``True``.
8185 Subclasses can override this to provide database-dependent
8186 quoting behavior for schema names.
8188 :param schema: string schema name
8189 """
8190 return self.quote(schema)
8192 def quote(self, ident: str) -> str:
8193 """Conditionally quote an identifier.
8195 The identifier is quoted if it is a reserved word, contains
8196 quote-necessary characters, or is an instance of
8197 :class:`.quoted_name` which includes ``quote`` set to ``True``.
8199 Subclasses can override this to provide database-dependent
8200 quoting behavior for identifier names.
8202 :param ident: string identifier
8203 """
8204 force = getattr(ident, "quote", None)
8206 if force is None:
8207 if ident in self._strings:
8208 return self._strings[ident]
8209 else:
8210 if self._requires_quotes(ident):
8211 self._strings[ident] = self.quote_identifier(ident)
8212 else:
8213 self._strings[ident] = ident
8214 return self._strings[ident]
8215 elif force:
8216 return self.quote_identifier(ident)
8217 else:
8218 return ident
8220 def format_collation(self, collation_name, collation_schema=None):
8221 if self.quote_case_sensitive_collations:
8222 name = self.quote(collation_name)
8223 else:
8224 name = collation_name
8226 if collation_schema is not None:
8227 return f"{self.quote_schema(collation_schema)}.{name}"
8228 else:
8229 return name
8231 def format_sequence(
8232 self, sequence: schema.Sequence, use_schema: bool = True
8233 ) -> str:
8234 name = self.quote(sequence.name)
8236 effective_schema = self.schema_for_object(sequence)
8238 if (
8239 not self.omit_schema
8240 and use_schema
8241 and effective_schema is not None
8242 ):
8243 name = self.quote_schema(effective_schema) + "." + name
8244 return name
8246 def format_label(
8247 self, label: Label[Any], name: Optional[str] = None
8248 ) -> str:
8249 return self.quote(name or label.name)
8251 def format_alias(
8252 self, alias: Optional[AliasedReturnsRows], name: Optional[str] = None
8253 ) -> str:
8254 if name is None:
8255 assert alias is not None
8256 return self.quote(alias.name)
8257 else:
8258 return self.quote(name)
8260 def format_savepoint(self, savepoint, name=None):
8261 # Running the savepoint name through quoting is unnecessary
8262 # for all known dialects. This is here to support potential
8263 # third party use cases
8264 ident = name or savepoint.ident
8265 if self._requires_quotes(ident):
8266 ident = self.quote_identifier(ident)
8267 return ident
8269 @util.preload_module("sqlalchemy.sql.naming")
8270 def format_constraint(
8271 self, constraint: Union[Constraint, Index], _alembic_quote: bool = True
8272 ) -> Optional[str]:
8273 naming = util.preloaded.sql_naming
8275 if constraint.name is _NONE_NAME:
8276 name = naming._constraint_name_for_table(
8277 constraint, constraint.table
8278 )
8280 if name is None:
8281 return None
8282 else:
8283 name = constraint.name
8285 assert name is not None
8286 if constraint.__visit_name__ == "index":
8287 return self.truncate_and_render_index_name(
8288 name, _alembic_quote=_alembic_quote
8289 )
8290 else:
8291 return self.truncate_and_render_constraint_name(
8292 name, _alembic_quote=_alembic_quote
8293 )
8295 def truncate_and_render_index_name(
8296 self, name: str, _alembic_quote: bool = True
8297 ) -> str:
8298 # calculate these at format time so that ad-hoc changes
8299 # to dialect.max_identifier_length etc. can be reflected
8300 # as IdentifierPreparer is long lived
8301 max_ = (
8302 self.dialect.max_index_name_length
8303 or self.dialect.max_identifier_length
8304 )
8305 return self._truncate_and_render_maxlen_name(
8306 name, max_, _alembic_quote
8307 )
8309 def truncate_and_render_constraint_name(
8310 self, name: str, _alembic_quote: bool = True
8311 ) -> str:
8312 # calculate these at format time so that ad-hoc changes
8313 # to dialect.max_identifier_length etc. can be reflected
8314 # as IdentifierPreparer is long lived
8315 max_ = (
8316 self.dialect.max_constraint_name_length
8317 or self.dialect.max_identifier_length
8318 )
8319 return self._truncate_and_render_maxlen_name(
8320 name, max_, _alembic_quote
8321 )
8323 def _truncate_and_render_maxlen_name(
8324 self, name: str, max_: int, _alembic_quote: bool
8325 ) -> str:
8326 if isinstance(name, elements._truncated_label):
8327 if len(name) > max_:
8328 name = name[0 : max_ - 8] + "_" + util.md5_hex(name)[-4:]
8329 else:
8330 self.dialect.validate_identifier(name)
8332 if not _alembic_quote:
8333 return name
8334 else:
8335 return self.quote(name)
8337 def format_index(self, index: Index) -> str:
8338 name = self.format_constraint(index)
8339 assert name is not None
8340 return name
8342 def format_table(
8343 self,
8344 table: FromClause,
8345 use_schema: bool = True,
8346 name: Optional[str] = None,
8347 ) -> str:
8348 """Prepare a quoted table and schema name."""
8349 if name is None:
8350 if TYPE_CHECKING:
8351 assert isinstance(table, NamedFromClause)
8352 name = table.name
8354 result = self.quote(name)
8356 effective_schema = self.schema_for_object(table)
8358 if not self.omit_schema and use_schema and effective_schema:
8359 result = self.quote_schema(effective_schema) + "." + result
8360 return result
8362 def format_schema(self, name):
8363 """Prepare a quoted schema name."""
8365 return self.quote(name)
8367 def format_label_name(
8368 self,
8369 name,
8370 anon_map=None,
8371 ):
8372 """Prepare a quoted column name."""
8374 if anon_map is not None and isinstance(
8375 name, elements._truncated_label
8376 ):
8377 name = name.apply_map(anon_map)
8379 return self.quote(name)
8381 def format_column(
8382 self,
8383 column: ColumnElement[Any],
8384 use_table: bool = False,
8385 name: Optional[str] = None,
8386 table_name: Optional[str] = None,
8387 use_schema: bool = False,
8388 anon_map: Optional[Mapping[str, Any]] = None,
8389 ) -> str:
8390 """Prepare a quoted column name."""
8392 if name is None:
8393 name = column.name
8394 assert name is not None
8396 if anon_map is not None and isinstance(
8397 name, elements._truncated_label
8398 ):
8399 name = name.apply_map(anon_map)
8401 if not getattr(column, "is_literal", False):
8402 if use_table:
8403 return (
8404 self.format_table(
8405 column.table, use_schema=use_schema, name=table_name
8406 )
8407 + "."
8408 + self.quote(name)
8409 )
8410 else:
8411 return self.quote(name)
8412 else:
8413 # literal textual elements get stuck into ColumnClause a lot,
8414 # which shouldn't get quoted
8416 if use_table:
8417 return (
8418 self.format_table(
8419 column.table, use_schema=use_schema, name=table_name
8420 )
8421 + "."
8422 + name
8423 )
8424 else:
8425 return name
8427 def format_table_seq(self, table, use_schema=True):
8428 """Format table name and schema as a tuple."""
8430 # Dialects with more levels in their fully qualified references
8431 # ('database', 'owner', etc.) could override this and return
8432 # a longer sequence.
8434 effective_schema = self.schema_for_object(table)
8436 if not self.omit_schema and use_schema and effective_schema:
8437 return (
8438 self.quote_schema(effective_schema),
8439 self.format_table(table, use_schema=False),
8440 )
8441 else:
8442 return (self.format_table(table, use_schema=False),)
8444 @util.memoized_property
8445 def _r_identifiers(self):
8446 initial, final, escaped_final = (
8447 re.escape(s)
8448 for s in (
8449 self.initial_quote,
8450 self.final_quote,
8451 self._escape_identifier(self.final_quote),
8452 )
8453 )
8454 r = re.compile(
8455 r"(?:"
8456 r"(?:%(initial)s((?:%(escaped)s|[^%(final)s])+)%(final)s"
8457 r"|([^\.]+))(?=\.|$))+"
8458 % {"initial": initial, "final": final, "escaped": escaped_final}
8459 )
8460 return r
8462 def unformat_identifiers(self, identifiers: str) -> Sequence[str]:
8463 """Unpack 'schema.table.column'-like strings into components."""
8465 r = self._r_identifiers
8466 return [
8467 self._unescape_identifier(i)
8468 for i in [a or b for a, b in r.findall(identifiers)]
8469 ]