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 FrozenSet
45from typing import Iterable
46from typing import Iterator
47from typing import List
48from typing import Mapping
49from typing import MutableMapping
50from typing import NamedTuple
51from typing import NoReturn
52from typing import Optional
53from typing import Pattern
54from typing import Sequence
55from typing import Set
56from typing import Tuple
57from typing import Type
58from typing import TYPE_CHECKING
59from typing import Union
61from . import base
62from . import coercions
63from . import crud
64from . import elements
65from . import functions
66from . import operators
67from . import roles
68from . import schema
69from . import selectable
70from . import sqltypes
71from . import util as sql_util
72from ._typing import is_column_element
73from ._typing import is_dml
74from .base import _de_clone
75from .base import _from_objects
76from .base import _NONE_NAME
77from .base import _SentinelDefaultCharacterization
78from .base import NO_ARG
79from .elements import quoted_name
80from .sqltypes import TupleType
81from .visitors import prefix_anon_map
82from .. import exc
83from .. import util
84from ..util import FastIntFlag
85from ..util.typing import Literal
86from ..util.typing import Protocol
87from ..util.typing import Self
88from ..util.typing import TypedDict
90if typing.TYPE_CHECKING:
91 from .annotation import _AnnotationDict
92 from .base import _AmbiguousTableNameMap
93 from .base import CompileState
94 from .base import Executable
95 from .cache_key import CacheKey
96 from .ddl import ExecutableDDLElement
97 from .dml import Insert
98 from .dml import Update
99 from .dml import UpdateBase
100 from .dml import UpdateDMLState
101 from .dml import ValuesBase
102 from .elements import _truncated_label
103 from .elements import BinaryExpression
104 from .elements import BindParameter
105 from .elements import ClauseElement
106 from .elements import ColumnClause
107 from .elements import ColumnElement
108 from .elements import False_
109 from .elements import Label
110 from .elements import Null
111 from .elements import True_
112 from .functions import Function
113 from .schema import CheckConstraint
114 from .schema import Column
115 from .schema import Constraint
116 from .schema import ForeignKeyConstraint
117 from .schema import IdentityOptions
118 from .schema import Index
119 from .schema import PrimaryKeyConstraint
120 from .schema import Table
121 from .schema import UniqueConstraint
122 from .selectable import _ColumnsClauseElement
123 from .selectable import AliasedReturnsRows
124 from .selectable import CompoundSelectState
125 from .selectable import CTE
126 from .selectable import FromClause
127 from .selectable import NamedFromClause
128 from .selectable import ReturnsRows
129 from .selectable import Select
130 from .selectable import SelectState
131 from .type_api import _BindProcessorType
132 from .type_api import TypeDecorator
133 from .type_api import TypeEngine
134 from .type_api import UserDefinedType
135 from .visitors import Visitable
136 from ..engine.cursor import CursorResultMetaData
137 from ..engine.interfaces import _CoreSingleExecuteParams
138 from ..engine.interfaces import _DBAPIAnyExecuteParams
139 from ..engine.interfaces import _DBAPIMultiExecuteParams
140 from ..engine.interfaces import _DBAPISingleExecuteParams
141 from ..engine.interfaces import _ExecuteOptions
142 from ..engine.interfaces import _GenericSetInputSizesType
143 from ..engine.interfaces import _MutableCoreSingleExecuteParams
144 from ..engine.interfaces import Dialect
145 from ..engine.interfaces import SchemaTranslateMapType
148_FromHintsType = Dict["FromClause", str]
150RESERVED_WORDS = {
151 "all",
152 "analyse",
153 "analyze",
154 "and",
155 "any",
156 "array",
157 "as",
158 "asc",
159 "asymmetric",
160 "authorization",
161 "between",
162 "binary",
163 "both",
164 "case",
165 "cast",
166 "check",
167 "collate",
168 "column",
169 "constraint",
170 "create",
171 "cross",
172 "current_date",
173 "current_role",
174 "current_time",
175 "current_timestamp",
176 "current_user",
177 "default",
178 "deferrable",
179 "desc",
180 "distinct",
181 "do",
182 "else",
183 "end",
184 "except",
185 "false",
186 "for",
187 "foreign",
188 "freeze",
189 "from",
190 "full",
191 "grant",
192 "group",
193 "having",
194 "ilike",
195 "in",
196 "initially",
197 "inner",
198 "intersect",
199 "into",
200 "is",
201 "isnull",
202 "join",
203 "leading",
204 "left",
205 "like",
206 "limit",
207 "localtime",
208 "localtimestamp",
209 "natural",
210 "new",
211 "not",
212 "notnull",
213 "null",
214 "off",
215 "offset",
216 "old",
217 "on",
218 "only",
219 "or",
220 "order",
221 "outer",
222 "overlaps",
223 "placing",
224 "primary",
225 "references",
226 "right",
227 "select",
228 "session_user",
229 "set",
230 "similar",
231 "some",
232 "symmetric",
233 "table",
234 "then",
235 "to",
236 "trailing",
237 "true",
238 "union",
239 "unique",
240 "user",
241 "using",
242 "verbose",
243 "when",
244 "where",
245}
247LEGAL_CHARACTERS = re.compile(r"^[A-Z0-9_$]+$", re.I)
248LEGAL_CHARACTERS_PLUS_SPACE = re.compile(r"^[A-Z0-9_ $]+$", re.I)
249ILLEGAL_INITIAL_CHARACTERS = {str(x) for x in range(0, 10)}.union(["$"])
251FK_ON_DELETE = re.compile(
252 r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I
253)
254FK_ON_UPDATE = re.compile(
255 r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I
256)
257FK_INITIALLY = re.compile(r"^(?:DEFERRED|IMMEDIATE)$", re.I)
258BIND_PARAMS = re.compile(r"(?<![:\w\$\x5c]):([\w\$]+)(?![:\w\$])", re.UNICODE)
259BIND_PARAMS_ESC = re.compile(r"\x5c(:[\w\$]*)(?![:\w\$])", re.UNICODE)
261_pyformat_template = "%%(%(name)s)s"
262BIND_TEMPLATES = {
263 "pyformat": _pyformat_template,
264 "qmark": "?",
265 "format": "%%s",
266 "numeric": ":[_POSITION]",
267 "numeric_dollar": "$[_POSITION]",
268 "named": ":%(name)s",
269}
272OPERATORS = {
273 # binary
274 operators.and_: " AND ",
275 operators.or_: " OR ",
276 operators.add: " + ",
277 operators.mul: " * ",
278 operators.sub: " - ",
279 operators.mod: " % ",
280 operators.neg: "-",
281 operators.lt: " < ",
282 operators.le: " <= ",
283 operators.ne: " != ",
284 operators.gt: " > ",
285 operators.ge: " >= ",
286 operators.eq: " = ",
287 operators.is_distinct_from: " IS DISTINCT FROM ",
288 operators.is_not_distinct_from: " IS NOT DISTINCT FROM ",
289 operators.concat_op: " || ",
290 operators.match_op: " MATCH ",
291 operators.not_match_op: " NOT MATCH ",
292 operators.in_op: " IN ",
293 operators.not_in_op: " NOT IN ",
294 operators.comma_op: ", ",
295 operators.from_: " FROM ",
296 operators.as_: " AS ",
297 operators.is_: " IS ",
298 operators.is_not: " IS NOT ",
299 operators.collate: " COLLATE ",
300 # unary
301 operators.exists: "EXISTS ",
302 operators.distinct_op: "DISTINCT ",
303 operators.inv: "NOT ",
304 operators.any_op: "ANY ",
305 operators.all_op: "ALL ",
306 # modifiers
307 operators.desc_op: " DESC",
308 operators.asc_op: " ASC",
309 operators.nulls_first_op: " NULLS FIRST",
310 operators.nulls_last_op: " NULLS LAST",
311 # bitwise
312 operators.bitwise_xor_op: " ^ ",
313 operators.bitwise_or_op: " | ",
314 operators.bitwise_and_op: " & ",
315 operators.bitwise_not_op: "~",
316 operators.bitwise_lshift_op: " << ",
317 operators.bitwise_rshift_op: " >> ",
318}
320FUNCTIONS: Dict[Type[Function[Any]], str] = {
321 functions.coalesce: "coalesce",
322 functions.current_date: "CURRENT_DATE",
323 functions.current_time: "CURRENT_TIME",
324 functions.current_timestamp: "CURRENT_TIMESTAMP",
325 functions.current_user: "CURRENT_USER",
326 functions.localtime: "LOCALTIME",
327 functions.localtimestamp: "LOCALTIMESTAMP",
328 functions.random: "random",
329 functions.sysdate: "sysdate",
330 functions.session_user: "SESSION_USER",
331 functions.user: "USER",
332 functions.cube: "CUBE",
333 functions.rollup: "ROLLUP",
334 functions.grouping_sets: "GROUPING SETS",
335}
338EXTRACT_MAP = {
339 "month": "month",
340 "day": "day",
341 "year": "year",
342 "second": "second",
343 "hour": "hour",
344 "doy": "doy",
345 "minute": "minute",
346 "quarter": "quarter",
347 "dow": "dow",
348 "week": "week",
349 "epoch": "epoch",
350 "milliseconds": "milliseconds",
351 "microseconds": "microseconds",
352 "timezone_hour": "timezone_hour",
353 "timezone_minute": "timezone_minute",
354}
356COMPOUND_KEYWORDS = {
357 selectable._CompoundSelectKeyword.UNION: "UNION",
358 selectable._CompoundSelectKeyword.UNION_ALL: "UNION ALL",
359 selectable._CompoundSelectKeyword.EXCEPT: "EXCEPT",
360 selectable._CompoundSelectKeyword.EXCEPT_ALL: "EXCEPT ALL",
361 selectable._CompoundSelectKeyword.INTERSECT: "INTERSECT",
362 selectable._CompoundSelectKeyword.INTERSECT_ALL: "INTERSECT ALL",
363}
366class ResultColumnsEntry(NamedTuple):
367 """Tracks a column expression that is expected to be represented
368 in the result rows for this statement.
370 This normally refers to the columns clause of a SELECT statement
371 but may also refer to a RETURNING clause, as well as for dialect-specific
372 emulations.
374 """
376 keyname: str
377 """string name that's expected in cursor.description"""
379 name: str
380 """column name, may be labeled"""
382 objects: Tuple[Any, ...]
383 """sequence of objects that should be able to locate this column
384 in a RowMapping. This is typically string names and aliases
385 as well as Column objects.
387 """
389 type: TypeEngine[Any]
390 """Datatype to be associated with this column. This is where
391 the "result processing" logic directly links the compiled statement
392 to the rows that come back from the cursor.
394 """
397class _ResultMapAppender(Protocol):
398 def __call__(
399 self,
400 keyname: str,
401 name: str,
402 objects: Sequence[Any],
403 type_: TypeEngine[Any],
404 ) -> None: ...
407# integer indexes into ResultColumnsEntry used by cursor.py.
408# some profiling showed integer access faster than named tuple
409RM_RENDERED_NAME: Literal[0] = 0
410RM_NAME: Literal[1] = 1
411RM_OBJECTS: Literal[2] = 2
412RM_TYPE: Literal[3] = 3
415class _BaseCompilerStackEntry(TypedDict):
416 asfrom_froms: Set[FromClause]
417 correlate_froms: Set[FromClause]
418 selectable: ReturnsRows
421class _CompilerStackEntry(_BaseCompilerStackEntry, total=False):
422 compile_state: CompileState
423 need_result_map_for_nested: bool
424 need_result_map_for_compound: bool
425 select_0: ReturnsRows
426 insert_from_select: Select[Any]
429class ExpandedState(NamedTuple):
430 """represents state to use when producing "expanded" and
431 "post compile" bound parameters for a statement.
433 "expanded" parameters are parameters that are generated at
434 statement execution time to suit a number of parameters passed, the most
435 prominent example being the individual elements inside of an IN expression.
437 "post compile" parameters are parameters where the SQL literal value
438 will be rendered into the SQL statement at execution time, rather than
439 being passed as separate parameters to the driver.
441 To create an :class:`.ExpandedState` instance, use the
442 :meth:`.SQLCompiler.construct_expanded_state` method on any
443 :class:`.SQLCompiler` instance.
445 """
447 statement: str
448 """String SQL statement with parameters fully expanded"""
450 parameters: _CoreSingleExecuteParams
451 """Parameter dictionary with parameters fully expanded.
453 For a statement that uses named parameters, this dictionary will map
454 exactly to the names in the statement. For a statement that uses
455 positional parameters, the :attr:`.ExpandedState.positional_parameters`
456 will yield a tuple with the positional parameter set.
458 """
460 processors: Mapping[str, _BindProcessorType[Any]]
461 """mapping of bound value processors"""
463 positiontup: Optional[Sequence[str]]
464 """Sequence of string names indicating the order of positional
465 parameters"""
467 parameter_expansion: Mapping[str, List[str]]
468 """Mapping representing the intermediary link from original parameter
469 name to list of "expanded" parameter names, for those parameters that
470 were expanded."""
472 @property
473 def positional_parameters(self) -> Tuple[Any, ...]:
474 """Tuple of positional parameters, for statements that were compiled
475 using a positional paramstyle.
477 """
478 if self.positiontup is None:
479 raise exc.InvalidRequestError(
480 "statement does not use a positional paramstyle"
481 )
482 return tuple(self.parameters[key] for key in self.positiontup)
484 @property
485 def additional_parameters(self) -> _CoreSingleExecuteParams:
486 """synonym for :attr:`.ExpandedState.parameters`."""
487 return self.parameters
490class _InsertManyValues(NamedTuple):
491 """represents state to use for executing an "insertmanyvalues" statement.
493 The primary consumers of this object are the
494 :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and
495 :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods.
497 .. versionadded:: 2.0
499 """
501 is_default_expr: bool
502 """if True, the statement is of the form
503 ``INSERT INTO TABLE DEFAULT VALUES``, and can't be rewritten as a "batch"
505 """
507 single_values_expr: str
508 """The rendered "values" clause of the INSERT statement.
510 This is typically the parenthesized section e.g. "(?, ?, ?)" or similar.
511 The insertmanyvalues logic uses this string as a search and replace
512 target.
514 """
516 insert_crud_params: List[crud._CrudParamElementStr]
517 """List of Column / bind names etc. used while rewriting the statement"""
519 num_positional_params_counted: int
520 """the number of bound parameters in a single-row statement.
522 This count may be larger or smaller than the actual number of columns
523 targeted in the INSERT, as it accommodates for SQL expressions
524 in the values list that may have zero or more parameters embedded
525 within them.
527 This count is part of what's used to organize rewritten parameter lists
528 when batching.
530 """
532 sort_by_parameter_order: bool = False
533 """if the deterministic_returnined_order parameter were used on the
534 insert.
536 All of the attributes following this will only be used if this is True.
538 """
540 includes_upsert_behaviors: bool = False
541 """if True, we have to accommodate for upsert behaviors.
543 This will in some cases downgrade "insertmanyvalues" that requests
544 deterministic ordering.
546 """
548 sentinel_columns: Optional[Sequence[Column[Any]]] = None
549 """List of sentinel columns that were located.
551 This list is only here if the INSERT asked for
552 sort_by_parameter_order=True,
553 and dialect-appropriate sentinel columns were located.
555 .. versionadded:: 2.0.10
557 """
559 num_sentinel_columns: int = 0
560 """how many sentinel columns are in the above list, if any.
562 This is the same as
563 ``len(sentinel_columns) if sentinel_columns is not None else 0``
565 """
567 sentinel_param_keys: Optional[Sequence[str]] = None
568 """parameter str keys in each param dictionary / tuple
569 that would link to the client side "sentinel" values for that row, which
570 we can use to match up parameter sets to result rows.
572 This is only present if sentinel_columns is present and the INSERT
573 statement actually refers to client side values for these sentinel
574 columns.
576 .. versionadded:: 2.0.10
578 .. versionchanged:: 2.0.29 - the sequence is now string dictionary keys
579 only, used against the "compiled parameteters" collection before
580 the parameters were converted by bound parameter processors
582 """
584 implicit_sentinel: bool = False
585 """if True, we have exactly one sentinel column and it uses a server side
586 value, currently has to generate an incrementing integer value.
588 The dialect in question would have asserted that it supports receiving
589 these values back and sorting on that value as a means of guaranteeing
590 correlation with the incoming parameter list.
592 .. versionadded:: 2.0.10
594 """
596 has_upsert_bound_parameters: bool = False
597 """if True, the upsert SET clause contains bound parameters that will
598 receive their values from the parameters dict (i.e., parametrized
599 bindparams where value is None and callable is None).
601 This means we can't batch multiple rows in a single statement, since
602 each row would need different values in the SET clause but there's only
603 one SET clause per statement. See issue #13130.
605 .. versionadded:: 2.0.37
607 """
609 embed_values_counter: bool = False
610 """Whether to embed an incrementing integer counter in each parameter
611 set within the VALUES clause as parameters are batched over.
613 This is only used for a specific INSERT..SELECT..VALUES..RETURNING syntax
614 where a subquery is used to produce value tuples. Current support
615 includes PostgreSQL, Microsoft SQL Server.
617 .. versionadded:: 2.0.10
619 """
622class _InsertManyValuesBatch(NamedTuple):
623 """represents an individual batch SQL statement for insertmanyvalues.
625 This is passed through the
626 :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and
627 :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods out
628 to the :class:`.Connection` within the
629 :meth:`.Connection._exec_insertmany_context` method.
631 .. versionadded:: 2.0.10
633 """
635 replaced_statement: str
636 replaced_parameters: _DBAPIAnyExecuteParams
637 processed_setinputsizes: Optional[_GenericSetInputSizesType]
638 batch: Sequence[_DBAPISingleExecuteParams]
639 sentinel_values: Sequence[Tuple[Any, ...]]
640 current_batch_size: int
641 batchnum: int
642 total_batches: int
643 rows_sorted: bool
644 is_downgraded: bool
647class InsertmanyvaluesSentinelOpts(FastIntFlag):
648 """bitflag enum indicating styles of PK defaults
649 which can work as implicit sentinel columns
651 """
653 NOT_SUPPORTED = 1
654 AUTOINCREMENT = 2
655 IDENTITY = 4
656 SEQUENCE = 8
658 ANY_AUTOINCREMENT = AUTOINCREMENT | IDENTITY | SEQUENCE
659 _SUPPORTED_OR_NOT = NOT_SUPPORTED | ANY_AUTOINCREMENT
661 USE_INSERT_FROM_SELECT = 16
662 RENDER_SELECT_COL_CASTS = 64
665class CompilerState(IntEnum):
666 COMPILING = 0
667 """statement is present, compilation phase in progress"""
669 STRING_APPLIED = 1
670 """statement is present, string form of the statement has been applied.
672 Additional processors by subclasses may still be pending.
674 """
676 NO_STATEMENT = 2
677 """compiler does not have a statement to compile, is used
678 for method access"""
681class Linting(IntEnum):
682 """represent preferences for the 'SQL linting' feature.
684 this feature currently includes support for flagging cartesian products
685 in SQL statements.
687 """
689 NO_LINTING = 0
690 "Disable all linting."
692 COLLECT_CARTESIAN_PRODUCTS = 1
693 """Collect data on FROMs and cartesian products and gather into
694 'self.from_linter'"""
696 WARN_LINTING = 2
697 "Emit warnings for linters that find problems"
699 FROM_LINTING = COLLECT_CARTESIAN_PRODUCTS | WARN_LINTING
700 """Warn for cartesian products; combines COLLECT_CARTESIAN_PRODUCTS
701 and WARN_LINTING"""
704NO_LINTING, COLLECT_CARTESIAN_PRODUCTS, WARN_LINTING, FROM_LINTING = tuple(
705 Linting
706)
709class FromLinter(collections.namedtuple("FromLinter", ["froms", "edges"])):
710 """represents current state for the "cartesian product" detection
711 feature."""
713 def lint(self, start=None):
714 froms = self.froms
715 if not froms:
716 return None, None
718 edges = set(self.edges)
719 the_rest = set(froms)
721 if start is not None:
722 start_with = start
723 the_rest.remove(start_with)
724 else:
725 start_with = the_rest.pop()
727 stack = collections.deque([start_with])
729 while stack and the_rest:
730 node = stack.popleft()
731 the_rest.discard(node)
733 # comparison of nodes in edges here is based on hash equality, as
734 # there are "annotated" elements that match the non-annotated ones.
735 # to remove the need for in-python hash() calls, use native
736 # containment routines (e.g. "node in edge", "edge.index(node)")
737 to_remove = {edge for edge in edges if node in edge}
739 # appendleft the node in each edge that is not
740 # the one that matched.
741 stack.extendleft(edge[not edge.index(node)] for edge in to_remove)
742 edges.difference_update(to_remove)
744 # FROMS left over? boom
745 if the_rest:
746 return the_rest, start_with
747 else:
748 return None, None
750 def warn(self, stmt_type="SELECT"):
751 the_rest, start_with = self.lint()
753 # FROMS left over? boom
754 if the_rest:
755 froms = the_rest
756 if froms:
757 template = (
758 "{stmt_type} statement has a cartesian product between "
759 "FROM element(s) {froms} and "
760 'FROM element "{start}". Apply join condition(s) '
761 "between each element to resolve."
762 )
763 froms_str = ", ".join(
764 f'"{self.froms[from_]}"' for from_ in froms
765 )
766 message = template.format(
767 stmt_type=stmt_type,
768 froms=froms_str,
769 start=self.froms[start_with],
770 )
772 util.warn(message)
775class Compiled:
776 """Represent a compiled SQL or DDL expression.
778 The ``__str__`` method of the ``Compiled`` object should produce
779 the actual text of the statement. ``Compiled`` objects are
780 specific to their underlying database dialect, and also may
781 or may not be specific to the columns referenced within a
782 particular set of bind parameters. In no case should the
783 ``Compiled`` object be dependent on the actual values of those
784 bind parameters, even though it may reference those values as
785 defaults.
786 """
788 statement: Optional[ClauseElement] = None
789 "The statement to compile."
790 string: str = ""
791 "The string representation of the ``statement``"
793 state: CompilerState
794 """description of the compiler's state"""
796 is_sql = False
797 is_ddl = False
799 _cached_metadata: Optional[CursorResultMetaData] = None
801 _result_columns: Optional[List[ResultColumnsEntry]] = None
803 schema_translate_map: Optional[SchemaTranslateMapType] = None
805 execution_options: _ExecuteOptions = util.EMPTY_DICT
806 """
807 Execution options propagated from the statement. In some cases,
808 sub-elements of the statement can modify these.
809 """
811 preparer: IdentifierPreparer
813 _annotations: _AnnotationDict = util.EMPTY_DICT
815 compile_state: Optional[CompileState] = None
816 """Optional :class:`.CompileState` object that maintains additional
817 state used by the compiler.
819 Major executable objects such as :class:`_expression.Insert`,
820 :class:`_expression.Update`, :class:`_expression.Delete`,
821 :class:`_expression.Select` will generate this
822 state when compiled in order to calculate additional information about the
823 object. For the top level object that is to be executed, the state can be
824 stored here where it can also have applicability towards result set
825 processing.
827 .. versionadded:: 1.4
829 """
831 dml_compile_state: Optional[CompileState] = None
832 """Optional :class:`.CompileState` assigned at the same point that
833 .isinsert, .isupdate, or .isdelete is assigned.
835 This will normally be the same object as .compile_state, with the
836 exception of cases like the :class:`.ORMFromStatementCompileState`
837 object.
839 .. versionadded:: 1.4.40
841 """
843 cache_key: Optional[CacheKey] = None
844 """The :class:`.CacheKey` that was generated ahead of creating this
845 :class:`.Compiled` object.
847 This is used for routines that need access to the original
848 :class:`.CacheKey` instance generated when the :class:`.Compiled`
849 instance was first cached, typically in order to reconcile
850 the original list of :class:`.BindParameter` objects with a
851 per-statement list that's generated on each call.
853 """
855 _gen_time: float
856 """Generation time of this :class:`.Compiled`, used for reporting
857 cache stats."""
859 def __init__(
860 self,
861 dialect: Dialect,
862 statement: Optional[ClauseElement],
863 schema_translate_map: Optional[SchemaTranslateMapType] = None,
864 render_schema_translate: bool = False,
865 compile_kwargs: Mapping[str, Any] = util.immutabledict(),
866 ):
867 """Construct a new :class:`.Compiled` object.
869 :param dialect: :class:`.Dialect` to compile against.
871 :param statement: :class:`_expression.ClauseElement` to be compiled.
873 :param schema_translate_map: dictionary of schema names to be
874 translated when forming the resultant SQL
876 .. seealso::
878 :ref:`schema_translating`
880 :param compile_kwargs: additional kwargs that will be
881 passed to the initial call to :meth:`.Compiled.process`.
884 """
885 self.dialect = dialect
886 self.preparer = self.dialect.identifier_preparer
887 if schema_translate_map:
888 self.schema_translate_map = schema_translate_map
889 self.preparer = self.preparer._with_schema_translate(
890 schema_translate_map
891 )
893 if statement is not None:
894 self.state = CompilerState.COMPILING
895 self.statement = statement
896 self.can_execute = statement.supports_execution
897 self._annotations = statement._annotations
898 if self.can_execute:
899 if TYPE_CHECKING:
900 assert isinstance(statement, Executable)
901 self.execution_options = statement._execution_options
902 self.string = self.process(self.statement, **compile_kwargs)
904 if render_schema_translate:
905 assert schema_translate_map is not None
906 self.string = self.preparer._render_schema_translates(
907 self.string, schema_translate_map
908 )
910 self.state = CompilerState.STRING_APPLIED
911 else:
912 self.state = CompilerState.NO_STATEMENT
914 self._gen_time = perf_counter()
916 def __init_subclass__(cls) -> None:
917 cls._init_compiler_cls()
918 return super().__init_subclass__()
920 @classmethod
921 def _init_compiler_cls(cls):
922 pass
924 def _execute_on_connection(
925 self, connection, distilled_params, execution_options
926 ):
927 if self.can_execute:
928 return connection._execute_compiled(
929 self, distilled_params, execution_options
930 )
931 else:
932 raise exc.ObjectNotExecutableError(self.statement)
934 def visit_unsupported_compilation(self, element, err, **kw):
935 raise exc.UnsupportedCompilationError(self, type(element)) from err
937 @property
938 def sql_compiler(self) -> SQLCompiler:
939 """Return a Compiled that is capable of processing SQL expressions.
941 If this compiler is one, it would likely just return 'self'.
943 """
945 raise NotImplementedError()
947 def process(self, obj: Visitable, **kwargs: Any) -> str:
948 return obj._compiler_dispatch(self, **kwargs)
950 def __str__(self) -> str:
951 """Return the string text of the generated SQL or DDL."""
953 if self.state is CompilerState.STRING_APPLIED:
954 return self.string
955 else:
956 return ""
958 def construct_params(
959 self,
960 params: Optional[_CoreSingleExecuteParams] = None,
961 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
962 escape_names: bool = True,
963 ) -> Optional[_MutableCoreSingleExecuteParams]:
964 """Return the bind params for this compiled object.
966 :param params: a dict of string/object pairs whose values will
967 override bind values compiled in to the
968 statement.
969 """
971 raise NotImplementedError()
973 @property
974 def params(self):
975 """Return the bind params for this compiled object."""
976 return self.construct_params()
979class TypeCompiler(util.EnsureKWArg):
980 """Produces DDL specification for TypeEngine objects."""
982 ensure_kwarg = r"visit_\w+"
984 def __init__(self, dialect: Dialect):
985 self.dialect = dialect
987 def process(self, type_: TypeEngine[Any], **kw: Any) -> str:
988 if (
989 type_._variant_mapping
990 and self.dialect.name in type_._variant_mapping
991 ):
992 type_ = type_._variant_mapping[self.dialect.name]
993 return type_._compiler_dispatch(self, **kw)
995 def visit_unsupported_compilation(
996 self, element: Any, err: Exception, **kw: Any
997 ) -> NoReturn:
998 raise exc.UnsupportedCompilationError(self, element) from err
1001# this was a Visitable, but to allow accurate detection of
1002# column elements this is actually a column element
1003class _CompileLabel(
1004 roles.BinaryElementRole[Any], elements.CompilerColumnElement
1005):
1006 """lightweight label object which acts as an expression.Label."""
1008 __visit_name__ = "label"
1009 __slots__ = "element", "name", "_alt_names"
1011 def __init__(self, col, name, alt_names=()):
1012 self.element = col
1013 self.name = name
1014 self._alt_names = (col,) + alt_names
1016 @property
1017 def proxy_set(self):
1018 return self.element.proxy_set
1020 @property
1021 def type(self):
1022 return self.element.type
1024 def self_group(self, **kw):
1025 return self
1028class ilike_case_insensitive(
1029 roles.BinaryElementRole[Any], elements.CompilerColumnElement
1030):
1031 """produce a wrapping element for a case-insensitive portion of
1032 an ILIKE construct.
1034 The construct usually renders the ``lower()`` function, but on
1035 PostgreSQL will pass silently with the assumption that "ILIKE"
1036 is being used.
1038 .. versionadded:: 2.0
1040 """
1042 __visit_name__ = "ilike_case_insensitive_operand"
1043 __slots__ = "element", "comparator"
1045 def __init__(self, element):
1046 self.element = element
1047 self.comparator = element.comparator
1049 @property
1050 def proxy_set(self):
1051 return self.element.proxy_set
1053 @property
1054 def type(self):
1055 return self.element.type
1057 def self_group(self, **kw):
1058 return self
1060 def _with_binary_element_type(self, type_):
1061 return ilike_case_insensitive(
1062 self.element._with_binary_element_type(type_)
1063 )
1066class SQLCompiler(Compiled):
1067 """Default implementation of :class:`.Compiled`.
1069 Compiles :class:`_expression.ClauseElement` objects into SQL strings.
1071 """
1073 extract_map = EXTRACT_MAP
1075 bindname_escape_characters: ClassVar[Mapping[str, str]] = (
1076 util.immutabledict(
1077 {
1078 "%": "P",
1079 "(": "A",
1080 ")": "Z",
1081 ":": "C",
1082 ".": "_",
1083 "[": "_",
1084 "]": "_",
1085 " ": "_",
1086 }
1087 )
1088 )
1089 """A mapping (e.g. dict or similar) containing a lookup of
1090 characters keyed to replacement characters which will be applied to all
1091 'bind names' used in SQL statements as a form of 'escaping'; the given
1092 characters are replaced entirely with the 'replacement' character when
1093 rendered in the SQL statement, and a similar translation is performed
1094 on the incoming names used in parameter dictionaries passed to methods
1095 like :meth:`_engine.Connection.execute`.
1097 This allows bound parameter names used in :func:`_sql.bindparam` and
1098 other constructs to have any arbitrary characters present without any
1099 concern for characters that aren't allowed at all on the target database.
1101 Third party dialects can establish their own dictionary here to replace the
1102 default mapping, which will ensure that the particular characters in the
1103 mapping will never appear in a bound parameter name.
1105 The dictionary is evaluated at **class creation time**, so cannot be
1106 modified at runtime; it must be present on the class when the class
1107 is first declared.
1109 Note that for dialects that have additional bound parameter rules such
1110 as additional restrictions on leading characters, the
1111 :meth:`_sql.SQLCompiler.bindparam_string` method may need to be augmented.
1112 See the cx_Oracle compiler for an example of this.
1114 .. versionadded:: 2.0.0rc1
1116 """
1118 _bind_translate_re: ClassVar[Pattern[str]]
1119 _bind_translate_chars: ClassVar[Mapping[str, str]]
1121 is_sql = True
1123 compound_keywords = COMPOUND_KEYWORDS
1125 isdelete: bool = False
1126 isinsert: bool = False
1127 isupdate: bool = False
1128 """class-level defaults which can be set at the instance
1129 level to define if this Compiled instance represents
1130 INSERT/UPDATE/DELETE
1131 """
1133 postfetch: Optional[List[Column[Any]]]
1134 """list of columns that can be post-fetched after INSERT or UPDATE to
1135 receive server-updated values"""
1137 insert_prefetch: Sequence[Column[Any]] = ()
1138 """list of columns for which default values should be evaluated before
1139 an INSERT takes place"""
1141 update_prefetch: Sequence[Column[Any]] = ()
1142 """list of columns for which onupdate default values should be evaluated
1143 before an UPDATE takes place"""
1145 implicit_returning: Optional[Sequence[ColumnElement[Any]]] = None
1146 """list of "implicit" returning columns for a toplevel INSERT or UPDATE
1147 statement, used to receive newly generated values of columns.
1149 .. versionadded:: 2.0 ``implicit_returning`` replaces the previous
1150 ``returning`` collection, which was not a generalized RETURNING
1151 collection and instead was in fact specific to the "implicit returning"
1152 feature.
1154 """
1156 isplaintext: bool = False
1158 binds: Dict[str, BindParameter[Any]]
1159 """a dictionary of bind parameter keys to BindParameter instances."""
1161 bind_names: Dict[BindParameter[Any], str]
1162 """a dictionary of BindParameter instances to "compiled" names
1163 that are actually present in the generated SQL"""
1165 stack: List[_CompilerStackEntry]
1166 """major statements such as SELECT, INSERT, UPDATE, DELETE are
1167 tracked in this stack using an entry format."""
1169 returning_precedes_values: bool = False
1170 """set to True classwide to generate RETURNING
1171 clauses before the VALUES or WHERE clause (i.e. MSSQL)
1172 """
1174 render_table_with_column_in_update_from: bool = False
1175 """set to True classwide to indicate the SET clause
1176 in a multi-table UPDATE statement should qualify
1177 columns with the table name (i.e. MySQL only)
1178 """
1180 ansi_bind_rules: bool = False
1181 """SQL 92 doesn't allow bind parameters to be used
1182 in the columns clause of a SELECT, nor does it allow
1183 ambiguous expressions like "? = ?". A compiler
1184 subclass can set this flag to False if the target
1185 driver/DB enforces this
1186 """
1188 bindtemplate: str
1189 """template to render bound parameters based on paramstyle."""
1191 compilation_bindtemplate: str
1192 """template used by compiler to render parameters before positional
1193 paramstyle application"""
1195 _numeric_binds_identifier_char: str
1196 """Character that's used to as the identifier of a numerical bind param.
1197 For example if this char is set to ``$``, numerical binds will be rendered
1198 in the form ``$1, $2, $3``.
1199 """
1201 _result_columns: List[ResultColumnsEntry]
1202 """relates label names in the final SQL to a tuple of local
1203 column/label name, ColumnElement object (if any) and
1204 TypeEngine. CursorResult uses this for type processing and
1205 column targeting"""
1207 _textual_ordered_columns: bool = False
1208 """tell the result object that the column names as rendered are important,
1209 but they are also "ordered" vs. what is in the compiled object here.
1211 As of 1.4.42 this condition is only present when the statement is a
1212 TextualSelect, e.g. text("....").columns(...), where it is required
1213 that the columns are considered positionally and not by name.
1215 """
1217 _ad_hoc_textual: bool = False
1218 """tell the result that we encountered text() or '*' constructs in the
1219 middle of the result columns, but we also have compiled columns, so
1220 if the number of columns in cursor.description does not match how many
1221 expressions we have, that means we can't rely on positional at all and
1222 should match on name.
1224 """
1226 _ordered_columns: bool = True
1227 """
1228 if False, means we can't be sure the list of entries
1229 in _result_columns is actually the rendered order. Usually
1230 True unless using an unordered TextualSelect.
1231 """
1233 _loose_column_name_matching: bool = False
1234 """tell the result object that the SQL statement is textual, wants to match
1235 up to Column objects, and may be using the ._tq_label in the SELECT rather
1236 than the base name.
1238 """
1240 _numeric_binds: bool = False
1241 """
1242 True if paramstyle is "numeric". This paramstyle is trickier than
1243 all the others.
1245 """
1247 _render_postcompile: bool = False
1248 """
1249 whether to render out POSTCOMPILE params during the compile phase.
1251 This attribute is used only for end-user invocation of stmt.compile();
1252 it's never used for actual statement execution, where instead the
1253 dialect internals access and render the internal postcompile structure
1254 directly.
1256 """
1258 _post_compile_expanded_state: Optional[ExpandedState] = None
1259 """When render_postcompile is used, the ``ExpandedState`` used to create
1260 the "expanded" SQL is assigned here, and then used by the ``.params``
1261 accessor and ``.construct_params()`` methods for their return values.
1263 .. versionadded:: 2.0.0rc1
1265 """
1267 _pre_expanded_string: Optional[str] = None
1268 """Stores the original string SQL before 'post_compile' is applied,
1269 for cases where 'post_compile' were used.
1271 """
1273 _pre_expanded_positiontup: Optional[List[str]] = None
1275 _insertmanyvalues: Optional[_InsertManyValues] = None
1277 _insert_crud_params: Optional[crud._CrudParamSequence] = None
1279 literal_execute_params: FrozenSet[BindParameter[Any]] = frozenset()
1280 """bindparameter objects that are rendered as literal values at statement
1281 execution time.
1283 """
1285 post_compile_params: FrozenSet[BindParameter[Any]] = frozenset()
1286 """bindparameter objects that are rendered as bound parameter placeholders
1287 at statement execution time.
1289 """
1291 escaped_bind_names: util.immutabledict[str, str] = util.EMPTY_DICT
1292 """Late escaping of bound parameter names that has to be converted
1293 to the original name when looking in the parameter dictionary.
1295 """
1297 has_out_parameters = False
1298 """if True, there are bindparam() objects that have the isoutparam
1299 flag set."""
1301 postfetch_lastrowid = False
1302 """if True, and this in insert, use cursor.lastrowid to populate
1303 result.inserted_primary_key. """
1305 _cache_key_bind_match: Optional[
1306 Tuple[
1307 Dict[
1308 BindParameter[Any],
1309 List[BindParameter[Any]],
1310 ],
1311 Dict[
1312 str,
1313 BindParameter[Any],
1314 ],
1315 ]
1316 ] = None
1317 """a mapping that will relate the BindParameter object we compile
1318 to those that are part of the extracted collection of parameters
1319 in the cache key, if we were given a cache key.
1321 """
1323 positiontup: Optional[List[str]] = None
1324 """for a compiled construct that uses a positional paramstyle, will be
1325 a sequence of strings, indicating the names of bound parameters in order.
1327 This is used in order to render bound parameters in their correct order,
1328 and is combined with the :attr:`_sql.Compiled.params` dictionary to
1329 render parameters.
1331 This sequence always contains the unescaped name of the parameters.
1333 .. seealso::
1335 :ref:`faq_sql_expression_string` - includes a usage example for
1336 debugging use cases.
1338 """
1339 _values_bindparam: Optional[List[str]] = None
1341 _visited_bindparam: Optional[List[str]] = None
1343 inline: bool = False
1345 ctes: Optional[MutableMapping[CTE, str]]
1347 # Detect same CTE references - Dict[(level, name), cte]
1348 # Level is required for supporting nesting
1349 ctes_by_level_name: Dict[Tuple[int, str], CTE]
1351 # To retrieve key/level in ctes_by_level_name -
1352 # Dict[cte_reference, (level, cte_name, cte_opts)]
1353 level_name_by_cte: Dict[CTE, Tuple[int, str, selectable._CTEOpts]]
1355 ctes_recursive: bool
1357 _post_compile_pattern = re.compile(r"__\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]")
1358 _pyformat_pattern = re.compile(r"%\(([^)]+?)\)s")
1359 _positional_pattern = re.compile(
1360 f"{_pyformat_pattern.pattern}|{_post_compile_pattern.pattern}"
1361 )
1363 @classmethod
1364 def _init_compiler_cls(cls):
1365 cls._init_bind_translate()
1367 @classmethod
1368 def _init_bind_translate(cls):
1369 reg = re.escape("".join(cls.bindname_escape_characters))
1370 cls._bind_translate_re = re.compile(f"[{reg}]")
1371 cls._bind_translate_chars = cls.bindname_escape_characters
1373 def __init__(
1374 self,
1375 dialect: Dialect,
1376 statement: Optional[ClauseElement],
1377 cache_key: Optional[CacheKey] = None,
1378 column_keys: Optional[Sequence[str]] = None,
1379 for_executemany: bool = False,
1380 linting: Linting = NO_LINTING,
1381 _supporting_against: Optional[SQLCompiler] = None,
1382 **kwargs: Any,
1383 ):
1384 """Construct a new :class:`.SQLCompiler` object.
1386 :param dialect: :class:`.Dialect` to be used
1388 :param statement: :class:`_expression.ClauseElement` to be compiled
1390 :param column_keys: a list of column names to be compiled into an
1391 INSERT or UPDATE statement.
1393 :param for_executemany: whether INSERT / UPDATE statements should
1394 expect that they are to be invoked in an "executemany" style,
1395 which may impact how the statement will be expected to return the
1396 values of defaults and autoincrement / sequences and similar.
1397 Depending on the backend and driver in use, support for retrieving
1398 these values may be disabled which means SQL expressions may
1399 be rendered inline, RETURNING may not be rendered, etc.
1401 :param kwargs: additional keyword arguments to be consumed by the
1402 superclass.
1404 """
1405 self.column_keys = column_keys
1407 self.cache_key = cache_key
1409 if cache_key:
1410 cksm = {b.key: b for b in cache_key[1]}
1411 ckbm = {b: [b] for b in cache_key[1]}
1412 self._cache_key_bind_match = (ckbm, cksm)
1414 # compile INSERT/UPDATE defaults/sequences to expect executemany
1415 # style execution, which may mean no pre-execute of defaults,
1416 # or no RETURNING
1417 self.for_executemany = for_executemany
1419 self.linting = linting
1421 # a dictionary of bind parameter keys to BindParameter
1422 # instances.
1423 self.binds = {}
1425 # a dictionary of BindParameter instances to "compiled" names
1426 # that are actually present in the generated SQL
1427 self.bind_names = util.column_dict()
1429 # stack which keeps track of nested SELECT statements
1430 self.stack = []
1432 self._result_columns = []
1434 # true if the paramstyle is positional
1435 self.positional = dialect.positional
1436 if self.positional:
1437 self._numeric_binds = nb = dialect.paramstyle.startswith("numeric")
1438 if nb:
1439 self._numeric_binds_identifier_char = (
1440 "$" if dialect.paramstyle == "numeric_dollar" else ":"
1441 )
1443 self.compilation_bindtemplate = _pyformat_template
1444 else:
1445 self.compilation_bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
1447 self.ctes = None
1449 self.label_length = (
1450 dialect.label_length or dialect.max_identifier_length
1451 )
1453 # a map which tracks "anonymous" identifiers that are created on
1454 # the fly here
1455 self.anon_map = prefix_anon_map()
1457 # a map which tracks "truncated" names based on
1458 # dialect.label_length or dialect.max_identifier_length
1459 self.truncated_names: Dict[Tuple[str, str], str] = {}
1460 self._truncated_counters: Dict[str, int] = {}
1462 Compiled.__init__(self, dialect, statement, **kwargs)
1464 if self.isinsert or self.isupdate or self.isdelete:
1465 if TYPE_CHECKING:
1466 assert isinstance(statement, UpdateBase)
1468 if self.isinsert or self.isupdate:
1469 if TYPE_CHECKING:
1470 assert isinstance(statement, ValuesBase)
1471 if statement._inline:
1472 self.inline = True
1473 elif self.for_executemany and (
1474 not self.isinsert
1475 or (
1476 self.dialect.insert_executemany_returning
1477 and statement._return_defaults
1478 )
1479 ):
1480 self.inline = True
1482 self.bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
1484 if _supporting_against:
1485 self.__dict__.update(
1486 {
1487 k: v
1488 for k, v in _supporting_against.__dict__.items()
1489 if k
1490 not in {
1491 "state",
1492 "dialect",
1493 "preparer",
1494 "positional",
1495 "_numeric_binds",
1496 "compilation_bindtemplate",
1497 "bindtemplate",
1498 }
1499 }
1500 )
1502 if self.state is CompilerState.STRING_APPLIED:
1503 if self.positional:
1504 if self._numeric_binds:
1505 self._process_numeric()
1506 else:
1507 self._process_positional()
1509 if self._render_postcompile:
1510 parameters = self.construct_params(
1511 escape_names=False,
1512 _no_postcompile=True,
1513 )
1515 self._process_parameters_for_postcompile(
1516 parameters, _populate_self=True
1517 )
1519 @property
1520 def insert_single_values_expr(self) -> Optional[str]:
1521 """When an INSERT is compiled with a single set of parameters inside
1522 a VALUES expression, the string is assigned here, where it can be
1523 used for insert batching schemes to rewrite the VALUES expression.
1525 .. versionadded:: 1.3.8
1527 .. versionchanged:: 2.0 This collection is no longer used by
1528 SQLAlchemy's built-in dialects, in favor of the currently
1529 internal ``_insertmanyvalues`` collection that is used only by
1530 :class:`.SQLCompiler`.
1532 """
1533 if self._insertmanyvalues is None:
1534 return None
1535 else:
1536 return self._insertmanyvalues.single_values_expr
1538 @util.ro_memoized_property
1539 def effective_returning(self) -> Optional[Sequence[ColumnElement[Any]]]:
1540 """The effective "returning" columns for INSERT, UPDATE or DELETE.
1542 This is either the so-called "implicit returning" columns which are
1543 calculated by the compiler on the fly, or those present based on what's
1544 present in ``self.statement._returning`` (expanded into individual
1545 columns using the ``._all_selected_columns`` attribute) i.e. those set
1546 explicitly using the :meth:`.UpdateBase.returning` method.
1548 .. versionadded:: 2.0
1550 """
1551 if self.implicit_returning:
1552 return self.implicit_returning
1553 elif self.statement is not None and is_dml(self.statement):
1554 return [
1555 c
1556 for c in self.statement._all_selected_columns
1557 if is_column_element(c)
1558 ]
1560 else:
1561 return None
1563 @property
1564 def returning(self):
1565 """backwards compatibility; returns the
1566 effective_returning collection.
1568 """
1569 return self.effective_returning
1571 @property
1572 def current_executable(self):
1573 """Return the current 'executable' that is being compiled.
1575 This is currently the :class:`_sql.Select`, :class:`_sql.Insert`,
1576 :class:`_sql.Update`, :class:`_sql.Delete`,
1577 :class:`_sql.CompoundSelect` object that is being compiled.
1578 Specifically it's assigned to the ``self.stack`` list of elements.
1580 When a statement like the above is being compiled, it normally
1581 is also assigned to the ``.statement`` attribute of the
1582 :class:`_sql.Compiler` object. However, all SQL constructs are
1583 ultimately nestable, and this attribute should never be consulted
1584 by a ``visit_`` method, as it is not guaranteed to be assigned
1585 nor guaranteed to correspond to the current statement being compiled.
1587 .. versionadded:: 1.3.21
1589 For compatibility with previous versions, use the following
1590 recipe::
1592 statement = getattr(self, "current_executable", False)
1593 if statement is False:
1594 statement = self.stack[-1]["selectable"]
1596 For versions 1.4 and above, ensure only .current_executable
1597 is used; the format of "self.stack" may change.
1600 """
1601 try:
1602 return self.stack[-1]["selectable"]
1603 except IndexError as ie:
1604 raise IndexError("Compiler does not have a stack entry") from ie
1606 @property
1607 def prefetch(self):
1608 return list(self.insert_prefetch) + list(self.update_prefetch)
1610 @util.memoized_property
1611 def _global_attributes(self) -> Dict[Any, Any]:
1612 return {}
1614 @util.memoized_instancemethod
1615 def _init_cte_state(self) -> MutableMapping[CTE, str]:
1616 """Initialize collections related to CTEs only if
1617 a CTE is located, to save on the overhead of
1618 these collections otherwise.
1620 """
1621 # collect CTEs to tack on top of a SELECT
1622 # To store the query to print - Dict[cte, text_query]
1623 ctes: MutableMapping[CTE, str] = util.OrderedDict()
1624 self.ctes = ctes
1626 # Detect same CTE references - Dict[(level, name), cte]
1627 # Level is required for supporting nesting
1628 self.ctes_by_level_name = {}
1630 # To retrieve key/level in ctes_by_level_name -
1631 # Dict[cte_reference, (level, cte_name, cte_opts)]
1632 self.level_name_by_cte = {}
1634 self.ctes_recursive = False
1636 return ctes
1638 @contextlib.contextmanager
1639 def _nested_result(self):
1640 """special API to support the use case of 'nested result sets'"""
1641 result_columns, ordered_columns = (
1642 self._result_columns,
1643 self._ordered_columns,
1644 )
1645 self._result_columns, self._ordered_columns = [], False
1647 try:
1648 if self.stack:
1649 entry = self.stack[-1]
1650 entry["need_result_map_for_nested"] = True
1651 else:
1652 entry = None
1653 yield self._result_columns, self._ordered_columns
1654 finally:
1655 if entry:
1656 entry.pop("need_result_map_for_nested")
1657 self._result_columns, self._ordered_columns = (
1658 result_columns,
1659 ordered_columns,
1660 )
1662 def _process_positional(self):
1663 assert not self.positiontup
1664 assert self.state is CompilerState.STRING_APPLIED
1665 assert not self._numeric_binds
1667 if self.dialect.paramstyle == "format":
1668 placeholder = "%s"
1669 else:
1670 assert self.dialect.paramstyle == "qmark"
1671 placeholder = "?"
1673 positions = []
1675 def find_position(m: re.Match[str]) -> str:
1676 normal_bind = m.group(1)
1677 if normal_bind:
1678 positions.append(normal_bind)
1679 return placeholder
1680 else:
1681 # this a post-compile bind
1682 positions.append(m.group(2))
1683 return m.group(0)
1685 self.string = re.sub(
1686 self._positional_pattern, find_position, self.string
1687 )
1689 if self.escaped_bind_names:
1690 reverse_escape = {v: k for k, v in self.escaped_bind_names.items()}
1691 assert len(self.escaped_bind_names) == len(reverse_escape)
1692 self.positiontup = [
1693 reverse_escape.get(name, name) for name in positions
1694 ]
1695 else:
1696 self.positiontup = positions
1698 if self._insertmanyvalues:
1699 positions = []
1701 single_values_expr = re.sub(
1702 self._positional_pattern,
1703 find_position,
1704 self._insertmanyvalues.single_values_expr,
1705 )
1706 insert_crud_params = [
1707 (
1708 v[0],
1709 v[1],
1710 re.sub(self._positional_pattern, find_position, v[2]),
1711 v[3],
1712 )
1713 for v in self._insertmanyvalues.insert_crud_params
1714 ]
1716 self._insertmanyvalues = self._insertmanyvalues._replace(
1717 single_values_expr=single_values_expr,
1718 insert_crud_params=insert_crud_params,
1719 )
1721 def _process_numeric(self):
1722 assert self._numeric_binds
1723 assert self.state is CompilerState.STRING_APPLIED
1725 num = 1
1726 param_pos: Dict[str, str] = {}
1727 order: Iterable[str]
1728 if self._insertmanyvalues and self._values_bindparam is not None:
1729 # bindparams that are not in values are always placed first.
1730 # this avoids the need of changing them when using executemany
1731 # values () ()
1732 order = itertools.chain(
1733 (
1734 name
1735 for name in self.bind_names.values()
1736 if name not in self._values_bindparam
1737 ),
1738 self.bind_names.values(),
1739 )
1740 else:
1741 order = self.bind_names.values()
1743 for bind_name in order:
1744 if bind_name in param_pos:
1745 continue
1746 bind = self.binds[bind_name]
1747 if (
1748 bind in self.post_compile_params
1749 or bind in self.literal_execute_params
1750 ):
1751 # set to None to just mark the in positiontup, it will not
1752 # be replaced below.
1753 param_pos[bind_name] = None # type: ignore
1754 else:
1755 ph = f"{self._numeric_binds_identifier_char}{num}"
1756 num += 1
1757 param_pos[bind_name] = ph
1759 self.next_numeric_pos = num
1761 self.positiontup = list(param_pos)
1762 if self.escaped_bind_names:
1763 len_before = len(param_pos)
1764 param_pos = {
1765 self.escaped_bind_names.get(name, name): pos
1766 for name, pos in param_pos.items()
1767 }
1768 assert len(param_pos) == len_before
1770 # Can't use format here since % chars are not escaped.
1771 self.string = self._pyformat_pattern.sub(
1772 lambda m: param_pos[m.group(1)], self.string
1773 )
1775 if self._insertmanyvalues:
1776 single_values_expr = (
1777 # format is ok here since single_values_expr includes only
1778 # place-holders
1779 self._insertmanyvalues.single_values_expr
1780 % param_pos
1781 )
1782 insert_crud_params = [
1783 (v[0], v[1], "%s", v[3])
1784 for v in self._insertmanyvalues.insert_crud_params
1785 ]
1787 self._insertmanyvalues = self._insertmanyvalues._replace(
1788 # This has the numbers (:1, :2)
1789 single_values_expr=single_values_expr,
1790 # The single binds are instead %s so they can be formatted
1791 insert_crud_params=insert_crud_params,
1792 )
1794 @util.memoized_property
1795 def _bind_processors(
1796 self,
1797 ) -> MutableMapping[
1798 str, Union[_BindProcessorType[Any], Sequence[_BindProcessorType[Any]]]
1799 ]:
1800 # mypy is not able to see the two value types as the above Union,
1801 # it just sees "object". don't know how to resolve
1802 return {
1803 key: value # type: ignore
1804 for key, value in (
1805 (
1806 self.bind_names[bindparam],
1807 (
1808 bindparam.type._cached_bind_processor(self.dialect)
1809 if not bindparam.type._is_tuple_type
1810 else tuple(
1811 elem_type._cached_bind_processor(self.dialect)
1812 for elem_type in cast(
1813 TupleType, bindparam.type
1814 ).types
1815 )
1816 ),
1817 )
1818 for bindparam in self.bind_names
1819 # literal_execute parameters are rendered into the SQL string
1820 # via their literal_processor and never bound as values, so
1821 # they do not need a bind processor. Skipping them also avoids
1822 # invoking bind-processor construction that may require the
1823 # DBAPI to be present (see asyncpg, psycopgcffi cases),
1824 # facilitating testing.
1825 if bindparam not in self.literal_execute_params
1826 )
1827 if value is not None
1828 }
1830 def is_subquery(self):
1831 return len(self.stack) > 1
1833 @property
1834 def sql_compiler(self) -> Self:
1835 return self
1837 def construct_expanded_state(
1838 self,
1839 params: Optional[_CoreSingleExecuteParams] = None,
1840 escape_names: bool = True,
1841 ) -> ExpandedState:
1842 """Return a new :class:`.ExpandedState` for a given parameter set.
1844 For queries that use "expanding" or other late-rendered parameters,
1845 this method will provide for both the finalized SQL string as well
1846 as the parameters that would be used for a particular parameter set.
1848 .. versionadded:: 2.0.0rc1
1850 """
1851 parameters = self.construct_params(
1852 params,
1853 escape_names=escape_names,
1854 _no_postcompile=True,
1855 )
1856 return self._process_parameters_for_postcompile(
1857 parameters,
1858 )
1860 def construct_params(
1861 self,
1862 params: Optional[_CoreSingleExecuteParams] = None,
1863 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
1864 escape_names: bool = True,
1865 _group_number: Optional[int] = None,
1866 _check: bool = True,
1867 _no_postcompile: bool = False,
1868 ) -> _MutableCoreSingleExecuteParams:
1869 """return a dictionary of bind parameter keys and values"""
1871 if self._render_postcompile and not _no_postcompile:
1872 assert self._post_compile_expanded_state is not None
1873 if not params:
1874 return dict(self._post_compile_expanded_state.parameters)
1875 else:
1876 raise exc.InvalidRequestError(
1877 "can't construct new parameters when render_postcompile "
1878 "is used; the statement is hard-linked to the original "
1879 "parameters. Use construct_expanded_state to generate a "
1880 "new statement and parameters."
1881 )
1883 has_escaped_names = escape_names and bool(self.escaped_bind_names)
1885 if extracted_parameters:
1886 # related the bound parameters collected in the original cache key
1887 # to those collected in the incoming cache key. They will not have
1888 # matching names but they will line up positionally in the same
1889 # way. The parameters present in self.bind_names may be clones of
1890 # these original cache key params in the case of DML but the .key
1891 # will be guaranteed to match.
1892 if self.cache_key is None:
1893 raise exc.CompileError(
1894 "This compiled object has no original cache key; "
1895 "can't pass extracted_parameters to construct_params"
1896 )
1897 else:
1898 orig_extracted = self.cache_key[1]
1900 ckbm_tuple = self._cache_key_bind_match
1901 assert ckbm_tuple is not None
1902 ckbm, _ = ckbm_tuple
1903 resolved_extracted = {
1904 bind: extracted
1905 for b, extracted in zip(orig_extracted, extracted_parameters)
1906 for bind in ckbm[b]
1907 }
1908 else:
1909 resolved_extracted = None
1911 if params:
1912 pd = {}
1913 for bindparam, name in self.bind_names.items():
1914 escaped_name = (
1915 self.escaped_bind_names.get(name, name)
1916 if has_escaped_names
1917 else name
1918 )
1920 if bindparam.key in params:
1921 pd[escaped_name] = params[bindparam.key]
1922 elif name in params:
1923 pd[escaped_name] = params[name]
1925 elif _check and bindparam.required:
1926 if _group_number:
1927 raise exc.InvalidRequestError(
1928 "A value is required for bind parameter %r, "
1929 "in parameter group %d"
1930 % (bindparam.key, _group_number),
1931 code="cd3x",
1932 )
1933 else:
1934 raise exc.InvalidRequestError(
1935 "A value is required for bind parameter %r"
1936 % bindparam.key,
1937 code="cd3x",
1938 )
1939 else:
1940 if resolved_extracted:
1941 value_param = resolved_extracted.get(
1942 bindparam, bindparam
1943 )
1944 else:
1945 value_param = bindparam
1947 if bindparam.callable:
1948 pd[escaped_name] = value_param.effective_value
1949 else:
1950 pd[escaped_name] = value_param.value
1951 return pd
1952 else:
1953 pd = {}
1954 for bindparam, name in self.bind_names.items():
1955 escaped_name = (
1956 self.escaped_bind_names.get(name, name)
1957 if has_escaped_names
1958 else name
1959 )
1961 if _check and bindparam.required:
1962 if _group_number:
1963 raise exc.InvalidRequestError(
1964 "A value is required for bind parameter %r, "
1965 "in parameter group %d"
1966 % (bindparam.key, _group_number),
1967 code="cd3x",
1968 )
1969 else:
1970 raise exc.InvalidRequestError(
1971 "A value is required for bind parameter %r"
1972 % bindparam.key,
1973 code="cd3x",
1974 )
1976 if resolved_extracted:
1977 value_param = resolved_extracted.get(bindparam, bindparam)
1978 else:
1979 value_param = bindparam
1981 if bindparam.callable:
1982 pd[escaped_name] = value_param.effective_value
1983 else:
1984 pd[escaped_name] = value_param.value
1986 return pd
1988 @util.memoized_instancemethod
1989 def _get_set_input_sizes_lookup(self):
1990 dialect = self.dialect
1992 include_types = dialect.include_set_input_sizes
1993 exclude_types = dialect.exclude_set_input_sizes
1995 dbapi = dialect.dbapi
1997 def lookup_type(typ):
1998 dbtype = typ._unwrapped_dialect_impl(dialect).get_dbapi_type(dbapi)
2000 if (
2001 dbtype is not None
2002 and (exclude_types is None or dbtype not in exclude_types)
2003 and (include_types is None or dbtype in include_types)
2004 ):
2005 return dbtype
2006 else:
2007 return None
2009 inputsizes = {}
2011 literal_execute_params = self.literal_execute_params
2013 for bindparam in self.bind_names:
2014 if bindparam in literal_execute_params:
2015 continue
2017 if bindparam.type._is_tuple_type:
2018 inputsizes[bindparam] = [
2019 lookup_type(typ)
2020 for typ in cast(TupleType, bindparam.type).types
2021 ]
2022 else:
2023 inputsizes[bindparam] = lookup_type(bindparam.type)
2025 return inputsizes
2027 @property
2028 def params(self):
2029 """Return the bind param dictionary embedded into this
2030 compiled object, for those values that are present.
2032 .. seealso::
2034 :ref:`faq_sql_expression_string` - includes a usage example for
2035 debugging use cases.
2037 """
2038 return self.construct_params(_check=False)
2040 def _process_parameters_for_postcompile(
2041 self,
2042 parameters: _MutableCoreSingleExecuteParams,
2043 _populate_self: bool = False,
2044 ) -> ExpandedState:
2045 """handle special post compile parameters.
2047 These include:
2049 * "expanding" parameters -typically IN tuples that are rendered
2050 on a per-parameter basis for an otherwise fixed SQL statement string.
2052 * literal_binds compiled with the literal_execute flag. Used for
2053 things like SQL Server "TOP N" where the driver does not accommodate
2054 N as a bound parameter.
2056 """
2058 expanded_parameters = {}
2059 new_positiontup: Optional[List[str]]
2061 pre_expanded_string = self._pre_expanded_string
2062 if pre_expanded_string is None:
2063 pre_expanded_string = self.string
2065 if self.positional:
2066 new_positiontup = []
2068 pre_expanded_positiontup = self._pre_expanded_positiontup
2069 if pre_expanded_positiontup is None:
2070 pre_expanded_positiontup = self.positiontup
2072 else:
2073 new_positiontup = pre_expanded_positiontup = None
2075 processors = self._bind_processors
2076 single_processors = cast(
2077 "Mapping[str, _BindProcessorType[Any]]", processors
2078 )
2079 tuple_processors = cast(
2080 "Mapping[str, Sequence[_BindProcessorType[Any]]]", processors
2081 )
2083 new_processors: Dict[str, _BindProcessorType[Any]] = {}
2085 replacement_expressions: Dict[str, Any] = {}
2086 to_update_sets: Dict[str, Any] = {}
2088 # notes:
2089 # *unescaped* parameter names in:
2090 # self.bind_names, self.binds, self._bind_processors, self.positiontup
2091 #
2092 # *escaped* parameter names in:
2093 # construct_params(), replacement_expressions
2095 numeric_positiontup: Optional[List[str]] = None
2097 if self.positional and pre_expanded_positiontup is not None:
2098 names: Iterable[str] = pre_expanded_positiontup
2099 if self._numeric_binds:
2100 numeric_positiontup = []
2101 else:
2102 names = self.bind_names.values()
2104 ebn = self.escaped_bind_names
2105 for name in names:
2106 escaped_name = ebn.get(name, name) if ebn else name
2107 parameter = self.binds[name]
2109 if parameter in self.literal_execute_params:
2110 if escaped_name not in replacement_expressions:
2111 replacement_expressions[escaped_name] = (
2112 self.render_literal_bindparam(
2113 parameter,
2114 render_literal_value=parameters.pop(escaped_name),
2115 )
2116 )
2117 continue
2119 if parameter in self.post_compile_params:
2120 if escaped_name in replacement_expressions:
2121 to_update = to_update_sets[escaped_name]
2122 values = None
2123 else:
2124 # we are removing the parameter from parameters
2125 # because it is a list value, which is not expected by
2126 # TypeEngine objects that would otherwise be asked to
2127 # process it. the single name is being replaced with
2128 # individual numbered parameters for each value in the
2129 # param.
2130 #
2131 # note we are also inserting *escaped* parameter names
2132 # into the given dictionary. default dialect will
2133 # use these param names directly as they will not be
2134 # in the escaped_bind_names dictionary.
2135 values = parameters.pop(name)
2137 leep_res = self._literal_execute_expanding_parameter(
2138 escaped_name, parameter, values
2139 )
2140 to_update, replacement_expr = leep_res
2142 to_update_sets[escaped_name] = to_update
2143 replacement_expressions[escaped_name] = replacement_expr
2145 if not parameter.literal_execute:
2146 parameters.update(to_update)
2147 if parameter.type._is_tuple_type:
2148 assert values is not None
2149 new_processors.update(
2150 (
2151 "%s_%s_%s" % (name, i, j),
2152 tuple_processors[name][j - 1],
2153 )
2154 for i, tuple_element in enumerate(values, 1)
2155 for j, _ in enumerate(tuple_element, 1)
2156 if name in tuple_processors
2157 and tuple_processors[name][j - 1] is not None
2158 )
2159 else:
2160 new_processors.update(
2161 (key, single_processors[name])
2162 for key, _ in to_update
2163 if name in single_processors
2164 )
2165 if numeric_positiontup is not None:
2166 numeric_positiontup.extend(
2167 name for name, _ in to_update
2168 )
2169 elif new_positiontup is not None:
2170 # to_update has escaped names, but that's ok since
2171 # these are new names, that aren't in the
2172 # escaped_bind_names dict.
2173 new_positiontup.extend(name for name, _ in to_update)
2174 expanded_parameters[name] = [
2175 expand_key for expand_key, _ in to_update
2176 ]
2177 elif new_positiontup is not None:
2178 new_positiontup.append(name)
2180 def process_expanding(m):
2181 key = m.group(1)
2182 expr = replacement_expressions[key]
2184 # if POSTCOMPILE included a bind_expression, render that
2185 # around each element
2186 if m.group(2):
2187 tok = m.group(2).split("~~")
2188 be_left, be_right = tok[1], tok[3]
2189 expr = ", ".join(
2190 "%s%s%s" % (be_left, exp, be_right)
2191 for exp in expr.split(", ")
2192 )
2193 return expr
2195 statement = re.sub(
2196 self._post_compile_pattern, process_expanding, pre_expanded_string
2197 )
2199 if numeric_positiontup is not None:
2200 assert new_positiontup is not None
2201 param_pos = {
2202 key: f"{self._numeric_binds_identifier_char}{num}"
2203 for num, key in enumerate(
2204 numeric_positiontup, self.next_numeric_pos
2205 )
2206 }
2207 # Can't use format here since % chars are not escaped.
2208 statement = self._pyformat_pattern.sub(
2209 lambda m: param_pos[m.group(1)], statement
2210 )
2211 new_positiontup.extend(numeric_positiontup)
2213 expanded_state = ExpandedState(
2214 statement,
2215 parameters,
2216 new_processors,
2217 new_positiontup,
2218 expanded_parameters,
2219 )
2221 if _populate_self:
2222 # this is for the "render_postcompile" flag, which is not
2223 # otherwise used internally and is for end-user debugging and
2224 # special use cases.
2225 self._pre_expanded_string = pre_expanded_string
2226 self._pre_expanded_positiontup = pre_expanded_positiontup
2227 self.string = expanded_state.statement
2228 self.positiontup = (
2229 list(expanded_state.positiontup or ())
2230 if self.positional
2231 else None
2232 )
2233 self._post_compile_expanded_state = expanded_state
2235 return expanded_state
2237 @util.preload_module("sqlalchemy.engine.cursor")
2238 def _create_result_map(self):
2239 """utility method used for unit tests only."""
2240 cursor = util.preloaded.engine_cursor
2241 return cursor.CursorResultMetaData._create_description_match_map(
2242 self._result_columns
2243 )
2245 # assigned by crud.py for insert/update statements
2246 _get_bind_name_for_col: _BindNameForColProtocol
2248 @util.memoized_property
2249 def _within_exec_param_key_getter(self) -> Callable[[Any], str]:
2250 getter = self._get_bind_name_for_col
2251 return getter
2253 @util.memoized_property
2254 @util.preload_module("sqlalchemy.engine.result")
2255 def _inserted_primary_key_from_lastrowid_getter(self):
2256 result = util.preloaded.engine_result
2258 param_key_getter = self._within_exec_param_key_getter
2260 assert self.compile_state is not None
2261 statement = self.compile_state.statement
2263 if TYPE_CHECKING:
2264 assert isinstance(statement, Insert)
2266 table = statement.table
2268 getters = [
2269 (operator.methodcaller("get", param_key_getter(col), None), col)
2270 for col in table.primary_key
2271 ]
2273 autoinc_getter = None
2274 autoinc_col = table._autoincrement_column
2275 if autoinc_col is not None:
2276 # apply type post processors to the lastrowid
2277 lastrowid_processor = autoinc_col.type._cached_result_processor(
2278 self.dialect, None
2279 )
2280 autoinc_key = param_key_getter(autoinc_col)
2282 # if a bind value is present for the autoincrement column
2283 # in the parameters, we need to do the logic dictated by
2284 # #7998; honor a non-None user-passed parameter over lastrowid.
2285 # previously in the 1.4 series we weren't fetching lastrowid
2286 # at all if the key were present in the parameters
2287 if autoinc_key in self.binds:
2289 def _autoinc_getter(lastrowid, parameters):
2290 param_value = parameters.get(autoinc_key, lastrowid)
2291 if param_value is not None:
2292 # they supplied non-None parameter, use that.
2293 # SQLite at least is observed to return the wrong
2294 # cursor.lastrowid for INSERT..ON CONFLICT so it
2295 # can't be used in all cases
2296 return param_value
2297 else:
2298 # use lastrowid
2299 return lastrowid
2301 # work around mypy https://github.com/python/mypy/issues/14027
2302 autoinc_getter = _autoinc_getter
2304 else:
2305 lastrowid_processor = None
2307 row_fn = result.result_tuple([col.key for col in table.primary_key])
2309 def get(lastrowid, parameters):
2310 """given cursor.lastrowid value and the parameters used for INSERT,
2311 return a "row" that represents the primary key, either by
2312 using the "lastrowid" or by extracting values from the parameters
2313 that were sent along with the INSERT.
2315 """
2316 if lastrowid_processor is not None:
2317 lastrowid = lastrowid_processor(lastrowid)
2319 if lastrowid is None:
2320 return row_fn(getter(parameters) for getter, col in getters)
2321 else:
2322 return row_fn(
2323 (
2324 (
2325 autoinc_getter(lastrowid, parameters)
2326 if autoinc_getter is not None
2327 else lastrowid
2328 )
2329 if col is autoinc_col
2330 else getter(parameters)
2331 )
2332 for getter, col in getters
2333 )
2335 return get
2337 @util.memoized_property
2338 @util.preload_module("sqlalchemy.engine.result")
2339 def _inserted_primary_key_from_returning_getter(self):
2340 result = util.preloaded.engine_result
2342 assert self.compile_state is not None
2343 statement = self.compile_state.statement
2345 if TYPE_CHECKING:
2346 assert isinstance(statement, Insert)
2348 param_key_getter = self._within_exec_param_key_getter
2349 table = statement.table
2351 returning = self.implicit_returning
2352 assert returning is not None
2353 ret = {col: idx for idx, col in enumerate(returning)}
2355 getters = cast(
2356 "List[Tuple[Callable[[Any], Any], bool]]",
2357 [
2358 (
2359 (operator.itemgetter(ret[col]), True)
2360 if col in ret
2361 else (
2362 operator.methodcaller(
2363 "get", param_key_getter(col), None
2364 ),
2365 False,
2366 )
2367 )
2368 for col in table.primary_key
2369 ],
2370 )
2372 row_fn = result.result_tuple([col.key for col in table.primary_key])
2374 def get(row, parameters):
2375 return row_fn(
2376 getter(row) if use_row else getter(parameters)
2377 for getter, use_row in getters
2378 )
2380 return get
2382 def default_from(self) -> str:
2383 """Called when a SELECT statement has no froms, and no FROM clause is
2384 to be appended.
2386 Gives Oracle Database a chance to tack on a ``FROM DUAL`` to the string
2387 output.
2389 """
2390 return ""
2392 def visit_override_binds(self, override_binds, **kw):
2393 """SQL compile the nested element of an _OverrideBinds with
2394 bindparams swapped out.
2396 The _OverrideBinds is not normally expected to be compiled; it
2397 is meant to be used when an already cached statement is to be used,
2398 the compilation was already performed, and only the bound params should
2399 be swapped in at execution time.
2401 However, there are test cases that exericise this object, and
2402 additionally the ORM subquery loader is known to feed in expressions
2403 which include this construct into new queries (discovered in #11173),
2404 so it has to do the right thing at compile time as well.
2406 """
2408 # get SQL text first
2409 sqltext = override_binds.element._compiler_dispatch(self, **kw)
2411 # for a test compile that is not for caching, change binds after the
2412 # fact. note that we don't try to
2413 # swap the bindparam as we compile, because our element may be
2414 # elsewhere in the statement already (e.g. a subquery or perhaps a
2415 # CTE) and was already visited / compiled. See
2416 # test_relationship_criteria.py ->
2417 # test_selectinload_local_criteria_subquery
2418 for k in override_binds.translate:
2419 if k not in self.binds:
2420 continue
2421 bp = self.binds[k]
2423 # so this would work, just change the value of bp in place.
2424 # but we dont want to mutate things outside.
2425 # bp.value = override_binds.translate[bp.key]
2426 # continue
2428 # instead, need to replace bp with new_bp or otherwise accommodate
2429 # in all internal collections
2430 new_bp = bp._with_value(
2431 override_binds.translate[bp.key],
2432 maintain_key=True,
2433 required=False,
2434 )
2436 name = self.bind_names[bp]
2437 self.binds[k] = self.binds[name] = new_bp
2438 self.bind_names[new_bp] = name
2439 self.bind_names.pop(bp, None)
2441 if bp in self.post_compile_params:
2442 self.post_compile_params |= {new_bp}
2443 if bp in self.literal_execute_params:
2444 self.literal_execute_params |= {new_bp}
2446 ckbm_tuple = self._cache_key_bind_match
2447 if ckbm_tuple:
2448 ckbm, cksm = ckbm_tuple
2449 for bp in bp._cloned_set:
2450 if bp.key in cksm:
2451 cb = cksm[bp.key]
2452 ckbm[cb].append(new_bp)
2454 return sqltext
2456 def visit_grouping(self, grouping, asfrom=False, **kwargs):
2457 return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")"
2459 def visit_select_statement_grouping(self, grouping, **kwargs):
2460 return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")"
2462 def visit_label_reference(
2463 self, element, within_columns_clause=False, **kwargs
2464 ):
2465 if self.stack and self.dialect.supports_simple_order_by_label:
2466 try:
2467 compile_state = cast(
2468 "Union[SelectState, CompoundSelectState]",
2469 self.stack[-1]["compile_state"],
2470 )
2471 except KeyError as ke:
2472 raise exc.CompileError(
2473 "Can't resolve label reference for ORDER BY / "
2474 "GROUP BY / DISTINCT etc."
2475 ) from ke
2477 (
2478 with_cols,
2479 only_froms,
2480 only_cols,
2481 ) = compile_state._label_resolve_dict
2482 if within_columns_clause:
2483 resolve_dict = only_froms
2484 else:
2485 resolve_dict = only_cols
2487 # this can be None in the case that a _label_reference()
2488 # were subject to a replacement operation, in which case
2489 # the replacement of the Label element may have changed
2490 # to something else like a ColumnClause expression.
2491 order_by_elem = element.element._order_by_label_element
2493 if (
2494 order_by_elem is not None
2495 and order_by_elem.name in resolve_dict
2496 and order_by_elem.shares_lineage(
2497 resolve_dict[order_by_elem.name]
2498 )
2499 ):
2500 kwargs["render_label_as_label"] = (
2501 element.element._order_by_label_element
2502 )
2503 return self.process(
2504 element.element,
2505 within_columns_clause=within_columns_clause,
2506 **kwargs,
2507 )
2509 def visit_textual_label_reference(
2510 self, element, within_columns_clause=False, **kwargs
2511 ):
2512 if not self.stack:
2513 # compiling the element outside of the context of a SELECT
2514 return self.process(element._text_clause)
2516 try:
2517 compile_state = cast(
2518 "Union[SelectState, CompoundSelectState]",
2519 self.stack[-1]["compile_state"],
2520 )
2521 except KeyError as ke:
2522 coercions._no_text_coercion(
2523 element.element,
2524 extra=(
2525 "Can't resolve label reference for ORDER BY / "
2526 "GROUP BY / DISTINCT etc."
2527 ),
2528 exc_cls=exc.CompileError,
2529 err=ke,
2530 )
2532 with_cols, only_froms, only_cols = compile_state._label_resolve_dict
2533 try:
2534 if within_columns_clause:
2535 col = only_froms[element.element]
2536 else:
2537 col = with_cols[element.element]
2538 except KeyError as err:
2539 coercions._no_text_coercion(
2540 element.element,
2541 extra=(
2542 "Can't resolve label reference for ORDER BY / "
2543 "GROUP BY / DISTINCT etc."
2544 ),
2545 exc_cls=exc.CompileError,
2546 err=err,
2547 )
2548 else:
2549 kwargs["render_label_as_label"] = col
2550 return self.process(
2551 col, within_columns_clause=within_columns_clause, **kwargs
2552 )
2554 def visit_label(
2555 self,
2556 label,
2557 add_to_result_map=None,
2558 within_label_clause=False,
2559 within_columns_clause=False,
2560 render_label_as_label=None,
2561 result_map_targets=(),
2562 **kw,
2563 ):
2564 # only render labels within the columns clause
2565 # or ORDER BY clause of a select. dialect-specific compilers
2566 # can modify this behavior.
2567 render_label_with_as = (
2568 within_columns_clause and not within_label_clause
2569 )
2570 render_label_only = render_label_as_label is label
2572 if render_label_only or render_label_with_as:
2573 if isinstance(label.name, elements._truncated_label):
2574 labelname = self._truncated_identifier("colident", label.name)
2575 else:
2576 labelname = label.name
2578 if render_label_with_as:
2579 if add_to_result_map is not None:
2580 add_to_result_map(
2581 labelname,
2582 label.name,
2583 (label, labelname) + label._alt_names + result_map_targets,
2584 label.type,
2585 )
2586 return (
2587 label.element._compiler_dispatch(
2588 self,
2589 within_columns_clause=True,
2590 within_label_clause=True,
2591 **kw,
2592 )
2593 + OPERATORS[operators.as_]
2594 + self.preparer.format_label(label, labelname)
2595 )
2596 elif render_label_only:
2597 return self.preparer.format_label(label, labelname)
2598 else:
2599 return label.element._compiler_dispatch(
2600 self, within_columns_clause=False, **kw
2601 )
2603 def _fallback_column_name(self, column):
2604 raise exc.CompileError(
2605 "Cannot compile Column object until its 'name' is assigned."
2606 )
2608 def visit_lambda_element(self, element, **kw):
2609 sql_element = element._resolved
2610 return self.process(sql_element, **kw)
2612 def visit_column(
2613 self,
2614 column: ColumnClause[Any],
2615 add_to_result_map: Optional[_ResultMapAppender] = None,
2616 include_table: bool = True,
2617 result_map_targets: Tuple[Any, ...] = (),
2618 ambiguous_table_name_map: Optional[_AmbiguousTableNameMap] = None,
2619 **kwargs: Any,
2620 ) -> str:
2621 name = orig_name = column.name
2622 if name is None:
2623 name = self._fallback_column_name(column)
2625 is_literal = column.is_literal
2626 if not is_literal and isinstance(name, elements._truncated_label):
2627 name = self._truncated_identifier("colident", name)
2629 if add_to_result_map is not None:
2630 targets = (column, name, column.key) + result_map_targets
2631 if column._tq_label:
2632 targets += (column._tq_label,)
2634 add_to_result_map(name, orig_name, targets, column.type)
2636 if is_literal:
2637 # note we are not currently accommodating for
2638 # literal_column(quoted_name('ident', True)) here
2639 name = self.escape_literal_column(name)
2640 else:
2641 name = self.preparer.quote(name)
2642 table = column.table
2643 if table is None or not include_table or not table.named_with_column:
2644 return name
2645 else:
2646 effective_schema = self.preparer.schema_for_object(table)
2648 if effective_schema:
2649 schema_prefix = (
2650 self.preparer.quote_schema(effective_schema) + "."
2651 )
2652 else:
2653 schema_prefix = ""
2655 if TYPE_CHECKING:
2656 assert isinstance(table, NamedFromClause)
2657 tablename = table.name
2659 if (
2660 not effective_schema
2661 and ambiguous_table_name_map
2662 and tablename in ambiguous_table_name_map
2663 ):
2664 tablename = ambiguous_table_name_map[tablename]
2666 if isinstance(tablename, elements._truncated_label):
2667 tablename = self._truncated_identifier("alias", tablename)
2669 return schema_prefix + self.preparer.quote(tablename) + "." + name
2671 def visit_collation(self, element, **kw):
2672 return self.preparer.format_collation(element.collation)
2674 def visit_fromclause(self, fromclause, **kwargs):
2675 return fromclause.name
2677 def visit_index(self, index, **kwargs):
2678 return index.name
2680 def visit_typeclause(self, typeclause, **kw):
2681 kw["type_expression"] = typeclause
2682 kw["identifier_preparer"] = self.preparer
2683 return self.dialect.type_compiler_instance.process(
2684 typeclause.type, **kw
2685 )
2687 def post_process_text(self, text):
2688 if self.preparer._double_percents:
2689 text = text.replace("%", "%%")
2690 return text
2692 def escape_literal_column(self, text):
2693 if self.preparer._double_percents:
2694 text = text.replace("%", "%%")
2695 return text
2697 def visit_textclause(self, textclause, add_to_result_map=None, **kw):
2698 def do_bindparam(m):
2699 name = m.group(1)
2700 if name in textclause._bindparams:
2701 return self.process(textclause._bindparams[name], **kw)
2702 else:
2703 return self.bindparam_string(name, **kw)
2705 if not self.stack:
2706 self.isplaintext = True
2708 if add_to_result_map:
2709 # text() object is present in the columns clause of a
2710 # select(). Add a no-name entry to the result map so that
2711 # row[text()] produces a result
2712 add_to_result_map(None, None, (textclause,), sqltypes.NULLTYPE)
2714 # un-escape any \:params
2715 return BIND_PARAMS_ESC.sub(
2716 lambda m: m.group(1),
2717 BIND_PARAMS.sub(
2718 do_bindparam, self.post_process_text(textclause.text)
2719 ),
2720 )
2722 def visit_textual_select(
2723 self, taf, compound_index=None, asfrom=False, **kw
2724 ):
2725 toplevel = not self.stack
2726 entry = self._default_stack_entry if toplevel else self.stack[-1]
2728 new_entry: _CompilerStackEntry = {
2729 "correlate_froms": set(),
2730 "asfrom_froms": set(),
2731 "selectable": taf,
2732 }
2733 self.stack.append(new_entry)
2735 if taf._independent_ctes:
2736 self._dispatch_independent_ctes(taf, kw)
2738 populate_result_map = (
2739 toplevel
2740 or (
2741 compound_index == 0
2742 and entry.get("need_result_map_for_compound", False)
2743 )
2744 or entry.get("need_result_map_for_nested", False)
2745 )
2747 if populate_result_map:
2748 self._ordered_columns = self._textual_ordered_columns = (
2749 taf.positional
2750 )
2752 # enable looser result column matching when the SQL text links to
2753 # Column objects by name only
2754 self._loose_column_name_matching = not taf.positional and bool(
2755 taf.column_args
2756 )
2758 for c in taf.column_args:
2759 self.process(
2760 c,
2761 within_columns_clause=True,
2762 add_to_result_map=self._add_to_result_map,
2763 )
2765 text = self.process(taf.element, **kw)
2766 if self.ctes:
2767 nesting_level = len(self.stack) if not toplevel else None
2768 text = self._render_cte_clause(nesting_level=nesting_level) + text
2770 self.stack.pop(-1)
2772 return text
2774 def visit_null(self, expr: Null, **kw: Any) -> str:
2775 return "NULL"
2777 def visit_true(self, expr: True_, **kw: Any) -> str:
2778 if self.dialect.supports_native_boolean:
2779 return "true"
2780 else:
2781 return "1"
2783 def visit_false(self, expr: False_, **kw: Any) -> str:
2784 if self.dialect.supports_native_boolean:
2785 return "false"
2786 else:
2787 return "0"
2789 def _generate_delimited_list(self, elements, separator, **kw):
2790 return separator.join(
2791 s
2792 for s in (c._compiler_dispatch(self, **kw) for c in elements)
2793 if s
2794 )
2796 def _generate_delimited_and_list(self, clauses, **kw):
2797 lcc, clauses = elements.BooleanClauseList._process_clauses_for_boolean(
2798 operators.and_,
2799 elements.True_._singleton,
2800 elements.False_._singleton,
2801 clauses,
2802 )
2803 if lcc == 1:
2804 return clauses[0]._compiler_dispatch(self, **kw)
2805 else:
2806 separator = OPERATORS[operators.and_]
2807 return separator.join(
2808 s
2809 for s in (c._compiler_dispatch(self, **kw) for c in clauses)
2810 if s
2811 )
2813 def visit_tuple(self, clauselist, **kw):
2814 return "(%s)" % self.visit_clauselist(clauselist, **kw)
2816 def visit_clauselist(self, clauselist, **kw):
2817 sep = clauselist.operator
2818 if sep is None:
2819 sep = " "
2820 else:
2821 sep = OPERATORS[clauselist.operator]
2823 return self._generate_delimited_list(clauselist.clauses, sep, **kw)
2825 def visit_expression_clauselist(self, clauselist, **kw):
2826 operator_ = clauselist.operator
2828 disp = self._get_operator_dispatch(
2829 operator_, "expression_clauselist", None
2830 )
2831 if disp:
2832 return disp(clauselist, operator_, **kw)
2834 try:
2835 opstring = OPERATORS[operator_]
2836 except KeyError as err:
2837 raise exc.UnsupportedCompilationError(self, operator_) from err
2838 else:
2839 kw["_in_operator_expression"] = True
2840 return self._generate_delimited_list(
2841 clauselist.clauses, opstring, **kw
2842 )
2844 def visit_case(self, clause, **kwargs):
2845 x = "CASE "
2846 if clause.value is not None:
2847 x += clause.value._compiler_dispatch(self, **kwargs) + " "
2848 for cond, result in clause.whens:
2849 x += (
2850 "WHEN "
2851 + cond._compiler_dispatch(self, **kwargs)
2852 + " THEN "
2853 + result._compiler_dispatch(self, **kwargs)
2854 + " "
2855 )
2856 if clause.else_ is not None:
2857 x += (
2858 "ELSE " + clause.else_._compiler_dispatch(self, **kwargs) + " "
2859 )
2860 x += "END"
2861 return x
2863 def visit_type_coerce(self, type_coerce, **kw):
2864 return type_coerce.typed_expression._compiler_dispatch(self, **kw)
2866 def visit_cast(self, cast, **kwargs):
2867 type_clause = cast.typeclause._compiler_dispatch(self, **kwargs)
2868 match = re.match("(.*)( COLLATE .*)", type_clause)
2869 return "CAST(%s AS %s)%s" % (
2870 cast.clause._compiler_dispatch(self, **kwargs),
2871 match.group(1) if match else type_clause,
2872 match.group(2) if match else "",
2873 )
2875 def _format_frame_clause(self, range_, **kw):
2876 return "%s AND %s" % (
2877 (
2878 "UNBOUNDED PRECEDING"
2879 if range_[0] is elements.RANGE_UNBOUNDED
2880 else (
2881 "CURRENT ROW"
2882 if range_[0] is elements.RANGE_CURRENT
2883 else (
2884 "%s PRECEDING"
2885 % (
2886 self.process(
2887 elements.literal(abs(range_[0])), **kw
2888 ),
2889 )
2890 if range_[0] < 0
2891 else "%s FOLLOWING"
2892 % (self.process(elements.literal(range_[0]), **kw),)
2893 )
2894 )
2895 ),
2896 (
2897 "UNBOUNDED FOLLOWING"
2898 if range_[1] is elements.RANGE_UNBOUNDED
2899 else (
2900 "CURRENT ROW"
2901 if range_[1] is elements.RANGE_CURRENT
2902 else (
2903 "%s PRECEDING"
2904 % (
2905 self.process(
2906 elements.literal(abs(range_[1])), **kw
2907 ),
2908 )
2909 if range_[1] < 0
2910 else "%s FOLLOWING"
2911 % (self.process(elements.literal(range_[1]), **kw),)
2912 )
2913 )
2914 ),
2915 )
2917 def visit_over(self, over, **kwargs):
2918 text = over.element._compiler_dispatch(self, **kwargs)
2919 if over.range_ is not None:
2920 range_ = "RANGE BETWEEN %s" % self._format_frame_clause(
2921 over.range_, **kwargs
2922 )
2923 elif over.rows is not None:
2924 range_ = "ROWS BETWEEN %s" % self._format_frame_clause(
2925 over.rows, **kwargs
2926 )
2927 elif over.groups is not None:
2928 range_ = "GROUPS BETWEEN %s" % self._format_frame_clause(
2929 over.groups, **kwargs
2930 )
2931 else:
2932 range_ = None
2934 return "%s OVER (%s)" % (
2935 text,
2936 " ".join(
2937 [
2938 "%s BY %s"
2939 % (word, clause._compiler_dispatch(self, **kwargs))
2940 for word, clause in (
2941 ("PARTITION", over.partition_by),
2942 ("ORDER", over.order_by),
2943 )
2944 if clause is not None and len(clause)
2945 ]
2946 + ([range_] if range_ else [])
2947 ),
2948 )
2950 def visit_withingroup(self, withingroup, **kwargs):
2951 return "%s WITHIN GROUP (ORDER BY %s)" % (
2952 withingroup.element._compiler_dispatch(self, **kwargs),
2953 withingroup.order_by._compiler_dispatch(self, **kwargs),
2954 )
2956 def visit_funcfilter(self, funcfilter, **kwargs):
2957 return "%s FILTER (WHERE %s)" % (
2958 funcfilter.func._compiler_dispatch(self, **kwargs),
2959 funcfilter.criterion._compiler_dispatch(self, **kwargs),
2960 )
2962 def visit_extract(self, extract, **kwargs):
2963 field = self.extract_map.get(extract.field, extract.field)
2964 return "EXTRACT(%s FROM %s)" % (
2965 field,
2966 extract.expr._compiler_dispatch(self, **kwargs),
2967 )
2969 def visit_scalar_function_column(self, element, **kw):
2970 compiled_fn = self.visit_function(element.fn, **kw)
2971 compiled_col = self.visit_column(element, **kw)
2972 return "(%s).%s" % (compiled_fn, compiled_col)
2974 def visit_function(
2975 self,
2976 func: Function[Any],
2977 add_to_result_map: Optional[_ResultMapAppender] = None,
2978 **kwargs: Any,
2979 ) -> str:
2980 if add_to_result_map is not None:
2981 add_to_result_map(func.name, func.name, (func.name,), func.type)
2983 disp = getattr(self, "visit_%s_func" % func.name.lower(), None)
2985 text: str
2987 if disp:
2988 text = disp(func, **kwargs)
2989 else:
2990 name = FUNCTIONS.get(func._deannotate().__class__, None)
2991 if name:
2992 if func._has_args:
2993 name += "%(expr)s"
2994 else:
2995 name = func.name
2996 name = (
2997 self.preparer.quote(name)
2998 if self.preparer._requires_quotes_illegal_chars(name)
2999 or isinstance(name, elements.quoted_name)
3000 else name
3001 )
3002 name = name + "%(expr)s"
3003 text = ".".join(
3004 [
3005 (
3006 self.preparer.quote(tok)
3007 if self.preparer._requires_quotes_illegal_chars(tok)
3008 or isinstance(name, elements.quoted_name)
3009 else tok
3010 )
3011 for tok in func.packagenames
3012 ]
3013 + [name]
3014 ) % {"expr": self.function_argspec(func, **kwargs)}
3016 if func._with_ordinality:
3017 text += " WITH ORDINALITY"
3018 return text
3020 def visit_next_value_func(self, next_value, **kw):
3021 return self.visit_sequence(next_value.sequence)
3023 def visit_sequence(self, sequence, **kw):
3024 raise NotImplementedError(
3025 "Dialect '%s' does not support sequence increments."
3026 % self.dialect.name
3027 )
3029 def function_argspec(self, func: Function[Any], **kwargs: Any) -> str:
3030 return func.clause_expr._compiler_dispatch(self, **kwargs)
3032 def visit_compound_select(
3033 self, cs, asfrom=False, compound_index=None, **kwargs
3034 ):
3035 toplevel = not self.stack
3037 compile_state = cs._compile_state_factory(cs, self, **kwargs)
3039 if toplevel and not self.compile_state:
3040 self.compile_state = compile_state
3042 compound_stmt = compile_state.statement
3044 entry = self._default_stack_entry if toplevel else self.stack[-1]
3045 need_result_map = toplevel or (
3046 not compound_index
3047 and entry.get("need_result_map_for_compound", False)
3048 )
3050 # indicates there is already a CompoundSelect in play
3051 if compound_index == 0:
3052 entry["select_0"] = cs
3054 self.stack.append(
3055 {
3056 "correlate_froms": entry["correlate_froms"],
3057 "asfrom_froms": entry["asfrom_froms"],
3058 "selectable": cs,
3059 "compile_state": compile_state,
3060 "need_result_map_for_compound": need_result_map,
3061 }
3062 )
3064 if compound_stmt._independent_ctes:
3065 self._dispatch_independent_ctes(compound_stmt, kwargs)
3067 keyword = self.compound_keywords[cs.keyword]
3069 text = (" " + keyword + " ").join(
3070 (
3071 c._compiler_dispatch(
3072 self, asfrom=asfrom, compound_index=i, **kwargs
3073 )
3074 for i, c in enumerate(cs.selects)
3075 )
3076 )
3078 kwargs["include_table"] = False
3079 text += self.group_by_clause(cs, **dict(asfrom=asfrom, **kwargs))
3080 text += self.order_by_clause(cs, **kwargs)
3081 if cs._has_row_limiting_clause:
3082 text += self._row_limit_clause(cs, **kwargs)
3084 if self.ctes:
3085 nesting_level = len(self.stack) if not toplevel else None
3086 text = (
3087 self._render_cte_clause(
3088 nesting_level=nesting_level,
3089 include_following_stack=True,
3090 )
3091 + text
3092 )
3094 self.stack.pop(-1)
3095 return text
3097 def _row_limit_clause(self, cs, **kwargs):
3098 if cs._fetch_clause is not None:
3099 return self.fetch_clause(cs, **kwargs)
3100 else:
3101 return self.limit_clause(cs, **kwargs)
3103 def _get_operator_dispatch(self, operator_, qualifier1, qualifier2):
3104 attrname = "visit_%s_%s%s" % (
3105 operator_.__name__,
3106 qualifier1,
3107 "_" + qualifier2 if qualifier2 else "",
3108 )
3109 return getattr(self, attrname, None)
3111 def visit_unary(
3112 self, unary, add_to_result_map=None, result_map_targets=(), **kw
3113 ):
3114 if add_to_result_map is not None:
3115 result_map_targets += (unary,)
3116 kw["add_to_result_map"] = add_to_result_map
3117 kw["result_map_targets"] = result_map_targets
3119 if unary.operator:
3120 if unary.modifier:
3121 raise exc.CompileError(
3122 "Unary expression does not support operator "
3123 "and modifier simultaneously"
3124 )
3125 disp = self._get_operator_dispatch(
3126 unary.operator, "unary", "operator"
3127 )
3128 if disp:
3129 return disp(unary, unary.operator, **kw)
3130 else:
3131 return self._generate_generic_unary_operator(
3132 unary, OPERATORS[unary.operator], **kw
3133 )
3134 elif unary.modifier:
3135 disp = self._get_operator_dispatch(
3136 unary.modifier, "unary", "modifier"
3137 )
3138 if disp:
3139 return disp(unary, unary.modifier, **kw)
3140 else:
3141 return self._generate_generic_unary_modifier(
3142 unary, OPERATORS[unary.modifier], **kw
3143 )
3144 else:
3145 raise exc.CompileError(
3146 "Unary expression has no operator or modifier"
3147 )
3149 def visit_truediv_binary(self, binary, operator, **kw):
3150 if self.dialect.div_is_floordiv:
3151 return (
3152 self.process(binary.left, **kw)
3153 + " / "
3154 # TODO: would need a fast cast again here,
3155 # unless we want to use an implicit cast like "+ 0.0"
3156 + self.process(
3157 elements.Cast(
3158 binary.right,
3159 (
3160 binary.right.type
3161 if binary.right.type._type_affinity
3162 is sqltypes.Numeric
3163 else sqltypes.Numeric()
3164 ),
3165 ),
3166 **kw,
3167 )
3168 )
3169 else:
3170 return (
3171 self.process(binary.left, **kw)
3172 + " / "
3173 + self.process(binary.right, **kw)
3174 )
3176 def visit_floordiv_binary(self, binary, operator, **kw):
3177 if (
3178 self.dialect.div_is_floordiv
3179 and binary.right.type._type_affinity is sqltypes.Integer
3180 and binary.left.type._type_affinity is sqltypes.Integer
3181 ):
3182 return (
3183 self.process(binary.left, **kw)
3184 + " / "
3185 + self.process(binary.right, **kw)
3186 )
3187 else:
3188 return "FLOOR(%s)" % (
3189 self.process(binary.left, **kw)
3190 + " / "
3191 + self.process(binary.right, **kw)
3192 )
3194 def visit_is_true_unary_operator(self, element, operator, **kw):
3195 if (
3196 element._is_implicitly_boolean
3197 or self.dialect.supports_native_boolean
3198 ):
3199 return self.process(element.element, **kw)
3200 else:
3201 return "%s = 1" % self.process(element.element, **kw)
3203 def visit_is_false_unary_operator(self, element, operator, **kw):
3204 if (
3205 element._is_implicitly_boolean
3206 or self.dialect.supports_native_boolean
3207 ):
3208 return "NOT %s" % self.process(element.element, **kw)
3209 else:
3210 return "%s = 0" % self.process(element.element, **kw)
3212 def visit_not_match_op_binary(self, binary, operator, **kw):
3213 return "NOT %s" % self.visit_binary(
3214 binary, override_operator=operators.match_op
3215 )
3217 def visit_not_in_op_binary(self, binary, operator, **kw):
3218 # The brackets are required in the NOT IN operation because the empty
3219 # case is handled using the form "(col NOT IN (null) OR 1 = 1)".
3220 # The presence of the OR makes the brackets required.
3221 return "(%s)" % self._generate_generic_binary(
3222 binary, OPERATORS[operator], **kw
3223 )
3225 def visit_empty_set_op_expr(self, type_, expand_op, **kw):
3226 if expand_op is operators.not_in_op:
3227 if len(type_) > 1:
3228 return "(%s)) OR (1 = 1" % (
3229 ", ".join("NULL" for element in type_)
3230 )
3231 else:
3232 return "NULL) OR (1 = 1"
3233 elif expand_op is operators.in_op:
3234 if len(type_) > 1:
3235 return "(%s)) AND (1 != 1" % (
3236 ", ".join("NULL" for element in type_)
3237 )
3238 else:
3239 return "NULL) AND (1 != 1"
3240 else:
3241 return self.visit_empty_set_expr(type_)
3243 def visit_empty_set_expr(self, element_types, **kw):
3244 raise NotImplementedError(
3245 "Dialect '%s' does not support empty set expression."
3246 % self.dialect.name
3247 )
3249 def _literal_execute_expanding_parameter_literal_binds(
3250 self, parameter, values, bind_expression_template=None
3251 ):
3252 typ_dialect_impl = parameter.type._unwrapped_dialect_impl(self.dialect)
3254 if not values:
3255 # empty IN expression. note we don't need to use
3256 # bind_expression_template here because there are no
3257 # expressions to render.
3259 if typ_dialect_impl._is_tuple_type:
3260 replacement_expression = (
3261 "VALUES " if self.dialect.tuple_in_values else ""
3262 ) + self.visit_empty_set_op_expr(
3263 parameter.type.types, parameter.expand_op
3264 )
3266 else:
3267 replacement_expression = self.visit_empty_set_op_expr(
3268 [parameter.type], parameter.expand_op
3269 )
3271 elif typ_dialect_impl._is_tuple_type or (
3272 typ_dialect_impl._isnull
3273 and isinstance(values[0], collections_abc.Sequence)
3274 and not isinstance(values[0], (str, bytes))
3275 ):
3276 if typ_dialect_impl._has_bind_expression:
3277 raise NotImplementedError(
3278 "bind_expression() on TupleType not supported with "
3279 "literal_binds"
3280 )
3282 replacement_expression = (
3283 "VALUES " if self.dialect.tuple_in_values else ""
3284 ) + ", ".join(
3285 "(%s)"
3286 % (
3287 ", ".join(
3288 self.render_literal_value(value, param_type)
3289 for value, param_type in zip(
3290 tuple_element, parameter.type.types
3291 )
3292 )
3293 )
3294 for i, tuple_element in enumerate(values)
3295 )
3296 else:
3297 if bind_expression_template:
3298 post_compile_pattern = self._post_compile_pattern
3299 m = post_compile_pattern.search(bind_expression_template)
3300 assert m and m.group(
3301 2
3302 ), "unexpected format for expanding parameter"
3304 tok = m.group(2).split("~~")
3305 be_left, be_right = tok[1], tok[3]
3306 replacement_expression = ", ".join(
3307 "%s%s%s"
3308 % (
3309 be_left,
3310 self.render_literal_value(value, parameter.type),
3311 be_right,
3312 )
3313 for value in values
3314 )
3315 else:
3316 replacement_expression = ", ".join(
3317 self.render_literal_value(value, parameter.type)
3318 for value in values
3319 )
3321 return (), replacement_expression
3323 def _literal_execute_expanding_parameter(self, name, parameter, values):
3324 if parameter.literal_execute:
3325 return self._literal_execute_expanding_parameter_literal_binds(
3326 parameter, values
3327 )
3329 dialect = self.dialect
3330 typ_dialect_impl = parameter.type._unwrapped_dialect_impl(dialect)
3332 if self._numeric_binds:
3333 bind_template = self.compilation_bindtemplate
3334 else:
3335 bind_template = self.bindtemplate
3337 if (
3338 self.dialect._bind_typing_render_casts
3339 and typ_dialect_impl.render_bind_cast
3340 ):
3342 def _render_bindtemplate(name):
3343 return self.render_bind_cast(
3344 parameter.type,
3345 typ_dialect_impl,
3346 bind_template % {"name": name},
3347 )
3349 else:
3351 def _render_bindtemplate(name):
3352 return bind_template % {"name": name}
3354 if not values:
3355 to_update = []
3356 if typ_dialect_impl._is_tuple_type:
3357 replacement_expression = self.visit_empty_set_op_expr(
3358 parameter.type.types, parameter.expand_op
3359 )
3360 else:
3361 replacement_expression = self.visit_empty_set_op_expr(
3362 [parameter.type], parameter.expand_op
3363 )
3365 elif typ_dialect_impl._is_tuple_type or (
3366 typ_dialect_impl._isnull
3367 and isinstance(values[0], collections_abc.Sequence)
3368 and not isinstance(values[0], (str, bytes))
3369 ):
3370 assert not typ_dialect_impl._is_array
3371 to_update = [
3372 ("%s_%s_%s" % (name, i, j), value)
3373 for i, tuple_element in enumerate(values, 1)
3374 for j, value in enumerate(tuple_element, 1)
3375 ]
3377 replacement_expression = (
3378 "VALUES " if dialect.tuple_in_values else ""
3379 ) + ", ".join(
3380 "(%s)"
3381 % (
3382 ", ".join(
3383 _render_bindtemplate(
3384 to_update[i * len(tuple_element) + j][0]
3385 )
3386 for j, value in enumerate(tuple_element)
3387 )
3388 )
3389 for i, tuple_element in enumerate(values)
3390 )
3391 else:
3392 to_update = [
3393 ("%s_%s" % (name, i), value)
3394 for i, value in enumerate(values, 1)
3395 ]
3396 replacement_expression = ", ".join(
3397 _render_bindtemplate(key) for key, value in to_update
3398 )
3400 return to_update, replacement_expression
3402 def visit_binary(
3403 self,
3404 binary,
3405 override_operator=None,
3406 eager_grouping=False,
3407 from_linter=None,
3408 lateral_from_linter=None,
3409 **kw,
3410 ):
3411 if from_linter and operators.is_comparison(binary.operator):
3412 if lateral_from_linter is not None:
3413 enclosing_lateral = kw["enclosing_lateral"]
3414 lateral_from_linter.edges.update(
3415 itertools.product(
3416 _de_clone(
3417 binary.left._from_objects + [enclosing_lateral]
3418 ),
3419 _de_clone(
3420 binary.right._from_objects + [enclosing_lateral]
3421 ),
3422 )
3423 )
3424 else:
3425 from_linter.edges.update(
3426 itertools.product(
3427 _de_clone(binary.left._from_objects),
3428 _de_clone(binary.right._from_objects),
3429 )
3430 )
3432 # don't allow "? = ?" to render
3433 if (
3434 self.ansi_bind_rules
3435 and isinstance(binary.left, elements.BindParameter)
3436 and isinstance(binary.right, elements.BindParameter)
3437 ):
3438 kw["literal_execute"] = True
3440 operator_ = override_operator or binary.operator
3441 disp = self._get_operator_dispatch(operator_, "binary", None)
3442 if disp:
3443 return disp(binary, operator_, **kw)
3444 else:
3445 try:
3446 opstring = OPERATORS[operator_]
3447 except KeyError as err:
3448 raise exc.UnsupportedCompilationError(self, operator_) from err
3449 else:
3450 return self._generate_generic_binary(
3451 binary,
3452 opstring,
3453 from_linter=from_linter,
3454 lateral_from_linter=lateral_from_linter,
3455 **kw,
3456 )
3458 def visit_function_as_comparison_op_binary(self, element, operator, **kw):
3459 return self.process(element.sql_function, **kw)
3461 def visit_mod_binary(self, binary, operator, **kw):
3462 if self.preparer._double_percents:
3463 return (
3464 self.process(binary.left, **kw)
3465 + " %% "
3466 + self.process(binary.right, **kw)
3467 )
3468 else:
3469 return (
3470 self.process(binary.left, **kw)
3471 + " % "
3472 + self.process(binary.right, **kw)
3473 )
3475 def visit_custom_op_binary(self, element, operator, **kw):
3476 kw["eager_grouping"] = operator.eager_grouping
3477 return self._generate_generic_binary(
3478 element,
3479 " " + self.escape_literal_column(operator.opstring) + " ",
3480 **kw,
3481 )
3483 def visit_custom_op_unary_operator(self, element, operator, **kw):
3484 return self._generate_generic_unary_operator(
3485 element, self.escape_literal_column(operator.opstring) + " ", **kw
3486 )
3488 def visit_custom_op_unary_modifier(self, element, operator, **kw):
3489 return self._generate_generic_unary_modifier(
3490 element, " " + self.escape_literal_column(operator.opstring), **kw
3491 )
3493 def _generate_generic_binary(
3494 self,
3495 binary: BinaryExpression[Any],
3496 opstring: str,
3497 eager_grouping: bool = False,
3498 **kw: Any,
3499 ) -> str:
3500 _in_operator_expression = kw.get("_in_operator_expression", False)
3502 kw["_in_operator_expression"] = True
3503 kw["_binary_op"] = binary.operator
3504 text = (
3505 binary.left._compiler_dispatch(
3506 self, eager_grouping=eager_grouping, **kw
3507 )
3508 + opstring
3509 + binary.right._compiler_dispatch(
3510 self, eager_grouping=eager_grouping, **kw
3511 )
3512 )
3514 if _in_operator_expression and eager_grouping:
3515 text = "(%s)" % text
3516 return text
3518 def _generate_generic_unary_operator(self, unary, opstring, **kw):
3519 return opstring + unary.element._compiler_dispatch(self, **kw)
3521 def _generate_generic_unary_modifier(self, unary, opstring, **kw):
3522 return unary.element._compiler_dispatch(self, **kw) + opstring
3524 @util.memoized_property
3525 def _like_percent_literal(self):
3526 return elements.literal_column("'%'", type_=sqltypes.STRINGTYPE)
3528 def visit_ilike_case_insensitive_operand(self, element, **kw):
3529 return f"lower({element.element._compiler_dispatch(self, **kw)})"
3531 def visit_contains_op_binary(self, binary, operator, **kw):
3532 binary = binary._clone()
3533 percent = self._like_percent_literal
3534 binary.right = percent.concat(binary.right).concat(percent)
3535 return self.visit_like_op_binary(binary, operator, **kw)
3537 def visit_not_contains_op_binary(self, binary, operator, **kw):
3538 binary = binary._clone()
3539 percent = self._like_percent_literal
3540 binary.right = percent.concat(binary.right).concat(percent)
3541 return self.visit_not_like_op_binary(binary, operator, **kw)
3543 def visit_icontains_op_binary(self, binary, operator, **kw):
3544 binary = binary._clone()
3545 percent = self._like_percent_literal
3546 binary.left = ilike_case_insensitive(binary.left)
3547 binary.right = percent.concat(
3548 ilike_case_insensitive(binary.right)
3549 ).concat(percent)
3550 return self.visit_ilike_op_binary(binary, operator, **kw)
3552 def visit_not_icontains_op_binary(self, binary, operator, **kw):
3553 binary = binary._clone()
3554 percent = self._like_percent_literal
3555 binary.left = ilike_case_insensitive(binary.left)
3556 binary.right = percent.concat(
3557 ilike_case_insensitive(binary.right)
3558 ).concat(percent)
3559 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3561 def visit_startswith_op_binary(self, binary, operator, **kw):
3562 binary = binary._clone()
3563 percent = self._like_percent_literal
3564 binary.right = percent._rconcat(binary.right)
3565 return self.visit_like_op_binary(binary, operator, **kw)
3567 def visit_not_startswith_op_binary(self, binary, operator, **kw):
3568 binary = binary._clone()
3569 percent = self._like_percent_literal
3570 binary.right = percent._rconcat(binary.right)
3571 return self.visit_not_like_op_binary(binary, operator, **kw)
3573 def visit_istartswith_op_binary(self, binary, operator, **kw):
3574 binary = binary._clone()
3575 percent = self._like_percent_literal
3576 binary.left = ilike_case_insensitive(binary.left)
3577 binary.right = percent._rconcat(ilike_case_insensitive(binary.right))
3578 return self.visit_ilike_op_binary(binary, operator, **kw)
3580 def visit_not_istartswith_op_binary(self, binary, operator, **kw):
3581 binary = binary._clone()
3582 percent = self._like_percent_literal
3583 binary.left = ilike_case_insensitive(binary.left)
3584 binary.right = percent._rconcat(ilike_case_insensitive(binary.right))
3585 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3587 def visit_endswith_op_binary(self, binary, operator, **kw):
3588 binary = binary._clone()
3589 percent = self._like_percent_literal
3590 binary.right = percent.concat(binary.right)
3591 return self.visit_like_op_binary(binary, operator, **kw)
3593 def visit_not_endswith_op_binary(self, binary, operator, **kw):
3594 binary = binary._clone()
3595 percent = self._like_percent_literal
3596 binary.right = percent.concat(binary.right)
3597 return self.visit_not_like_op_binary(binary, operator, **kw)
3599 def visit_iendswith_op_binary(self, binary, operator, **kw):
3600 binary = binary._clone()
3601 percent = self._like_percent_literal
3602 binary.left = ilike_case_insensitive(binary.left)
3603 binary.right = percent.concat(ilike_case_insensitive(binary.right))
3604 return self.visit_ilike_op_binary(binary, operator, **kw)
3606 def visit_not_iendswith_op_binary(self, binary, operator, **kw):
3607 binary = binary._clone()
3608 percent = self._like_percent_literal
3609 binary.left = ilike_case_insensitive(binary.left)
3610 binary.right = percent.concat(ilike_case_insensitive(binary.right))
3611 return self.visit_not_ilike_op_binary(binary, operator, **kw)
3613 def visit_like_op_binary(self, binary, operator, **kw):
3614 escape = binary.modifiers.get("escape", None)
3616 return "%s LIKE %s" % (
3617 binary.left._compiler_dispatch(self, **kw),
3618 binary.right._compiler_dispatch(self, **kw),
3619 ) + (
3620 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
3621 if escape is not None
3622 else ""
3623 )
3625 def visit_not_like_op_binary(self, binary, operator, **kw):
3626 escape = binary.modifiers.get("escape", None)
3627 return "%s NOT LIKE %s" % (
3628 binary.left._compiler_dispatch(self, **kw),
3629 binary.right._compiler_dispatch(self, **kw),
3630 ) + (
3631 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
3632 if escape is not None
3633 else ""
3634 )
3636 def visit_ilike_op_binary(self, binary, operator, **kw):
3637 if operator is operators.ilike_op:
3638 binary = binary._clone()
3639 binary.left = ilike_case_insensitive(binary.left)
3640 binary.right = ilike_case_insensitive(binary.right)
3641 # else we assume ilower() has been applied
3643 return self.visit_like_op_binary(binary, operator, **kw)
3645 def visit_not_ilike_op_binary(self, binary, operator, **kw):
3646 if operator is operators.not_ilike_op:
3647 binary = binary._clone()
3648 binary.left = ilike_case_insensitive(binary.left)
3649 binary.right = ilike_case_insensitive(binary.right)
3650 # else we assume ilower() has been applied
3652 return self.visit_not_like_op_binary(binary, operator, **kw)
3654 def visit_between_op_binary(self, binary, operator, **kw):
3655 symmetric = binary.modifiers.get("symmetric", False)
3656 return self._generate_generic_binary(
3657 binary, " BETWEEN SYMMETRIC " if symmetric else " BETWEEN ", **kw
3658 )
3660 def visit_not_between_op_binary(self, binary, operator, **kw):
3661 symmetric = binary.modifiers.get("symmetric", False)
3662 return self._generate_generic_binary(
3663 binary,
3664 " NOT BETWEEN SYMMETRIC " if symmetric else " NOT BETWEEN ",
3665 **kw,
3666 )
3668 def visit_regexp_match_op_binary(
3669 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3670 ) -> str:
3671 raise exc.CompileError(
3672 "%s dialect does not support regular expressions"
3673 % self.dialect.name
3674 )
3676 def visit_not_regexp_match_op_binary(
3677 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3678 ) -> str:
3679 raise exc.CompileError(
3680 "%s dialect does not support regular expressions"
3681 % self.dialect.name
3682 )
3684 def visit_regexp_replace_op_binary(
3685 self, binary: BinaryExpression[Any], operator: Any, **kw: Any
3686 ) -> str:
3687 raise exc.CompileError(
3688 "%s dialect does not support regular expression replacements"
3689 % self.dialect.name
3690 )
3692 def visit_bindparam(
3693 self,
3694 bindparam,
3695 within_columns_clause=False,
3696 literal_binds=False,
3697 skip_bind_expression=False,
3698 literal_execute=False,
3699 render_postcompile=False,
3700 is_upsert_set=False,
3701 **kwargs,
3702 ):
3703 # Detect parametrized bindparams in upsert SET clause for issue #13130
3704 if (
3705 is_upsert_set
3706 and bindparam.value is None
3707 and bindparam.callable is None
3708 and self._insertmanyvalues is not None
3709 ):
3710 self._insertmanyvalues = self._insertmanyvalues._replace(
3711 has_upsert_bound_parameters=True
3712 )
3714 if not skip_bind_expression:
3715 impl = bindparam.type.dialect_impl(self.dialect)
3716 if impl._has_bind_expression:
3717 bind_expression = impl.bind_expression(bindparam)
3718 wrapped = self.process(
3719 bind_expression,
3720 skip_bind_expression=True,
3721 within_columns_clause=within_columns_clause,
3722 literal_binds=literal_binds and not bindparam.expanding,
3723 literal_execute=literal_execute,
3724 render_postcompile=render_postcompile,
3725 **kwargs,
3726 )
3727 if bindparam.expanding:
3728 # for postcompile w/ expanding, move the "wrapped" part
3729 # of this into the inside
3731 m = re.match(
3732 r"^(.*)\(__\[POSTCOMPILE_(\S+?)\]\)(.*)$", wrapped
3733 )
3734 assert m, "unexpected format for expanding parameter"
3735 wrapped = "(__[POSTCOMPILE_%s~~%s~~REPL~~%s~~])" % (
3736 m.group(2),
3737 m.group(1),
3738 m.group(3),
3739 )
3741 if literal_binds:
3742 ret = self.render_literal_bindparam(
3743 bindparam,
3744 within_columns_clause=True,
3745 bind_expression_template=wrapped,
3746 **kwargs,
3747 )
3748 return "(%s)" % ret
3750 return wrapped
3752 if not literal_binds:
3753 literal_execute = (
3754 literal_execute
3755 or bindparam.literal_execute
3756 or (within_columns_clause and self.ansi_bind_rules)
3757 )
3758 post_compile = literal_execute or bindparam.expanding
3759 else:
3760 post_compile = False
3762 if literal_binds:
3763 ret = self.render_literal_bindparam(
3764 bindparam, within_columns_clause=True, **kwargs
3765 )
3766 if bindparam.expanding:
3767 ret = "(%s)" % ret
3768 return ret
3770 name = self._truncate_bindparam(bindparam)
3772 if name in self.binds:
3773 existing = self.binds[name]
3774 if existing is not bindparam:
3775 if (
3776 (existing.unique or bindparam.unique)
3777 and not existing.proxy_set.intersection(
3778 bindparam.proxy_set
3779 )
3780 and not existing._cloned_set.intersection(
3781 bindparam._cloned_set
3782 )
3783 ):
3784 raise exc.CompileError(
3785 "Bind parameter '%s' conflicts with "
3786 "unique bind parameter of the same name" % name
3787 )
3788 elif existing.expanding != bindparam.expanding:
3789 raise exc.CompileError(
3790 "Can't reuse bound parameter name '%s' in both "
3791 "'expanding' (e.g. within an IN expression) and "
3792 "non-expanding contexts. If this parameter is to "
3793 "receive a list/array value, set 'expanding=True' on "
3794 "it for expressions that aren't IN, otherwise use "
3795 "a different parameter name." % (name,)
3796 )
3797 elif existing._is_crud or bindparam._is_crud:
3798 if existing._is_crud and bindparam._is_crud:
3799 # TODO: this condition is not well understood.
3800 # see tests in test/sql/test_update.py
3801 raise exc.CompileError(
3802 "Encountered unsupported case when compiling an "
3803 "INSERT or UPDATE statement. If this is a "
3804 "multi-table "
3805 "UPDATE statement, please provide string-named "
3806 "arguments to the "
3807 "values() method with distinct names; support for "
3808 "multi-table UPDATE statements that "
3809 "target multiple tables for UPDATE is very "
3810 "limited",
3811 )
3812 else:
3813 raise exc.CompileError(
3814 f"bindparam() name '{bindparam.key}' is reserved "
3815 "for automatic usage in the VALUES or SET "
3816 "clause of this "
3817 "insert/update statement. Please use a "
3818 "name other than column name when using "
3819 "bindparam() "
3820 "with insert() or update() (for example, "
3821 f"'b_{bindparam.key}')."
3822 )
3824 self.binds[bindparam.key] = self.binds[name] = bindparam
3826 # if we are given a cache key that we're going to match against,
3827 # relate the bindparam here to one that is most likely present
3828 # in the "extracted params" portion of the cache key. this is used
3829 # to set up a positional mapping that is used to determine the
3830 # correct parameters for a subsequent use of this compiled with
3831 # a different set of parameter values. here, we accommodate for
3832 # parameters that may have been cloned both before and after the cache
3833 # key was been generated.
3834 ckbm_tuple = self._cache_key_bind_match
3836 if ckbm_tuple:
3837 ckbm, cksm = ckbm_tuple
3838 for bp in bindparam._cloned_set:
3839 if bp.key in cksm:
3840 cb = cksm[bp.key]
3841 ckbm[cb].append(bindparam)
3843 if bindparam.isoutparam:
3844 self.has_out_parameters = True
3846 if post_compile:
3847 if render_postcompile:
3848 self._render_postcompile = True
3850 if literal_execute:
3851 self.literal_execute_params |= {bindparam}
3852 else:
3853 self.post_compile_params |= {bindparam}
3855 ret = self.bindparam_string(
3856 name,
3857 post_compile=post_compile,
3858 expanding=bindparam.expanding,
3859 bindparam_type=bindparam.type,
3860 **kwargs,
3861 )
3863 if bindparam.expanding:
3864 ret = "(%s)" % ret
3866 return ret
3868 def render_bind_cast(self, type_, dbapi_type, sqltext):
3869 raise NotImplementedError()
3871 def render_literal_bindparam(
3872 self,
3873 bindparam,
3874 render_literal_value=NO_ARG,
3875 bind_expression_template=None,
3876 **kw,
3877 ):
3878 if render_literal_value is not NO_ARG:
3879 value = render_literal_value
3880 else:
3881 if bindparam.value is None and bindparam.callable is None:
3882 op = kw.get("_binary_op", None)
3883 if op and op not in (operators.is_, operators.is_not):
3884 util.warn_limited(
3885 "Bound parameter '%s' rendering literal NULL in a SQL "
3886 "expression; comparisons to NULL should not use "
3887 "operators outside of 'is' or 'is not'",
3888 (bindparam.key,),
3889 )
3890 return self.process(sqltypes.NULLTYPE, **kw)
3891 value = bindparam.effective_value
3893 if bindparam.expanding:
3894 leep = self._literal_execute_expanding_parameter_literal_binds
3895 to_update, replacement_expr = leep(
3896 bindparam,
3897 value,
3898 bind_expression_template=bind_expression_template,
3899 )
3900 return replacement_expr
3901 else:
3902 return self.render_literal_value(value, bindparam.type)
3904 def render_literal_value(
3905 self, value: Any, type_: sqltypes.TypeEngine[Any]
3906 ) -> str:
3907 """Render the value of a bind parameter as a quoted literal.
3909 This is used for statement sections that do not accept bind parameters
3910 on the target driver/database.
3912 This should be implemented by subclasses using the quoting services
3913 of the DBAPI.
3915 """
3917 if value is None and not type_.should_evaluate_none:
3918 # issue #10535 - handle NULL in the compiler without placing
3919 # this onto each type, except for "evaluate None" types
3920 # (e.g. JSON)
3921 return self.process(elements.Null._instance())
3923 processor = type_._cached_literal_processor(self.dialect)
3924 if processor:
3925 try:
3926 return processor(value)
3927 except Exception as e:
3928 raise exc.CompileError(
3929 f"Could not render literal value "
3930 f'"{sql_util._repr_single_value(value)}" '
3931 f"with datatype "
3932 f"{type_}; see parent stack trace for "
3933 "more detail."
3934 ) from e
3936 else:
3937 raise exc.CompileError(
3938 f"No literal value renderer is available for literal value "
3939 f'"{sql_util._repr_single_value(value)}" '
3940 f"with datatype {type_}"
3941 )
3943 def _truncate_bindparam(self, bindparam):
3944 if bindparam in self.bind_names:
3945 return self.bind_names[bindparam]
3947 bind_name = bindparam.key
3948 if isinstance(bind_name, elements._truncated_label):
3949 bind_name = self._truncated_identifier("bindparam", bind_name)
3951 # add to bind_names for translation
3952 self.bind_names[bindparam] = bind_name
3954 return bind_name
3956 def _truncated_identifier(
3957 self, ident_class: str, name: _truncated_label
3958 ) -> str:
3959 if (ident_class, name) in self.truncated_names:
3960 return self.truncated_names[(ident_class, name)]
3962 anonname = name.apply_map(self.anon_map)
3964 if len(anonname) > self.label_length - 6:
3965 counter = self._truncated_counters.get(ident_class, 1)
3966 truncname = (
3967 anonname[0 : max(self.label_length - 6, 0)]
3968 + "_"
3969 + hex(counter)[2:]
3970 )
3971 self._truncated_counters[ident_class] = counter + 1
3972 else:
3973 truncname = anonname
3974 self.truncated_names[(ident_class, name)] = truncname
3975 return truncname
3977 def _anonymize(self, name: str) -> str:
3978 return name % self.anon_map
3980 def bindparam_string(
3981 self,
3982 name: str,
3983 post_compile: bool = False,
3984 expanding: bool = False,
3985 escaped_from: Optional[str] = None,
3986 bindparam_type: Optional[TypeEngine[Any]] = None,
3987 accumulate_bind_names: Optional[Set[str]] = None,
3988 visited_bindparam: Optional[List[str]] = None,
3989 **kw: Any,
3990 ) -> str:
3991 # TODO: accumulate_bind_names is passed by crud.py to gather
3992 # names on a per-value basis, visited_bindparam is passed by
3993 # visit_insert() to collect all parameters in the statement.
3994 # see if this gathering can be simplified somehow
3995 if accumulate_bind_names is not None:
3996 accumulate_bind_names.add(name)
3997 if visited_bindparam is not None:
3998 visited_bindparam.append(name)
4000 if not escaped_from:
4001 if self._bind_translate_re.search(name):
4002 # not quite the translate use case as we want to
4003 # also get a quick boolean if we even found
4004 # unusual characters in the name
4005 new_name = self._bind_translate_re.sub(
4006 lambda m: self._bind_translate_chars[m.group(0)],
4007 name,
4008 )
4009 escaped_from = name
4010 name = new_name
4012 if escaped_from:
4013 self.escaped_bind_names = self.escaped_bind_names.union(
4014 {escaped_from: name}
4015 )
4016 if post_compile:
4017 ret = "__[POSTCOMPILE_%s]" % name
4018 if expanding:
4019 # for expanding, bound parameters or literal values will be
4020 # rendered per item
4021 return ret
4023 # otherwise, for non-expanding "literal execute", apply
4024 # bind casts as determined by the datatype
4025 if bindparam_type is not None:
4026 type_impl = bindparam_type._unwrapped_dialect_impl(
4027 self.dialect
4028 )
4029 if type_impl.render_literal_cast:
4030 ret = self.render_bind_cast(bindparam_type, type_impl, ret)
4031 return ret
4032 elif self.state is CompilerState.COMPILING:
4033 ret = self.compilation_bindtemplate % {"name": name}
4034 else:
4035 ret = self.bindtemplate % {"name": name}
4037 if (
4038 bindparam_type is not None
4039 and self.dialect._bind_typing_render_casts
4040 ):
4041 type_impl = bindparam_type._unwrapped_dialect_impl(self.dialect)
4042 if type_impl.render_bind_cast:
4043 ret = self.render_bind_cast(bindparam_type, type_impl, ret)
4045 return ret
4047 def _dispatch_independent_ctes(self, stmt, kw):
4048 local_kw = kw.copy()
4049 local_kw.pop("cte_opts", None)
4050 for cte, opt in zip(
4051 stmt._independent_ctes, stmt._independent_ctes_opts
4052 ):
4053 cte._compiler_dispatch(self, cte_opts=opt, **local_kw)
4055 def visit_cte(
4056 self,
4057 cte: CTE,
4058 asfrom: bool = False,
4059 ashint: bool = False,
4060 fromhints: Optional[_FromHintsType] = None,
4061 visiting_cte: Optional[CTE] = None,
4062 from_linter: Optional[FromLinter] = None,
4063 cte_opts: selectable._CTEOpts = selectable._CTEOpts(False),
4064 **kwargs: Any,
4065 ) -> Optional[str]:
4066 self_ctes = self._init_cte_state()
4067 assert self_ctes is self.ctes
4069 kwargs["visiting_cte"] = cte
4071 cte_name = cte.name
4073 if isinstance(cte_name, elements._truncated_label):
4074 cte_name = self._truncated_identifier("alias", cte_name)
4076 is_new_cte = True
4077 embedded_in_current_named_cte = False
4079 _reference_cte = cte._get_reference_cte()
4081 nesting = cte.nesting or cte_opts.nesting
4083 # check for CTE already encountered
4084 if _reference_cte in self.level_name_by_cte:
4085 cte_level, _, existing_cte_opts = self.level_name_by_cte[
4086 _reference_cte
4087 ]
4088 assert _ == cte_name
4090 cte_level_name = (cte_level, cte_name)
4091 existing_cte = self.ctes_by_level_name[cte_level_name]
4093 # check if we are receiving it here with a specific
4094 # "nest_here" location; if so, move it to this location
4096 if cte_opts.nesting:
4097 if existing_cte_opts.nesting:
4098 raise exc.CompileError(
4099 "CTE is stated as 'nest_here' in "
4100 "more than one location"
4101 )
4103 old_level_name = (cte_level, cte_name)
4104 cte_level = len(self.stack) if nesting else 1
4105 cte_level_name = new_level_name = (cte_level, cte_name)
4107 del self.ctes_by_level_name[old_level_name]
4108 self.ctes_by_level_name[new_level_name] = existing_cte
4109 self.level_name_by_cte[_reference_cte] = new_level_name + (
4110 cte_opts,
4111 )
4113 else:
4114 cte_level = len(self.stack) if nesting else 1
4115 cte_level_name = (cte_level, cte_name)
4117 if cte_level_name in self.ctes_by_level_name:
4118 existing_cte = self.ctes_by_level_name[cte_level_name]
4119 else:
4120 existing_cte = None
4122 if existing_cte is not None:
4123 embedded_in_current_named_cte = visiting_cte is existing_cte
4125 # we've generated a same-named CTE that we are enclosed in,
4126 # or this is the same CTE. just return the name.
4127 if cte is existing_cte._restates or cte is existing_cte:
4128 is_new_cte = False
4129 elif existing_cte is cte._restates:
4130 # we've generated a same-named CTE that is
4131 # enclosed in us - we take precedence, so
4132 # discard the text for the "inner".
4133 del self_ctes[existing_cte]
4135 existing_cte_reference_cte = existing_cte._get_reference_cte()
4137 assert existing_cte_reference_cte is _reference_cte
4138 assert existing_cte_reference_cte is existing_cte
4140 del self.level_name_by_cte[existing_cte_reference_cte]
4141 else:
4142 if (
4143 # if the two CTEs have the same hash, which we expect
4144 # here means that one/both is an annotated of the other
4145 (hash(cte) == hash(existing_cte))
4146 # or...
4147 or (
4148 (
4149 # if they are clones, i.e. they came from the ORM
4150 # or some other visit method
4151 cte._is_clone_of is not None
4152 or existing_cte._is_clone_of is not None
4153 )
4154 # and are deep-copy identical
4155 and cte.compare(existing_cte)
4156 )
4157 ):
4158 # then consider these two CTEs the same
4159 is_new_cte = False
4160 else:
4161 # otherwise these are two CTEs that either will render
4162 # differently, or were indicated separately by the user,
4163 # with the same name
4164 raise exc.CompileError(
4165 "Multiple, unrelated CTEs found with "
4166 "the same name: %r" % cte_name
4167 )
4169 if not asfrom and not is_new_cte:
4170 return None
4172 if cte._cte_alias is not None:
4173 pre_alias_cte = cte._cte_alias
4174 cte_pre_alias_name = cte._cte_alias.name
4175 if isinstance(cte_pre_alias_name, elements._truncated_label):
4176 cte_pre_alias_name = self._truncated_identifier(
4177 "alias", cte_pre_alias_name
4178 )
4179 else:
4180 pre_alias_cte = cte
4181 cte_pre_alias_name = None
4183 if is_new_cte:
4184 self.ctes_by_level_name[cte_level_name] = cte
4185 self.level_name_by_cte[_reference_cte] = cte_level_name + (
4186 cte_opts,
4187 )
4189 if pre_alias_cte not in self.ctes:
4190 self.visit_cte(pre_alias_cte, **kwargs)
4192 if not cte_pre_alias_name and cte not in self_ctes:
4193 if cte.recursive:
4194 self.ctes_recursive = True
4195 text = self.preparer.format_alias(cte, cte_name)
4196 if cte.recursive or cte.element.name_cte_columns:
4197 col_source = cte.element
4199 # TODO: can we get at the .columns_plus_names collection
4200 # that is already (or will be?) generated for the SELECT
4201 # rather than calling twice?
4202 recur_cols = [
4203 # TODO: proxy_name is not technically safe,
4204 # see test_cte->
4205 # test_with_recursive_no_name_currently_buggy. not
4206 # clear what should be done with such a case
4207 fallback_label_name or proxy_name
4208 for (
4209 _,
4210 proxy_name,
4211 fallback_label_name,
4212 c,
4213 repeated,
4214 ) in (col_source._generate_columns_plus_names(True))
4215 if not repeated
4216 ]
4218 text += "(%s)" % (
4219 ", ".join(
4220 self.preparer.format_label_name(
4221 ident, anon_map=self.anon_map
4222 )
4223 for ident in recur_cols
4224 )
4225 )
4227 assert kwargs.get("subquery", False) is False
4229 if not self.stack:
4230 # toplevel, this is a stringify of the
4231 # cte directly. just compile the inner
4232 # the way alias() does.
4233 return cte.element._compiler_dispatch(
4234 self, asfrom=asfrom, **kwargs
4235 )
4236 else:
4237 prefixes = self._generate_prefixes(
4238 cte, cte._prefixes, **kwargs
4239 )
4240 inner = cte.element._compiler_dispatch(
4241 self, asfrom=True, **kwargs
4242 )
4244 text += " AS %s\n(%s)" % (prefixes, inner)
4246 if cte._suffixes:
4247 text += " " + self._generate_prefixes(
4248 cte, cte._suffixes, **kwargs
4249 )
4251 self_ctes[cte] = text
4253 if asfrom:
4254 if from_linter:
4255 from_linter.froms[cte._de_clone()] = cte_name
4257 if not is_new_cte and embedded_in_current_named_cte:
4258 return self.preparer.format_alias(cte, cte_name)
4260 if cte_pre_alias_name:
4261 text = self.preparer.format_alias(cte, cte_pre_alias_name)
4262 if self.preparer._requires_quotes(cte_name):
4263 cte_name = self.preparer.quote(cte_name)
4264 text += self.get_render_as_alias_suffix(cte_name)
4265 return text # type: ignore[no-any-return]
4266 else:
4267 return self.preparer.format_alias(cte, cte_name)
4269 return None
4271 def visit_table_valued_alias(self, element, **kw):
4272 if element.joins_implicitly:
4273 kw["from_linter"] = None
4274 if element._is_lateral:
4275 return self.visit_lateral(element, **kw)
4276 else:
4277 return self.visit_alias(element, **kw)
4279 def visit_table_valued_column(self, element, **kw):
4280 return self.visit_column(element, **kw)
4282 def visit_alias(
4283 self,
4284 alias,
4285 asfrom=False,
4286 ashint=False,
4287 iscrud=False,
4288 fromhints=None,
4289 subquery=False,
4290 lateral=False,
4291 enclosing_alias=None,
4292 from_linter=None,
4293 **kwargs,
4294 ):
4295 if lateral:
4296 if "enclosing_lateral" not in kwargs:
4297 # if lateral is set and enclosing_lateral is not
4298 # present, we assume we are being called directly
4299 # from visit_lateral() and we need to set enclosing_lateral.
4300 assert alias._is_lateral
4301 kwargs["enclosing_lateral"] = alias
4303 # for lateral objects, we track a second from_linter that is...
4304 # lateral! to the level above us.
4305 if (
4306 from_linter
4307 and "lateral_from_linter" not in kwargs
4308 and "enclosing_lateral" in kwargs
4309 ):
4310 kwargs["lateral_from_linter"] = from_linter
4312 if enclosing_alias is not None and enclosing_alias.element is alias:
4313 inner = alias.element._compiler_dispatch(
4314 self,
4315 asfrom=asfrom,
4316 ashint=ashint,
4317 iscrud=iscrud,
4318 fromhints=fromhints,
4319 lateral=lateral,
4320 enclosing_alias=alias,
4321 **kwargs,
4322 )
4323 if subquery and (asfrom or lateral):
4324 inner = "(%s)" % (inner,)
4325 return inner
4326 else:
4327 kwargs["enclosing_alias"] = alias
4329 if asfrom or ashint:
4330 if isinstance(alias.name, elements._truncated_label):
4331 alias_name = self._truncated_identifier("alias", alias.name)
4332 else:
4333 alias_name = alias.name
4335 if ashint:
4336 return self.preparer.format_alias(alias, alias_name)
4337 elif asfrom:
4338 if from_linter:
4339 from_linter.froms[alias._de_clone()] = alias_name
4341 inner = alias.element._compiler_dispatch(
4342 self, asfrom=True, lateral=lateral, **kwargs
4343 )
4344 if subquery:
4345 inner = "(%s)" % (inner,)
4347 ret = inner + self.get_render_as_alias_suffix(
4348 self.preparer.format_alias(alias, alias_name)
4349 )
4351 if alias._supports_derived_columns and alias._render_derived:
4352 ret += "(%s)" % (
4353 ", ".join(
4354 "%s%s"
4355 % (
4356 self.preparer.quote(col.name),
4357 (
4358 " %s"
4359 % self.dialect.type_compiler_instance.process(
4360 col.type, **kwargs
4361 )
4362 if alias._render_derived_w_types
4363 else ""
4364 ),
4365 )
4366 for col in alias.c
4367 )
4368 )
4370 if fromhints and alias in fromhints:
4371 ret = self.format_from_hint_text(
4372 ret, alias, fromhints[alias], iscrud
4373 )
4375 return ret
4376 else:
4377 # note we cancel the "subquery" flag here as well
4378 return alias.element._compiler_dispatch(
4379 self, lateral=lateral, **kwargs
4380 )
4382 def visit_subquery(self, subquery, **kw):
4383 kw["subquery"] = True
4384 return self.visit_alias(subquery, **kw)
4386 def visit_lateral(self, lateral_, **kw):
4387 kw["lateral"] = True
4388 return "LATERAL %s" % self.visit_alias(lateral_, **kw)
4390 def visit_tablesample(self, tablesample, asfrom=False, **kw):
4391 text = "%s TABLESAMPLE %s" % (
4392 self.visit_alias(tablesample, asfrom=True, **kw),
4393 tablesample._get_method()._compiler_dispatch(self, **kw),
4394 )
4396 if tablesample.seed is not None:
4397 text += " REPEATABLE (%s)" % (
4398 tablesample.seed._compiler_dispatch(self, **kw)
4399 )
4401 return text
4403 def _render_values(self, element, **kw):
4404 kw.setdefault("literal_binds", element.literal_binds)
4405 tuples = ", ".join(
4406 self.process(
4407 elements.Tuple(
4408 types=element._column_types, *elem
4409 ).self_group(),
4410 **kw,
4411 )
4412 for chunk in element._data
4413 for elem in chunk
4414 )
4415 return f"VALUES {tuples}"
4417 def visit_values(
4418 self, element, asfrom=False, from_linter=None, visiting_cte=None, **kw
4419 ):
4421 if element._independent_ctes:
4422 self._dispatch_independent_ctes(element, kw)
4424 v = self._render_values(element, **kw)
4426 if element._unnamed:
4427 name = None
4428 elif isinstance(element.name, elements._truncated_label):
4429 name = self._truncated_identifier("values", element.name)
4430 else:
4431 name = element.name
4433 if element._is_lateral:
4434 lateral = "LATERAL "
4435 else:
4436 lateral = ""
4438 if asfrom:
4439 if from_linter:
4440 from_linter.froms[element._de_clone()] = (
4441 name if name is not None else "(unnamed VALUES element)"
4442 )
4444 if visiting_cte is not None and visiting_cte.element is element:
4445 if element._is_lateral:
4446 raise exc.CompileError(
4447 "Can't use a LATERAL VALUES expression inside of a CTE"
4448 )
4449 elif name:
4450 kw["include_table"] = False
4451 v = "%s(%s)%s (%s)" % (
4452 lateral,
4453 v,
4454 self.get_render_as_alias_suffix(self.preparer.quote(name)),
4455 (
4456 ", ".join(
4457 c._compiler_dispatch(self, **kw)
4458 for c in element.columns
4459 )
4460 ),
4461 )
4462 else:
4463 v = "%s(%s)" % (lateral, v)
4464 return v
4466 def visit_scalar_values(self, element, **kw):
4467 return f"({self._render_values(element, **kw)})"
4469 def get_render_as_alias_suffix(self, alias_name_text):
4470 return " AS " + alias_name_text
4472 def _add_to_result_map(
4473 self,
4474 keyname: str,
4475 name: str,
4476 objects: Tuple[Any, ...],
4477 type_: TypeEngine[Any],
4478 ) -> None:
4480 # note objects must be non-empty for cursor.py to handle the
4481 # collection properly
4482 assert objects
4484 if keyname is None or keyname == "*":
4485 self._ordered_columns = False
4486 self._ad_hoc_textual = True
4487 if type_._is_tuple_type:
4488 raise exc.CompileError(
4489 "Most backends don't support SELECTing "
4490 "from a tuple() object. If this is an ORM query, "
4491 "consider using the Bundle object."
4492 )
4493 self._result_columns.append(
4494 ResultColumnsEntry(keyname, name, objects, type_)
4495 )
4497 def _label_returning_column(
4498 self, stmt, column, populate_result_map, column_clause_args=None, **kw
4499 ):
4500 """Render a column with necessary labels inside of a RETURNING clause.
4502 This method is provided for individual dialects in place of calling
4503 the _label_select_column method directly, so that the two use cases
4504 of RETURNING vs. SELECT can be disambiguated going forward.
4506 .. versionadded:: 1.4.21
4508 """
4509 return self._label_select_column(
4510 None,
4511 column,
4512 populate_result_map,
4513 False,
4514 {} if column_clause_args is None else column_clause_args,
4515 **kw,
4516 )
4518 def _label_select_column(
4519 self,
4520 select,
4521 column,
4522 populate_result_map,
4523 asfrom,
4524 column_clause_args,
4525 name=None,
4526 proxy_name=None,
4527 fallback_label_name=None,
4528 within_columns_clause=True,
4529 column_is_repeated=False,
4530 need_column_expressions=False,
4531 include_table=True,
4532 ):
4533 """produce labeled columns present in a select()."""
4534 impl = column.type.dialect_impl(self.dialect)
4536 if impl._has_column_expression and (
4537 need_column_expressions or populate_result_map
4538 ):
4539 col_expr = impl.column_expression(column)
4540 else:
4541 col_expr = column
4543 if populate_result_map:
4544 # pass an "add_to_result_map" callable into the compilation
4545 # of embedded columns. this collects information about the
4546 # column as it will be fetched in the result and is coordinated
4547 # with cursor.description when the query is executed.
4548 add_to_result_map = self._add_to_result_map
4550 # if the SELECT statement told us this column is a repeat,
4551 # wrap the callable with one that prevents the addition of the
4552 # targets
4553 if column_is_repeated:
4554 _add_to_result_map = add_to_result_map
4556 def add_to_result_map(keyname, name, objects, type_):
4557 _add_to_result_map(keyname, name, (keyname,), type_)
4559 # if we redefined col_expr for type expressions, wrap the
4560 # callable with one that adds the original column to the targets
4561 elif col_expr is not column:
4562 _add_to_result_map = add_to_result_map
4564 def add_to_result_map(keyname, name, objects, type_):
4565 _add_to_result_map(
4566 keyname, name, (column,) + objects, type_
4567 )
4569 else:
4570 add_to_result_map = None
4572 # this method is used by some of the dialects for RETURNING,
4573 # which has different inputs. _label_returning_column was added
4574 # as the better target for this now however for 1.4 we will keep
4575 # _label_select_column directly compatible with this use case.
4576 # these assertions right now set up the current expected inputs
4577 assert within_columns_clause, (
4578 "_label_select_column is only relevant within "
4579 "the columns clause of a SELECT or RETURNING"
4580 )
4581 result_expr: Union[elements.Label[Any], _CompileLabel]
4583 if isinstance(column, elements.Label):
4584 if col_expr is not column:
4585 result_expr = _CompileLabel(
4586 col_expr, column.name, alt_names=(column.element,)
4587 )
4588 else:
4589 result_expr = col_expr
4591 elif name:
4592 # here, _columns_plus_names has determined there's an explicit
4593 # label name we need to use. this is the default for
4594 # tablenames_plus_columnnames as well as when columns are being
4595 # deduplicated on name
4597 assert (
4598 proxy_name is not None
4599 ), "proxy_name is required if 'name' is passed"
4601 result_expr = _CompileLabel(
4602 col_expr,
4603 name,
4604 alt_names=(
4605 proxy_name,
4606 # this is a hack to allow legacy result column lookups
4607 # to work as they did before; this goes away in 2.0.
4608 # TODO: this only seems to be tested indirectly
4609 # via test/orm/test_deprecations.py. should be a
4610 # resultset test for this
4611 column._tq_label,
4612 ),
4613 )
4614 else:
4615 # determine here whether this column should be rendered in
4616 # a labelled context or not, as we were given no required label
4617 # name from the caller. Here we apply heuristics based on the kind
4618 # of SQL expression involved.
4620 if col_expr is not column:
4621 # type-specific expression wrapping the given column,
4622 # so we render a label
4623 render_with_label = True
4624 elif isinstance(column, elements.ColumnClause):
4625 # table-bound column, we render its name as a label if we are
4626 # inside of a subquery only
4627 render_with_label = (
4628 asfrom
4629 and not column.is_literal
4630 and column.table is not None
4631 )
4632 elif isinstance(column, elements.TextClause):
4633 render_with_label = False
4634 elif isinstance(column, elements.UnaryExpression):
4635 # unary expression. notes added as of #12681
4636 #
4637 # By convention, the visit_unary() method
4638 # itself does not add an entry to the result map, and relies
4639 # upon either the inner expression creating a result map
4640 # entry, or if not, by creating a label here that produces
4641 # the result map entry. Where that happens is based on whether
4642 # or not the element immediately inside the unary is a
4643 # NamedColumn subclass or not.
4644 #
4645 # Now, this also impacts how the SELECT is written; if
4646 # we decide to generate a label here, we get the usual
4647 # "~(x+y) AS anon_1" thing in the columns clause. If we
4648 # don't, we don't get an AS at all, we get like
4649 # "~table.column".
4650 #
4651 # But here is the important thing as of modernish (like 1.4)
4652 # versions of SQLAlchemy - **whether or not the AS <label>
4653 # is present in the statement is not actually important**.
4654 # We target result columns **positionally** for a fully
4655 # compiled ``Select()`` object; before 1.4 we needed those
4656 # labels to match in cursor.description etc etc but now it
4657 # really doesn't matter.
4658 # So really, we could set render_with_label True in all cases.
4659 # Or we could just have visit_unary() populate the result map
4660 # in all cases.
4661 #
4662 # What we're doing here is strictly trying to not rock the
4663 # boat too much with when we do/don't render "AS label";
4664 # labels being present helps in the edge cases that we
4665 # "fall back" to named cursor.description matching, labels
4666 # not being present for columns keeps us from having awkward
4667 # phrases like "SELECT DISTINCT table.x AS x".
4668 render_with_label = (
4669 (
4670 # exception case to detect if we render "not boolean"
4671 # as "not <col>" for native boolean or "<col> = 1"
4672 # for non-native boolean. this is controlled by
4673 # visit_is_<true|false>_unary_operator
4674 column.operator
4675 in (operators.is_false, operators.is_true)
4676 and not self.dialect.supports_native_boolean
4677 )
4678 or column._wraps_unnamed_column()
4679 or asfrom
4680 )
4681 elif (
4682 # general class of expressions that don't have a SQL-column
4683 # addressable name. includes scalar selects, bind parameters,
4684 # SQL functions, others
4685 not isinstance(column, elements.NamedColumn)
4686 # deeper check that indicates there's no natural "name" to
4687 # this element, which accommodates for custom SQL constructs
4688 # that might have a ".name" attribute (but aren't SQL
4689 # functions) but are not implementing this more recently added
4690 # base class. in theory the "NamedColumn" check should be
4691 # enough, however here we seek to maintain legacy behaviors
4692 # as well.
4693 and column._non_anon_label is None
4694 ):
4695 render_with_label = True
4696 else:
4697 render_with_label = False
4699 if render_with_label:
4700 if not fallback_label_name:
4701 # used by the RETURNING case right now. we generate it
4702 # here as 3rd party dialects may be referring to
4703 # _label_select_column method directly instead of the
4704 # just-added _label_returning_column method
4705 assert not column_is_repeated
4706 fallback_label_name = column._anon_name_label
4708 fallback_label_name = (
4709 elements._truncated_label(fallback_label_name)
4710 if not isinstance(
4711 fallback_label_name, elements._truncated_label
4712 )
4713 else fallback_label_name
4714 )
4716 result_expr = _CompileLabel(
4717 col_expr, fallback_label_name, alt_names=(proxy_name,)
4718 )
4719 else:
4720 result_expr = col_expr
4722 column_clause_args.update(
4723 within_columns_clause=within_columns_clause,
4724 add_to_result_map=add_to_result_map,
4725 include_table=include_table,
4726 )
4727 return result_expr._compiler_dispatch(self, **column_clause_args)
4729 def format_from_hint_text(self, sqltext, table, hint, iscrud):
4730 hinttext = self.get_from_hint_text(table, hint)
4731 if hinttext:
4732 sqltext += " " + hinttext
4733 return sqltext
4735 def get_select_hint_text(self, byfroms):
4736 return None
4738 def get_from_hint_text(
4739 self, table: FromClause, text: Optional[str]
4740 ) -> Optional[str]:
4741 return None
4743 def get_crud_hint_text(self, table, text):
4744 return None
4746 def get_statement_hint_text(self, hint_texts):
4747 return " ".join(hint_texts)
4749 _default_stack_entry: _CompilerStackEntry
4751 if not typing.TYPE_CHECKING:
4752 _default_stack_entry = util.immutabledict(
4753 [("correlate_froms", frozenset()), ("asfrom_froms", frozenset())]
4754 )
4756 def _display_froms_for_select(
4757 self, select_stmt, asfrom, lateral=False, **kw
4758 ):
4759 # utility method to help external dialects
4760 # get the correct from list for a select.
4761 # specifically the oracle dialect needs this feature
4762 # right now.
4763 toplevel = not self.stack
4764 entry = self._default_stack_entry if toplevel else self.stack[-1]
4766 compile_state = select_stmt._compile_state_factory(select_stmt, self)
4768 correlate_froms = entry["correlate_froms"]
4769 asfrom_froms = entry["asfrom_froms"]
4771 if asfrom and not lateral:
4772 froms = compile_state._get_display_froms(
4773 explicit_correlate_froms=correlate_froms.difference(
4774 asfrom_froms
4775 ),
4776 implicit_correlate_froms=(),
4777 )
4778 else:
4779 froms = compile_state._get_display_froms(
4780 explicit_correlate_froms=correlate_froms,
4781 implicit_correlate_froms=asfrom_froms,
4782 )
4783 return froms
4785 translate_select_structure: Any = None
4786 """if not ``None``, should be a callable which accepts ``(select_stmt,
4787 **kw)`` and returns a select object. this is used for structural changes
4788 mostly to accommodate for LIMIT/OFFSET schemes
4790 """
4792 def visit_select(
4793 self,
4794 select_stmt,
4795 asfrom=False,
4796 insert_into=False,
4797 fromhints=None,
4798 compound_index=None,
4799 select_wraps_for=None,
4800 lateral=False,
4801 from_linter=None,
4802 **kwargs,
4803 ):
4804 assert select_wraps_for is None, (
4805 "SQLAlchemy 1.4 requires use of "
4806 "the translate_select_structure hook for structural "
4807 "translations of SELECT objects"
4808 )
4810 # initial setup of SELECT. the compile_state_factory may now
4811 # be creating a totally different SELECT from the one that was
4812 # passed in. for ORM use this will convert from an ORM-state
4813 # SELECT to a regular "Core" SELECT. other composed operations
4814 # such as computation of joins will be performed.
4816 kwargs["within_columns_clause"] = False
4818 compile_state = select_stmt._compile_state_factory(
4819 select_stmt, self, **kwargs
4820 )
4821 kwargs["ambiguous_table_name_map"] = (
4822 compile_state._ambiguous_table_name_map
4823 )
4825 select_stmt = compile_state.statement
4827 toplevel = not self.stack
4829 if toplevel and not self.compile_state:
4830 self.compile_state = compile_state
4832 is_embedded_select = compound_index is not None or insert_into
4834 # translate step for Oracle, SQL Server which often need to
4835 # restructure the SELECT to allow for LIMIT/OFFSET and possibly
4836 # other conditions
4837 if self.translate_select_structure:
4838 new_select_stmt = self.translate_select_structure(
4839 select_stmt, asfrom=asfrom, **kwargs
4840 )
4842 # if SELECT was restructured, maintain a link to the originals
4843 # and assemble a new compile state
4844 if new_select_stmt is not select_stmt:
4845 compile_state_wraps_for = compile_state
4846 select_wraps_for = select_stmt
4847 select_stmt = new_select_stmt
4849 compile_state = select_stmt._compile_state_factory(
4850 select_stmt, self, **kwargs
4851 )
4852 select_stmt = compile_state.statement
4854 entry = self._default_stack_entry if toplevel else self.stack[-1]
4856 populate_result_map = need_column_expressions = (
4857 toplevel
4858 or entry.get("need_result_map_for_compound", False)
4859 or entry.get("need_result_map_for_nested", False)
4860 )
4862 # indicates there is a CompoundSelect in play and we are not the
4863 # first select
4864 if compound_index:
4865 populate_result_map = False
4867 # this was first proposed as part of #3372; however, it is not
4868 # reached in current tests and could possibly be an assertion
4869 # instead.
4870 if not populate_result_map and "add_to_result_map" in kwargs:
4871 del kwargs["add_to_result_map"]
4873 froms = self._setup_select_stack(
4874 select_stmt, compile_state, entry, asfrom, lateral, compound_index
4875 )
4877 column_clause_args = kwargs.copy()
4878 column_clause_args.update(
4879 {"within_label_clause": False, "within_columns_clause": False}
4880 )
4882 text = "SELECT " # we're off to a good start !
4884 if select_stmt._hints:
4885 hint_text, byfrom = self._setup_select_hints(select_stmt)
4886 if hint_text:
4887 text += hint_text + " "
4888 else:
4889 byfrom = None
4891 if select_stmt._independent_ctes:
4892 self._dispatch_independent_ctes(select_stmt, kwargs)
4894 if select_stmt._prefixes:
4895 text += self._generate_prefixes(
4896 select_stmt, select_stmt._prefixes, **kwargs
4897 )
4899 text += self.get_select_precolumns(select_stmt, **kwargs)
4900 # the actual list of columns to print in the SELECT column list.
4901 inner_columns = [
4902 c
4903 for c in [
4904 self._label_select_column(
4905 select_stmt,
4906 column,
4907 populate_result_map,
4908 asfrom,
4909 column_clause_args,
4910 name=name,
4911 proxy_name=proxy_name,
4912 fallback_label_name=fallback_label_name,
4913 column_is_repeated=repeated,
4914 need_column_expressions=need_column_expressions,
4915 )
4916 for (
4917 name,
4918 proxy_name,
4919 fallback_label_name,
4920 column,
4921 repeated,
4922 ) in compile_state.columns_plus_names
4923 ]
4924 if c is not None
4925 ]
4927 if populate_result_map and select_wraps_for is not None:
4928 # if this select was generated from translate_select,
4929 # rewrite the targeted columns in the result map
4931 translate = dict(
4932 zip(
4933 [
4934 name
4935 for (
4936 key,
4937 proxy_name,
4938 fallback_label_name,
4939 name,
4940 repeated,
4941 ) in compile_state.columns_plus_names
4942 ],
4943 [
4944 name
4945 for (
4946 key,
4947 proxy_name,
4948 fallback_label_name,
4949 name,
4950 repeated,
4951 ) in compile_state_wraps_for.columns_plus_names
4952 ],
4953 )
4954 )
4956 self._result_columns = [
4957 ResultColumnsEntry(
4958 key, name, tuple(translate.get(o, o) for o in obj), type_
4959 )
4960 for key, name, obj, type_ in self._result_columns
4961 ]
4963 text = self._compose_select_body(
4964 text,
4965 select_stmt,
4966 compile_state,
4967 inner_columns,
4968 froms,
4969 byfrom,
4970 toplevel,
4971 kwargs,
4972 )
4974 if select_stmt._statement_hints:
4975 per_dialect = [
4976 ht
4977 for (dialect_name, ht) in select_stmt._statement_hints
4978 if dialect_name in ("*", self.dialect.name)
4979 ]
4980 if per_dialect:
4981 text += " " + self.get_statement_hint_text(per_dialect)
4983 # In compound query, CTEs are shared at the compound level
4984 if self.ctes and (not is_embedded_select or toplevel):
4985 nesting_level = len(self.stack) if not toplevel else None
4986 text = self._render_cte_clause(nesting_level=nesting_level) + text
4988 if select_stmt._suffixes:
4989 text += " " + self._generate_prefixes(
4990 select_stmt, select_stmt._suffixes, **kwargs
4991 )
4993 self.stack.pop(-1)
4995 return text
4997 def _setup_select_hints(
4998 self, select: Select[Any]
4999 ) -> Tuple[str, _FromHintsType]:
5000 byfrom = {
5001 from_: hinttext
5002 % {"name": from_._compiler_dispatch(self, ashint=True)}
5003 for (from_, dialect), hinttext in select._hints.items()
5004 if dialect in ("*", self.dialect.name)
5005 }
5006 hint_text = self.get_select_hint_text(byfrom)
5007 return hint_text, byfrom
5009 def _setup_select_stack(
5010 self, select, compile_state, entry, asfrom, lateral, compound_index
5011 ):
5012 correlate_froms = entry["correlate_froms"]
5013 asfrom_froms = entry["asfrom_froms"]
5015 if compound_index == 0:
5016 entry["select_0"] = select
5017 elif compound_index:
5018 select_0 = entry["select_0"]
5019 numcols = len(select_0._all_selected_columns)
5021 if len(compile_state.columns_plus_names) != numcols:
5022 raise exc.CompileError(
5023 "All selectables passed to "
5024 "CompoundSelect must have identical numbers of "
5025 "columns; select #%d has %d columns, select "
5026 "#%d has %d"
5027 % (
5028 1,
5029 numcols,
5030 compound_index + 1,
5031 len(select._all_selected_columns),
5032 )
5033 )
5035 if asfrom and not lateral:
5036 froms = compile_state._get_display_froms(
5037 explicit_correlate_froms=correlate_froms.difference(
5038 asfrom_froms
5039 ),
5040 implicit_correlate_froms=(),
5041 )
5042 else:
5043 froms = compile_state._get_display_froms(
5044 explicit_correlate_froms=correlate_froms,
5045 implicit_correlate_froms=asfrom_froms,
5046 )
5048 new_correlate_froms = set(_from_objects(*froms))
5049 all_correlate_froms = new_correlate_froms.union(correlate_froms)
5051 new_entry: _CompilerStackEntry = {
5052 "asfrom_froms": new_correlate_froms,
5053 "correlate_froms": all_correlate_froms,
5054 "selectable": select,
5055 "compile_state": compile_state,
5056 }
5057 self.stack.append(new_entry)
5059 return froms
5061 def _compose_select_body(
5062 self,
5063 text,
5064 select,
5065 compile_state,
5066 inner_columns,
5067 froms,
5068 byfrom,
5069 toplevel,
5070 kwargs,
5071 ):
5072 text += ", ".join(inner_columns)
5074 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
5075 from_linter = FromLinter({}, set())
5076 warn_linting = self.linting & WARN_LINTING
5077 if toplevel:
5078 self.from_linter = from_linter
5079 else:
5080 from_linter = None
5081 warn_linting = False
5083 # adjust the whitespace for no inner columns, part of #9440,
5084 # so that a no-col SELECT comes out as "SELECT WHERE..." or
5085 # "SELECT FROM ...".
5086 # while it would be better to have built the SELECT starting string
5087 # without trailing whitespace first, then add whitespace only if inner
5088 # cols were present, this breaks compatibility with various custom
5089 # compilation schemes that are currently being tested.
5090 if not inner_columns:
5091 text = text.rstrip()
5093 if froms:
5094 text += " \nFROM "
5096 if select._hints:
5097 text += ", ".join(
5098 [
5099 f._compiler_dispatch(
5100 self,
5101 asfrom=True,
5102 fromhints=byfrom,
5103 from_linter=from_linter,
5104 **kwargs,
5105 )
5106 for f in froms
5107 ]
5108 )
5109 else:
5110 text += ", ".join(
5111 [
5112 f._compiler_dispatch(
5113 self,
5114 asfrom=True,
5115 from_linter=from_linter,
5116 **kwargs,
5117 )
5118 for f in froms
5119 ]
5120 )
5121 else:
5122 text += self.default_from()
5124 if select._where_criteria:
5125 t = self._generate_delimited_and_list(
5126 select._where_criteria, from_linter=from_linter, **kwargs
5127 )
5128 if t:
5129 text += " \nWHERE " + t
5131 if warn_linting:
5132 assert from_linter is not None
5133 from_linter.warn()
5135 if select._group_by_clauses:
5136 text += self.group_by_clause(select, **kwargs)
5138 if select._having_criteria:
5139 t = self._generate_delimited_and_list(
5140 select._having_criteria, **kwargs
5141 )
5142 if t:
5143 text += " \nHAVING " + t
5145 if select._order_by_clauses:
5146 text += self.order_by_clause(select, **kwargs)
5148 if select._has_row_limiting_clause:
5149 text += self._row_limit_clause(select, **kwargs)
5151 if select._for_update_arg is not None:
5152 text += self.for_update_clause(select, **kwargs)
5154 return text
5156 def _generate_prefixes(self, stmt, prefixes, **kw):
5157 clause = " ".join(
5158 prefix._compiler_dispatch(self, **kw)
5159 for prefix, dialect_name in prefixes
5160 if dialect_name in (None, "*") or dialect_name == self.dialect.name
5161 )
5162 if clause:
5163 clause += " "
5164 return clause
5166 def _render_cte_clause(
5167 self,
5168 nesting_level=None,
5169 include_following_stack=False,
5170 ):
5171 """
5172 include_following_stack
5173 Also render the nesting CTEs on the next stack. Useful for
5174 SQL structures like UNION or INSERT that can wrap SELECT
5175 statements containing nesting CTEs.
5176 """
5177 if not self.ctes:
5178 return ""
5180 ctes: MutableMapping[CTE, str]
5182 if nesting_level and nesting_level > 1:
5183 ctes = util.OrderedDict()
5184 for cte in list(self.ctes.keys()):
5185 cte_level, cte_name, cte_opts = self.level_name_by_cte[
5186 cte._get_reference_cte()
5187 ]
5188 nesting = cte.nesting or cte_opts.nesting
5189 is_rendered_level = cte_level == nesting_level or (
5190 include_following_stack and cte_level == nesting_level + 1
5191 )
5192 if not (nesting and is_rendered_level):
5193 continue
5195 ctes[cte] = self.ctes[cte]
5197 else:
5198 ctes = self.ctes
5200 if not ctes:
5201 return ""
5202 ctes_recursive = any([cte.recursive for cte in ctes])
5204 cte_text = self.get_cte_preamble(ctes_recursive) + " "
5205 cte_text += ", \n".join([txt for txt in ctes.values()])
5206 cte_text += "\n "
5208 if nesting_level and nesting_level > 1:
5209 for cte in list(ctes.keys()):
5210 cte_level, cte_name, cte_opts = self.level_name_by_cte[
5211 cte._get_reference_cte()
5212 ]
5213 del self.ctes[cte]
5214 del self.ctes_by_level_name[(cte_level, cte_name)]
5215 del self.level_name_by_cte[cte._get_reference_cte()]
5217 return cte_text
5219 def get_cte_preamble(self, recursive):
5220 if recursive:
5221 return "WITH RECURSIVE"
5222 else:
5223 return "WITH"
5225 def get_select_precolumns(self, select: Select[Any], **kw: Any) -> str:
5226 """Called when building a ``SELECT`` statement, position is just
5227 before column list.
5229 """
5230 if select._distinct_on:
5231 util.warn_deprecated(
5232 "DISTINCT ON is currently supported only by the PostgreSQL "
5233 "dialect. Use of DISTINCT ON for other backends is currently "
5234 "silently ignored, however this usage is deprecated, and will "
5235 "raise CompileError in a future release for all backends "
5236 "that do not support this syntax.",
5237 version="1.4",
5238 )
5239 return "DISTINCT " if select._distinct else ""
5241 def group_by_clause(self, select, **kw):
5242 """allow dialects to customize how GROUP BY is rendered."""
5244 group_by = self._generate_delimited_list(
5245 select._group_by_clauses, OPERATORS[operators.comma_op], **kw
5246 )
5247 if group_by:
5248 return " GROUP BY " + group_by
5249 else:
5250 return ""
5252 def order_by_clause(self, select, **kw):
5253 """allow dialects to customize how ORDER BY is rendered."""
5255 order_by = self._generate_delimited_list(
5256 select._order_by_clauses, OPERATORS[operators.comma_op], **kw
5257 )
5259 if order_by:
5260 return " ORDER BY " + order_by
5261 else:
5262 return ""
5264 def for_update_clause(self, select, **kw):
5265 return " FOR UPDATE"
5267 def returning_clause(
5268 self,
5269 stmt: UpdateBase,
5270 returning_cols: Sequence[_ColumnsClauseElement],
5271 *,
5272 populate_result_map: bool,
5273 **kw: Any,
5274 ) -> str:
5275 columns = [
5276 self._label_returning_column(
5277 stmt,
5278 column,
5279 populate_result_map,
5280 fallback_label_name=fallback_label_name,
5281 column_is_repeated=repeated,
5282 name=name,
5283 proxy_name=proxy_name,
5284 **kw,
5285 )
5286 for (
5287 name,
5288 proxy_name,
5289 fallback_label_name,
5290 column,
5291 repeated,
5292 ) in stmt._generate_columns_plus_names(
5293 True, cols=base._select_iterables(returning_cols)
5294 )
5295 ]
5297 return "RETURNING " + ", ".join(columns)
5299 def limit_clause(self, select, **kw):
5300 text = ""
5301 if select._limit_clause is not None:
5302 text += "\n LIMIT " + self.process(select._limit_clause, **kw)
5303 if select._offset_clause is not None:
5304 if select._limit_clause is None:
5305 text += "\n LIMIT -1"
5306 text += " OFFSET " + self.process(select._offset_clause, **kw)
5307 return text
5309 def fetch_clause(
5310 self,
5311 select,
5312 fetch_clause=None,
5313 require_offset=False,
5314 use_literal_execute_for_simple_int=False,
5315 **kw,
5316 ):
5317 if fetch_clause is None:
5318 fetch_clause = select._fetch_clause
5319 fetch_clause_options = select._fetch_clause_options
5320 else:
5321 fetch_clause_options = {"percent": False, "with_ties": False}
5323 text = ""
5325 if select._offset_clause is not None:
5326 offset_clause = select._offset_clause
5327 if (
5328 use_literal_execute_for_simple_int
5329 and select._simple_int_clause(offset_clause)
5330 ):
5331 offset_clause = offset_clause.render_literal_execute()
5332 offset_str = self.process(offset_clause, **kw)
5333 text += "\n OFFSET %s ROWS" % offset_str
5334 elif require_offset:
5335 text += "\n OFFSET 0 ROWS"
5337 if fetch_clause is not None:
5338 if (
5339 use_literal_execute_for_simple_int
5340 and select._simple_int_clause(fetch_clause)
5341 ):
5342 fetch_clause = fetch_clause.render_literal_execute()
5343 text += "\n FETCH FIRST %s%s ROWS %s" % (
5344 self.process(fetch_clause, **kw),
5345 " PERCENT" if fetch_clause_options["percent"] else "",
5346 "WITH TIES" if fetch_clause_options["with_ties"] else "ONLY",
5347 )
5348 return text
5350 def visit_table(
5351 self,
5352 table,
5353 asfrom=False,
5354 iscrud=False,
5355 ashint=False,
5356 fromhints=None,
5357 use_schema=True,
5358 from_linter=None,
5359 ambiguous_table_name_map=None,
5360 enclosing_alias=None,
5361 **kwargs,
5362 ):
5363 if from_linter:
5364 from_linter.froms[table] = table.fullname
5366 if asfrom or ashint:
5367 effective_schema = self.preparer.schema_for_object(table)
5369 if use_schema and effective_schema:
5370 ret = (
5371 self.preparer.quote_schema(effective_schema)
5372 + "."
5373 + self.preparer.quote(table.name)
5374 )
5375 else:
5376 ret = self.preparer.quote(table.name)
5378 if (
5379 (
5380 enclosing_alias is None
5381 or enclosing_alias.element is not table
5382 )
5383 and not effective_schema
5384 and ambiguous_table_name_map
5385 and table.name in ambiguous_table_name_map
5386 ):
5387 anon_name = self._truncated_identifier(
5388 "alias", ambiguous_table_name_map[table.name]
5389 )
5391 ret = ret + self.get_render_as_alias_suffix(
5392 self.preparer.format_alias(None, anon_name)
5393 )
5395 if fromhints and table in fromhints:
5396 ret = self.format_from_hint_text(
5397 ret, table, fromhints[table], iscrud
5398 )
5399 return ret
5400 else:
5401 return ""
5403 def visit_join(self, join, asfrom=False, from_linter=None, **kwargs):
5404 if from_linter:
5405 from_linter.edges.update(
5406 itertools.product(
5407 _de_clone(join.left._from_objects),
5408 _de_clone(join.right._from_objects),
5409 )
5410 )
5412 if join.full:
5413 join_type = " FULL OUTER JOIN "
5414 elif join.isouter:
5415 join_type = " LEFT OUTER JOIN "
5416 else:
5417 join_type = " JOIN "
5418 return (
5419 join.left._compiler_dispatch(
5420 self, asfrom=True, from_linter=from_linter, **kwargs
5421 )
5422 + join_type
5423 + join.right._compiler_dispatch(
5424 self, asfrom=True, from_linter=from_linter, **kwargs
5425 )
5426 + " ON "
5427 # TODO: likely need asfrom=True here?
5428 + join.onclause._compiler_dispatch(
5429 self, from_linter=from_linter, **kwargs
5430 )
5431 )
5433 def _setup_crud_hints(self, stmt, table_text):
5434 dialect_hints = {
5435 table: hint_text
5436 for (table, dialect), hint_text in stmt._hints.items()
5437 if dialect in ("*", self.dialect.name)
5438 }
5439 if stmt.table in dialect_hints:
5440 table_text = self.format_from_hint_text(
5441 table_text, stmt.table, dialect_hints[stmt.table], True
5442 )
5443 return dialect_hints, table_text
5445 # within the realm of "insertmanyvalues sentinel columns",
5446 # these lookups match different kinds of Column() configurations
5447 # to specific backend capabilities. they are broken into two
5448 # lookups, one for autoincrement columns and the other for non
5449 # autoincrement columns
5450 _sentinel_col_non_autoinc_lookup = util.immutabledict(
5451 {
5452 _SentinelDefaultCharacterization.CLIENTSIDE: (
5453 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5454 ),
5455 _SentinelDefaultCharacterization.SENTINEL_DEFAULT: (
5456 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5457 ),
5458 _SentinelDefaultCharacterization.NONE: (
5459 InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT
5460 ),
5461 _SentinelDefaultCharacterization.IDENTITY: (
5462 InsertmanyvaluesSentinelOpts.IDENTITY
5463 ),
5464 _SentinelDefaultCharacterization.SEQUENCE: (
5465 InsertmanyvaluesSentinelOpts.SEQUENCE
5466 ),
5467 }
5468 )
5469 _sentinel_col_autoinc_lookup = _sentinel_col_non_autoinc_lookup.union(
5470 {
5471 _SentinelDefaultCharacterization.NONE: (
5472 InsertmanyvaluesSentinelOpts.AUTOINCREMENT
5473 ),
5474 }
5475 )
5477 def _get_sentinel_column_for_table(
5478 self, table: Table
5479 ) -> Optional[Sequence[Column[Any]]]:
5480 """given a :class:`.Table`, return a usable sentinel column or
5481 columns for this dialect if any.
5483 Return None if no sentinel columns could be identified, or raise an
5484 error if a column was marked as a sentinel explicitly but isn't
5485 compatible with this dialect.
5487 """
5489 sentinel_opts = self.dialect.insertmanyvalues_implicit_sentinel
5490 sentinel_characteristics = table._sentinel_column_characteristics
5492 sent_cols = sentinel_characteristics.columns
5494 if sent_cols is None:
5495 return None
5497 if sentinel_characteristics.is_autoinc:
5498 bitmask = self._sentinel_col_autoinc_lookup.get(
5499 sentinel_characteristics.default_characterization, 0
5500 )
5501 else:
5502 bitmask = self._sentinel_col_non_autoinc_lookup.get(
5503 sentinel_characteristics.default_characterization, 0
5504 )
5506 if sentinel_opts & bitmask:
5507 return sent_cols
5509 if sentinel_characteristics.is_explicit:
5510 # a column was explicitly marked as insert_sentinel=True,
5511 # however it is not compatible with this dialect. they should
5512 # not indicate this column as a sentinel if they need to include
5513 # this dialect.
5515 # TODO: do we want non-primary key explicit sentinel cols
5516 # that can gracefully degrade for some backends?
5517 # insert_sentinel="degrade" perhaps. not for the initial release.
5518 # I am hoping people are generally not dealing with this sentinel
5519 # business at all.
5521 # if is_explicit is True, there will be only one sentinel column.
5523 raise exc.InvalidRequestError(
5524 f"Column {sent_cols[0]} can't be explicitly "
5525 "marked as a sentinel column when using the "
5526 f"{self.dialect.name} dialect, as the "
5527 "particular type of default generation on this column is "
5528 "not currently compatible with this dialect's specific "
5529 f"INSERT..RETURNING syntax which can receive the "
5530 "server-generated value in "
5531 "a deterministic way. To remove this error, remove "
5532 "insert_sentinel=True from primary key autoincrement "
5533 "columns; these columns are automatically used as "
5534 "sentinels for supported dialects in any case."
5535 )
5537 return None
5539 def _deliver_insertmanyvalues_batches(
5540 self,
5541 statement: str,
5542 parameters: _DBAPIMultiExecuteParams,
5543 compiled_parameters: List[_MutableCoreSingleExecuteParams],
5544 generic_setinputsizes: Optional[_GenericSetInputSizesType],
5545 batch_size: int,
5546 sort_by_parameter_order: bool,
5547 schema_translate_map: Optional[SchemaTranslateMapType],
5548 ) -> Iterator[_InsertManyValuesBatch]:
5549 imv = self._insertmanyvalues
5550 assert imv is not None
5552 if not imv.sentinel_param_keys:
5553 _sentinel_from_params = None
5554 else:
5555 _sentinel_from_params = operator.itemgetter(
5556 *imv.sentinel_param_keys
5557 )
5559 lenparams = len(parameters)
5560 if imv.is_default_expr and not self.dialect.supports_default_metavalue:
5561 # backend doesn't support
5562 # INSERT INTO table (pk_col) VALUES (DEFAULT), (DEFAULT), ...
5563 # at the moment this is basically SQL Server due to
5564 # not being able to use DEFAULT for identity column
5565 # just yield out that many single statements! still
5566 # faster than a whole connection.execute() call ;)
5567 #
5568 # note we still are taking advantage of the fact that we know
5569 # we are using RETURNING. The generalized approach of fetching
5570 # cursor.lastrowid etc. still goes through the more heavyweight
5571 # "ExecutionContext per statement" system as it isn't usable
5572 # as a generic "RETURNING" approach
5573 use_row_at_a_time = True
5574 downgraded = False
5575 elif not self.dialect.supports_multivalues_insert or (
5576 sort_by_parameter_order
5577 and self._result_columns
5578 and (
5579 imv.sentinel_columns is None
5580 or (
5581 imv.includes_upsert_behaviors
5582 and not imv.embed_values_counter
5583 )
5584 )
5585 ):
5586 # deterministic order was requested and the compiler could
5587 # not organize sentinel columns for this dialect/statement.
5588 # use row at a time. Note: if embed_values_counter is True,
5589 # the counter itself provides the ordering capability we need,
5590 # so we can use batch mode even with upsert behaviors.
5591 use_row_at_a_time = True
5592 downgraded = True
5593 elif (
5594 imv.has_upsert_bound_parameters
5595 and not imv.embed_values_counter
5596 and self._result_columns
5597 ):
5598 # For upsert behaviors (ON CONFLICT DO UPDATE, etc.) with RETURNING
5599 # and parametrized bindparams in the SET clause, we must use
5600 # row-at-a-time. Batching multiple rows in a single statement
5601 # doesn't work when the SET clause contains bound parameters that
5602 # will receive different values per row, as there's only one SET
5603 # clause per statement. See issue #13130.
5604 use_row_at_a_time = True
5605 downgraded = True
5606 else:
5607 use_row_at_a_time = False
5608 downgraded = False
5610 if use_row_at_a_time:
5611 for batchnum, (param, compiled_param) in enumerate(
5612 cast(
5613 "Sequence[Tuple[_DBAPISingleExecuteParams, _MutableCoreSingleExecuteParams]]", # noqa: E501
5614 zip(parameters, compiled_parameters),
5615 ),
5616 1,
5617 ):
5618 yield _InsertManyValuesBatch(
5619 statement,
5620 param,
5621 generic_setinputsizes,
5622 [param],
5623 (
5624 [_sentinel_from_params(compiled_param)]
5625 if _sentinel_from_params
5626 else []
5627 ),
5628 1,
5629 batchnum,
5630 lenparams,
5631 sort_by_parameter_order,
5632 downgraded,
5633 )
5634 return
5636 if schema_translate_map:
5637 rst = functools.partial(
5638 self.preparer._render_schema_translates,
5639 schema_translate_map=schema_translate_map,
5640 )
5641 else:
5642 rst = None
5644 imv_single_values_expr = imv.single_values_expr
5645 if rst:
5646 imv_single_values_expr = rst(imv_single_values_expr)
5648 executemany_values = f"({imv_single_values_expr})"
5649 statement = statement.replace(executemany_values, "__EXECMANY_TOKEN__")
5651 # Use optional insertmanyvalues_max_parameters
5652 # to further shrink the batch size so that there are no more than
5653 # insertmanyvalues_max_parameters params.
5654 # Currently used by SQL Server, which limits statements to 2100 bound
5655 # parameters (actually 2099).
5656 max_params = self.dialect.insertmanyvalues_max_parameters
5657 if max_params:
5658 total_num_of_params = len(self.bind_names)
5659 num_params_per_batch = len(imv.insert_crud_params)
5660 num_params_outside_of_batch = (
5661 total_num_of_params - num_params_per_batch
5662 )
5663 batch_size = min(
5664 batch_size,
5665 (
5666 (max_params - num_params_outside_of_batch)
5667 // num_params_per_batch
5668 ),
5669 )
5671 batches = cast("List[Sequence[Any]]", list(parameters))
5672 compiled_batches = cast(
5673 "List[Sequence[Any]]", list(compiled_parameters)
5674 )
5676 processed_setinputsizes: Optional[_GenericSetInputSizesType] = None
5677 batchnum = 1
5678 total_batches = lenparams // batch_size + (
5679 1 if lenparams % batch_size else 0
5680 )
5682 insert_crud_params = imv.insert_crud_params
5683 assert insert_crud_params is not None
5685 if rst:
5686 insert_crud_params = [
5687 (col, key, rst(expr), st)
5688 for col, key, expr, st in insert_crud_params
5689 ]
5691 escaped_bind_names: Mapping[str, str]
5692 expand_pos_lower_index = expand_pos_upper_index = 0
5694 if not self.positional:
5695 if self.escaped_bind_names:
5696 escaped_bind_names = self.escaped_bind_names
5697 else:
5698 escaped_bind_names = {}
5700 all_keys = set(parameters[0])
5702 def apply_placeholders(keys, formatted):
5703 for key in keys:
5704 key = escaped_bind_names.get(key, key)
5705 formatted = formatted.replace(
5706 self.bindtemplate % {"name": key},
5707 self.bindtemplate
5708 % {"name": f"{key}__EXECMANY_INDEX__"},
5709 )
5710 return formatted
5712 if imv.embed_values_counter:
5713 imv_values_counter = ", _IMV_VALUES_COUNTER"
5714 else:
5715 imv_values_counter = ""
5716 formatted_values_clause = f"""({', '.join(
5717 apply_placeholders(bind_keys, formatted)
5718 for _, _, formatted, bind_keys in insert_crud_params
5719 )}{imv_values_counter})"""
5721 keys_to_replace = all_keys.intersection(
5722 escaped_bind_names.get(key, key)
5723 for _, _, _, bind_keys in insert_crud_params
5724 for key in bind_keys
5725 )
5726 base_parameters = {
5727 key: parameters[0][key]
5728 for key in all_keys.difference(keys_to_replace)
5729 }
5731 executemany_values_w_comma = ""
5732 else:
5733 formatted_values_clause = ""
5734 keys_to_replace = set()
5735 base_parameters = {}
5737 if imv.embed_values_counter:
5738 executemany_values_w_comma = (
5739 f"({imv_single_values_expr}, _IMV_VALUES_COUNTER), "
5740 )
5741 else:
5742 executemany_values_w_comma = f"({imv_single_values_expr}), "
5744 all_names_we_will_expand: Set[str] = set()
5745 for elem in imv.insert_crud_params:
5746 all_names_we_will_expand.update(elem[3])
5748 # get the start and end position in a particular list
5749 # of parameters where we will be doing the "expanding".
5750 # statements can have params on either side or both sides,
5751 # given RETURNING and CTEs
5752 if all_names_we_will_expand:
5753 positiontup = self.positiontup
5754 assert positiontup is not None
5756 all_expand_positions = {
5757 idx
5758 for idx, name in enumerate(positiontup)
5759 if name in all_names_we_will_expand
5760 }
5761 expand_pos_lower_index = min(all_expand_positions)
5762 expand_pos_upper_index = max(all_expand_positions) + 1
5763 assert (
5764 len(all_expand_positions)
5765 == expand_pos_upper_index - expand_pos_lower_index
5766 )
5768 if self._numeric_binds:
5769 escaped = re.escape(self._numeric_binds_identifier_char)
5770 executemany_values_w_comma = re.sub(
5771 rf"{escaped}\d+", "%s", executemany_values_w_comma
5772 )
5774 while batches:
5775 batch = batches[0:batch_size]
5776 compiled_batch = compiled_batches[0:batch_size]
5778 batches[0:batch_size] = []
5779 compiled_batches[0:batch_size] = []
5781 if batches:
5782 current_batch_size = batch_size
5783 else:
5784 current_batch_size = len(batch)
5786 if generic_setinputsizes:
5787 # if setinputsizes is present, expand this collection to
5788 # suit the batch length as well
5789 # currently this will be mssql+pyodbc for internal dialects
5790 processed_setinputsizes = [
5791 (new_key, len_, typ)
5792 for new_key, len_, typ in (
5793 (f"{key}_{index}", len_, typ)
5794 for index in range(current_batch_size)
5795 for key, len_, typ in generic_setinputsizes
5796 )
5797 ]
5799 replaced_parameters: Any
5800 if self.positional:
5801 num_ins_params = imv.num_positional_params_counted
5803 batch_iterator: Iterable[Sequence[Any]]
5804 extra_params_left: Sequence[Any]
5805 extra_params_right: Sequence[Any]
5807 if num_ins_params == len(batch[0]):
5808 extra_params_left = extra_params_right = ()
5809 batch_iterator = batch
5810 else:
5811 extra_params_left = batch[0][:expand_pos_lower_index]
5812 extra_params_right = batch[0][expand_pos_upper_index:]
5813 batch_iterator = (
5814 b[expand_pos_lower_index:expand_pos_upper_index]
5815 for b in batch
5816 )
5818 if imv.embed_values_counter:
5819 expanded_values_string = (
5820 "".join(
5821 executemany_values_w_comma.replace(
5822 "_IMV_VALUES_COUNTER", str(i)
5823 )
5824 for i, _ in enumerate(batch)
5825 )
5826 )[:-2]
5827 else:
5828 expanded_values_string = (
5829 (executemany_values_w_comma * current_batch_size)
5830 )[:-2]
5832 if self._numeric_binds and num_ins_params > 0:
5833 # numeric will always number the parameters inside of
5834 # VALUES (and thus order self.positiontup) to be higher
5835 # than non-VALUES parameters, no matter where in the
5836 # statement those non-VALUES parameters appear (this is
5837 # ensured in _process_numeric by numbering first all
5838 # params that are not in _values_bindparam)
5839 # therefore all extra params are always
5840 # on the left side and numbered lower than the VALUES
5841 # parameters
5842 assert not extra_params_right
5844 start = expand_pos_lower_index + 1
5845 end = num_ins_params * (current_batch_size) + start
5847 # need to format here, since statement may contain
5848 # unescaped %, while values_string contains just (%s, %s)
5849 positions = tuple(
5850 f"{self._numeric_binds_identifier_char}{i}"
5851 for i in range(start, end)
5852 )
5853 expanded_values_string = expanded_values_string % positions
5855 replaced_statement = statement.replace(
5856 "__EXECMANY_TOKEN__", expanded_values_string
5857 )
5859 replaced_parameters = tuple(
5860 itertools.chain.from_iterable(batch_iterator)
5861 )
5863 replaced_parameters = (
5864 extra_params_left
5865 + replaced_parameters
5866 + extra_params_right
5867 )
5869 else:
5870 replaced_values_clauses = []
5871 replaced_parameters = base_parameters.copy()
5873 for i, param in enumerate(batch):
5874 fmv = formatted_values_clause.replace(
5875 "EXECMANY_INDEX__", str(i)
5876 )
5877 if imv.embed_values_counter:
5878 fmv = fmv.replace("_IMV_VALUES_COUNTER", str(i))
5880 replaced_values_clauses.append(fmv)
5881 replaced_parameters.update(
5882 {f"{key}__{i}": param[key] for key in keys_to_replace}
5883 )
5885 replaced_statement = statement.replace(
5886 "__EXECMANY_TOKEN__",
5887 ", ".join(replaced_values_clauses),
5888 )
5890 yield _InsertManyValuesBatch(
5891 replaced_statement,
5892 replaced_parameters,
5893 processed_setinputsizes,
5894 batch,
5895 (
5896 [_sentinel_from_params(cb) for cb in compiled_batch]
5897 if _sentinel_from_params
5898 else []
5899 ),
5900 current_batch_size,
5901 batchnum,
5902 total_batches,
5903 sort_by_parameter_order,
5904 False,
5905 )
5906 batchnum += 1
5908 def visit_insert(
5909 self, insert_stmt, visited_bindparam=None, visiting_cte=None, **kw
5910 ):
5911 compile_state = insert_stmt._compile_state_factory(
5912 insert_stmt, self, **kw
5913 )
5914 insert_stmt = compile_state.statement
5916 if visiting_cte is not None:
5917 kw["visiting_cte"] = visiting_cte
5918 toplevel = False
5919 else:
5920 toplevel = not self.stack
5922 if toplevel:
5923 self.isinsert = True
5924 if not self.dml_compile_state:
5925 self.dml_compile_state = compile_state
5926 if not self.compile_state:
5927 self.compile_state = compile_state
5929 self.stack.append(
5930 {
5931 "correlate_froms": set(),
5932 "asfrom_froms": set(),
5933 "selectable": insert_stmt,
5934 }
5935 )
5937 counted_bindparam = 0
5939 # reset any incoming "visited_bindparam" collection
5940 visited_bindparam = None
5942 # for positional, insertmanyvalues needs to know how many
5943 # bound parameters are in the VALUES sequence; there's no simple
5944 # rule because default expressions etc. can have zero or more
5945 # params inside them. After multiple attempts to figure this out,
5946 # this very simplistic "count after" works and is
5947 # likely the least amount of callcounts, though looks clumsy
5948 if self.positional and visiting_cte is None:
5949 # if we are inside a CTE, don't count parameters
5950 # here since they won't be for insertmanyvalues. keep
5951 # visited_bindparam at None so no counting happens.
5952 # see #9173
5953 visited_bindparam = []
5955 crud_params_struct = crud._get_crud_params(
5956 self,
5957 insert_stmt,
5958 compile_state,
5959 toplevel,
5960 visited_bindparam=visited_bindparam,
5961 **kw,
5962 )
5964 if self.positional and visited_bindparam is not None:
5965 counted_bindparam = len(visited_bindparam)
5966 if self._numeric_binds:
5967 if self._values_bindparam is not None:
5968 self._values_bindparam += visited_bindparam
5969 else:
5970 self._values_bindparam = visited_bindparam
5972 crud_params_single = crud_params_struct.single_params
5974 if (
5975 not crud_params_single
5976 and not self.dialect.supports_default_values
5977 and not self.dialect.supports_default_metavalue
5978 and not self.dialect.supports_empty_insert
5979 ):
5980 raise exc.CompileError(
5981 "The '%s' dialect with current database "
5982 "version settings does not support empty "
5983 "inserts." % self.dialect.name
5984 )
5986 if compile_state._has_multi_parameters:
5987 if not self.dialect.supports_multivalues_insert:
5988 raise exc.CompileError(
5989 "The '%s' dialect with current database "
5990 "version settings does not support "
5991 "in-place multirow inserts." % self.dialect.name
5992 )
5993 elif (
5994 self.implicit_returning or insert_stmt._returning
5995 ) and insert_stmt._sort_by_parameter_order:
5996 raise exc.CompileError(
5997 "RETURNING cannot be deterministically sorted when "
5998 "using an INSERT which includes multi-row values()."
5999 )
6000 crud_params_single = crud_params_struct.single_params
6001 else:
6002 crud_params_single = crud_params_struct.single_params
6004 preparer = self.preparer
6005 supports_default_values = self.dialect.supports_default_values
6007 text = "INSERT "
6009 if insert_stmt._prefixes:
6010 text += self._generate_prefixes(
6011 insert_stmt, insert_stmt._prefixes, **kw
6012 )
6014 text += "INTO "
6015 table_text = preparer.format_table(insert_stmt.table)
6017 if insert_stmt._hints:
6018 _, table_text = self._setup_crud_hints(insert_stmt, table_text)
6020 if insert_stmt._independent_ctes:
6021 self._dispatch_independent_ctes(insert_stmt, kw)
6023 text += table_text
6025 if crud_params_single or not supports_default_values:
6026 text += " (%s)" % ", ".join(
6027 [expr for _, expr, _, _ in crud_params_single]
6028 )
6030 # look for insertmanyvalues attributes that would have been configured
6031 # by crud.py as it scanned through the columns to be part of the
6032 # INSERT
6033 use_insertmanyvalues = crud_params_struct.use_insertmanyvalues
6034 named_sentinel_params: Optional[Sequence[str]] = None
6035 add_sentinel_cols = None
6036 implicit_sentinel = False
6038 returning_cols = self.implicit_returning or insert_stmt._returning
6039 if returning_cols:
6040 add_sentinel_cols = crud_params_struct.use_sentinel_columns
6041 if add_sentinel_cols is not None:
6042 assert use_insertmanyvalues
6044 # search for the sentinel column explicitly present
6045 # in the INSERT columns list, and additionally check that
6046 # this column has a bound parameter name set up that's in the
6047 # parameter list. If both of these cases are present, it means
6048 # we will have a client side value for the sentinel in each
6049 # parameter set.
6051 _params_by_col = {
6052 col: param_names
6053 for col, _, _, param_names in crud_params_single
6054 }
6055 named_sentinel_params = []
6056 for _add_sentinel_col in add_sentinel_cols:
6057 if _add_sentinel_col not in _params_by_col:
6058 named_sentinel_params = None
6059 break
6060 param_name = self._within_exec_param_key_getter(
6061 _add_sentinel_col
6062 )
6063 if param_name not in _params_by_col[_add_sentinel_col]:
6064 named_sentinel_params = None
6065 break
6066 named_sentinel_params.append(param_name)
6068 if named_sentinel_params is None:
6069 # if we are not going to have a client side value for
6070 # the sentinel in the parameter set, that means it's
6071 # an autoincrement, an IDENTITY, or a server-side SQL
6072 # expression like nextval('seqname'). So this is
6073 # an "implicit" sentinel; we will look for it in
6074 # RETURNING
6075 # only, and then sort on it. For this case on PG,
6076 # SQL Server we have to use a special INSERT form
6077 # that guarantees the server side function lines up with
6078 # the entries in the VALUES.
6079 if (
6080 self.dialect.insertmanyvalues_implicit_sentinel
6081 & InsertmanyvaluesSentinelOpts.ANY_AUTOINCREMENT
6082 ):
6083 implicit_sentinel = True
6084 else:
6085 # here, we are not using a sentinel at all
6086 # and we are likely the SQLite dialect.
6087 # The first add_sentinel_col that we have should not
6088 # be marked as "insert_sentinel=True". if it was,
6089 # an error should have been raised in
6090 # _get_sentinel_column_for_table.
6091 assert not add_sentinel_cols[0]._insert_sentinel, (
6092 "sentinel selection rules should have prevented "
6093 "us from getting here for this dialect"
6094 )
6096 # always put the sentinel columns last. even if they are
6097 # in the returning list already, they will be there twice
6098 # then.
6099 returning_cols = list(returning_cols) + list(add_sentinel_cols)
6101 returning_clause = self.returning_clause(
6102 insert_stmt,
6103 returning_cols,
6104 populate_result_map=toplevel,
6105 )
6107 if self.returning_precedes_values:
6108 text += " " + returning_clause
6110 else:
6111 returning_clause = None
6113 if insert_stmt.select is not None:
6114 # placed here by crud.py
6115 select_text = self.process(
6116 self.stack[-1]["insert_from_select"], insert_into=True, **kw
6117 )
6119 if self.ctes and self.dialect.cte_follows_insert:
6120 nesting_level = len(self.stack) if not toplevel else None
6121 text += " %s%s" % (
6122 self._render_cte_clause(
6123 nesting_level=nesting_level,
6124 include_following_stack=True,
6125 ),
6126 select_text,
6127 )
6128 else:
6129 text += " %s" % select_text
6130 elif not crud_params_single and supports_default_values:
6131 text += " DEFAULT VALUES"
6132 if use_insertmanyvalues:
6133 self._insertmanyvalues = _InsertManyValues(
6134 True,
6135 self.dialect.default_metavalue_token,
6136 crud_params_single,
6137 counted_bindparam,
6138 sort_by_parameter_order=(
6139 insert_stmt._sort_by_parameter_order
6140 ),
6141 includes_upsert_behaviors=(
6142 insert_stmt._post_values_clause is not None
6143 ),
6144 sentinel_columns=add_sentinel_cols,
6145 num_sentinel_columns=(
6146 len(add_sentinel_cols) if add_sentinel_cols else 0
6147 ),
6148 implicit_sentinel=implicit_sentinel,
6149 )
6150 elif compile_state._has_multi_parameters:
6151 text += " VALUES %s" % (
6152 ", ".join(
6153 "(%s)"
6154 % (", ".join(value for _, _, value, _ in crud_param_set))
6155 for crud_param_set in crud_params_struct.all_multi_params
6156 ),
6157 )
6158 elif use_insertmanyvalues:
6159 if (
6160 implicit_sentinel
6161 and (
6162 self.dialect.insertmanyvalues_implicit_sentinel
6163 & InsertmanyvaluesSentinelOpts.USE_INSERT_FROM_SELECT
6164 )
6165 # this is checking if we have
6166 # INSERT INTO table (id) VALUES (DEFAULT).
6167 and not (crud_params_struct.is_default_metavalue_only)
6168 ):
6169 # if we have a sentinel column that is server generated,
6170 # then for selected backends render the VALUES list as a
6171 # subquery. This is the orderable form supported by
6172 # PostgreSQL and in fewer cases SQL Server
6173 embed_sentinel_value = True
6175 render_bind_casts = (
6176 self.dialect.insertmanyvalues_implicit_sentinel
6177 & InsertmanyvaluesSentinelOpts.RENDER_SELECT_COL_CASTS
6178 )
6180 add_sentinel_set = add_sentinel_cols or ()
6182 insert_single_values_expr = ", ".join(
6183 [
6184 value
6185 for col, _, value, _ in crud_params_single
6186 if col not in add_sentinel_set
6187 ]
6188 )
6190 colnames = ", ".join(
6191 f"p{i}"
6192 for i, cp in enumerate(crud_params_single)
6193 if cp[0] not in add_sentinel_set
6194 )
6196 if render_bind_casts:
6197 # render casts for the SELECT list. For PG, we are
6198 # already rendering bind casts in the parameter list,
6199 # selectively for the more "tricky" types like ARRAY.
6200 # however, even for the "easy" types, if the parameter
6201 # is NULL for every entry, PG gives up and says
6202 # "it must be TEXT", which fails for other easy types
6203 # like ints. So we cast on this side too.
6204 colnames_w_cast = ", ".join(
6205 (
6206 self.render_bind_cast(
6207 col.type,
6208 col.type._unwrapped_dialect_impl(self.dialect),
6209 f"p{i}",
6210 )
6211 if col not in add_sentinel_set
6212 else expr
6213 )
6214 for i, (col, _, expr, _) in enumerate(
6215 crud_params_single
6216 )
6217 )
6218 else:
6219 colnames_w_cast = ", ".join(
6220 (f"p{i}" if col not in add_sentinel_set else expr)
6221 for i, (col, _, expr, _) in enumerate(
6222 crud_params_single
6223 )
6224 )
6226 insert_crud_params = [
6227 elem
6228 for elem in crud_params_single
6229 if elem[0] not in add_sentinel_set
6230 ]
6232 text += (
6233 f" SELECT {colnames_w_cast} FROM "
6234 f"(VALUES ({insert_single_values_expr})) "
6235 f"AS imp_sen({colnames}, sen_counter) "
6236 "ORDER BY sen_counter"
6237 )
6239 else:
6240 # otherwise, if no sentinel or backend doesn't support
6241 # orderable subquery form, use a plain VALUES list
6242 embed_sentinel_value = False
6243 insert_crud_params = crud_params_single
6244 insert_single_values_expr = ", ".join(
6245 [value for _, _, value, _ in crud_params_single]
6246 )
6248 text += f" VALUES ({insert_single_values_expr})"
6250 self._insertmanyvalues = _InsertManyValues(
6251 is_default_expr=False,
6252 single_values_expr=insert_single_values_expr,
6253 insert_crud_params=insert_crud_params,
6254 num_positional_params_counted=counted_bindparam,
6255 sort_by_parameter_order=(insert_stmt._sort_by_parameter_order),
6256 includes_upsert_behaviors=(
6257 insert_stmt._post_values_clause is not None
6258 ),
6259 sentinel_columns=add_sentinel_cols,
6260 num_sentinel_columns=(
6261 len(add_sentinel_cols) if add_sentinel_cols else 0
6262 ),
6263 sentinel_param_keys=named_sentinel_params,
6264 implicit_sentinel=implicit_sentinel,
6265 embed_values_counter=embed_sentinel_value,
6266 )
6268 else:
6269 insert_single_values_expr = ", ".join(
6270 [value for _, _, value, _ in crud_params_single]
6271 )
6273 text += f" VALUES ({insert_single_values_expr})"
6275 if insert_stmt._post_values_clause is not None:
6276 post_values_clause = self.process(
6277 insert_stmt._post_values_clause, **kw
6278 )
6279 if post_values_clause:
6280 text += " " + post_values_clause
6282 if returning_clause and not self.returning_precedes_values:
6283 text += " " + returning_clause
6285 if self.ctes and not self.dialect.cte_follows_insert:
6286 nesting_level = len(self.stack) if not toplevel else None
6287 text = (
6288 self._render_cte_clause(
6289 nesting_level=nesting_level,
6290 include_following_stack=True,
6291 )
6292 + text
6293 )
6295 self.stack.pop(-1)
6297 return text
6299 def update_limit_clause(self, update_stmt):
6300 """Provide a hook for MySQL to add LIMIT to the UPDATE"""
6301 return None
6303 def delete_limit_clause(self, delete_stmt):
6304 """Provide a hook for MySQL to add LIMIT to the DELETE"""
6305 return None
6307 def update_tables_clause(self, update_stmt, from_table, extra_froms, **kw):
6308 """Provide a hook to override the initial table clause
6309 in an UPDATE statement.
6311 MySQL overrides this.
6313 """
6314 kw["asfrom"] = True
6315 return from_table._compiler_dispatch(self, iscrud=True, **kw)
6317 def update_from_clause(
6318 self, update_stmt, from_table, extra_froms, from_hints, **kw
6319 ):
6320 """Provide a hook to override the generation of an
6321 UPDATE..FROM clause.
6323 MySQL and MSSQL override this.
6325 """
6326 raise NotImplementedError(
6327 "This backend does not support multiple-table "
6328 "criteria within UPDATE"
6329 )
6331 def visit_update(
6332 self,
6333 update_stmt: Update,
6334 visiting_cte: Optional[CTE] = None,
6335 **kw: Any,
6336 ) -> str:
6337 compile_state = update_stmt._compile_state_factory(
6338 update_stmt, self, **kw
6339 )
6340 if TYPE_CHECKING:
6341 assert isinstance(compile_state, UpdateDMLState)
6342 update_stmt = compile_state.statement # type: ignore[assignment]
6344 if visiting_cte is not None:
6345 kw["visiting_cte"] = visiting_cte
6346 toplevel = False
6347 else:
6348 toplevel = not self.stack
6350 if toplevel:
6351 self.isupdate = True
6352 if not self.dml_compile_state:
6353 self.dml_compile_state = compile_state
6354 if not self.compile_state:
6355 self.compile_state = compile_state
6357 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
6358 from_linter = FromLinter({}, set())
6359 warn_linting = self.linting & WARN_LINTING
6360 if toplevel:
6361 self.from_linter = from_linter
6362 else:
6363 from_linter = None
6364 warn_linting = False
6366 extra_froms = compile_state._extra_froms
6367 is_multitable = bool(extra_froms)
6369 if is_multitable:
6370 # main table might be a JOIN
6371 main_froms = set(_from_objects(update_stmt.table))
6372 render_extra_froms = [
6373 f for f in extra_froms if f not in main_froms
6374 ]
6375 correlate_froms = main_froms.union(extra_froms)
6376 else:
6377 render_extra_froms = []
6378 correlate_froms = {update_stmt.table}
6380 self.stack.append(
6381 {
6382 "correlate_froms": correlate_froms,
6383 "asfrom_froms": correlate_froms,
6384 "selectable": update_stmt,
6385 }
6386 )
6388 text = "UPDATE "
6390 if update_stmt._prefixes:
6391 text += self._generate_prefixes(
6392 update_stmt, update_stmt._prefixes, **kw
6393 )
6395 table_text = self.update_tables_clause(
6396 update_stmt,
6397 update_stmt.table,
6398 render_extra_froms,
6399 from_linter=from_linter,
6400 **kw,
6401 )
6402 crud_params_struct = crud._get_crud_params(
6403 self, update_stmt, compile_state, toplevel, **kw
6404 )
6405 crud_params = crud_params_struct.single_params
6407 if update_stmt._hints:
6408 dialect_hints, table_text = self._setup_crud_hints(
6409 update_stmt, table_text
6410 )
6411 else:
6412 dialect_hints = None
6414 if update_stmt._independent_ctes:
6415 self._dispatch_independent_ctes(update_stmt, kw)
6417 text += table_text
6419 text += " SET "
6420 text += ", ".join(
6421 expr + "=" + value
6422 for _, expr, value, _ in cast(
6423 "List[Tuple[Any, str, str, Any]]", crud_params
6424 )
6425 )
6427 if self.implicit_returning or update_stmt._returning:
6428 if self.returning_precedes_values:
6429 text += " " + self.returning_clause(
6430 update_stmt,
6431 self.implicit_returning or update_stmt._returning,
6432 populate_result_map=toplevel,
6433 )
6435 if extra_froms:
6436 extra_from_text = self.update_from_clause(
6437 update_stmt,
6438 update_stmt.table,
6439 render_extra_froms,
6440 dialect_hints,
6441 from_linter=from_linter,
6442 **kw,
6443 )
6444 if extra_from_text:
6445 text += " " + extra_from_text
6447 if update_stmt._where_criteria:
6448 t = self._generate_delimited_and_list(
6449 update_stmt._where_criteria, from_linter=from_linter, **kw
6450 )
6451 if t:
6452 text += " WHERE " + t
6454 limit_clause = self.update_limit_clause(update_stmt)
6455 if limit_clause:
6456 text += " " + limit_clause
6458 if (
6459 self.implicit_returning or update_stmt._returning
6460 ) and not self.returning_precedes_values:
6461 text += " " + self.returning_clause(
6462 update_stmt,
6463 self.implicit_returning or update_stmt._returning,
6464 populate_result_map=toplevel,
6465 )
6467 if self.ctes:
6468 nesting_level = len(self.stack) if not toplevel else None
6469 text = self._render_cte_clause(nesting_level=nesting_level) + text
6471 if warn_linting:
6472 assert from_linter is not None
6473 from_linter.warn(stmt_type="UPDATE")
6475 self.stack.pop(-1)
6477 return text # type: ignore[no-any-return]
6479 def delete_extra_from_clause(
6480 self, delete_stmt, from_table, extra_froms, from_hints, **kw
6481 ):
6482 """Provide a hook to override the generation of an
6483 DELETE..FROM clause.
6485 This can be used to implement DELETE..USING for example.
6487 MySQL and MSSQL override this.
6489 """
6490 raise NotImplementedError(
6491 "This backend does not support multiple-table "
6492 "criteria within DELETE"
6493 )
6495 def delete_table_clause(self, delete_stmt, from_table, extra_froms, **kw):
6496 return from_table._compiler_dispatch(
6497 self, asfrom=True, iscrud=True, **kw
6498 )
6500 def visit_delete(self, delete_stmt, visiting_cte=None, **kw):
6501 compile_state = delete_stmt._compile_state_factory(
6502 delete_stmt, self, **kw
6503 )
6504 delete_stmt = compile_state.statement
6506 if visiting_cte is not None:
6507 kw["visiting_cte"] = visiting_cte
6508 toplevel = False
6509 else:
6510 toplevel = not self.stack
6512 if toplevel:
6513 self.isdelete = True
6514 if not self.dml_compile_state:
6515 self.dml_compile_state = compile_state
6516 if not self.compile_state:
6517 self.compile_state = compile_state
6519 if self.linting & COLLECT_CARTESIAN_PRODUCTS:
6520 from_linter = FromLinter({}, set())
6521 warn_linting = self.linting & WARN_LINTING
6522 if toplevel:
6523 self.from_linter = from_linter
6524 else:
6525 from_linter = None
6526 warn_linting = False
6528 extra_froms = compile_state._extra_froms
6530 correlate_froms = {delete_stmt.table}.union(extra_froms)
6531 self.stack.append(
6532 {
6533 "correlate_froms": correlate_froms,
6534 "asfrom_froms": correlate_froms,
6535 "selectable": delete_stmt,
6536 }
6537 )
6539 text = "DELETE "
6541 if delete_stmt._prefixes:
6542 text += self._generate_prefixes(
6543 delete_stmt, delete_stmt._prefixes, **kw
6544 )
6546 text += "FROM "
6548 try:
6549 table_text = self.delete_table_clause(
6550 delete_stmt,
6551 delete_stmt.table,
6552 extra_froms,
6553 from_linter=from_linter,
6554 )
6555 except TypeError:
6556 # anticipate 3rd party dialects that don't include **kw
6557 # TODO: remove in 2.1
6558 table_text = self.delete_table_clause(
6559 delete_stmt, delete_stmt.table, extra_froms
6560 )
6561 if from_linter:
6562 _ = self.process(delete_stmt.table, from_linter=from_linter)
6564 crud._get_crud_params(self, delete_stmt, compile_state, toplevel, **kw)
6566 if delete_stmt._hints:
6567 dialect_hints, table_text = self._setup_crud_hints(
6568 delete_stmt, table_text
6569 )
6570 else:
6571 dialect_hints = None
6573 if delete_stmt._independent_ctes:
6574 self._dispatch_independent_ctes(delete_stmt, kw)
6576 text += table_text
6578 if (
6579 self.implicit_returning or delete_stmt._returning
6580 ) and self.returning_precedes_values:
6581 text += " " + self.returning_clause(
6582 delete_stmt,
6583 self.implicit_returning or delete_stmt._returning,
6584 populate_result_map=toplevel,
6585 )
6587 if extra_froms:
6588 extra_from_text = self.delete_extra_from_clause(
6589 delete_stmt,
6590 delete_stmt.table,
6591 extra_froms,
6592 dialect_hints,
6593 from_linter=from_linter,
6594 **kw,
6595 )
6596 if extra_from_text:
6597 text += " " + extra_from_text
6599 if delete_stmt._where_criteria:
6600 t = self._generate_delimited_and_list(
6601 delete_stmt._where_criteria, from_linter=from_linter, **kw
6602 )
6603 if t:
6604 text += " WHERE " + t
6606 limit_clause = self.delete_limit_clause(delete_stmt)
6607 if limit_clause:
6608 text += " " + limit_clause
6610 if (
6611 self.implicit_returning or delete_stmt._returning
6612 ) and not self.returning_precedes_values:
6613 text += " " + self.returning_clause(
6614 delete_stmt,
6615 self.implicit_returning or delete_stmt._returning,
6616 populate_result_map=toplevel,
6617 )
6619 if self.ctes:
6620 nesting_level = len(self.stack) if not toplevel else None
6621 text = self._render_cte_clause(nesting_level=nesting_level) + text
6623 if warn_linting:
6624 assert from_linter is not None
6625 from_linter.warn(stmt_type="DELETE")
6627 self.stack.pop(-1)
6629 return text
6631 def visit_savepoint(self, savepoint_stmt, **kw):
6632 return "SAVEPOINT %s" % self.preparer.format_savepoint(savepoint_stmt)
6634 def visit_rollback_to_savepoint(self, savepoint_stmt, **kw):
6635 return "ROLLBACK TO SAVEPOINT %s" % self.preparer.format_savepoint(
6636 savepoint_stmt
6637 )
6639 def visit_release_savepoint(self, savepoint_stmt, **kw):
6640 return "RELEASE SAVEPOINT %s" % self.preparer.format_savepoint(
6641 savepoint_stmt
6642 )
6645class StrSQLCompiler(SQLCompiler):
6646 """A :class:`.SQLCompiler` subclass which allows a small selection
6647 of non-standard SQL features to render into a string value.
6649 The :class:`.StrSQLCompiler` is invoked whenever a Core expression
6650 element is directly stringified without calling upon the
6651 :meth:`_expression.ClauseElement.compile` method.
6652 It can render a limited set
6653 of non-standard SQL constructs to assist in basic stringification,
6654 however for more substantial custom or dialect-specific SQL constructs,
6655 it will be necessary to make use of
6656 :meth:`_expression.ClauseElement.compile`
6657 directly.
6659 .. seealso::
6661 :ref:`faq_sql_expression_string`
6663 """
6665 def get_select_precolumns(self, select: Select[Any], **kw: Any) -> str:
6666 return "DISTINCT " if select._distinct else ""
6668 def _fallback_column_name(self, column):
6669 return "<name unknown>"
6671 @util.preload_module("sqlalchemy.engine.url")
6672 def visit_unsupported_compilation(self, element, err, **kw):
6673 if element.stringify_dialect != "default":
6674 url = util.preloaded.engine_url
6675 dialect = url.URL.create(element.stringify_dialect).get_dialect()()
6677 compiler = dialect.statement_compiler(
6678 dialect, None, _supporting_against=self
6679 )
6680 if not isinstance(compiler, StrSQLCompiler):
6681 return compiler.process(element, **kw)
6683 return super().visit_unsupported_compilation(element, err)
6685 def visit_getitem_binary(self, binary, operator, **kw):
6686 return "%s[%s]" % (
6687 self.process(binary.left, **kw),
6688 self.process(binary.right, **kw),
6689 )
6691 def visit_json_getitem_op_binary(self, binary, operator, **kw):
6692 return self.visit_getitem_binary(binary, operator, **kw)
6694 def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
6695 return self.visit_getitem_binary(binary, operator, **kw)
6697 def visit_sequence(self, sequence, **kw):
6698 return (
6699 f"<next sequence value: {self.preparer.format_sequence(sequence)}>"
6700 )
6702 def returning_clause(
6703 self,
6704 stmt: UpdateBase,
6705 returning_cols: Sequence[_ColumnsClauseElement],
6706 *,
6707 populate_result_map: bool,
6708 **kw: Any,
6709 ) -> str:
6710 columns = [
6711 self._label_select_column(None, c, True, False, {})
6712 for c in base._select_iterables(returning_cols)
6713 ]
6714 return "RETURNING " + ", ".join(columns)
6716 def update_from_clause(
6717 self, update_stmt, from_table, extra_froms, from_hints, **kw
6718 ):
6719 kw["asfrom"] = True
6720 return "FROM " + ", ".join(
6721 t._compiler_dispatch(self, fromhints=from_hints, **kw)
6722 for t in extra_froms
6723 )
6725 def delete_extra_from_clause(
6726 self, delete_stmt, from_table, extra_froms, from_hints, **kw
6727 ):
6728 kw["asfrom"] = True
6729 return ", " + ", ".join(
6730 t._compiler_dispatch(self, fromhints=from_hints, **kw)
6731 for t in extra_froms
6732 )
6734 def visit_empty_set_expr(self, element_types, **kw):
6735 return "SELECT 1 WHERE 1!=1"
6737 def get_from_hint_text(self, table, text):
6738 return "[%s]" % text
6740 def visit_regexp_match_op_binary(self, binary, operator, **kw):
6741 return self._generate_generic_binary(binary, " <regexp> ", **kw)
6743 def visit_not_regexp_match_op_binary(self, binary, operator, **kw):
6744 return self._generate_generic_binary(binary, " <not regexp> ", **kw)
6746 def visit_regexp_replace_op_binary(self, binary, operator, **kw):
6747 return "<regexp replace>(%s, %s)" % (
6748 binary.left._compiler_dispatch(self, **kw),
6749 binary.right._compiler_dispatch(self, **kw),
6750 )
6752 def visit_try_cast(self, cast, **kwargs):
6753 return "TRY_CAST(%s AS %s)" % (
6754 cast.clause._compiler_dispatch(self, **kwargs),
6755 cast.typeclause._compiler_dispatch(self, **kwargs),
6756 )
6759class DDLCompiler(Compiled):
6760 is_ddl = True
6762 if TYPE_CHECKING:
6764 def __init__(
6765 self,
6766 dialect: Dialect,
6767 statement: ExecutableDDLElement,
6768 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
6769 render_schema_translate: bool = ...,
6770 compile_kwargs: Mapping[str, Any] = ...,
6771 ): ...
6773 @util.ro_memoized_property
6774 def sql_compiler(self) -> SQLCompiler:
6775 return self.dialect.statement_compiler(
6776 self.dialect, None, schema_translate_map=self.schema_translate_map
6777 )
6779 @util.memoized_property
6780 def type_compiler(self):
6781 return self.dialect.type_compiler_instance
6783 def construct_params(
6784 self,
6785 params: Optional[_CoreSingleExecuteParams] = None,
6786 extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None,
6787 escape_names: bool = True,
6788 ) -> Optional[_MutableCoreSingleExecuteParams]:
6789 return None
6791 def visit_ddl(self, ddl, **kwargs):
6792 # table events can substitute table and schema name
6793 context = ddl.context
6794 if isinstance(ddl.target, schema.Table):
6795 context = context.copy()
6797 preparer = self.preparer
6798 path = preparer.format_table_seq(ddl.target)
6799 if len(path) == 1:
6800 table, sch = path[0], ""
6801 else:
6802 table, sch = path[-1], path[0]
6804 context.setdefault("table", table)
6805 context.setdefault("schema", sch)
6806 context.setdefault("fullname", preparer.format_table(ddl.target))
6808 return self.sql_compiler.post_process_text(ddl.statement % context)
6810 def visit_create_schema(self, create, **kw):
6811 text = "CREATE SCHEMA "
6812 if create.if_not_exists:
6813 text += "IF NOT EXISTS "
6814 return text + self.preparer.format_schema(create.element)
6816 def visit_drop_schema(self, drop, **kw):
6817 text = "DROP SCHEMA "
6818 if drop.if_exists:
6819 text += "IF EXISTS "
6820 text += self.preparer.format_schema(drop.element)
6821 if drop.cascade:
6822 text += " CASCADE"
6823 return text
6825 def visit_create_table(self, create, **kw):
6826 table = create.element
6827 preparer = self.preparer
6829 text = "\nCREATE "
6830 if table._prefixes:
6831 text += " ".join(table._prefixes) + " "
6833 text += "TABLE "
6834 if create.if_not_exists:
6835 text += "IF NOT EXISTS "
6837 text += preparer.format_table(table) + " "
6839 create_table_suffix = self.create_table_suffix(table)
6840 if create_table_suffix:
6841 text += create_table_suffix + " "
6843 text += "("
6845 separator = "\n"
6847 # if only one primary key, specify it along with the column
6848 first_pk = False
6849 for create_column in create.columns:
6850 column = create_column.element
6851 try:
6852 processed = self.process(
6853 create_column, first_pk=column.primary_key and not first_pk
6854 )
6855 if processed is not None:
6856 text += separator
6857 separator = ", \n"
6858 text += "\t" + processed
6859 if column.primary_key:
6860 first_pk = True
6861 except exc.CompileError as ce:
6862 raise exc.CompileError(
6863 "(in table '%s', column '%s'): %s"
6864 % (table.description, column.name, ce.args[0])
6865 ) from ce
6867 const = self.create_table_constraints(
6868 table,
6869 _include_foreign_key_constraints=create.include_foreign_key_constraints, # noqa
6870 )
6871 if const:
6872 text += separator + "\t" + const
6874 text += "\n)%s\n\n" % self.post_create_table(table)
6875 return text
6877 def visit_create_column(self, create, first_pk=False, **kw):
6878 column = create.element
6880 if column.system:
6881 return None
6883 text = self.get_column_specification(column, first_pk=first_pk)
6884 const = " ".join(
6885 self.process(constraint) for constraint in column.constraints
6886 )
6887 if const:
6888 text += " " + const
6890 return text
6892 def create_table_constraints(
6893 self, table, _include_foreign_key_constraints=None, **kw
6894 ):
6895 # On some DB order is significant: visit PK first, then the
6896 # other constraints (engine.ReflectionTest.testbasic failed on FB2)
6897 constraints = []
6898 if table.primary_key:
6899 constraints.append(table.primary_key)
6901 all_fkcs = table.foreign_key_constraints
6902 if _include_foreign_key_constraints is not None:
6903 omit_fkcs = all_fkcs.difference(_include_foreign_key_constraints)
6904 else:
6905 omit_fkcs = set()
6907 constraints.extend(
6908 [
6909 c
6910 for c in table._sorted_constraints
6911 if c is not table.primary_key and c not in omit_fkcs
6912 ]
6913 )
6915 return ", \n\t".join(
6916 p
6917 for p in (
6918 self.process(constraint)
6919 for constraint in constraints
6920 if (constraint._should_create_for_compiler(self))
6921 and (
6922 not self.dialect.supports_alter
6923 or not getattr(constraint, "use_alter", False)
6924 )
6925 )
6926 if p is not None
6927 )
6929 def visit_drop_table(self, drop, **kw):
6930 text = "\nDROP TABLE "
6931 if drop.if_exists:
6932 text += "IF EXISTS "
6933 return text + self.preparer.format_table(drop.element)
6935 def visit_drop_view(self, drop, **kw):
6936 return "\nDROP VIEW " + self.preparer.format_table(drop.element)
6938 def _verify_index_table(self, index: Index) -> None:
6939 if index.table is None:
6940 raise exc.CompileError(
6941 "Index '%s' is not associated with any table." % index.name
6942 )
6944 def visit_create_index(
6945 self, create, include_schema=False, include_table_schema=True, **kw
6946 ):
6947 index = create.element
6948 self._verify_index_table(index)
6949 preparer = self.preparer
6950 text = "CREATE "
6951 if index.unique:
6952 text += "UNIQUE "
6953 if index.name is None:
6954 raise exc.CompileError(
6955 "CREATE INDEX requires that the index have a name"
6956 )
6958 text += "INDEX "
6959 if create.if_not_exists:
6960 text += "IF NOT EXISTS "
6962 text += "%s ON %s (%s)" % (
6963 self._prepared_index_name(index, include_schema=include_schema),
6964 preparer.format_table(
6965 index.table, use_schema=include_table_schema
6966 ),
6967 ", ".join(
6968 self.sql_compiler.process(
6969 expr, include_table=False, literal_binds=True
6970 )
6971 for expr in index.expressions
6972 ),
6973 )
6974 return text
6976 def visit_drop_index(self, drop, **kw):
6977 index = drop.element
6979 if index.name is None:
6980 raise exc.CompileError(
6981 "DROP INDEX requires that the index have a name"
6982 )
6983 text = "\nDROP INDEX "
6984 if drop.if_exists:
6985 text += "IF EXISTS "
6987 return text + self._prepared_index_name(index, include_schema=True)
6989 def _prepared_index_name(
6990 self, index: Index, include_schema: bool = False
6991 ) -> str:
6992 if index.table is not None:
6993 effective_schema = self.preparer.schema_for_object(index.table)
6994 else:
6995 effective_schema = None
6996 if include_schema and effective_schema:
6997 schema_name = self.preparer.quote_schema(effective_schema)
6998 else:
6999 schema_name = None
7001 index_name: str = self.preparer.format_index(index)
7003 if schema_name:
7004 index_name = schema_name + "." + index_name
7005 return index_name
7007 def visit_add_constraint(self, create, **kw):
7008 return "ALTER TABLE %s ADD %s" % (
7009 self.preparer.format_table(create.element.table),
7010 self.process(create.element),
7011 )
7013 def visit_set_table_comment(self, create, **kw):
7014 return "COMMENT ON TABLE %s IS %s" % (
7015 self.preparer.format_table(create.element),
7016 self.sql_compiler.render_literal_value(
7017 create.element.comment, sqltypes.String()
7018 ),
7019 )
7021 def visit_drop_table_comment(self, drop, **kw):
7022 return "COMMENT ON TABLE %s IS NULL" % self.preparer.format_table(
7023 drop.element
7024 )
7026 def visit_set_column_comment(self, create, **kw):
7027 return "COMMENT ON COLUMN %s IS %s" % (
7028 self.preparer.format_column(
7029 create.element, use_table=True, use_schema=True
7030 ),
7031 self.sql_compiler.render_literal_value(
7032 create.element.comment, sqltypes.String()
7033 ),
7034 )
7036 def visit_drop_column_comment(self, drop, **kw):
7037 return "COMMENT ON COLUMN %s IS NULL" % self.preparer.format_column(
7038 drop.element, use_table=True
7039 )
7041 def visit_set_constraint_comment(self, create, **kw):
7042 raise exc.UnsupportedCompilationError(self, type(create))
7044 def visit_drop_constraint_comment(self, drop, **kw):
7045 raise exc.UnsupportedCompilationError(self, type(drop))
7047 def get_identity_options(self, identity_options: IdentityOptions) -> str:
7048 text = []
7049 if identity_options.increment is not None:
7050 text.append("INCREMENT BY %d" % identity_options.increment)
7051 if identity_options.start is not None:
7052 text.append("START WITH %d" % identity_options.start)
7053 if identity_options.minvalue is not None:
7054 text.append("MINVALUE %d" % identity_options.minvalue)
7055 if identity_options.maxvalue is not None:
7056 text.append("MAXVALUE %d" % identity_options.maxvalue)
7057 if identity_options.nominvalue is not None:
7058 text.append("NO MINVALUE")
7059 if identity_options.nomaxvalue is not None:
7060 text.append("NO MAXVALUE")
7061 if identity_options.cache is not None:
7062 text.append("CACHE %d" % identity_options.cache)
7063 if identity_options.cycle is not None:
7064 text.append("CYCLE" if identity_options.cycle else "NO CYCLE")
7065 return " ".join(text)
7067 def visit_create_sequence(self, create, prefix=None, **kw):
7068 text = "CREATE SEQUENCE "
7069 if create.if_not_exists:
7070 text += "IF NOT EXISTS "
7071 text += self.preparer.format_sequence(create.element)
7073 if prefix:
7074 text += prefix
7075 options = self.get_identity_options(create.element)
7076 if options:
7077 text += " " + options
7078 return text
7080 def visit_drop_sequence(self, drop, **kw):
7081 text = "DROP SEQUENCE "
7082 if drop.if_exists:
7083 text += "IF EXISTS "
7084 return text + self.preparer.format_sequence(drop.element)
7086 def visit_drop_constraint(self, drop, **kw):
7087 constraint = drop.element
7088 if constraint.name is not None:
7089 formatted_name = self.preparer.format_constraint(constraint)
7090 else:
7091 formatted_name = None
7093 if formatted_name is None:
7094 raise exc.CompileError(
7095 "Can't emit DROP CONSTRAINT for constraint %r; "
7096 "it has no name" % drop.element
7097 )
7098 return "ALTER TABLE %s DROP CONSTRAINT %s%s%s" % (
7099 self.preparer.format_table(drop.element.table),
7100 "IF EXISTS " if drop.if_exists else "",
7101 formatted_name,
7102 " CASCADE" if drop.cascade else "",
7103 )
7105 def get_column_specification(self, column, **kwargs):
7106 colspec = (
7107 self.preparer.format_column(column)
7108 + " "
7109 + self.dialect.type_compiler_instance.process(
7110 column.type, type_expression=column
7111 )
7112 )
7113 default = self.get_column_default_string(column)
7114 if default is not None:
7115 colspec += " DEFAULT " + default
7117 if column.computed is not None:
7118 colspec += " " + self.process(column.computed)
7120 if (
7121 column.identity is not None
7122 and self.dialect.supports_identity_columns
7123 ):
7124 colspec += " " + self.process(column.identity)
7126 if not column.nullable and (
7127 not column.identity or not self.dialect.supports_identity_columns
7128 ):
7129 colspec += " NOT NULL"
7130 return colspec
7132 def create_table_suffix(self, table):
7133 return ""
7135 def post_create_table(self, table):
7136 return ""
7138 def get_column_default_string(self, column: Column[Any]) -> Optional[str]:
7139 if isinstance(column.server_default, schema.DefaultClause):
7140 return self.render_default_string(column.server_default.arg)
7141 else:
7142 return None
7144 def render_default_string(self, default: Union[Visitable, str]) -> str:
7145 if isinstance(default, str):
7146 return self.sql_compiler.render_literal_value(
7147 default, sqltypes.STRINGTYPE
7148 )
7149 else:
7150 return self.sql_compiler.process(default, literal_binds=True)
7152 def visit_table_or_column_check_constraint(self, constraint, **kw):
7153 if constraint.is_column_level:
7154 return self.visit_column_check_constraint(constraint)
7155 else:
7156 return self.visit_check_constraint(constraint)
7158 def visit_check_constraint(self, constraint, **kw):
7159 text = self.define_constraint_preamble(constraint, **kw)
7160 text += self.define_check_body(constraint, **kw)
7161 text += self.define_constraint_deferrability(constraint)
7162 return text
7164 def visit_column_check_constraint(self, constraint, **kw):
7165 text = self.define_constraint_preamble(constraint, **kw)
7166 text += self.define_check_body(constraint, **kw)
7167 text += self.define_constraint_deferrability(constraint)
7168 return text
7170 def visit_primary_key_constraint(
7171 self, constraint: PrimaryKeyConstraint, **kw: Any
7172 ) -> str:
7173 if len(constraint) == 0:
7174 return ""
7175 text = self.define_constraint_preamble(constraint, **kw)
7176 text += self.define_primary_key_body(constraint, **kw)
7177 text += self.define_constraint_deferrability(constraint)
7178 return text
7180 def visit_foreign_key_constraint(
7181 self, constraint: ForeignKeyConstraint, **kw: Any
7182 ) -> str:
7183 text = self.define_constraint_preamble(constraint, **kw)
7184 text += self.define_foreign_key_body(constraint, **kw)
7185 text += self.define_constraint_match(constraint)
7186 text += self.define_constraint_cascades(constraint)
7187 text += self.define_constraint_deferrability(constraint)
7188 return text
7190 def define_constraint_remote_table(self, constraint, table, preparer):
7191 """Format the remote table clause of a CREATE CONSTRAINT clause."""
7193 return preparer.format_table(table)
7195 def visit_unique_constraint(
7196 self, constraint: UniqueConstraint, **kw: Any
7197 ) -> str:
7198 if len(constraint) == 0:
7199 return ""
7200 text = self.define_constraint_preamble(constraint, **kw)
7201 text += self.define_unique_body(constraint, **kw)
7202 text += self.define_constraint_deferrability(constraint)
7203 return text
7205 def define_constraint_preamble(
7206 self, constraint: Constraint, **kw: Any
7207 ) -> str:
7208 text = ""
7209 if constraint.name is not None:
7210 formatted_name = self.preparer.format_constraint(constraint)
7211 if formatted_name is not None:
7212 text += "CONSTRAINT %s " % formatted_name
7213 return text
7215 def define_primary_key_body(
7216 self, constraint: PrimaryKeyConstraint, **kw: Any
7217 ) -> str:
7218 text = ""
7219 text += "PRIMARY KEY "
7220 text += "(%s)" % ", ".join(
7221 self.preparer.quote(c.name)
7222 for c in (
7223 constraint.columns_autoinc_first
7224 if constraint._implicit_generated
7225 else constraint.columns
7226 )
7227 )
7228 return text
7230 def define_foreign_key_body(
7231 self, constraint: ForeignKeyConstraint, **kw: Any
7232 ) -> str:
7233 preparer = self.preparer
7234 remote_table = list(constraint.elements)[0].column.table
7235 text = "FOREIGN KEY(%s) REFERENCES %s (%s)" % (
7236 ", ".join(
7237 preparer.quote(f.parent.name) for f in constraint.elements
7238 ),
7239 self.define_constraint_remote_table(
7240 constraint, remote_table, preparer
7241 ),
7242 ", ".join(
7243 preparer.quote(f.column.name) for f in constraint.elements
7244 ),
7245 )
7246 return text
7248 def define_unique_body(
7249 self, constraint: UniqueConstraint, **kw: Any
7250 ) -> str:
7251 text = "UNIQUE %s(%s)" % (
7252 self.define_unique_constraint_distinct(constraint, **kw),
7253 ", ".join(self.preparer.quote(c.name) for c in constraint),
7254 )
7255 return text
7257 def define_check_body(self, constraint: CheckConstraint, **kw: Any) -> str:
7258 text = "CHECK (%s)" % self.sql_compiler.process(
7259 constraint.sqltext, include_table=False, literal_binds=True
7260 )
7261 return text
7263 def define_unique_constraint_distinct(
7264 self, constraint: UniqueConstraint, **kw: Any
7265 ) -> str:
7266 return ""
7268 def define_constraint_cascades(
7269 self, constraint: ForeignKeyConstraint
7270 ) -> str:
7271 text = ""
7272 if constraint.ondelete is not None:
7273 text += self.define_constraint_ondelete_cascade(constraint)
7275 if constraint.onupdate is not None:
7276 text += self.define_constraint_onupdate_cascade(constraint)
7277 return text
7279 def define_constraint_ondelete_cascade(
7280 self, constraint: ForeignKeyConstraint
7281 ) -> str:
7282 return " ON DELETE %s" % self.preparer.validate_sql_phrase(
7283 constraint.ondelete, FK_ON_DELETE
7284 )
7286 def define_constraint_onupdate_cascade(
7287 self, constraint: ForeignKeyConstraint
7288 ) -> str:
7289 return " ON UPDATE %s" % self.preparer.validate_sql_phrase(
7290 constraint.onupdate, FK_ON_UPDATE
7291 )
7293 def define_constraint_deferrability(self, constraint: Constraint) -> str:
7294 text = ""
7295 if constraint.deferrable is not None:
7296 if constraint.deferrable:
7297 text += " DEFERRABLE"
7298 else:
7299 text += " NOT DEFERRABLE"
7300 if constraint.initially is not None:
7301 text += " INITIALLY %s" % self.preparer.validate_sql_phrase(
7302 constraint.initially, FK_INITIALLY
7303 )
7304 return text
7306 def define_constraint_match(self, constraint: ForeignKeyConstraint) -> str:
7307 text = ""
7308 if constraint.match is not None:
7309 text += " MATCH %s" % constraint.match
7310 return text
7312 def visit_computed_column(self, generated, **kw):
7313 text = "GENERATED ALWAYS AS (%s)" % self.sql_compiler.process(
7314 generated.sqltext, include_table=False, literal_binds=True
7315 )
7316 if generated.persisted is True:
7317 text += " STORED"
7318 elif generated.persisted is False:
7319 text += " VIRTUAL"
7320 return text
7322 def visit_identity_column(self, identity, **kw):
7323 text = "GENERATED %s AS IDENTITY" % (
7324 "ALWAYS" if identity.always else "BY DEFAULT",
7325 )
7326 options = self.get_identity_options(identity)
7327 if options:
7328 text += " (%s)" % options
7329 return text
7332class GenericTypeCompiler(TypeCompiler):
7333 def visit_FLOAT(self, type_: sqltypes.Float[Any], **kw: Any) -> str:
7334 return "FLOAT"
7336 def visit_DOUBLE(self, type_: sqltypes.Double[Any], **kw: Any) -> str:
7337 return "DOUBLE"
7339 def visit_DOUBLE_PRECISION(
7340 self, type_: sqltypes.DOUBLE_PRECISION[Any], **kw: Any
7341 ) -> str:
7342 return "DOUBLE PRECISION"
7344 def visit_REAL(self, type_: sqltypes.REAL[Any], **kw: Any) -> str:
7345 return "REAL"
7347 def visit_NUMERIC(self, type_: sqltypes.Numeric[Any], **kw: Any) -> str:
7348 if type_.precision is None:
7349 return "NUMERIC"
7350 elif type_.scale is None:
7351 return "NUMERIC(%(precision)s)" % {"precision": type_.precision}
7352 else:
7353 return "NUMERIC(%(precision)s, %(scale)s)" % {
7354 "precision": type_.precision,
7355 "scale": type_.scale,
7356 }
7358 def visit_DECIMAL(self, type_: sqltypes.DECIMAL[Any], **kw: Any) -> str:
7359 if type_.precision is None:
7360 return "DECIMAL"
7361 elif type_.scale is None:
7362 return "DECIMAL(%(precision)s)" % {"precision": type_.precision}
7363 else:
7364 return "DECIMAL(%(precision)s, %(scale)s)" % {
7365 "precision": type_.precision,
7366 "scale": type_.scale,
7367 }
7369 def visit_INTEGER(self, type_: sqltypes.Integer, **kw: Any) -> str:
7370 return "INTEGER"
7372 def visit_SMALLINT(self, type_: sqltypes.SmallInteger, **kw: Any) -> str:
7373 return "SMALLINT"
7375 def visit_BIGINT(self, type_: sqltypes.BigInteger, **kw: Any) -> str:
7376 return "BIGINT"
7378 def visit_TIMESTAMP(self, type_: sqltypes.TIMESTAMP, **kw: Any) -> str:
7379 return "TIMESTAMP"
7381 def visit_DATETIME(self, type_: sqltypes.DateTime, **kw: Any) -> str:
7382 return "DATETIME"
7384 def visit_DATE(self, type_: sqltypes.Date, **kw: Any) -> str:
7385 return "DATE"
7387 def visit_TIME(self, type_: sqltypes.Time, **kw: Any) -> str:
7388 return "TIME"
7390 def visit_CLOB(self, type_: sqltypes.CLOB, **kw: Any) -> str:
7391 return "CLOB"
7393 def visit_NCLOB(self, type_: sqltypes.Text, **kw: Any) -> str:
7394 return "NCLOB"
7396 def _render_string_type(
7397 self, name: str, length: Optional[int], collation: Optional[str]
7398 ) -> str:
7399 text = name
7400 if length:
7401 text += f"({length})"
7402 if collation:
7403 text += f' COLLATE "{collation}"'
7404 return text
7406 def visit_CHAR(self, type_: sqltypes.CHAR, **kw: Any) -> str:
7407 return self._render_string_type("CHAR", type_.length, type_.collation)
7409 def visit_NCHAR(self, type_: sqltypes.NCHAR, **kw: Any) -> str:
7410 return self._render_string_type("NCHAR", type_.length, type_.collation)
7412 def visit_VARCHAR(self, type_: sqltypes.String, **kw: Any) -> str:
7413 return self._render_string_type(
7414 "VARCHAR", type_.length, type_.collation
7415 )
7417 def visit_NVARCHAR(self, type_: sqltypes.NVARCHAR, **kw: Any) -> str:
7418 return self._render_string_type(
7419 "NVARCHAR", type_.length, type_.collation
7420 )
7422 def visit_TEXT(self, type_: sqltypes.Text, **kw: Any) -> str:
7423 return self._render_string_type("TEXT", type_.length, type_.collation)
7425 def visit_UUID(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
7426 return "UUID"
7428 def visit_BLOB(self, type_: sqltypes.LargeBinary, **kw: Any) -> str:
7429 return "BLOB"
7431 def visit_BINARY(self, type_: sqltypes.BINARY, **kw: Any) -> str:
7432 return "BINARY" + (type_.length and "(%d)" % type_.length or "")
7434 def visit_VARBINARY(self, type_: sqltypes.VARBINARY, **kw: Any) -> str:
7435 return "VARBINARY" + (type_.length and "(%d)" % type_.length or "")
7437 def visit_BOOLEAN(self, type_: sqltypes.Boolean, **kw: Any) -> str:
7438 return "BOOLEAN"
7440 def visit_uuid(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
7441 if not type_.native_uuid or not self.dialect.supports_native_uuid:
7442 return self._render_string_type("CHAR", length=32, collation=None)
7443 else:
7444 return self.visit_UUID(type_, **kw)
7446 def visit_large_binary(
7447 self, type_: sqltypes.LargeBinary, **kw: Any
7448 ) -> str:
7449 return self.visit_BLOB(type_, **kw)
7451 def visit_boolean(self, type_: sqltypes.Boolean, **kw: Any) -> str:
7452 return self.visit_BOOLEAN(type_, **kw)
7454 def visit_time(self, type_: sqltypes.Time, **kw: Any) -> str:
7455 return self.visit_TIME(type_, **kw)
7457 def visit_datetime(self, type_: sqltypes.DateTime, **kw: Any) -> str:
7458 return self.visit_DATETIME(type_, **kw)
7460 def visit_date(self, type_: sqltypes.Date, **kw: Any) -> str:
7461 return self.visit_DATE(type_, **kw)
7463 def visit_big_integer(self, type_: sqltypes.BigInteger, **kw: Any) -> str:
7464 return self.visit_BIGINT(type_, **kw)
7466 def visit_small_integer(
7467 self, type_: sqltypes.SmallInteger, **kw: Any
7468 ) -> str:
7469 return self.visit_SMALLINT(type_, **kw)
7471 def visit_integer(self, type_: sqltypes.Integer, **kw: Any) -> str:
7472 return self.visit_INTEGER(type_, **kw)
7474 def visit_real(self, type_: sqltypes.REAL[Any], **kw: Any) -> str:
7475 return self.visit_REAL(type_, **kw)
7477 def visit_float(self, type_: sqltypes.Float[Any], **kw: Any) -> str:
7478 return self.visit_FLOAT(type_, **kw)
7480 def visit_double(self, type_: sqltypes.Double[Any], **kw: Any) -> str:
7481 return self.visit_DOUBLE(type_, **kw)
7483 def visit_numeric(self, type_: sqltypes.Numeric[Any], **kw: Any) -> str:
7484 return self.visit_NUMERIC(type_, **kw)
7486 def visit_string(self, type_: sqltypes.String, **kw: Any) -> str:
7487 return self.visit_VARCHAR(type_, **kw)
7489 def visit_unicode(self, type_: sqltypes.Unicode, **kw: Any) -> str:
7490 return self.visit_VARCHAR(type_, **kw)
7492 def visit_text(self, type_: sqltypes.Text, **kw: Any) -> str:
7493 return self.visit_TEXT(type_, **kw)
7495 def visit_unicode_text(
7496 self, type_: sqltypes.UnicodeText, **kw: Any
7497 ) -> str:
7498 return self.visit_TEXT(type_, **kw)
7500 def visit_enum(self, type_: sqltypes.Enum, **kw: Any) -> str:
7501 return self.visit_VARCHAR(type_, **kw)
7503 def visit_null(self, type_, **kw):
7504 raise exc.CompileError(
7505 "Can't generate DDL for %r; "
7506 "did you forget to specify a "
7507 "type on this Column?" % type_
7508 )
7510 def visit_type_decorator(
7511 self, type_: TypeDecorator[Any], **kw: Any
7512 ) -> str:
7513 return self.process(type_.type_engine(self.dialect), **kw)
7515 def visit_user_defined(
7516 self, type_: UserDefinedType[Any], **kw: Any
7517 ) -> str:
7518 return type_.get_col_spec(**kw)
7521class StrSQLTypeCompiler(GenericTypeCompiler):
7522 def process(self, type_, **kw):
7523 try:
7524 _compiler_dispatch = type_._compiler_dispatch
7525 except AttributeError:
7526 return self._visit_unknown(type_, **kw)
7527 else:
7528 return _compiler_dispatch(self, **kw)
7530 def __getattr__(self, key):
7531 if key.startswith("visit_"):
7532 return self._visit_unknown
7533 else:
7534 raise AttributeError(key)
7536 def _visit_unknown(self, type_, **kw):
7537 if type_.__class__.__name__ == type_.__class__.__name__.upper():
7538 return type_.__class__.__name__
7539 else:
7540 return repr(type_)
7542 def visit_null(self, type_, **kw):
7543 return "NULL"
7545 def visit_user_defined(self, type_, **kw):
7546 try:
7547 get_col_spec = type_.get_col_spec
7548 except AttributeError:
7549 return repr(type_)
7550 else:
7551 return get_col_spec(**kw)
7554class _SchemaForObjectCallable(Protocol):
7555 def __call__(self, __obj: Any) -> str: ...
7558class _BindNameForColProtocol(Protocol):
7559 def __call__(self, col: ColumnClause[Any]) -> str: ...
7562class IdentifierPreparer:
7563 """Handle quoting and case-folding of identifiers based on options."""
7565 reserved_words = RESERVED_WORDS
7567 legal_characters = LEGAL_CHARACTERS
7569 illegal_initial_characters = ILLEGAL_INITIAL_CHARACTERS
7571 initial_quote: str
7573 final_quote: str
7575 _strings: MutableMapping[str, str]
7577 schema_for_object: _SchemaForObjectCallable = operator.attrgetter("schema")
7578 """Return the .schema attribute for an object.
7580 For the default IdentifierPreparer, the schema for an object is always
7581 the value of the ".schema" attribute. if the preparer is replaced
7582 with one that has a non-empty schema_translate_map, the value of the
7583 ".schema" attribute is rendered a symbol that will be converted to a
7584 real schema name from the mapping post-compile.
7586 """
7588 _includes_none_schema_translate: bool = False
7590 def __init__(
7591 self,
7592 dialect: Dialect,
7593 initial_quote: str = '"',
7594 final_quote: Optional[str] = None,
7595 escape_quote: str = '"',
7596 quote_case_sensitive_collations: bool = True,
7597 omit_schema: bool = False,
7598 ):
7599 """Construct a new ``IdentifierPreparer`` object.
7601 initial_quote
7602 Character that begins a delimited identifier.
7604 final_quote
7605 Character that ends a delimited identifier. Defaults to
7606 `initial_quote`.
7608 omit_schema
7609 Prevent prepending schema name. Useful for databases that do
7610 not support schemae.
7611 """
7613 self.dialect = dialect
7614 self.initial_quote = initial_quote
7615 self.final_quote = final_quote or self.initial_quote
7616 self.escape_quote = escape_quote
7617 self.escape_to_quote = self.escape_quote * 2
7618 self.omit_schema = omit_schema
7619 self.quote_case_sensitive_collations = quote_case_sensitive_collations
7620 self._strings = {}
7621 self._double_percents = self.dialect.paramstyle in (
7622 "format",
7623 "pyformat",
7624 )
7626 def _with_schema_translate(self, schema_translate_map):
7627 prep = self.__class__.__new__(self.__class__)
7628 prep.__dict__.update(self.__dict__)
7630 includes_none = None in schema_translate_map
7632 def symbol_getter(obj):
7633 name = obj.schema
7634 if obj._use_schema_map and (name is not None or includes_none):
7635 if name is not None and ("[" in name or "]" in name):
7636 raise exc.CompileError(
7637 "Square bracket characters ([]) not supported "
7638 "in schema translate name '%s'" % name
7639 )
7640 return quoted_name(
7641 "__[SCHEMA_%s]" % (name or "_none"), quote=False
7642 )
7643 else:
7644 return obj.schema
7646 prep.schema_for_object = symbol_getter
7647 prep._includes_none_schema_translate = includes_none
7648 return prep
7650 def _render_schema_translates(
7651 self, statement: str, schema_translate_map: SchemaTranslateMapType
7652 ) -> str:
7653 d = schema_translate_map
7654 if None in d:
7655 if not self._includes_none_schema_translate:
7656 raise exc.InvalidRequestError(
7657 "schema translate map which previously did not have "
7658 "`None` present as a key now has `None` present; compiled "
7659 "statement may lack adequate placeholders. Please use "
7660 "consistent keys in successive "
7661 "schema_translate_map dictionaries."
7662 )
7664 d["_none"] = d[None] # type: ignore[index]
7666 def replace(m):
7667 name = m.group(2)
7668 if name in d:
7669 effective_schema = d[name]
7670 else:
7671 if name in (None, "_none"):
7672 raise exc.InvalidRequestError(
7673 "schema translate map which previously had `None` "
7674 "present as a key now no longer has it present; don't "
7675 "know how to apply schema for compiled statement. "
7676 "Please use consistent keys in successive "
7677 "schema_translate_map dictionaries."
7678 )
7679 effective_schema = name
7681 if not effective_schema:
7682 effective_schema = self.dialect.default_schema_name
7683 if not effective_schema:
7684 # TODO: no coverage here
7685 raise exc.CompileError(
7686 "Dialect has no default schema name; can't "
7687 "use None as dynamic schema target."
7688 )
7689 return self.quote_schema(effective_schema)
7691 return re.sub(r"(__\[SCHEMA_([^\]]+)\])", replace, statement)
7693 def _escape_identifier(self, value: str) -> str:
7694 """Escape an identifier.
7696 Subclasses should override this to provide database-dependent
7697 escaping behavior.
7698 """
7700 value = value.replace(self.escape_quote, self.escape_to_quote)
7701 if self._double_percents:
7702 value = value.replace("%", "%%")
7703 return value
7705 def _unescape_identifier(self, value: str) -> str:
7706 """Canonicalize an escaped identifier.
7708 Subclasses should override this to provide database-dependent
7709 unescaping behavior that reverses _escape_identifier.
7710 """
7712 return value.replace(self.escape_to_quote, self.escape_quote)
7714 def validate_sql_phrase(self, element, reg):
7715 """keyword sequence filter.
7717 a filter for elements that are intended to represent keyword sequences,
7718 such as "INITIALLY", "INITIALLY DEFERRED", etc. no special characters
7719 should be present.
7721 .. versionadded:: 1.3
7723 """
7725 if element is not None and not reg.match(element):
7726 raise exc.CompileError(
7727 "Unexpected SQL phrase: %r (matching against %r)"
7728 % (element, reg.pattern)
7729 )
7730 return element
7732 def quote_identifier(self, value: str) -> str:
7733 """Quote an identifier.
7735 Subclasses should override this to provide database-dependent
7736 quoting behavior.
7737 """
7739 return (
7740 self.initial_quote
7741 + self._escape_identifier(value)
7742 + self.final_quote
7743 )
7745 def _requires_quotes(self, value: str) -> bool:
7746 """Return True if the given identifier requires quoting."""
7747 lc_value = value.lower()
7748 return (
7749 lc_value in self.reserved_words
7750 or value[0] in self.illegal_initial_characters
7751 or not self.legal_characters.match(str(value))
7752 or (lc_value != value)
7753 )
7755 def _requires_quotes_illegal_chars(self, value):
7756 """Return True if the given identifier requires quoting, but
7757 not taking case convention into account."""
7758 return not self.legal_characters.match(str(value))
7760 def quote_schema(self, schema: str, force: Any = None) -> str:
7761 """Conditionally quote a schema name.
7764 The name is quoted if it is a reserved word, contains quote-necessary
7765 characters, or is an instance of :class:`.quoted_name` which includes
7766 ``quote`` set to ``True``.
7768 Subclasses can override this to provide database-dependent
7769 quoting behavior for schema names.
7771 :param schema: string schema name
7772 :param force: unused
7774 .. deprecated:: 0.9
7776 The :paramref:`.IdentifierPreparer.quote_schema.force`
7777 parameter is deprecated and will be removed in a future
7778 release. This flag has no effect on the behavior of the
7779 :meth:`.IdentifierPreparer.quote` method; please refer to
7780 :class:`.quoted_name`.
7782 """
7783 if force is not None:
7784 # not using the util.deprecated_params() decorator in this
7785 # case because of the additional function call overhead on this
7786 # very performance-critical spot.
7787 util.warn_deprecated(
7788 "The IdentifierPreparer.quote_schema.force parameter is "
7789 "deprecated and will be removed in a future release. This "
7790 "flag has no effect on the behavior of the "
7791 "IdentifierPreparer.quote method; please refer to "
7792 "quoted_name().",
7793 # deprecated 0.9. warning from 1.3
7794 version="0.9",
7795 )
7797 return self.quote(schema)
7799 def quote(self, ident: str, force: Any = None) -> str:
7800 """Conditionally quote an identifier.
7802 The identifier is quoted if it is a reserved word, contains
7803 quote-necessary characters, or is an instance of
7804 :class:`.quoted_name` which includes ``quote`` set to ``True``.
7806 Subclasses can override this to provide database-dependent
7807 quoting behavior for identifier names.
7809 :param ident: string identifier
7810 :param force: unused
7812 .. deprecated:: 0.9
7814 The :paramref:`.IdentifierPreparer.quote.force`
7815 parameter is deprecated and will be removed in a future
7816 release. This flag has no effect on the behavior of the
7817 :meth:`.IdentifierPreparer.quote` method; please refer to
7818 :class:`.quoted_name`.
7820 """
7821 if force is not None:
7822 # not using the util.deprecated_params() decorator in this
7823 # case because of the additional function call overhead on this
7824 # very performance-critical spot.
7825 util.warn_deprecated(
7826 "The IdentifierPreparer.quote.force parameter is "
7827 "deprecated and will be removed in a future release. This "
7828 "flag has no effect on the behavior of the "
7829 "IdentifierPreparer.quote method; please refer to "
7830 "quoted_name().",
7831 # deprecated 0.9. warning from 1.3
7832 version="0.9",
7833 )
7835 force = getattr(ident, "quote", None)
7837 if force is None:
7838 if ident in self._strings:
7839 return self._strings[ident]
7840 else:
7841 if self._requires_quotes(ident):
7842 self._strings[ident] = self.quote_identifier(ident)
7843 else:
7844 self._strings[ident] = ident
7845 return self._strings[ident]
7846 elif force:
7847 return self.quote_identifier(ident)
7848 else:
7849 return ident
7851 def format_collation(self, collation_name):
7852 if self.quote_case_sensitive_collations:
7853 return self.quote(collation_name)
7854 else:
7855 return collation_name
7857 def format_sequence(
7858 self, sequence: schema.Sequence, use_schema: bool = True
7859 ) -> str:
7860 name = self.quote(sequence.name)
7862 effective_schema = self.schema_for_object(sequence)
7864 if (
7865 not self.omit_schema
7866 and use_schema
7867 and effective_schema is not None
7868 ):
7869 name = self.quote_schema(effective_schema) + "." + name
7870 return name
7872 def format_label(
7873 self, label: Label[Any], name: Optional[str] = None
7874 ) -> str:
7875 return self.quote(name or label.name)
7877 def format_alias(
7878 self, alias: Optional[AliasedReturnsRows], name: Optional[str] = None
7879 ) -> str:
7880 if name is None:
7881 assert alias is not None
7882 return self.quote(alias.name)
7883 else:
7884 return self.quote(name)
7886 def format_savepoint(self, savepoint, name=None):
7887 # Running the savepoint name through quoting is unnecessary
7888 # for all known dialects. This is here to support potential
7889 # third party use cases
7890 ident = name or savepoint.ident
7891 if self._requires_quotes(ident):
7892 ident = self.quote_identifier(ident)
7893 return ident
7895 @util.preload_module("sqlalchemy.sql.naming")
7896 def format_constraint(
7897 self, constraint: Union[Constraint, Index], _alembic_quote: bool = True
7898 ) -> Optional[str]:
7899 naming = util.preloaded.sql_naming
7901 if constraint.name is _NONE_NAME:
7902 name = naming._constraint_name_for_table(
7903 constraint, constraint.table
7904 )
7906 if name is None:
7907 return None
7908 else:
7909 name = constraint.name
7911 assert name is not None
7912 if constraint.__visit_name__ == "index":
7913 return self.truncate_and_render_index_name(
7914 name, _alembic_quote=_alembic_quote
7915 )
7916 else:
7917 return self.truncate_and_render_constraint_name(
7918 name, _alembic_quote=_alembic_quote
7919 )
7921 def truncate_and_render_index_name(
7922 self, name: str, _alembic_quote: bool = True
7923 ) -> str:
7924 # calculate these at format time so that ad-hoc changes
7925 # to dialect.max_identifier_length etc. can be reflected
7926 # as IdentifierPreparer is long lived
7927 max_ = (
7928 self.dialect.max_index_name_length
7929 or self.dialect.max_identifier_length
7930 )
7931 return self._truncate_and_render_maxlen_name(
7932 name, max_, _alembic_quote
7933 )
7935 def truncate_and_render_constraint_name(
7936 self, name: str, _alembic_quote: bool = True
7937 ) -> str:
7938 # calculate these at format time so that ad-hoc changes
7939 # to dialect.max_identifier_length etc. can be reflected
7940 # as IdentifierPreparer is long lived
7941 max_ = (
7942 self.dialect.max_constraint_name_length
7943 or self.dialect.max_identifier_length
7944 )
7945 return self._truncate_and_render_maxlen_name(
7946 name, max_, _alembic_quote
7947 )
7949 def _truncate_and_render_maxlen_name(
7950 self, name: str, max_: int, _alembic_quote: bool
7951 ) -> str:
7952 if isinstance(name, elements._truncated_label):
7953 if len(name) > max_:
7954 name = name[0 : max_ - 8] + "_" + util.md5_hex(name)[-4:]
7955 else:
7956 self.dialect.validate_identifier(name)
7958 if not _alembic_quote:
7959 return name
7960 else:
7961 return self.quote(name)
7963 def format_index(self, index: Index) -> str:
7964 name = self.format_constraint(index)
7965 assert name is not None
7966 return name
7968 def format_table(
7969 self,
7970 table: FromClause,
7971 use_schema: bool = True,
7972 name: Optional[str] = None,
7973 ) -> str:
7974 """Prepare a quoted table and schema name."""
7975 if name is None:
7976 if TYPE_CHECKING:
7977 assert isinstance(table, NamedFromClause)
7978 name = table.name
7980 result = self.quote(name)
7982 effective_schema = self.schema_for_object(table)
7984 if not self.omit_schema and use_schema and effective_schema:
7985 result = self.quote_schema(effective_schema) + "." + result
7986 return result
7988 def format_schema(self, name):
7989 """Prepare a quoted schema name."""
7991 return self.quote(name)
7993 def format_label_name(
7994 self,
7995 name,
7996 anon_map=None,
7997 ):
7998 """Prepare a quoted column name."""
8000 if anon_map is not None and isinstance(
8001 name, elements._truncated_label
8002 ):
8003 name = name.apply_map(anon_map)
8005 return self.quote(name)
8007 def format_column(
8008 self,
8009 column: ColumnElement[Any],
8010 use_table: bool = False,
8011 name: Optional[str] = None,
8012 table_name: Optional[str] = None,
8013 use_schema: bool = False,
8014 anon_map: Optional[Mapping[str, Any]] = None,
8015 ) -> str:
8016 """Prepare a quoted column name."""
8018 if name is None:
8019 name = column.name
8020 assert name is not None
8022 if anon_map is not None and isinstance(
8023 name, elements._truncated_label
8024 ):
8025 name = name.apply_map(anon_map)
8027 if not getattr(column, "is_literal", False):
8028 if use_table:
8029 return (
8030 self.format_table(
8031 column.table, use_schema=use_schema, name=table_name
8032 )
8033 + "."
8034 + self.quote(name)
8035 )
8036 else:
8037 return self.quote(name)
8038 else:
8039 # literal textual elements get stuck into ColumnClause a lot,
8040 # which shouldn't get quoted
8042 if use_table:
8043 return (
8044 self.format_table(
8045 column.table, use_schema=use_schema, name=table_name
8046 )
8047 + "."
8048 + name
8049 )
8050 else:
8051 return name
8053 def format_table_seq(self, table, use_schema=True):
8054 """Format table name and schema as a tuple."""
8056 # Dialects with more levels in their fully qualified references
8057 # ('database', 'owner', etc.) could override this and return
8058 # a longer sequence.
8060 effective_schema = self.schema_for_object(table)
8062 if not self.omit_schema and use_schema and effective_schema:
8063 return (
8064 self.quote_schema(effective_schema),
8065 self.format_table(table, use_schema=False),
8066 )
8067 else:
8068 return (self.format_table(table, use_schema=False),)
8070 @util.memoized_property
8071 def _r_identifiers(self):
8072 initial, final, escaped_final = (
8073 re.escape(s)
8074 for s in (
8075 self.initial_quote,
8076 self.final_quote,
8077 self._escape_identifier(self.final_quote),
8078 )
8079 )
8080 r = re.compile(
8081 r"(?:"
8082 r"(?:%(initial)s((?:%(escaped)s|[^%(final)s])+)%(final)s"
8083 r"|([^\.]+))(?=\.|$))+"
8084 % {"initial": initial, "final": final, "escaped": escaped_final}
8085 )
8086 return r
8088 def unformat_identifiers(self, identifiers: str) -> Sequence[str]:
8089 """Unpack 'schema.table.column'-like strings into components."""
8091 r = self._r_identifiers
8092 return [
8093 self._unescape_identifier(i)
8094 for i in [a or b for a, b in r.findall(identifiers)]
8095 ]