diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/sqlalchemy/sql/compiler.py')
| -rw-r--r-- | venv/lib/python3.11/site-packages/sqlalchemy/sql/compiler.py | 7811 |
1 files changed, 0 insertions, 7811 deletions
diff --git a/venv/lib/python3.11/site-packages/sqlalchemy/sql/compiler.py b/venv/lib/python3.11/site-packages/sqlalchemy/sql/compiler.py deleted file mode 100644 index c354ba8..0000000 --- a/venv/lib/python3.11/site-packages/sqlalchemy/sql/compiler.py +++ /dev/null @@ -1,7811 +0,0 @@ -# sql/compiler.py -# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors -# <see AUTHORS file> -# -# This module is part of SQLAlchemy and is released under -# the MIT License: https://www.opensource.org/licenses/mit-license.php -# mypy: allow-untyped-defs, allow-untyped-calls - -"""Base SQL and DDL compiler implementations. - -Classes provided include: - -:class:`.compiler.SQLCompiler` - renders SQL -strings - -:class:`.compiler.DDLCompiler` - renders DDL -(data definition language) strings - -:class:`.compiler.GenericTypeCompiler` - renders -type specification strings. - -To generate user-defined SQL strings, see -:doc:`/ext/compiler`. - -""" -from __future__ import annotations - -import collections -import collections.abc as collections_abc -import contextlib -from enum import IntEnum -import functools -import itertools -import operator -import re -from time import perf_counter -import typing -from typing import Any -from typing import Callable -from typing import cast -from typing import ClassVar -from typing import Dict -from typing import FrozenSet -from typing import Iterable -from typing import Iterator -from typing import List -from typing import Mapping -from typing import MutableMapping -from typing import NamedTuple -from typing import NoReturn -from typing import Optional -from typing import Pattern -from typing import Sequence -from typing import Set -from typing import Tuple -from typing import Type -from typing import TYPE_CHECKING -from typing import Union - -from . import base -from . import coercions -from . import crud -from . import elements -from . import functions -from . import operators -from . import roles -from . import schema -from . import selectable -from . import sqltypes -from . import util as sql_util -from ._typing import is_column_element -from ._typing import is_dml -from .base import _de_clone -from .base import _from_objects -from .base import _NONE_NAME -from .base import _SentinelDefaultCharacterization -from .base import Executable -from .base import NO_ARG -from .elements import ClauseElement -from .elements import quoted_name -from .schema import Column -from .sqltypes import TupleType -from .type_api import TypeEngine -from .visitors import prefix_anon_map -from .visitors import Visitable -from .. import exc -from .. import util -from ..util import FastIntFlag -from ..util.typing import Literal -from ..util.typing import Protocol -from ..util.typing import TypedDict - -if typing.TYPE_CHECKING: - from .annotation import _AnnotationDict - from .base import _AmbiguousTableNameMap - from .base import CompileState - from .cache_key import CacheKey - from .ddl import ExecutableDDLElement - from .dml import Insert - from .dml import UpdateBase - from .dml import ValuesBase - from .elements import _truncated_label - from .elements import BindParameter - from .elements import ColumnClause - from .elements import ColumnElement - from .elements import Label - from .functions import Function - from .schema import Table - from .selectable import AliasedReturnsRows - from .selectable import CompoundSelectState - from .selectable import CTE - from .selectable import FromClause - from .selectable import NamedFromClause - from .selectable import ReturnsRows - from .selectable import Select - from .selectable import SelectState - from .type_api import _BindProcessorType - from ..engine.cursor import CursorResultMetaData - from ..engine.interfaces import _CoreSingleExecuteParams - from ..engine.interfaces import _DBAPIAnyExecuteParams - from ..engine.interfaces import _DBAPIMultiExecuteParams - from ..engine.interfaces import _DBAPISingleExecuteParams - from ..engine.interfaces import _ExecuteOptions - from ..engine.interfaces import _GenericSetInputSizesType - from ..engine.interfaces import _MutableCoreSingleExecuteParams - from ..engine.interfaces import Dialect - from ..engine.interfaces import SchemaTranslateMapType - -_FromHintsType = Dict["FromClause", str] - -RESERVED_WORDS = { - "all", - "analyse", - "analyze", - "and", - "any", - "array", - "as", - "asc", - "asymmetric", - "authorization", - "between", - "binary", - "both", - "case", - "cast", - "check", - "collate", - "column", - "constraint", - "create", - "cross", - "current_date", - "current_role", - "current_time", - "current_timestamp", - "current_user", - "default", - "deferrable", - "desc", - "distinct", - "do", - "else", - "end", - "except", - "false", - "for", - "foreign", - "freeze", - "from", - "full", - "grant", - "group", - "having", - "ilike", - "in", - "initially", - "inner", - "intersect", - "into", - "is", - "isnull", - "join", - "leading", - "left", - "like", - "limit", - "localtime", - "localtimestamp", - "natural", - "new", - "not", - "notnull", - "null", - "off", - "offset", - "old", - "on", - "only", - "or", - "order", - "outer", - "overlaps", - "placing", - "primary", - "references", - "right", - "select", - "session_user", - "set", - "similar", - "some", - "symmetric", - "table", - "then", - "to", - "trailing", - "true", - "union", - "unique", - "user", - "using", - "verbose", - "when", - "where", -} - -LEGAL_CHARACTERS = re.compile(r"^[A-Z0-9_$]+$", re.I) -LEGAL_CHARACTERS_PLUS_SPACE = re.compile(r"^[A-Z0-9_ $]+$", re.I) -ILLEGAL_INITIAL_CHARACTERS = {str(x) for x in range(0, 10)}.union(["$"]) - -FK_ON_DELETE = re.compile( - r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I -) -FK_ON_UPDATE = re.compile( - r"^(?:RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT)$", re.I -) -FK_INITIALLY = re.compile(r"^(?:DEFERRED|IMMEDIATE)$", re.I) -BIND_PARAMS = re.compile(r"(?<![:\w\$\x5c]):([\w\$]+)(?![:\w\$])", re.UNICODE) -BIND_PARAMS_ESC = re.compile(r"\x5c(:[\w\$]*)(?![:\w\$])", re.UNICODE) - -_pyformat_template = "%%(%(name)s)s" -BIND_TEMPLATES = { - "pyformat": _pyformat_template, - "qmark": "?", - "format": "%%s", - "numeric": ":[_POSITION]", - "numeric_dollar": "$[_POSITION]", - "named": ":%(name)s", -} - - -OPERATORS = { - # binary - operators.and_: " AND ", - operators.or_: " OR ", - operators.add: " + ", - operators.mul: " * ", - operators.sub: " - ", - operators.mod: " % ", - operators.neg: "-", - operators.lt: " < ", - operators.le: " <= ", - operators.ne: " != ", - operators.gt: " > ", - operators.ge: " >= ", - operators.eq: " = ", - operators.is_distinct_from: " IS DISTINCT FROM ", - operators.is_not_distinct_from: " IS NOT DISTINCT FROM ", - operators.concat_op: " || ", - operators.match_op: " MATCH ", - operators.not_match_op: " NOT MATCH ", - operators.in_op: " IN ", - operators.not_in_op: " NOT IN ", - operators.comma_op: ", ", - operators.from_: " FROM ", - operators.as_: " AS ", - operators.is_: " IS ", - operators.is_not: " IS NOT ", - operators.collate: " COLLATE ", - # unary - operators.exists: "EXISTS ", - operators.distinct_op: "DISTINCT ", - operators.inv: "NOT ", - operators.any_op: "ANY ", - operators.all_op: "ALL ", - # modifiers - operators.desc_op: " DESC", - operators.asc_op: " ASC", - operators.nulls_first_op: " NULLS FIRST", - operators.nulls_last_op: " NULLS LAST", - # bitwise - operators.bitwise_xor_op: " ^ ", - operators.bitwise_or_op: " | ", - operators.bitwise_and_op: " & ", - operators.bitwise_not_op: "~", - operators.bitwise_lshift_op: " << ", - operators.bitwise_rshift_op: " >> ", -} - -FUNCTIONS: Dict[Type[Function[Any]], str] = { - functions.coalesce: "coalesce", - functions.current_date: "CURRENT_DATE", - functions.current_time: "CURRENT_TIME", - functions.current_timestamp: "CURRENT_TIMESTAMP", - functions.current_user: "CURRENT_USER", - functions.localtime: "LOCALTIME", - functions.localtimestamp: "LOCALTIMESTAMP", - functions.random: "random", - functions.sysdate: "sysdate", - functions.session_user: "SESSION_USER", - functions.user: "USER", - functions.cube: "CUBE", - functions.rollup: "ROLLUP", - functions.grouping_sets: "GROUPING SETS", -} - - -EXTRACT_MAP = { - "month": "month", - "day": "day", - "year": "year", - "second": "second", - "hour": "hour", - "doy": "doy", - "minute": "minute", - "quarter": "quarter", - "dow": "dow", - "week": "week", - "epoch": "epoch", - "milliseconds": "milliseconds", - "microseconds": "microseconds", - "timezone_hour": "timezone_hour", - "timezone_minute": "timezone_minute", -} - -COMPOUND_KEYWORDS = { - selectable._CompoundSelectKeyword.UNION: "UNION", - selectable._CompoundSelectKeyword.UNION_ALL: "UNION ALL", - selectable._CompoundSelectKeyword.EXCEPT: "EXCEPT", - selectable._CompoundSelectKeyword.EXCEPT_ALL: "EXCEPT ALL", - selectable._CompoundSelectKeyword.INTERSECT: "INTERSECT", - selectable._CompoundSelectKeyword.INTERSECT_ALL: "INTERSECT ALL", -} - - -class ResultColumnsEntry(NamedTuple): - """Tracks a column expression that is expected to be represented - in the result rows for this statement. - - This normally refers to the columns clause of a SELECT statement - but may also refer to a RETURNING clause, as well as for dialect-specific - emulations. - - """ - - keyname: str - """string name that's expected in cursor.description""" - - name: str - """column name, may be labeled""" - - objects: Tuple[Any, ...] - """sequence of objects that should be able to locate this column - in a RowMapping. This is typically string names and aliases - as well as Column objects. - - """ - - type: TypeEngine[Any] - """Datatype to be associated with this column. This is where - the "result processing" logic directly links the compiled statement - to the rows that come back from the cursor. - - """ - - -class _ResultMapAppender(Protocol): - def __call__( - self, - keyname: str, - name: str, - objects: Sequence[Any], - type_: TypeEngine[Any], - ) -> None: ... - - -# integer indexes into ResultColumnsEntry used by cursor.py. -# some profiling showed integer access faster than named tuple -RM_RENDERED_NAME: Literal[0] = 0 -RM_NAME: Literal[1] = 1 -RM_OBJECTS: Literal[2] = 2 -RM_TYPE: Literal[3] = 3 - - -class _BaseCompilerStackEntry(TypedDict): - asfrom_froms: Set[FromClause] - correlate_froms: Set[FromClause] - selectable: ReturnsRows - - -class _CompilerStackEntry(_BaseCompilerStackEntry, total=False): - compile_state: CompileState - need_result_map_for_nested: bool - need_result_map_for_compound: bool - select_0: ReturnsRows - insert_from_select: Select[Any] - - -class ExpandedState(NamedTuple): - """represents state to use when producing "expanded" and - "post compile" bound parameters for a statement. - - "expanded" parameters are parameters that are generated at - statement execution time to suit a number of parameters passed, the most - prominent example being the individual elements inside of an IN expression. - - "post compile" parameters are parameters where the SQL literal value - will be rendered into the SQL statement at execution time, rather than - being passed as separate parameters to the driver. - - To create an :class:`.ExpandedState` instance, use the - :meth:`.SQLCompiler.construct_expanded_state` method on any - :class:`.SQLCompiler` instance. - - """ - - statement: str - """String SQL statement with parameters fully expanded""" - - parameters: _CoreSingleExecuteParams - """Parameter dictionary with parameters fully expanded. - - For a statement that uses named parameters, this dictionary will map - exactly to the names in the statement. For a statement that uses - positional parameters, the :attr:`.ExpandedState.positional_parameters` - will yield a tuple with the positional parameter set. - - """ - - processors: Mapping[str, _BindProcessorType[Any]] - """mapping of bound value processors""" - - positiontup: Optional[Sequence[str]] - """Sequence of string names indicating the order of positional - parameters""" - - parameter_expansion: Mapping[str, List[str]] - """Mapping representing the intermediary link from original parameter - name to list of "expanded" parameter names, for those parameters that - were expanded.""" - - @property - def positional_parameters(self) -> Tuple[Any, ...]: - """Tuple of positional parameters, for statements that were compiled - using a positional paramstyle. - - """ - if self.positiontup is None: - raise exc.InvalidRequestError( - "statement does not use a positional paramstyle" - ) - return tuple(self.parameters[key] for key in self.positiontup) - - @property - def additional_parameters(self) -> _CoreSingleExecuteParams: - """synonym for :attr:`.ExpandedState.parameters`.""" - return self.parameters - - -class _InsertManyValues(NamedTuple): - """represents state to use for executing an "insertmanyvalues" statement. - - The primary consumers of this object are the - :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and - :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods. - - .. versionadded:: 2.0 - - """ - - is_default_expr: bool - """if True, the statement is of the form - ``INSERT INTO TABLE DEFAULT VALUES``, and can't be rewritten as a "batch" - - """ - - single_values_expr: str - """The rendered "values" clause of the INSERT statement. - - This is typically the parenthesized section e.g. "(?, ?, ?)" or similar. - The insertmanyvalues logic uses this string as a search and replace - target. - - """ - - insert_crud_params: List[crud._CrudParamElementStr] - """List of Column / bind names etc. used while rewriting the statement""" - - num_positional_params_counted: int - """the number of bound parameters in a single-row statement. - - This count may be larger or smaller than the actual number of columns - targeted in the INSERT, as it accommodates for SQL expressions - in the values list that may have zero or more parameters embedded - within them. - - This count is part of what's used to organize rewritten parameter lists - when batching. - - """ - - sort_by_parameter_order: bool = False - """if the deterministic_returnined_order parameter were used on the - insert. - - All of the attributes following this will only be used if this is True. - - """ - - includes_upsert_behaviors: bool = False - """if True, we have to accommodate for upsert behaviors. - - This will in some cases downgrade "insertmanyvalues" that requests - deterministic ordering. - - """ - - sentinel_columns: Optional[Sequence[Column[Any]]] = None - """List of sentinel columns that were located. - - This list is only here if the INSERT asked for - sort_by_parameter_order=True, - and dialect-appropriate sentinel columns were located. - - .. versionadded:: 2.0.10 - - """ - - num_sentinel_columns: int = 0 - """how many sentinel columns are in the above list, if any. - - This is the same as - ``len(sentinel_columns) if sentinel_columns is not None else 0`` - - """ - - sentinel_param_keys: Optional[Sequence[str]] = None - """parameter str keys in each param dictionary / tuple - that would link to the client side "sentinel" values for that row, which - we can use to match up parameter sets to result rows. - - This is only present if sentinel_columns is present and the INSERT - statement actually refers to client side values for these sentinel - columns. - - .. versionadded:: 2.0.10 - - .. versionchanged:: 2.0.29 - the sequence is now string dictionary keys - only, used against the "compiled parameteters" collection before - the parameters were converted by bound parameter processors - - """ - - implicit_sentinel: bool = False - """if True, we have exactly one sentinel column and it uses a server side - value, currently has to generate an incrementing integer value. - - The dialect in question would have asserted that it supports receiving - these values back and sorting on that value as a means of guaranteeing - correlation with the incoming parameter list. - - .. versionadded:: 2.0.10 - - """ - - embed_values_counter: bool = False - """Whether to embed an incrementing integer counter in each parameter - set within the VALUES clause as parameters are batched over. - - This is only used for a specific INSERT..SELECT..VALUES..RETURNING syntax - where a subquery is used to produce value tuples. Current support - includes PostgreSQL, Microsoft SQL Server. - - .. versionadded:: 2.0.10 - - """ - - -class _InsertManyValuesBatch(NamedTuple): - """represents an individual batch SQL statement for insertmanyvalues. - - This is passed through the - :meth:`.SQLCompiler._deliver_insertmanyvalues_batches` and - :meth:`.DefaultDialect._deliver_insertmanyvalues_batches` methods out - to the :class:`.Connection` within the - :meth:`.Connection._exec_insertmany_context` method. - - .. versionadded:: 2.0.10 - - """ - - replaced_statement: str - replaced_parameters: _DBAPIAnyExecuteParams - processed_setinputsizes: Optional[_GenericSetInputSizesType] - batch: Sequence[_DBAPISingleExecuteParams] - sentinel_values: Sequence[Tuple[Any, ...]] - current_batch_size: int - batchnum: int - total_batches: int - rows_sorted: bool - is_downgraded: bool - - -class InsertmanyvaluesSentinelOpts(FastIntFlag): - """bitflag enum indicating styles of PK defaults - which can work as implicit sentinel columns - - """ - - NOT_SUPPORTED = 1 - AUTOINCREMENT = 2 - IDENTITY = 4 - SEQUENCE = 8 - - ANY_AUTOINCREMENT = AUTOINCREMENT | IDENTITY | SEQUENCE - _SUPPORTED_OR_NOT = NOT_SUPPORTED | ANY_AUTOINCREMENT - - USE_INSERT_FROM_SELECT = 16 - RENDER_SELECT_COL_CASTS = 64 - - -class CompilerState(IntEnum): - COMPILING = 0 - """statement is present, compilation phase in progress""" - - STRING_APPLIED = 1 - """statement is present, string form of the statement has been applied. - - Additional processors by subclasses may still be pending. - - """ - - NO_STATEMENT = 2 - """compiler does not have a statement to compile, is used - for method access""" - - -class Linting(IntEnum): - """represent preferences for the 'SQL linting' feature. - - this feature currently includes support for flagging cartesian products - in SQL statements. - - """ - - NO_LINTING = 0 - "Disable all linting." - - COLLECT_CARTESIAN_PRODUCTS = 1 - """Collect data on FROMs and cartesian products and gather into - 'self.from_linter'""" - - WARN_LINTING = 2 - "Emit warnings for linters that find problems" - - FROM_LINTING = COLLECT_CARTESIAN_PRODUCTS | WARN_LINTING - """Warn for cartesian products; combines COLLECT_CARTESIAN_PRODUCTS - and WARN_LINTING""" - - -NO_LINTING, COLLECT_CARTESIAN_PRODUCTS, WARN_LINTING, FROM_LINTING = tuple( - Linting -) - - -class FromLinter(collections.namedtuple("FromLinter", ["froms", "edges"])): - """represents current state for the "cartesian product" detection - feature.""" - - def lint(self, start=None): - froms = self.froms - if not froms: - return None, None - - edges = set(self.edges) - the_rest = set(froms) - - if start is not None: - start_with = start - the_rest.remove(start_with) - else: - start_with = the_rest.pop() - - stack = collections.deque([start_with]) - - while stack and the_rest: - node = stack.popleft() - the_rest.discard(node) - - # comparison of nodes in edges here is based on hash equality, as - # there are "annotated" elements that match the non-annotated ones. - # to remove the need for in-python hash() calls, use native - # containment routines (e.g. "node in edge", "edge.index(node)") - to_remove = {edge for edge in edges if node in edge} - - # appendleft the node in each edge that is not - # the one that matched. - stack.extendleft(edge[not edge.index(node)] for edge in to_remove) - edges.difference_update(to_remove) - - # FROMS left over? boom - if the_rest: - return the_rest, start_with - else: - return None, None - - def warn(self, stmt_type="SELECT"): - the_rest, start_with = self.lint() - - # FROMS left over? boom - if the_rest: - froms = the_rest - if froms: - template = ( - "{stmt_type} statement has a cartesian product between " - "FROM element(s) {froms} and " - 'FROM element "{start}". Apply join condition(s) ' - "between each element to resolve." - ) - froms_str = ", ".join( - f'"{self.froms[from_]}"' for from_ in froms - ) - message = template.format( - stmt_type=stmt_type, - froms=froms_str, - start=self.froms[start_with], - ) - - util.warn(message) - - -class Compiled: - """Represent a compiled SQL or DDL expression. - - The ``__str__`` method of the ``Compiled`` object should produce - the actual text of the statement. ``Compiled`` objects are - specific to their underlying database dialect, and also may - or may not be specific to the columns referenced within a - particular set of bind parameters. In no case should the - ``Compiled`` object be dependent on the actual values of those - bind parameters, even though it may reference those values as - defaults. - """ - - statement: Optional[ClauseElement] = None - "The statement to compile." - string: str = "" - "The string representation of the ``statement``" - - state: CompilerState - """description of the compiler's state""" - - is_sql = False - is_ddl = False - - _cached_metadata: Optional[CursorResultMetaData] = None - - _result_columns: Optional[List[ResultColumnsEntry]] = None - - schema_translate_map: Optional[SchemaTranslateMapType] = None - - execution_options: _ExecuteOptions = util.EMPTY_DICT - """ - Execution options propagated from the statement. In some cases, - sub-elements of the statement can modify these. - """ - - preparer: IdentifierPreparer - - _annotations: _AnnotationDict = util.EMPTY_DICT - - compile_state: Optional[CompileState] = None - """Optional :class:`.CompileState` object that maintains additional - state used by the compiler. - - Major executable objects such as :class:`_expression.Insert`, - :class:`_expression.Update`, :class:`_expression.Delete`, - :class:`_expression.Select` will generate this - state when compiled in order to calculate additional information about the - object. For the top level object that is to be executed, the state can be - stored here where it can also have applicability towards result set - processing. - - .. versionadded:: 1.4 - - """ - - dml_compile_state: Optional[CompileState] = None - """Optional :class:`.CompileState` assigned at the same point that - .isinsert, .isupdate, or .isdelete is assigned. - - This will normally be the same object as .compile_state, with the - exception of cases like the :class:`.ORMFromStatementCompileState` - object. - - .. versionadded:: 1.4.40 - - """ - - cache_key: Optional[CacheKey] = None - """The :class:`.CacheKey` that was generated ahead of creating this - :class:`.Compiled` object. - - This is used for routines that need access to the original - :class:`.CacheKey` instance generated when the :class:`.Compiled` - instance was first cached, typically in order to reconcile - the original list of :class:`.BindParameter` objects with a - per-statement list that's generated on each call. - - """ - - _gen_time: float - """Generation time of this :class:`.Compiled`, used for reporting - cache stats.""" - - def __init__( - self, - dialect: Dialect, - statement: Optional[ClauseElement], - schema_translate_map: Optional[SchemaTranslateMapType] = None, - render_schema_translate: bool = False, - compile_kwargs: Mapping[str, Any] = util.immutabledict(), - ): - """Construct a new :class:`.Compiled` object. - - :param dialect: :class:`.Dialect` to compile against. - - :param statement: :class:`_expression.ClauseElement` to be compiled. - - :param schema_translate_map: dictionary of schema names to be - translated when forming the resultant SQL - - .. seealso:: - - :ref:`schema_translating` - - :param compile_kwargs: additional kwargs that will be - passed to the initial call to :meth:`.Compiled.process`. - - - """ - self.dialect = dialect - self.preparer = self.dialect.identifier_preparer - if schema_translate_map: - self.schema_translate_map = schema_translate_map - self.preparer = self.preparer._with_schema_translate( - schema_translate_map - ) - - if statement is not None: - self.state = CompilerState.COMPILING - self.statement = statement - self.can_execute = statement.supports_execution - self._annotations = statement._annotations - if self.can_execute: - if TYPE_CHECKING: - assert isinstance(statement, Executable) - self.execution_options = statement._execution_options - self.string = self.process(self.statement, **compile_kwargs) - - if render_schema_translate: - self.string = self.preparer._render_schema_translates( - self.string, schema_translate_map - ) - - self.state = CompilerState.STRING_APPLIED - else: - self.state = CompilerState.NO_STATEMENT - - self._gen_time = perf_counter() - - def __init_subclass__(cls) -> None: - cls._init_compiler_cls() - return super().__init_subclass__() - - @classmethod - def _init_compiler_cls(cls): - pass - - def _execute_on_connection( - self, connection, distilled_params, execution_options - ): - if self.can_execute: - return connection._execute_compiled( - self, distilled_params, execution_options - ) - else: - raise exc.ObjectNotExecutableError(self.statement) - - def visit_unsupported_compilation(self, element, err, **kw): - raise exc.UnsupportedCompilationError(self, type(element)) from err - - @property - def sql_compiler(self): - """Return a Compiled that is capable of processing SQL expressions. - - If this compiler is one, it would likely just return 'self'. - - """ - - raise NotImplementedError() - - def process(self, obj: Visitable, **kwargs: Any) -> str: - return obj._compiler_dispatch(self, **kwargs) - - def __str__(self) -> str: - """Return the string text of the generated SQL or DDL.""" - - if self.state is CompilerState.STRING_APPLIED: - return self.string - else: - return "" - - def construct_params( - self, - params: Optional[_CoreSingleExecuteParams] = None, - extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None, - escape_names: bool = True, - ) -> Optional[_MutableCoreSingleExecuteParams]: - """Return the bind params for this compiled object. - - :param params: a dict of string/object pairs whose values will - override bind values compiled in to the - statement. - """ - - raise NotImplementedError() - - @property - def params(self): - """Return the bind params for this compiled object.""" - return self.construct_params() - - -class TypeCompiler(util.EnsureKWArg): - """Produces DDL specification for TypeEngine objects.""" - - ensure_kwarg = r"visit_\w+" - - def __init__(self, dialect: Dialect): - self.dialect = dialect - - def process(self, type_: TypeEngine[Any], **kw: Any) -> str: - if ( - type_._variant_mapping - and self.dialect.name in type_._variant_mapping - ): - type_ = type_._variant_mapping[self.dialect.name] - return type_._compiler_dispatch(self, **kw) - - def visit_unsupported_compilation( - self, element: Any, err: Exception, **kw: Any - ) -> NoReturn: - raise exc.UnsupportedCompilationError(self, element) from err - - -# this was a Visitable, but to allow accurate detection of -# column elements this is actually a column element -class _CompileLabel( - roles.BinaryElementRole[Any], elements.CompilerColumnElement -): - """lightweight label object which acts as an expression.Label.""" - - __visit_name__ = "label" - __slots__ = "element", "name", "_alt_names" - - def __init__(self, col, name, alt_names=()): - self.element = col - self.name = name - self._alt_names = (col,) + alt_names - - @property - def proxy_set(self): - return self.element.proxy_set - - @property - def type(self): - return self.element.type - - def self_group(self, **kw): - return self - - -class ilike_case_insensitive( - roles.BinaryElementRole[Any], elements.CompilerColumnElement -): - """produce a wrapping element for a case-insensitive portion of - an ILIKE construct. - - The construct usually renders the ``lower()`` function, but on - PostgreSQL will pass silently with the assumption that "ILIKE" - is being used. - - .. versionadded:: 2.0 - - """ - - __visit_name__ = "ilike_case_insensitive_operand" - __slots__ = "element", "comparator" - - def __init__(self, element): - self.element = element - self.comparator = element.comparator - - @property - def proxy_set(self): - return self.element.proxy_set - - @property - def type(self): - return self.element.type - - def self_group(self, **kw): - return self - - def _with_binary_element_type(self, type_): - return ilike_case_insensitive( - self.element._with_binary_element_type(type_) - ) - - -class SQLCompiler(Compiled): - """Default implementation of :class:`.Compiled`. - - Compiles :class:`_expression.ClauseElement` objects into SQL strings. - - """ - - extract_map = EXTRACT_MAP - - bindname_escape_characters: ClassVar[Mapping[str, str]] = ( - util.immutabledict( - { - "%": "P", - "(": "A", - ")": "Z", - ":": "C", - ".": "_", - "[": "_", - "]": "_", - " ": "_", - } - ) - ) - """A mapping (e.g. dict or similar) containing a lookup of - characters keyed to replacement characters which will be applied to all - 'bind names' used in SQL statements as a form of 'escaping'; the given - characters are replaced entirely with the 'replacement' character when - rendered in the SQL statement, and a similar translation is performed - on the incoming names used in parameter dictionaries passed to methods - like :meth:`_engine.Connection.execute`. - - This allows bound parameter names used in :func:`_sql.bindparam` and - other constructs to have any arbitrary characters present without any - concern for characters that aren't allowed at all on the target database. - - Third party dialects can establish their own dictionary here to replace the - default mapping, which will ensure that the particular characters in the - mapping will never appear in a bound parameter name. - - The dictionary is evaluated at **class creation time**, so cannot be - modified at runtime; it must be present on the class when the class - is first declared. - - Note that for dialects that have additional bound parameter rules such - as additional restrictions on leading characters, the - :meth:`_sql.SQLCompiler.bindparam_string` method may need to be augmented. - See the cx_Oracle compiler for an example of this. - - .. versionadded:: 2.0.0rc1 - - """ - - _bind_translate_re: ClassVar[Pattern[str]] - _bind_translate_chars: ClassVar[Mapping[str, str]] - - is_sql = True - - compound_keywords = COMPOUND_KEYWORDS - - isdelete: bool = False - isinsert: bool = False - isupdate: bool = False - """class-level defaults which can be set at the instance - level to define if this Compiled instance represents - INSERT/UPDATE/DELETE - """ - - postfetch: Optional[List[Column[Any]]] - """list of columns that can be post-fetched after INSERT or UPDATE to - receive server-updated values""" - - insert_prefetch: Sequence[Column[Any]] = () - """list of columns for which default values should be evaluated before - an INSERT takes place""" - - update_prefetch: Sequence[Column[Any]] = () - """list of columns for which onupdate default values should be evaluated - before an UPDATE takes place""" - - implicit_returning: Optional[Sequence[ColumnElement[Any]]] = None - """list of "implicit" returning columns for a toplevel INSERT or UPDATE - statement, used to receive newly generated values of columns. - - .. versionadded:: 2.0 ``implicit_returning`` replaces the previous - ``returning`` collection, which was not a generalized RETURNING - collection and instead was in fact specific to the "implicit returning" - feature. - - """ - - isplaintext: bool = False - - binds: Dict[str, BindParameter[Any]] - """a dictionary of bind parameter keys to BindParameter instances.""" - - bind_names: Dict[BindParameter[Any], str] - """a dictionary of BindParameter instances to "compiled" names - that are actually present in the generated SQL""" - - stack: List[_CompilerStackEntry] - """major statements such as SELECT, INSERT, UPDATE, DELETE are - tracked in this stack using an entry format.""" - - returning_precedes_values: bool = False - """set to True classwide to generate RETURNING - clauses before the VALUES or WHERE clause (i.e. MSSQL) - """ - - render_table_with_column_in_update_from: bool = False - """set to True classwide to indicate the SET clause - in a multi-table UPDATE statement should qualify - columns with the table name (i.e. MySQL only) - """ - - ansi_bind_rules: bool = False - """SQL 92 doesn't allow bind parameters to be used - in the columns clause of a SELECT, nor does it allow - ambiguous expressions like "? = ?". A compiler - subclass can set this flag to False if the target - driver/DB enforces this - """ - - bindtemplate: str - """template to render bound parameters based on paramstyle.""" - - compilation_bindtemplate: str - """template used by compiler to render parameters before positional - paramstyle application""" - - _numeric_binds_identifier_char: str - """Character that's used to as the identifier of a numerical bind param. - For example if this char is set to ``$``, numerical binds will be rendered - in the form ``$1, $2, $3``. - """ - - _result_columns: List[ResultColumnsEntry] - """relates label names in the final SQL to a tuple of local - column/label name, ColumnElement object (if any) and - TypeEngine. CursorResult uses this for type processing and - column targeting""" - - _textual_ordered_columns: bool = False - """tell the result object that the column names as rendered are important, - but they are also "ordered" vs. what is in the compiled object here. - - As of 1.4.42 this condition is only present when the statement is a - TextualSelect, e.g. text("....").columns(...), where it is required - that the columns are considered positionally and not by name. - - """ - - _ad_hoc_textual: bool = False - """tell the result that we encountered text() or '*' constructs in the - middle of the result columns, but we also have compiled columns, so - if the number of columns in cursor.description does not match how many - expressions we have, that means we can't rely on positional at all and - should match on name. - - """ - - _ordered_columns: bool = True - """ - if False, means we can't be sure the list of entries - in _result_columns is actually the rendered order. Usually - True unless using an unordered TextualSelect. - """ - - _loose_column_name_matching: bool = False - """tell the result object that the SQL statement is textual, wants to match - up to Column objects, and may be using the ._tq_label in the SELECT rather - than the base name. - - """ - - _numeric_binds: bool = False - """ - True if paramstyle is "numeric". This paramstyle is trickier than - all the others. - - """ - - _render_postcompile: bool = False - """ - whether to render out POSTCOMPILE params during the compile phase. - - This attribute is used only for end-user invocation of stmt.compile(); - it's never used for actual statement execution, where instead the - dialect internals access and render the internal postcompile structure - directly. - - """ - - _post_compile_expanded_state: Optional[ExpandedState] = None - """When render_postcompile is used, the ``ExpandedState`` used to create - the "expanded" SQL is assigned here, and then used by the ``.params`` - accessor and ``.construct_params()`` methods for their return values. - - .. versionadded:: 2.0.0rc1 - - """ - - _pre_expanded_string: Optional[str] = None - """Stores the original string SQL before 'post_compile' is applied, - for cases where 'post_compile' were used. - - """ - - _pre_expanded_positiontup: Optional[List[str]] = None - - _insertmanyvalues: Optional[_InsertManyValues] = None - - _insert_crud_params: Optional[crud._CrudParamSequence] = None - - literal_execute_params: FrozenSet[BindParameter[Any]] = frozenset() - """bindparameter objects that are rendered as literal values at statement - execution time. - - """ - - post_compile_params: FrozenSet[BindParameter[Any]] = frozenset() - """bindparameter objects that are rendered as bound parameter placeholders - at statement execution time. - - """ - - escaped_bind_names: util.immutabledict[str, str] = util.EMPTY_DICT - """Late escaping of bound parameter names that has to be converted - to the original name when looking in the parameter dictionary. - - """ - - has_out_parameters = False - """if True, there are bindparam() objects that have the isoutparam - flag set.""" - - postfetch_lastrowid = False - """if True, and this in insert, use cursor.lastrowid to populate - result.inserted_primary_key. """ - - _cache_key_bind_match: Optional[ - Tuple[ - Dict[ - BindParameter[Any], - List[BindParameter[Any]], - ], - Dict[ - str, - BindParameter[Any], - ], - ] - ] = None - """a mapping that will relate the BindParameter object we compile - to those that are part of the extracted collection of parameters - in the cache key, if we were given a cache key. - - """ - - positiontup: Optional[List[str]] = None - """for a compiled construct that uses a positional paramstyle, will be - a sequence of strings, indicating the names of bound parameters in order. - - This is used in order to render bound parameters in their correct order, - and is combined with the :attr:`_sql.Compiled.params` dictionary to - render parameters. - - This sequence always contains the unescaped name of the parameters. - - .. seealso:: - - :ref:`faq_sql_expression_string` - includes a usage example for - debugging use cases. - - """ - _values_bindparam: Optional[List[str]] = None - - _visited_bindparam: Optional[List[str]] = None - - inline: bool = False - - ctes: Optional[MutableMapping[CTE, str]] - - # Detect same CTE references - Dict[(level, name), cte] - # Level is required for supporting nesting - ctes_by_level_name: Dict[Tuple[int, str], CTE] - - # To retrieve key/level in ctes_by_level_name - - # Dict[cte_reference, (level, cte_name, cte_opts)] - level_name_by_cte: Dict[CTE, Tuple[int, str, selectable._CTEOpts]] - - ctes_recursive: bool - - _post_compile_pattern = re.compile(r"__\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]") - _pyformat_pattern = re.compile(r"%\(([^)]+?)\)s") - _positional_pattern = re.compile( - f"{_pyformat_pattern.pattern}|{_post_compile_pattern.pattern}" - ) - - @classmethod - def _init_compiler_cls(cls): - cls._init_bind_translate() - - @classmethod - def _init_bind_translate(cls): - reg = re.escape("".join(cls.bindname_escape_characters)) - cls._bind_translate_re = re.compile(f"[{reg}]") - cls._bind_translate_chars = cls.bindname_escape_characters - - def __init__( - self, - dialect: Dialect, - statement: Optional[ClauseElement], - cache_key: Optional[CacheKey] = None, - column_keys: Optional[Sequence[str]] = None, - for_executemany: bool = False, - linting: Linting = NO_LINTING, - _supporting_against: Optional[SQLCompiler] = None, - **kwargs: Any, - ): - """Construct a new :class:`.SQLCompiler` object. - - :param dialect: :class:`.Dialect` to be used - - :param statement: :class:`_expression.ClauseElement` to be compiled - - :param column_keys: a list of column names to be compiled into an - INSERT or UPDATE statement. - - :param for_executemany: whether INSERT / UPDATE statements should - expect that they are to be invoked in an "executemany" style, - which may impact how the statement will be expected to return the - values of defaults and autoincrement / sequences and similar. - Depending on the backend and driver in use, support for retrieving - these values may be disabled which means SQL expressions may - be rendered inline, RETURNING may not be rendered, etc. - - :param kwargs: additional keyword arguments to be consumed by the - superclass. - - """ - self.column_keys = column_keys - - self.cache_key = cache_key - - if cache_key: - cksm = {b.key: b for b in cache_key[1]} - ckbm = {b: [b] for b in cache_key[1]} - self._cache_key_bind_match = (ckbm, cksm) - - # compile INSERT/UPDATE defaults/sequences to expect executemany - # style execution, which may mean no pre-execute of defaults, - # or no RETURNING - self.for_executemany = for_executemany - - self.linting = linting - - # a dictionary of bind parameter keys to BindParameter - # instances. - self.binds = {} - - # a dictionary of BindParameter instances to "compiled" names - # that are actually present in the generated SQL - self.bind_names = util.column_dict() - - # stack which keeps track of nested SELECT statements - self.stack = [] - - self._result_columns = [] - - # true if the paramstyle is positional - self.positional = dialect.positional - if self.positional: - self._numeric_binds = nb = dialect.paramstyle.startswith("numeric") - if nb: - self._numeric_binds_identifier_char = ( - "$" if dialect.paramstyle == "numeric_dollar" else ":" - ) - - self.compilation_bindtemplate = _pyformat_template - else: - self.compilation_bindtemplate = BIND_TEMPLATES[dialect.paramstyle] - - self.ctes = None - - self.label_length = ( - dialect.label_length or dialect.max_identifier_length - ) - - # a map which tracks "anonymous" identifiers that are created on - # the fly here - self.anon_map = prefix_anon_map() - - # a map which tracks "truncated" names based on - # dialect.label_length or dialect.max_identifier_length - self.truncated_names: Dict[Tuple[str, str], str] = {} - self._truncated_counters: Dict[str, int] = {} - - Compiled.__init__(self, dialect, statement, **kwargs) - - if self.isinsert or self.isupdate or self.isdelete: - if TYPE_CHECKING: - assert isinstance(statement, UpdateBase) - - if self.isinsert or self.isupdate: - if TYPE_CHECKING: - assert isinstance(statement, ValuesBase) - if statement._inline: - self.inline = True - elif self.for_executemany and ( - not self.isinsert - or ( - self.dialect.insert_executemany_returning - and statement._return_defaults - ) - ): - self.inline = True - - self.bindtemplate = BIND_TEMPLATES[dialect.paramstyle] - - if _supporting_against: - self.__dict__.update( - { - k: v - for k, v in _supporting_against.__dict__.items() - if k - not in { - "state", - "dialect", - "preparer", - "positional", - "_numeric_binds", - "compilation_bindtemplate", - "bindtemplate", - } - } - ) - - if self.state is CompilerState.STRING_APPLIED: - if self.positional: - if self._numeric_binds: - self._process_numeric() - else: - self._process_positional() - - if self._render_postcompile: - parameters = self.construct_params( - escape_names=False, - _no_postcompile=True, - ) - - self._process_parameters_for_postcompile( - parameters, _populate_self=True - ) - - @property - def insert_single_values_expr(self) -> Optional[str]: - """When an INSERT is compiled with a single set of parameters inside - a VALUES expression, the string is assigned here, where it can be - used for insert batching schemes to rewrite the VALUES expression. - - .. versionadded:: 1.3.8 - - .. versionchanged:: 2.0 This collection is no longer used by - SQLAlchemy's built-in dialects, in favor of the currently - internal ``_insertmanyvalues`` collection that is used only by - :class:`.SQLCompiler`. - - """ - if self._insertmanyvalues is None: - return None - else: - return self._insertmanyvalues.single_values_expr - - @util.ro_memoized_property - def effective_returning(self) -> Optional[Sequence[ColumnElement[Any]]]: - """The effective "returning" columns for INSERT, UPDATE or DELETE. - - This is either the so-called "implicit returning" columns which are - calculated by the compiler on the fly, or those present based on what's - present in ``self.statement._returning`` (expanded into individual - columns using the ``._all_selected_columns`` attribute) i.e. those set - explicitly using the :meth:`.UpdateBase.returning` method. - - .. versionadded:: 2.0 - - """ - if self.implicit_returning: - return self.implicit_returning - elif self.statement is not None and is_dml(self.statement): - return [ - c - for c in self.statement._all_selected_columns - if is_column_element(c) - ] - - else: - return None - - @property - def returning(self): - """backwards compatibility; returns the - effective_returning collection. - - """ - return self.effective_returning - - @property - def current_executable(self): - """Return the current 'executable' that is being compiled. - - This is currently the :class:`_sql.Select`, :class:`_sql.Insert`, - :class:`_sql.Update`, :class:`_sql.Delete`, - :class:`_sql.CompoundSelect` object that is being compiled. - Specifically it's assigned to the ``self.stack`` list of elements. - - When a statement like the above is being compiled, it normally - is also assigned to the ``.statement`` attribute of the - :class:`_sql.Compiler` object. However, all SQL constructs are - ultimately nestable, and this attribute should never be consulted - by a ``visit_`` method, as it is not guaranteed to be assigned - nor guaranteed to correspond to the current statement being compiled. - - .. versionadded:: 1.3.21 - - For compatibility with previous versions, use the following - recipe:: - - statement = getattr(self, "current_executable", False) - if statement is False: - statement = self.stack[-1]["selectable"] - - For versions 1.4 and above, ensure only .current_executable - is used; the format of "self.stack" may change. - - - """ - try: - return self.stack[-1]["selectable"] - except IndexError as ie: - raise IndexError("Compiler does not have a stack entry") from ie - - @property - def prefetch(self): - return list(self.insert_prefetch) + list(self.update_prefetch) - - @util.memoized_property - def _global_attributes(self) -> Dict[Any, Any]: - return {} - - @util.memoized_instancemethod - def _init_cte_state(self) -> MutableMapping[CTE, str]: - """Initialize collections related to CTEs only if - a CTE is located, to save on the overhead of - these collections otherwise. - - """ - # collect CTEs to tack on top of a SELECT - # To store the query to print - Dict[cte, text_query] - ctes: MutableMapping[CTE, str] = util.OrderedDict() - self.ctes = ctes - - # Detect same CTE references - Dict[(level, name), cte] - # Level is required for supporting nesting - self.ctes_by_level_name = {} - - # To retrieve key/level in ctes_by_level_name - - # Dict[cte_reference, (level, cte_name, cte_opts)] - self.level_name_by_cte = {} - - self.ctes_recursive = False - - return ctes - - @contextlib.contextmanager - def _nested_result(self): - """special API to support the use case of 'nested result sets'""" - result_columns, ordered_columns = ( - self._result_columns, - self._ordered_columns, - ) - self._result_columns, self._ordered_columns = [], False - - try: - if self.stack: - entry = self.stack[-1] - entry["need_result_map_for_nested"] = True - else: - entry = None - yield self._result_columns, self._ordered_columns - finally: - if entry: - entry.pop("need_result_map_for_nested") - self._result_columns, self._ordered_columns = ( - result_columns, - ordered_columns, - ) - - def _process_positional(self): - assert not self.positiontup - assert self.state is CompilerState.STRING_APPLIED - assert not self._numeric_binds - - if self.dialect.paramstyle == "format": - placeholder = "%s" - else: - assert self.dialect.paramstyle == "qmark" - placeholder = "?" - - positions = [] - - def find_position(m: re.Match[str]) -> str: - normal_bind = m.group(1) - if normal_bind: - positions.append(normal_bind) - return placeholder - else: - # this a post-compile bind - positions.append(m.group(2)) - return m.group(0) - - self.string = re.sub( - self._positional_pattern, find_position, self.string - ) - - if self.escaped_bind_names: - reverse_escape = {v: k for k, v in self.escaped_bind_names.items()} - assert len(self.escaped_bind_names) == len(reverse_escape) - self.positiontup = [ - reverse_escape.get(name, name) for name in positions - ] - else: - self.positiontup = positions - - if self._insertmanyvalues: - positions = [] - - single_values_expr = re.sub( - self._positional_pattern, - find_position, - self._insertmanyvalues.single_values_expr, - ) - insert_crud_params = [ - ( - v[0], - v[1], - re.sub(self._positional_pattern, find_position, v[2]), - v[3], - ) - for v in self._insertmanyvalues.insert_crud_params - ] - - self._insertmanyvalues = self._insertmanyvalues._replace( - single_values_expr=single_values_expr, - insert_crud_params=insert_crud_params, - ) - - def _process_numeric(self): - assert self._numeric_binds - assert self.state is CompilerState.STRING_APPLIED - - num = 1 - param_pos: Dict[str, str] = {} - order: Iterable[str] - if self._insertmanyvalues and self._values_bindparam is not None: - # bindparams that are not in values are always placed first. - # this avoids the need of changing them when using executemany - # values () () - order = itertools.chain( - ( - name - for name in self.bind_names.values() - if name not in self._values_bindparam - ), - self.bind_names.values(), - ) - else: - order = self.bind_names.values() - - for bind_name in order: - if bind_name in param_pos: - continue - bind = self.binds[bind_name] - if ( - bind in self.post_compile_params - or bind in self.literal_execute_params - ): - # set to None to just mark the in positiontup, it will not - # be replaced below. - param_pos[bind_name] = None # type: ignore - else: - ph = f"{self._numeric_binds_identifier_char}{num}" - num += 1 - param_pos[bind_name] = ph - - self.next_numeric_pos = num - - self.positiontup = list(param_pos) - if self.escaped_bind_names: - len_before = len(param_pos) - param_pos = { - self.escaped_bind_names.get(name, name): pos - for name, pos in param_pos.items() - } - assert len(param_pos) == len_before - - # Can't use format here since % chars are not escaped. - self.string = self._pyformat_pattern.sub( - lambda m: param_pos[m.group(1)], self.string - ) - - if self._insertmanyvalues: - single_values_expr = ( - # format is ok here since single_values_expr includes only - # place-holders - self._insertmanyvalues.single_values_expr - % param_pos - ) - insert_crud_params = [ - (v[0], v[1], "%s", v[3]) - for v in self._insertmanyvalues.insert_crud_params - ] - - self._insertmanyvalues = self._insertmanyvalues._replace( - # This has the numbers (:1, :2) - single_values_expr=single_values_expr, - # The single binds are instead %s so they can be formatted - insert_crud_params=insert_crud_params, - ) - - @util.memoized_property - def _bind_processors( - self, - ) -> MutableMapping[ - str, Union[_BindProcessorType[Any], Sequence[_BindProcessorType[Any]]] - ]: - # mypy is not able to see the two value types as the above Union, - # it just sees "object". don't know how to resolve - return { - key: value # type: ignore - for key, value in ( - ( - self.bind_names[bindparam], - ( - bindparam.type._cached_bind_processor(self.dialect) - if not bindparam.type._is_tuple_type - else tuple( - elem_type._cached_bind_processor(self.dialect) - for elem_type in cast( - TupleType, bindparam.type - ).types - ) - ), - ) - for bindparam in self.bind_names - ) - if value is not None - } - - def is_subquery(self): - return len(self.stack) > 1 - - @property - def sql_compiler(self): - return self - - def construct_expanded_state( - self, - params: Optional[_CoreSingleExecuteParams] = None, - escape_names: bool = True, - ) -> ExpandedState: - """Return a new :class:`.ExpandedState` for a given parameter set. - - For queries that use "expanding" or other late-rendered parameters, - this method will provide for both the finalized SQL string as well - as the parameters that would be used for a particular parameter set. - - .. versionadded:: 2.0.0rc1 - - """ - parameters = self.construct_params( - params, - escape_names=escape_names, - _no_postcompile=True, - ) - return self._process_parameters_for_postcompile( - parameters, - ) - - def construct_params( - self, - params: Optional[_CoreSingleExecuteParams] = None, - extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None, - escape_names: bool = True, - _group_number: Optional[int] = None, - _check: bool = True, - _no_postcompile: bool = False, - ) -> _MutableCoreSingleExecuteParams: - """return a dictionary of bind parameter keys and values""" - - if self._render_postcompile and not _no_postcompile: - assert self._post_compile_expanded_state is not None - if not params: - return dict(self._post_compile_expanded_state.parameters) - else: - raise exc.InvalidRequestError( - "can't construct new parameters when render_postcompile " - "is used; the statement is hard-linked to the original " - "parameters. Use construct_expanded_state to generate a " - "new statement and parameters." - ) - - has_escaped_names = escape_names and bool(self.escaped_bind_names) - - if extracted_parameters: - # related the bound parameters collected in the original cache key - # to those collected in the incoming cache key. They will not have - # matching names but they will line up positionally in the same - # way. The parameters present in self.bind_names may be clones of - # these original cache key params in the case of DML but the .key - # will be guaranteed to match. - if self.cache_key is None: - raise exc.CompileError( - "This compiled object has no original cache key; " - "can't pass extracted_parameters to construct_params" - ) - else: - orig_extracted = self.cache_key[1] - - ckbm_tuple = self._cache_key_bind_match - assert ckbm_tuple is not None - ckbm, _ = ckbm_tuple - resolved_extracted = { - bind: extracted - for b, extracted in zip(orig_extracted, extracted_parameters) - for bind in ckbm[b] - } - else: - resolved_extracted = None - - if params: - pd = {} - for bindparam, name in self.bind_names.items(): - escaped_name = ( - self.escaped_bind_names.get(name, name) - if has_escaped_names - else name - ) - - if bindparam.key in params: - pd[escaped_name] = params[bindparam.key] - elif name in params: - pd[escaped_name] = params[name] - - elif _check and bindparam.required: - if _group_number: - raise exc.InvalidRequestError( - "A value is required for bind parameter %r, " - "in parameter group %d" - % (bindparam.key, _group_number), - code="cd3x", - ) - else: - raise exc.InvalidRequestError( - "A value is required for bind parameter %r" - % bindparam.key, - code="cd3x", - ) - else: - if resolved_extracted: - value_param = resolved_extracted.get( - bindparam, bindparam - ) - else: - value_param = bindparam - - if bindparam.callable: - pd[escaped_name] = value_param.effective_value - else: - pd[escaped_name] = value_param.value - return pd - else: - pd = {} - for bindparam, name in self.bind_names.items(): - escaped_name = ( - self.escaped_bind_names.get(name, name) - if has_escaped_names - else name - ) - - if _check and bindparam.required: - if _group_number: - raise exc.InvalidRequestError( - "A value is required for bind parameter %r, " - "in parameter group %d" - % (bindparam.key, _group_number), - code="cd3x", - ) - else: - raise exc.InvalidRequestError( - "A value is required for bind parameter %r" - % bindparam.key, - code="cd3x", - ) - - if resolved_extracted: - value_param = resolved_extracted.get(bindparam, bindparam) - else: - value_param = bindparam - - if bindparam.callable: - pd[escaped_name] = value_param.effective_value - else: - pd[escaped_name] = value_param.value - - return pd - - @util.memoized_instancemethod - def _get_set_input_sizes_lookup(self): - dialect = self.dialect - - include_types = dialect.include_set_input_sizes - exclude_types = dialect.exclude_set_input_sizes - - dbapi = dialect.dbapi - - def lookup_type(typ): - dbtype = typ._unwrapped_dialect_impl(dialect).get_dbapi_type(dbapi) - - if ( - dbtype is not None - and (exclude_types is None or dbtype not in exclude_types) - and (include_types is None or dbtype in include_types) - ): - return dbtype - else: - return None - - inputsizes = {} - - literal_execute_params = self.literal_execute_params - - for bindparam in self.bind_names: - if bindparam in literal_execute_params: - continue - - if bindparam.type._is_tuple_type: - inputsizes[bindparam] = [ - lookup_type(typ) - for typ in cast(TupleType, bindparam.type).types - ] - else: - inputsizes[bindparam] = lookup_type(bindparam.type) - - return inputsizes - - @property - def params(self): - """Return the bind param dictionary embedded into this - compiled object, for those values that are present. - - .. seealso:: - - :ref:`faq_sql_expression_string` - includes a usage example for - debugging use cases. - - """ - return self.construct_params(_check=False) - - def _process_parameters_for_postcompile( - self, - parameters: _MutableCoreSingleExecuteParams, - _populate_self: bool = False, - ) -> ExpandedState: - """handle special post compile parameters. - - These include: - - * "expanding" parameters -typically IN tuples that are rendered - on a per-parameter basis for an otherwise fixed SQL statement string. - - * literal_binds compiled with the literal_execute flag. Used for - things like SQL Server "TOP N" where the driver does not accommodate - N as a bound parameter. - - """ - - expanded_parameters = {} - new_positiontup: Optional[List[str]] - - pre_expanded_string = self._pre_expanded_string - if pre_expanded_string is None: - pre_expanded_string = self.string - - if self.positional: - new_positiontup = [] - - pre_expanded_positiontup = self._pre_expanded_positiontup - if pre_expanded_positiontup is None: - pre_expanded_positiontup = self.positiontup - - else: - new_positiontup = pre_expanded_positiontup = None - - processors = self._bind_processors - single_processors = cast( - "Mapping[str, _BindProcessorType[Any]]", processors - ) - tuple_processors = cast( - "Mapping[str, Sequence[_BindProcessorType[Any]]]", processors - ) - - new_processors: Dict[str, _BindProcessorType[Any]] = {} - - replacement_expressions: Dict[str, Any] = {} - to_update_sets: Dict[str, Any] = {} - - # notes: - # *unescaped* parameter names in: - # self.bind_names, self.binds, self._bind_processors, self.positiontup - # - # *escaped* parameter names in: - # construct_params(), replacement_expressions - - numeric_positiontup: Optional[List[str]] = None - - if self.positional and pre_expanded_positiontup is not None: - names: Iterable[str] = pre_expanded_positiontup - if self._numeric_binds: - numeric_positiontup = [] - else: - names = self.bind_names.values() - - ebn = self.escaped_bind_names - for name in names: - escaped_name = ebn.get(name, name) if ebn else name - parameter = self.binds[name] - - if parameter in self.literal_execute_params: - if escaped_name not in replacement_expressions: - replacement_expressions[escaped_name] = ( - self.render_literal_bindparam( - parameter, - render_literal_value=parameters.pop(escaped_name), - ) - ) - continue - - if parameter in self.post_compile_params: - if escaped_name in replacement_expressions: - to_update = to_update_sets[escaped_name] - values = None - else: - # we are removing the parameter from parameters - # because it is a list value, which is not expected by - # TypeEngine objects that would otherwise be asked to - # process it. the single name is being replaced with - # individual numbered parameters for each value in the - # param. - # - # note we are also inserting *escaped* parameter names - # into the given dictionary. default dialect will - # use these param names directly as they will not be - # in the escaped_bind_names dictionary. - values = parameters.pop(name) - - leep_res = self._literal_execute_expanding_parameter( - escaped_name, parameter, values - ) - (to_update, replacement_expr) = leep_res - - to_update_sets[escaped_name] = to_update - replacement_expressions[escaped_name] = replacement_expr - - if not parameter.literal_execute: - parameters.update(to_update) - if parameter.type._is_tuple_type: - assert values is not None - new_processors.update( - ( - "%s_%s_%s" % (name, i, j), - tuple_processors[name][j - 1], - ) - for i, tuple_element in enumerate(values, 1) - for j, _ in enumerate(tuple_element, 1) - if name in tuple_processors - and tuple_processors[name][j - 1] is not None - ) - else: - new_processors.update( - (key, single_processors[name]) - for key, _ in to_update - if name in single_processors - ) - if numeric_positiontup is not None: - numeric_positiontup.extend( - name for name, _ in to_update - ) - elif new_positiontup is not None: - # to_update has escaped names, but that's ok since - # these are new names, that aren't in the - # escaped_bind_names dict. - new_positiontup.extend(name for name, _ in to_update) - expanded_parameters[name] = [ - expand_key for expand_key, _ in to_update - ] - elif new_positiontup is not None: - new_positiontup.append(name) - - def process_expanding(m): - key = m.group(1) - expr = replacement_expressions[key] - - # if POSTCOMPILE included a bind_expression, render that - # around each element - if m.group(2): - tok = m.group(2).split("~~") - be_left, be_right = tok[1], tok[3] - expr = ", ".join( - "%s%s%s" % (be_left, exp, be_right) - for exp in expr.split(", ") - ) - return expr - - statement = re.sub( - self._post_compile_pattern, process_expanding, pre_expanded_string - ) - - if numeric_positiontup is not None: - assert new_positiontup is not None - param_pos = { - key: f"{self._numeric_binds_identifier_char}{num}" - for num, key in enumerate( - numeric_positiontup, self.next_numeric_pos - ) - } - # Can't use format here since % chars are not escaped. - statement = self._pyformat_pattern.sub( - lambda m: param_pos[m.group(1)], statement - ) - new_positiontup.extend(numeric_positiontup) - - expanded_state = ExpandedState( - statement, - parameters, - new_processors, - new_positiontup, - expanded_parameters, - ) - - if _populate_self: - # this is for the "render_postcompile" flag, which is not - # otherwise used internally and is for end-user debugging and - # special use cases. - self._pre_expanded_string = pre_expanded_string - self._pre_expanded_positiontup = pre_expanded_positiontup - self.string = expanded_state.statement - self.positiontup = ( - list(expanded_state.positiontup or ()) - if self.positional - else None - ) - self._post_compile_expanded_state = expanded_state - - return expanded_state - - @util.preload_module("sqlalchemy.engine.cursor") - def _create_result_map(self): - """utility method used for unit tests only.""" - cursor = util.preloaded.engine_cursor - return cursor.CursorResultMetaData._create_description_match_map( - self._result_columns - ) - - # assigned by crud.py for insert/update statements - _get_bind_name_for_col: _BindNameForColProtocol - - @util.memoized_property - def _within_exec_param_key_getter(self) -> Callable[[Any], str]: - getter = self._get_bind_name_for_col - return getter - - @util.memoized_property - @util.preload_module("sqlalchemy.engine.result") - def _inserted_primary_key_from_lastrowid_getter(self): - result = util.preloaded.engine_result - - param_key_getter = self._within_exec_param_key_getter - - assert self.compile_state is not None - statement = self.compile_state.statement - - if TYPE_CHECKING: - assert isinstance(statement, Insert) - - table = statement.table - - getters = [ - (operator.methodcaller("get", param_key_getter(col), None), col) - for col in table.primary_key - ] - - autoinc_getter = None - autoinc_col = table._autoincrement_column - if autoinc_col is not None: - # apply type post processors to the lastrowid - lastrowid_processor = autoinc_col.type._cached_result_processor( - self.dialect, None - ) - autoinc_key = param_key_getter(autoinc_col) - - # if a bind value is present for the autoincrement column - # in the parameters, we need to do the logic dictated by - # #7998; honor a non-None user-passed parameter over lastrowid. - # previously in the 1.4 series we weren't fetching lastrowid - # at all if the key were present in the parameters - if autoinc_key in self.binds: - - def _autoinc_getter(lastrowid, parameters): - param_value = parameters.get(autoinc_key, lastrowid) - if param_value is not None: - # they supplied non-None parameter, use that. - # SQLite at least is observed to return the wrong - # cursor.lastrowid for INSERT..ON CONFLICT so it - # can't be used in all cases - return param_value - else: - # use lastrowid - return lastrowid - - # work around mypy https://github.com/python/mypy/issues/14027 - autoinc_getter = _autoinc_getter - - else: - lastrowid_processor = None - - row_fn = result.result_tuple([col.key for col in table.primary_key]) - - def get(lastrowid, parameters): - """given cursor.lastrowid value and the parameters used for INSERT, - return a "row" that represents the primary key, either by - using the "lastrowid" or by extracting values from the parameters - that were sent along with the INSERT. - - """ - if lastrowid_processor is not None: - lastrowid = lastrowid_processor(lastrowid) - - if lastrowid is None: - return row_fn(getter(parameters) for getter, col in getters) - else: - return row_fn( - ( - ( - autoinc_getter(lastrowid, parameters) - if autoinc_getter is not None - else lastrowid - ) - if col is autoinc_col - else getter(parameters) - ) - for getter, col in getters - ) - - return get - - @util.memoized_property - @util.preload_module("sqlalchemy.engine.result") - def _inserted_primary_key_from_returning_getter(self): - if typing.TYPE_CHECKING: - from ..engine import result - else: - result = util.preloaded.engine_result - - assert self.compile_state is not None - statement = self.compile_state.statement - - if TYPE_CHECKING: - assert isinstance(statement, Insert) - - param_key_getter = self._within_exec_param_key_getter - table = statement.table - - returning = self.implicit_returning - assert returning is not None - ret = {col: idx for idx, col in enumerate(returning)} - - getters = cast( - "List[Tuple[Callable[[Any], Any], bool]]", - [ - ( - (operator.itemgetter(ret[col]), True) - if col in ret - else ( - operator.methodcaller( - "get", param_key_getter(col), None - ), - False, - ) - ) - for col in table.primary_key - ], - ) - - row_fn = result.result_tuple([col.key for col in table.primary_key]) - - def get(row, parameters): - return row_fn( - getter(row) if use_row else getter(parameters) - for getter, use_row in getters - ) - - return get - - def default_from(self): - """Called when a SELECT statement has no froms, and no FROM clause is - to be appended. - - Gives Oracle a chance to tack on a ``FROM DUAL`` to the string output. - - """ - return "" - - def visit_override_binds(self, override_binds, **kw): - """SQL compile the nested element of an _OverrideBinds with - bindparams swapped out. - - The _OverrideBinds is not normally expected to be compiled; it - is meant to be used when an already cached statement is to be used, - the compilation was already performed, and only the bound params should - be swapped in at execution time. - - However, there are test cases that exericise this object, and - additionally the ORM subquery loader is known to feed in expressions - which include this construct into new queries (discovered in #11173), - so it has to do the right thing at compile time as well. - - """ - - # get SQL text first - sqltext = override_binds.element._compiler_dispatch(self, **kw) - - # for a test compile that is not for caching, change binds after the - # fact. note that we don't try to - # swap the bindparam as we compile, because our element may be - # elsewhere in the statement already (e.g. a subquery or perhaps a - # CTE) and was already visited / compiled. See - # test_relationship_criteria.py -> - # test_selectinload_local_criteria_subquery - for k in override_binds.translate: - if k not in self.binds: - continue - bp = self.binds[k] - - # so this would work, just change the value of bp in place. - # but we dont want to mutate things outside. - # bp.value = override_binds.translate[bp.key] - # continue - - # instead, need to replace bp with new_bp or otherwise accommodate - # in all internal collections - new_bp = bp._with_value( - override_binds.translate[bp.key], - maintain_key=True, - required=False, - ) - - name = self.bind_names[bp] - self.binds[k] = self.binds[name] = new_bp - self.bind_names[new_bp] = name - self.bind_names.pop(bp, None) - - if bp in self.post_compile_params: - self.post_compile_params |= {new_bp} - if bp in self.literal_execute_params: - self.literal_execute_params |= {new_bp} - - ckbm_tuple = self._cache_key_bind_match - if ckbm_tuple: - ckbm, cksm = ckbm_tuple - for bp in bp._cloned_set: - if bp.key in cksm: - cb = cksm[bp.key] - ckbm[cb].append(new_bp) - - return sqltext - - def visit_grouping(self, grouping, asfrom=False, **kwargs): - return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")" - - def visit_select_statement_grouping(self, grouping, **kwargs): - return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")" - - def visit_label_reference( - self, element, within_columns_clause=False, **kwargs - ): - if self.stack and self.dialect.supports_simple_order_by_label: - try: - compile_state = cast( - "Union[SelectState, CompoundSelectState]", - self.stack[-1]["compile_state"], - ) - except KeyError as ke: - raise exc.CompileError( - "Can't resolve label reference for ORDER BY / " - "GROUP BY / DISTINCT etc." - ) from ke - - ( - with_cols, - only_froms, - only_cols, - ) = compile_state._label_resolve_dict - if within_columns_clause: - resolve_dict = only_froms - else: - resolve_dict = only_cols - - # this can be None in the case that a _label_reference() - # were subject to a replacement operation, in which case - # the replacement of the Label element may have changed - # to something else like a ColumnClause expression. - order_by_elem = element.element._order_by_label_element - - if ( - order_by_elem is not None - and order_by_elem.name in resolve_dict - and order_by_elem.shares_lineage( - resolve_dict[order_by_elem.name] - ) - ): - kwargs["render_label_as_label"] = ( - element.element._order_by_label_element - ) - return self.process( - element.element, - within_columns_clause=within_columns_clause, - **kwargs, - ) - - def visit_textual_label_reference( - self, element, within_columns_clause=False, **kwargs - ): - if not self.stack: - # compiling the element outside of the context of a SELECT - return self.process(element._text_clause) - - try: - compile_state = cast( - "Union[SelectState, CompoundSelectState]", - self.stack[-1]["compile_state"], - ) - except KeyError as ke: - coercions._no_text_coercion( - element.element, - extra=( - "Can't resolve label reference for ORDER BY / " - "GROUP BY / DISTINCT etc." - ), - exc_cls=exc.CompileError, - err=ke, - ) - - with_cols, only_froms, only_cols = compile_state._label_resolve_dict - try: - if within_columns_clause: - col = only_froms[element.element] - else: - col = with_cols[element.element] - except KeyError as err: - coercions._no_text_coercion( - element.element, - extra=( - "Can't resolve label reference for ORDER BY / " - "GROUP BY / DISTINCT etc." - ), - exc_cls=exc.CompileError, - err=err, - ) - else: - kwargs["render_label_as_label"] = col - return self.process( - col, within_columns_clause=within_columns_clause, **kwargs - ) - - def visit_label( - self, - label, - add_to_result_map=None, - within_label_clause=False, - within_columns_clause=False, - render_label_as_label=None, - result_map_targets=(), - **kw, - ): - # only render labels within the columns clause - # or ORDER BY clause of a select. dialect-specific compilers - # can modify this behavior. - render_label_with_as = ( - within_columns_clause and not within_label_clause - ) - render_label_only = render_label_as_label is label - - if render_label_only or render_label_with_as: - if isinstance(label.name, elements._truncated_label): - labelname = self._truncated_identifier("colident", label.name) - else: - labelname = label.name - - if render_label_with_as: - if add_to_result_map is not None: - add_to_result_map( - labelname, - label.name, - (label, labelname) + label._alt_names + result_map_targets, - label.type, - ) - return ( - label.element._compiler_dispatch( - self, - within_columns_clause=True, - within_label_clause=True, - **kw, - ) - + OPERATORS[operators.as_] - + self.preparer.format_label(label, labelname) - ) - elif render_label_only: - return self.preparer.format_label(label, labelname) - else: - return label.element._compiler_dispatch( - self, within_columns_clause=False, **kw - ) - - def _fallback_column_name(self, column): - raise exc.CompileError( - "Cannot compile Column object until its 'name' is assigned." - ) - - def visit_lambda_element(self, element, **kw): - sql_element = element._resolved - return self.process(sql_element, **kw) - - def visit_column( - self, - column: ColumnClause[Any], - add_to_result_map: Optional[_ResultMapAppender] = None, - include_table: bool = True, - result_map_targets: Tuple[Any, ...] = (), - ambiguous_table_name_map: Optional[_AmbiguousTableNameMap] = None, - **kwargs: Any, - ) -> str: - name = orig_name = column.name - if name is None: - name = self._fallback_column_name(column) - - is_literal = column.is_literal - if not is_literal and isinstance(name, elements._truncated_label): - name = self._truncated_identifier("colident", name) - - if add_to_result_map is not None: - targets = (column, name, column.key) + result_map_targets - if column._tq_label: - targets += (column._tq_label,) - - add_to_result_map(name, orig_name, targets, column.type) - - if is_literal: - # note we are not currently accommodating for - # literal_column(quoted_name('ident', True)) here - name = self.escape_literal_column(name) - else: - name = self.preparer.quote(name) - table = column.table - if table is None or not include_table or not table.named_with_column: - return name - else: - effective_schema = self.preparer.schema_for_object(table) - - if effective_schema: - schema_prefix = ( - self.preparer.quote_schema(effective_schema) + "." - ) - else: - schema_prefix = "" - - if TYPE_CHECKING: - assert isinstance(table, NamedFromClause) - tablename = table.name - - if ( - not effective_schema - and ambiguous_table_name_map - and tablename in ambiguous_table_name_map - ): - tablename = ambiguous_table_name_map[tablename] - - if isinstance(tablename, elements._truncated_label): - tablename = self._truncated_identifier("alias", tablename) - - return schema_prefix + self.preparer.quote(tablename) + "." + name - - def visit_collation(self, element, **kw): - return self.preparer.format_collation(element.collation) - - def visit_fromclause(self, fromclause, **kwargs): - return fromclause.name - - def visit_index(self, index, **kwargs): - return index.name - - def visit_typeclause(self, typeclause, **kw): - kw["type_expression"] = typeclause - kw["identifier_preparer"] = self.preparer - return self.dialect.type_compiler_instance.process( - typeclause.type, **kw - ) - - def post_process_text(self, text): - if self.preparer._double_percents: - text = text.replace("%", "%%") - return text - - def escape_literal_column(self, text): - if self.preparer._double_percents: - text = text.replace("%", "%%") - return text - - def visit_textclause(self, textclause, add_to_result_map=None, **kw): - def do_bindparam(m): - name = m.group(1) - if name in textclause._bindparams: - return self.process(textclause._bindparams[name], **kw) - else: - return self.bindparam_string(name, **kw) - - if not self.stack: - self.isplaintext = True - - if add_to_result_map: - # text() object is present in the columns clause of a - # select(). Add a no-name entry to the result map so that - # row[text()] produces a result - add_to_result_map(None, None, (textclause,), sqltypes.NULLTYPE) - - # un-escape any \:params - return BIND_PARAMS_ESC.sub( - lambda m: m.group(1), - BIND_PARAMS.sub( - do_bindparam, self.post_process_text(textclause.text) - ), - ) - - def visit_textual_select( - self, taf, compound_index=None, asfrom=False, **kw - ): - toplevel = not self.stack - entry = self._default_stack_entry if toplevel else self.stack[-1] - - new_entry: _CompilerStackEntry = { - "correlate_froms": set(), - "asfrom_froms": set(), - "selectable": taf, - } - self.stack.append(new_entry) - - if taf._independent_ctes: - self._dispatch_independent_ctes(taf, kw) - - populate_result_map = ( - toplevel - or ( - compound_index == 0 - and entry.get("need_result_map_for_compound", False) - ) - or entry.get("need_result_map_for_nested", False) - ) - - if populate_result_map: - self._ordered_columns = self._textual_ordered_columns = ( - taf.positional - ) - - # enable looser result column matching when the SQL text links to - # Column objects by name only - self._loose_column_name_matching = not taf.positional and bool( - taf.column_args - ) - - for c in taf.column_args: - self.process( - c, - within_columns_clause=True, - add_to_result_map=self._add_to_result_map, - ) - - text = self.process(taf.element, **kw) - if self.ctes: - nesting_level = len(self.stack) if not toplevel else None - text = self._render_cte_clause(nesting_level=nesting_level) + text - - self.stack.pop(-1) - - return text - - def visit_null(self, expr, **kw): - return "NULL" - - def visit_true(self, expr, **kw): - if self.dialect.supports_native_boolean: - return "true" - else: - return "1" - - def visit_false(self, expr, **kw): - if self.dialect.supports_native_boolean: - return "false" - else: - return "0" - - def _generate_delimited_list(self, elements, separator, **kw): - return separator.join( - s - for s in (c._compiler_dispatch(self, **kw) for c in elements) - if s - ) - - def _generate_delimited_and_list(self, clauses, **kw): - lcc, clauses = elements.BooleanClauseList._process_clauses_for_boolean( - operators.and_, - elements.True_._singleton, - elements.False_._singleton, - clauses, - ) - if lcc == 1: - return clauses[0]._compiler_dispatch(self, **kw) - else: - separator = OPERATORS[operators.and_] - return separator.join( - s - for s in (c._compiler_dispatch(self, **kw) for c in clauses) - if s - ) - - def visit_tuple(self, clauselist, **kw): - return "(%s)" % self.visit_clauselist(clauselist, **kw) - - def visit_clauselist(self, clauselist, **kw): - sep = clauselist.operator - if sep is None: - sep = " " - else: - sep = OPERATORS[clauselist.operator] - - return self._generate_delimited_list(clauselist.clauses, sep, **kw) - - def visit_expression_clauselist(self, clauselist, **kw): - operator_ = clauselist.operator - - disp = self._get_operator_dispatch( - operator_, "expression_clauselist", None - ) - if disp: - return disp(clauselist, operator_, **kw) - - try: - opstring = OPERATORS[operator_] - except KeyError as err: - raise exc.UnsupportedCompilationError(self, operator_) from err - else: - kw["_in_operator_expression"] = True - return self._generate_delimited_list( - clauselist.clauses, opstring, **kw - ) - - def visit_case(self, clause, **kwargs): - x = "CASE " - if clause.value is not None: - x += clause.value._compiler_dispatch(self, **kwargs) + " " - for cond, result in clause.whens: - x += ( - "WHEN " - + cond._compiler_dispatch(self, **kwargs) - + " THEN " - + result._compiler_dispatch(self, **kwargs) - + " " - ) - if clause.else_ is not None: - x += ( - "ELSE " + clause.else_._compiler_dispatch(self, **kwargs) + " " - ) - x += "END" - return x - - def visit_type_coerce(self, type_coerce, **kw): - return type_coerce.typed_expression._compiler_dispatch(self, **kw) - - def visit_cast(self, cast, **kwargs): - type_clause = cast.typeclause._compiler_dispatch(self, **kwargs) - match = re.match("(.*)( COLLATE .*)", type_clause) - return "CAST(%s AS %s)%s" % ( - cast.clause._compiler_dispatch(self, **kwargs), - match.group(1) if match else type_clause, - match.group(2) if match else "", - ) - - def _format_frame_clause(self, range_, **kw): - return "%s AND %s" % ( - ( - "UNBOUNDED PRECEDING" - if range_[0] is elements.RANGE_UNBOUNDED - else ( - "CURRENT ROW" - if range_[0] is elements.RANGE_CURRENT - else ( - "%s PRECEDING" - % ( - self.process( - elements.literal(abs(range_[0])), **kw - ), - ) - if range_[0] < 0 - else "%s FOLLOWING" - % (self.process(elements.literal(range_[0]), **kw),) - ) - ) - ), - ( - "UNBOUNDED FOLLOWING" - if range_[1] is elements.RANGE_UNBOUNDED - else ( - "CURRENT ROW" - if range_[1] is elements.RANGE_CURRENT - else ( - "%s PRECEDING" - % ( - self.process( - elements.literal(abs(range_[1])), **kw - ), - ) - if range_[1] < 0 - else "%s FOLLOWING" - % (self.process(elements.literal(range_[1]), **kw),) - ) - ) - ), - ) - - def visit_over(self, over, **kwargs): - text = over.element._compiler_dispatch(self, **kwargs) - if over.range_: - range_ = "RANGE BETWEEN %s" % self._format_frame_clause( - over.range_, **kwargs - ) - elif over.rows: - range_ = "ROWS BETWEEN %s" % self._format_frame_clause( - over.rows, **kwargs - ) - else: - range_ = None - - return "%s OVER (%s)" % ( - text, - " ".join( - [ - "%s BY %s" - % (word, clause._compiler_dispatch(self, **kwargs)) - for word, clause in ( - ("PARTITION", over.partition_by), - ("ORDER", over.order_by), - ) - if clause is not None and len(clause) - ] - + ([range_] if range_ else []) - ), - ) - - def visit_withingroup(self, withingroup, **kwargs): - return "%s WITHIN GROUP (ORDER BY %s)" % ( - withingroup.element._compiler_dispatch(self, **kwargs), - withingroup.order_by._compiler_dispatch(self, **kwargs), - ) - - def visit_funcfilter(self, funcfilter, **kwargs): - return "%s FILTER (WHERE %s)" % ( - funcfilter.func._compiler_dispatch(self, **kwargs), - funcfilter.criterion._compiler_dispatch(self, **kwargs), - ) - - def visit_extract(self, extract, **kwargs): - field = self.extract_map.get(extract.field, extract.field) - return "EXTRACT(%s FROM %s)" % ( - field, - extract.expr._compiler_dispatch(self, **kwargs), - ) - - def visit_scalar_function_column(self, element, **kw): - compiled_fn = self.visit_function(element.fn, **kw) - compiled_col = self.visit_column(element, **kw) - return "(%s).%s" % (compiled_fn, compiled_col) - - def visit_function( - self, - func: Function[Any], - add_to_result_map: Optional[_ResultMapAppender] = None, - **kwargs: Any, - ) -> str: - if add_to_result_map is not None: - add_to_result_map(func.name, func.name, (), func.type) - - disp = getattr(self, "visit_%s_func" % func.name.lower(), None) - - text: str - - if disp: - text = disp(func, **kwargs) - else: - name = FUNCTIONS.get(func._deannotate().__class__, None) - if name: - if func._has_args: - name += "%(expr)s" - else: - name = func.name - name = ( - self.preparer.quote(name) - if self.preparer._requires_quotes_illegal_chars(name) - or isinstance(name, elements.quoted_name) - else name - ) - name = name + "%(expr)s" - text = ".".join( - [ - ( - self.preparer.quote(tok) - if self.preparer._requires_quotes_illegal_chars(tok) - or isinstance(name, elements.quoted_name) - else tok - ) - for tok in func.packagenames - ] - + [name] - ) % {"expr": self.function_argspec(func, **kwargs)} - - if func._with_ordinality: - text += " WITH ORDINALITY" - return text - - def visit_next_value_func(self, next_value, **kw): - return self.visit_sequence(next_value.sequence) - - def visit_sequence(self, sequence, **kw): - raise NotImplementedError( - "Dialect '%s' does not support sequence increments." - % self.dialect.name - ) - - def function_argspec(self, func, **kwargs): - return func.clause_expr._compiler_dispatch(self, **kwargs) - - def visit_compound_select( - self, cs, asfrom=False, compound_index=None, **kwargs - ): - toplevel = not self.stack - - compile_state = cs._compile_state_factory(cs, self, **kwargs) - - if toplevel and not self.compile_state: - self.compile_state = compile_state - - compound_stmt = compile_state.statement - - entry = self._default_stack_entry if toplevel else self.stack[-1] - need_result_map = toplevel or ( - not compound_index - and entry.get("need_result_map_for_compound", False) - ) - - # indicates there is already a CompoundSelect in play - if compound_index == 0: - entry["select_0"] = cs - - self.stack.append( - { - "correlate_froms": entry["correlate_froms"], - "asfrom_froms": entry["asfrom_froms"], - "selectable": cs, - "compile_state": compile_state, - "need_result_map_for_compound": need_result_map, - } - ) - - if compound_stmt._independent_ctes: - self._dispatch_independent_ctes(compound_stmt, kwargs) - - keyword = self.compound_keywords[cs.keyword] - - text = (" " + keyword + " ").join( - ( - c._compiler_dispatch( - self, asfrom=asfrom, compound_index=i, **kwargs - ) - for i, c in enumerate(cs.selects) - ) - ) - - kwargs["include_table"] = False - text += self.group_by_clause(cs, **dict(asfrom=asfrom, **kwargs)) - text += self.order_by_clause(cs, **kwargs) - if cs._has_row_limiting_clause: - text += self._row_limit_clause(cs, **kwargs) - - if self.ctes: - nesting_level = len(self.stack) if not toplevel else None - text = ( - self._render_cte_clause( - nesting_level=nesting_level, - include_following_stack=True, - ) - + text - ) - - self.stack.pop(-1) - return text - - def _row_limit_clause(self, cs, **kwargs): - if cs._fetch_clause is not None: - return self.fetch_clause(cs, **kwargs) - else: - return self.limit_clause(cs, **kwargs) - - def _get_operator_dispatch(self, operator_, qualifier1, qualifier2): - attrname = "visit_%s_%s%s" % ( - operator_.__name__, - qualifier1, - "_" + qualifier2 if qualifier2 else "", - ) - return getattr(self, attrname, None) - - def visit_unary( - self, unary, add_to_result_map=None, result_map_targets=(), **kw - ): - if add_to_result_map is not None: - result_map_targets += (unary,) - kw["add_to_result_map"] = add_to_result_map - kw["result_map_targets"] = result_map_targets - - if unary.operator: - if unary.modifier: - raise exc.CompileError( - "Unary expression does not support operator " - "and modifier simultaneously" - ) - disp = self._get_operator_dispatch( - unary.operator, "unary", "operator" - ) - if disp: - return disp(unary, unary.operator, **kw) - else: - return self._generate_generic_unary_operator( - unary, OPERATORS[unary.operator], **kw - ) - elif unary.modifier: - disp = self._get_operator_dispatch( - unary.modifier, "unary", "modifier" - ) - if disp: - return disp(unary, unary.modifier, **kw) - else: - return self._generate_generic_unary_modifier( - unary, OPERATORS[unary.modifier], **kw - ) - else: - raise exc.CompileError( - "Unary expression has no operator or modifier" - ) - - def visit_truediv_binary(self, binary, operator, **kw): - if self.dialect.div_is_floordiv: - return ( - self.process(binary.left, **kw) - + " / " - # TODO: would need a fast cast again here, - # unless we want to use an implicit cast like "+ 0.0" - + self.process( - elements.Cast( - binary.right, - ( - binary.right.type - if binary.right.type._type_affinity - is sqltypes.Numeric - else sqltypes.Numeric() - ), - ), - **kw, - ) - ) - else: - return ( - self.process(binary.left, **kw) - + " / " - + self.process(binary.right, **kw) - ) - - def visit_floordiv_binary(self, binary, operator, **kw): - if ( - self.dialect.div_is_floordiv - and binary.right.type._type_affinity is sqltypes.Integer - ): - return ( - self.process(binary.left, **kw) - + " / " - + self.process(binary.right, **kw) - ) - else: - return "FLOOR(%s)" % ( - self.process(binary.left, **kw) - + " / " - + self.process(binary.right, **kw) - ) - - def visit_is_true_unary_operator(self, element, operator, **kw): - if ( - element._is_implicitly_boolean - or self.dialect.supports_native_boolean - ): - return self.process(element.element, **kw) - else: - return "%s = 1" % self.process(element.element, **kw) - - def visit_is_false_unary_operator(self, element, operator, **kw): - if ( - element._is_implicitly_boolean - or self.dialect.supports_native_boolean - ): - return "NOT %s" % self.process(element.element, **kw) - else: - return "%s = 0" % self.process(element.element, **kw) - - def visit_not_match_op_binary(self, binary, operator, **kw): - return "NOT %s" % self.visit_binary( - binary, override_operator=operators.match_op - ) - - def visit_not_in_op_binary(self, binary, operator, **kw): - # The brackets are required in the NOT IN operation because the empty - # case is handled using the form "(col NOT IN (null) OR 1 = 1)". - # The presence of the OR makes the brackets required. - return "(%s)" % self._generate_generic_binary( - binary, OPERATORS[operator], **kw - ) - - def visit_empty_set_op_expr(self, type_, expand_op, **kw): - if expand_op is operators.not_in_op: - if len(type_) > 1: - return "(%s)) OR (1 = 1" % ( - ", ".join("NULL" for element in type_) - ) - else: - return "NULL) OR (1 = 1" - elif expand_op is operators.in_op: - if len(type_) > 1: - return "(%s)) AND (1 != 1" % ( - ", ".join("NULL" for element in type_) - ) - else: - return "NULL) AND (1 != 1" - else: - return self.visit_empty_set_expr(type_) - - def visit_empty_set_expr(self, element_types, **kw): - raise NotImplementedError( - "Dialect '%s' does not support empty set expression." - % self.dialect.name - ) - - def _literal_execute_expanding_parameter_literal_binds( - self, parameter, values, bind_expression_template=None - ): - typ_dialect_impl = parameter.type._unwrapped_dialect_impl(self.dialect) - - if not values: - # empty IN expression. note we don't need to use - # bind_expression_template here because there are no - # expressions to render. - - if typ_dialect_impl._is_tuple_type: - replacement_expression = ( - "VALUES " if self.dialect.tuple_in_values else "" - ) + self.visit_empty_set_op_expr( - parameter.type.types, parameter.expand_op - ) - - else: - replacement_expression = self.visit_empty_set_op_expr( - [parameter.type], parameter.expand_op - ) - - elif typ_dialect_impl._is_tuple_type or ( - typ_dialect_impl._isnull - and isinstance(values[0], collections_abc.Sequence) - and not isinstance(values[0], (str, bytes)) - ): - if typ_dialect_impl._has_bind_expression: - raise NotImplementedError( - "bind_expression() on TupleType not supported with " - "literal_binds" - ) - - replacement_expression = ( - "VALUES " if self.dialect.tuple_in_values else "" - ) + ", ".join( - "(%s)" - % ( - ", ".join( - self.render_literal_value(value, param_type) - for value, param_type in zip( - tuple_element, parameter.type.types - ) - ) - ) - for i, tuple_element in enumerate(values) - ) - else: - if bind_expression_template: - post_compile_pattern = self._post_compile_pattern - m = post_compile_pattern.search(bind_expression_template) - assert m and m.group( - 2 - ), "unexpected format for expanding parameter" - - tok = m.group(2).split("~~") - be_left, be_right = tok[1], tok[3] - replacement_expression = ", ".join( - "%s%s%s" - % ( - be_left, - self.render_literal_value(value, parameter.type), - be_right, - ) - for value in values - ) - else: - replacement_expression = ", ".join( - self.render_literal_value(value, parameter.type) - for value in values - ) - - return (), replacement_expression - - def _literal_execute_expanding_parameter(self, name, parameter, values): - if parameter.literal_execute: - return self._literal_execute_expanding_parameter_literal_binds( - parameter, values - ) - - dialect = self.dialect - typ_dialect_impl = parameter.type._unwrapped_dialect_impl(dialect) - - if self._numeric_binds: - bind_template = self.compilation_bindtemplate - else: - bind_template = self.bindtemplate - - if ( - self.dialect._bind_typing_render_casts - and typ_dialect_impl.render_bind_cast - ): - - def _render_bindtemplate(name): - return self.render_bind_cast( - parameter.type, - typ_dialect_impl, - bind_template % {"name": name}, - ) - - else: - - def _render_bindtemplate(name): - return bind_template % {"name": name} - - if not values: - to_update = [] - if typ_dialect_impl._is_tuple_type: - replacement_expression = self.visit_empty_set_op_expr( - parameter.type.types, parameter.expand_op - ) - else: - replacement_expression = self.visit_empty_set_op_expr( - [parameter.type], parameter.expand_op - ) - - elif typ_dialect_impl._is_tuple_type or ( - typ_dialect_impl._isnull - and isinstance(values[0], collections_abc.Sequence) - and not isinstance(values[0], (str, bytes)) - ): - assert not typ_dialect_impl._is_array - to_update = [ - ("%s_%s_%s" % (name, i, j), value) - for i, tuple_element in enumerate(values, 1) - for j, value in enumerate(tuple_element, 1) - ] - - replacement_expression = ( - "VALUES " if dialect.tuple_in_values else "" - ) + ", ".join( - "(%s)" - % ( - ", ".join( - _render_bindtemplate( - to_update[i * len(tuple_element) + j][0] - ) - for j, value in enumerate(tuple_element) - ) - ) - for i, tuple_element in enumerate(values) - ) - else: - to_update = [ - ("%s_%s" % (name, i), value) - for i, value in enumerate(values, 1) - ] - replacement_expression = ", ".join( - _render_bindtemplate(key) for key, value in to_update - ) - - return to_update, replacement_expression - - def visit_binary( - self, - binary, - override_operator=None, - eager_grouping=False, - from_linter=None, - lateral_from_linter=None, - **kw, - ): - if from_linter and operators.is_comparison(binary.operator): - if lateral_from_linter is not None: - enclosing_lateral = kw["enclosing_lateral"] - lateral_from_linter.edges.update( - itertools.product( - _de_clone( - binary.left._from_objects + [enclosing_lateral] - ), - _de_clone( - binary.right._from_objects + [enclosing_lateral] - ), - ) - ) - else: - from_linter.edges.update( - itertools.product( - _de_clone(binary.left._from_objects), - _de_clone(binary.right._from_objects), - ) - ) - - # don't allow "? = ?" to render - if ( - self.ansi_bind_rules - and isinstance(binary.left, elements.BindParameter) - and isinstance(binary.right, elements.BindParameter) - ): - kw["literal_execute"] = True - - operator_ = override_operator or binary.operator - disp = self._get_operator_dispatch(operator_, "binary", None) - if disp: - return disp(binary, operator_, **kw) - else: - try: - opstring = OPERATORS[operator_] - except KeyError as err: - raise exc.UnsupportedCompilationError(self, operator_) from err - else: - return self._generate_generic_binary( - binary, - opstring, - from_linter=from_linter, - lateral_from_linter=lateral_from_linter, - **kw, - ) - - def visit_function_as_comparison_op_binary(self, element, operator, **kw): - return self.process(element.sql_function, **kw) - - def visit_mod_binary(self, binary, operator, **kw): - if self.preparer._double_percents: - return ( - self.process(binary.left, **kw) - + " %% " - + self.process(binary.right, **kw) - ) - else: - return ( - self.process(binary.left, **kw) - + " % " - + self.process(binary.right, **kw) - ) - - def visit_custom_op_binary(self, element, operator, **kw): - kw["eager_grouping"] = operator.eager_grouping - return self._generate_generic_binary( - element, - " " + self.escape_literal_column(operator.opstring) + " ", - **kw, - ) - - def visit_custom_op_unary_operator(self, element, operator, **kw): - return self._generate_generic_unary_operator( - element, self.escape_literal_column(operator.opstring) + " ", **kw - ) - - def visit_custom_op_unary_modifier(self, element, operator, **kw): - return self._generate_generic_unary_modifier( - element, " " + self.escape_literal_column(operator.opstring), **kw - ) - - def _generate_generic_binary( - self, binary, opstring, eager_grouping=False, **kw - ): - _in_operator_expression = kw.get("_in_operator_expression", False) - - kw["_in_operator_expression"] = True - kw["_binary_op"] = binary.operator - text = ( - binary.left._compiler_dispatch( - self, eager_grouping=eager_grouping, **kw - ) - + opstring - + binary.right._compiler_dispatch( - self, eager_grouping=eager_grouping, **kw - ) - ) - - if _in_operator_expression and eager_grouping: - text = "(%s)" % text - return text - - def _generate_generic_unary_operator(self, unary, opstring, **kw): - return opstring + unary.element._compiler_dispatch(self, **kw) - - def _generate_generic_unary_modifier(self, unary, opstring, **kw): - return unary.element._compiler_dispatch(self, **kw) + opstring - - @util.memoized_property - def _like_percent_literal(self): - return elements.literal_column("'%'", type_=sqltypes.STRINGTYPE) - - def visit_ilike_case_insensitive_operand(self, element, **kw): - return f"lower({element.element._compiler_dispatch(self, **kw)})" - - def visit_contains_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.right = percent.concat(binary.right).concat(percent) - return self.visit_like_op_binary(binary, operator, **kw) - - def visit_not_contains_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.right = percent.concat(binary.right).concat(percent) - return self.visit_not_like_op_binary(binary, operator, **kw) - - def visit_icontains_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.left = ilike_case_insensitive(binary.left) - binary.right = percent.concat( - ilike_case_insensitive(binary.right) - ).concat(percent) - return self.visit_ilike_op_binary(binary, operator, **kw) - - def visit_not_icontains_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.left = ilike_case_insensitive(binary.left) - binary.right = percent.concat( - ilike_case_insensitive(binary.right) - ).concat(percent) - return self.visit_not_ilike_op_binary(binary, operator, **kw) - - def visit_startswith_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.right = percent._rconcat(binary.right) - return self.visit_like_op_binary(binary, operator, **kw) - - def visit_not_startswith_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.right = percent._rconcat(binary.right) - return self.visit_not_like_op_binary(binary, operator, **kw) - - def visit_istartswith_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.left = ilike_case_insensitive(binary.left) - binary.right = percent._rconcat(ilike_case_insensitive(binary.right)) - return self.visit_ilike_op_binary(binary, operator, **kw) - - def visit_not_istartswith_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.left = ilike_case_insensitive(binary.left) - binary.right = percent._rconcat(ilike_case_insensitive(binary.right)) - return self.visit_not_ilike_op_binary(binary, operator, **kw) - - def visit_endswith_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.right = percent.concat(binary.right) - return self.visit_like_op_binary(binary, operator, **kw) - - def visit_not_endswith_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.right = percent.concat(binary.right) - return self.visit_not_like_op_binary(binary, operator, **kw) - - def visit_iendswith_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.left = ilike_case_insensitive(binary.left) - binary.right = percent.concat(ilike_case_insensitive(binary.right)) - return self.visit_ilike_op_binary(binary, operator, **kw) - - def visit_not_iendswith_op_binary(self, binary, operator, **kw): - binary = binary._clone() - percent = self._like_percent_literal - binary.left = ilike_case_insensitive(binary.left) - binary.right = percent.concat(ilike_case_insensitive(binary.right)) - return self.visit_not_ilike_op_binary(binary, operator, **kw) - - def visit_like_op_binary(self, binary, operator, **kw): - escape = binary.modifiers.get("escape", None) - - return "%s LIKE %s" % ( - binary.left._compiler_dispatch(self, **kw), - binary.right._compiler_dispatch(self, **kw), - ) + ( - " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE) - if escape is not None - else "" - ) - - def visit_not_like_op_binary(self, binary, operator, **kw): - escape = binary.modifiers.get("escape", None) - return "%s NOT LIKE %s" % ( - binary.left._compiler_dispatch(self, **kw), - binary.right._compiler_dispatch(self, **kw), - ) + ( - " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE) - if escape is not None - else "" - ) - - def visit_ilike_op_binary(self, binary, operator, **kw): - if operator is operators.ilike_op: - binary = binary._clone() - binary.left = ilike_case_insensitive(binary.left) - binary.right = ilike_case_insensitive(binary.right) - # else we assume ilower() has been applied - - return self.visit_like_op_binary(binary, operator, **kw) - - def visit_not_ilike_op_binary(self, binary, operator, **kw): - if operator is operators.not_ilike_op: - binary = binary._clone() - binary.left = ilike_case_insensitive(binary.left) - binary.right = ilike_case_insensitive(binary.right) - # else we assume ilower() has been applied - - return self.visit_not_like_op_binary(binary, operator, **kw) - - def visit_between_op_binary(self, binary, operator, **kw): - symmetric = binary.modifiers.get("symmetric", False) - return self._generate_generic_binary( - binary, " BETWEEN SYMMETRIC " if symmetric else " BETWEEN ", **kw - ) - - def visit_not_between_op_binary(self, binary, operator, **kw): - symmetric = binary.modifiers.get("symmetric", False) - return self._generate_generic_binary( - binary, - " NOT BETWEEN SYMMETRIC " if symmetric else " NOT BETWEEN ", - **kw, - ) - - def visit_regexp_match_op_binary(self, binary, operator, **kw): - raise exc.CompileError( - "%s dialect does not support regular expressions" - % self.dialect.name - ) - - def visit_not_regexp_match_op_binary(self, binary, operator, **kw): - raise exc.CompileError( - "%s dialect does not support regular expressions" - % self.dialect.name - ) - - def visit_regexp_replace_op_binary(self, binary, operator, **kw): - raise exc.CompileError( - "%s dialect does not support regular expression replacements" - % self.dialect.name - ) - - def visit_bindparam( - self, - bindparam, - within_columns_clause=False, - literal_binds=False, - skip_bind_expression=False, - literal_execute=False, - render_postcompile=False, - **kwargs, - ): - - if not skip_bind_expression: - impl = bindparam.type.dialect_impl(self.dialect) - if impl._has_bind_expression: - bind_expression = impl.bind_expression(bindparam) - wrapped = self.process( - bind_expression, - skip_bind_expression=True, - within_columns_clause=within_columns_clause, - literal_binds=literal_binds and not bindparam.expanding, - literal_execute=literal_execute, - render_postcompile=render_postcompile, - **kwargs, - ) - if bindparam.expanding: - # for postcompile w/ expanding, move the "wrapped" part - # of this into the inside - - m = re.match( - r"^(.*)\(__\[POSTCOMPILE_(\S+?)\]\)(.*)$", wrapped - ) - assert m, "unexpected format for expanding parameter" - wrapped = "(__[POSTCOMPILE_%s~~%s~~REPL~~%s~~])" % ( - m.group(2), - m.group(1), - m.group(3), - ) - - if literal_binds: - ret = self.render_literal_bindparam( - bindparam, - within_columns_clause=True, - bind_expression_template=wrapped, - **kwargs, - ) - return "(%s)" % ret - - return wrapped - - if not literal_binds: - literal_execute = ( - literal_execute - or bindparam.literal_execute - or (within_columns_clause and self.ansi_bind_rules) - ) - post_compile = literal_execute or bindparam.expanding - else: - post_compile = False - - if literal_binds: - ret = self.render_literal_bindparam( - bindparam, within_columns_clause=True, **kwargs - ) - if bindparam.expanding: - ret = "(%s)" % ret - return ret - - name = self._truncate_bindparam(bindparam) - - if name in self.binds: - existing = self.binds[name] - if existing is not bindparam: - if ( - (existing.unique or bindparam.unique) - and not existing.proxy_set.intersection( - bindparam.proxy_set - ) - and not existing._cloned_set.intersection( - bindparam._cloned_set - ) - ): - raise exc.CompileError( - "Bind parameter '%s' conflicts with " - "unique bind parameter of the same name" % name - ) - elif existing.expanding != bindparam.expanding: - raise exc.CompileError( - "Can't reuse bound parameter name '%s' in both " - "'expanding' (e.g. within an IN expression) and " - "non-expanding contexts. If this parameter is to " - "receive a list/array value, set 'expanding=True' on " - "it for expressions that aren't IN, otherwise use " - "a different parameter name." % (name,) - ) - elif existing._is_crud or bindparam._is_crud: - if existing._is_crud and bindparam._is_crud: - # TODO: this condition is not well understood. - # see tests in test/sql/test_update.py - raise exc.CompileError( - "Encountered unsupported case when compiling an " - "INSERT or UPDATE statement. If this is a " - "multi-table " - "UPDATE statement, please provide string-named " - "arguments to the " - "values() method with distinct names; support for " - "multi-table UPDATE statements that " - "target multiple tables for UPDATE is very " - "limited", - ) - else: - raise exc.CompileError( - f"bindparam() name '{bindparam.key}' is reserved " - "for automatic usage in the VALUES or SET " - "clause of this " - "insert/update statement. Please use a " - "name other than column name when using " - "bindparam() " - "with insert() or update() (for example, " - f"'b_{bindparam.key}')." - ) - - self.binds[bindparam.key] = self.binds[name] = bindparam - - # if we are given a cache key that we're going to match against, - # relate the bindparam here to one that is most likely present - # in the "extracted params" portion of the cache key. this is used - # to set up a positional mapping that is used to determine the - # correct parameters for a subsequent use of this compiled with - # a different set of parameter values. here, we accommodate for - # parameters that may have been cloned both before and after the cache - # key was been generated. - ckbm_tuple = self._cache_key_bind_match - - if ckbm_tuple: - ckbm, cksm = ckbm_tuple - for bp in bindparam._cloned_set: - if bp.key in cksm: - cb = cksm[bp.key] - ckbm[cb].append(bindparam) - - if bindparam.isoutparam: - self.has_out_parameters = True - - if post_compile: - if render_postcompile: - self._render_postcompile = True - - if literal_execute: - self.literal_execute_params |= {bindparam} - else: - self.post_compile_params |= {bindparam} - - ret = self.bindparam_string( - name, - post_compile=post_compile, - expanding=bindparam.expanding, - bindparam_type=bindparam.type, - **kwargs, - ) - - if bindparam.expanding: - ret = "(%s)" % ret - - return ret - - def render_bind_cast(self, type_, dbapi_type, sqltext): - raise NotImplementedError() - - def render_literal_bindparam( - self, - bindparam, - render_literal_value=NO_ARG, - bind_expression_template=None, - **kw, - ): - if render_literal_value is not NO_ARG: - value = render_literal_value - else: - if bindparam.value is None and bindparam.callable is None: - op = kw.get("_binary_op", None) - if op and op not in (operators.is_, operators.is_not): - util.warn_limited( - "Bound parameter '%s' rendering literal NULL in a SQL " - "expression; comparisons to NULL should not use " - "operators outside of 'is' or 'is not'", - (bindparam.key,), - ) - return self.process(sqltypes.NULLTYPE, **kw) - value = bindparam.effective_value - - if bindparam.expanding: - leep = self._literal_execute_expanding_parameter_literal_binds - to_update, replacement_expr = leep( - bindparam, - value, - bind_expression_template=bind_expression_template, - ) - return replacement_expr - else: - return self.render_literal_value(value, bindparam.type) - - def render_literal_value(self, value, type_): - """Render the value of a bind parameter as a quoted literal. - - This is used for statement sections that do not accept bind parameters - on the target driver/database. - - This should be implemented by subclasses using the quoting services - of the DBAPI. - - """ - - if value is None and not type_.should_evaluate_none: - # issue #10535 - handle NULL in the compiler without placing - # this onto each type, except for "evaluate None" types - # (e.g. JSON) - return self.process(elements.Null._instance()) - - processor = type_._cached_literal_processor(self.dialect) - if processor: - try: - return processor(value) - except Exception as e: - raise exc.CompileError( - f"Could not render literal value " - f'"{sql_util._repr_single_value(value)}" ' - f"with datatype " - f"{type_}; see parent stack trace for " - "more detail." - ) from e - - else: - raise exc.CompileError( - f"No literal value renderer is available for literal value " - f'"{sql_util._repr_single_value(value)}" ' - f"with datatype {type_}" - ) - - def _truncate_bindparam(self, bindparam): - if bindparam in self.bind_names: - return self.bind_names[bindparam] - - bind_name = bindparam.key - if isinstance(bind_name, elements._truncated_label): - bind_name = self._truncated_identifier("bindparam", bind_name) - - # add to bind_names for translation - self.bind_names[bindparam] = bind_name - - return bind_name - - def _truncated_identifier( - self, ident_class: str, name: _truncated_label - ) -> str: - if (ident_class, name) in self.truncated_names: - return self.truncated_names[(ident_class, name)] - - anonname = name.apply_map(self.anon_map) - - if len(anonname) > self.label_length - 6: - counter = self._truncated_counters.get(ident_class, 1) - truncname = ( - anonname[0 : max(self.label_length - 6, 0)] - + "_" - + hex(counter)[2:] - ) - self._truncated_counters[ident_class] = counter + 1 - else: - truncname = anonname - self.truncated_names[(ident_class, name)] = truncname - return truncname - - def _anonymize(self, name: str) -> str: - return name % self.anon_map - - def bindparam_string( - self, - name: str, - post_compile: bool = False, - expanding: bool = False, - escaped_from: Optional[str] = None, - bindparam_type: Optional[TypeEngine[Any]] = None, - accumulate_bind_names: Optional[Set[str]] = None, - visited_bindparam: Optional[List[str]] = None, - **kw: Any, - ) -> str: - # TODO: accumulate_bind_names is passed by crud.py to gather - # names on a per-value basis, visited_bindparam is passed by - # visit_insert() to collect all parameters in the statement. - # see if this gathering can be simplified somehow - if accumulate_bind_names is not None: - accumulate_bind_names.add(name) - if visited_bindparam is not None: - visited_bindparam.append(name) - - if not escaped_from: - if self._bind_translate_re.search(name): - # not quite the translate use case as we want to - # also get a quick boolean if we even found - # unusual characters in the name - new_name = self._bind_translate_re.sub( - lambda m: self._bind_translate_chars[m.group(0)], - name, - ) - escaped_from = name - name = new_name - - if escaped_from: - self.escaped_bind_names = self.escaped_bind_names.union( - {escaped_from: name} - ) - if post_compile: - ret = "__[POSTCOMPILE_%s]" % name - if expanding: - # for expanding, bound parameters or literal values will be - # rendered per item - return ret - - # otherwise, for non-expanding "literal execute", apply - # bind casts as determined by the datatype - if bindparam_type is not None: - type_impl = bindparam_type._unwrapped_dialect_impl( - self.dialect - ) - if type_impl.render_literal_cast: - ret = self.render_bind_cast(bindparam_type, type_impl, ret) - return ret - elif self.state is CompilerState.COMPILING: - ret = self.compilation_bindtemplate % {"name": name} - else: - ret = self.bindtemplate % {"name": name} - - if ( - bindparam_type is not None - and self.dialect._bind_typing_render_casts - ): - type_impl = bindparam_type._unwrapped_dialect_impl(self.dialect) - if type_impl.render_bind_cast: - ret = self.render_bind_cast(bindparam_type, type_impl, ret) - - return ret - - def _dispatch_independent_ctes(self, stmt, kw): - local_kw = kw.copy() - local_kw.pop("cte_opts", None) - for cte, opt in zip( - stmt._independent_ctes, stmt._independent_ctes_opts - ): - cte._compiler_dispatch(self, cte_opts=opt, **local_kw) - - def visit_cte( - self, - cte: CTE, - asfrom: bool = False, - ashint: bool = False, - fromhints: Optional[_FromHintsType] = None, - visiting_cte: Optional[CTE] = None, - from_linter: Optional[FromLinter] = None, - cte_opts: selectable._CTEOpts = selectable._CTEOpts(False), - **kwargs: Any, - ) -> Optional[str]: - self_ctes = self._init_cte_state() - assert self_ctes is self.ctes - - kwargs["visiting_cte"] = cte - - cte_name = cte.name - - if isinstance(cte_name, elements._truncated_label): - cte_name = self._truncated_identifier("alias", cte_name) - - is_new_cte = True - embedded_in_current_named_cte = False - - _reference_cte = cte._get_reference_cte() - - nesting = cte.nesting or cte_opts.nesting - - # check for CTE already encountered - if _reference_cte in self.level_name_by_cte: - cte_level, _, existing_cte_opts = self.level_name_by_cte[ - _reference_cte - ] - assert _ == cte_name - - cte_level_name = (cte_level, cte_name) - existing_cte = self.ctes_by_level_name[cte_level_name] - - # check if we are receiving it here with a specific - # "nest_here" location; if so, move it to this location - - if cte_opts.nesting: - if existing_cte_opts.nesting: - raise exc.CompileError( - "CTE is stated as 'nest_here' in " - "more than one location" - ) - - old_level_name = (cte_level, cte_name) - cte_level = len(self.stack) if nesting else 1 - cte_level_name = new_level_name = (cte_level, cte_name) - - del self.ctes_by_level_name[old_level_name] - self.ctes_by_level_name[new_level_name] = existing_cte - self.level_name_by_cte[_reference_cte] = new_level_name + ( - cte_opts, - ) - - else: - cte_level = len(self.stack) if nesting else 1 - cte_level_name = (cte_level, cte_name) - - if cte_level_name in self.ctes_by_level_name: - existing_cte = self.ctes_by_level_name[cte_level_name] - else: - existing_cte = None - - if existing_cte is not None: - embedded_in_current_named_cte = visiting_cte is existing_cte - - # we've generated a same-named CTE that we are enclosed in, - # or this is the same CTE. just return the name. - if cte is existing_cte._restates or cte is existing_cte: - is_new_cte = False - elif existing_cte is cte._restates: - # we've generated a same-named CTE that is - # enclosed in us - we take precedence, so - # discard the text for the "inner". - del self_ctes[existing_cte] - - existing_cte_reference_cte = existing_cte._get_reference_cte() - - assert existing_cte_reference_cte is _reference_cte - assert existing_cte_reference_cte is existing_cte - - del self.level_name_by_cte[existing_cte_reference_cte] - else: - # if the two CTEs are deep-copy identical, consider them - # the same, **if** they are clones, that is, they came from - # the ORM or other visit method - if ( - cte._is_clone_of is not None - or existing_cte._is_clone_of is not None - ) and cte.compare(existing_cte): - is_new_cte = False - else: - raise exc.CompileError( - "Multiple, unrelated CTEs found with " - "the same name: %r" % cte_name - ) - - if not asfrom and not is_new_cte: - return None - - if cte._cte_alias is not None: - pre_alias_cte = cte._cte_alias - cte_pre_alias_name = cte._cte_alias.name - if isinstance(cte_pre_alias_name, elements._truncated_label): - cte_pre_alias_name = self._truncated_identifier( - "alias", cte_pre_alias_name - ) - else: - pre_alias_cte = cte - cte_pre_alias_name = None - - if is_new_cte: - self.ctes_by_level_name[cte_level_name] = cte - self.level_name_by_cte[_reference_cte] = cte_level_name + ( - cte_opts, - ) - - if pre_alias_cte not in self.ctes: - self.visit_cte(pre_alias_cte, **kwargs) - - if not cte_pre_alias_name and cte not in self_ctes: - if cte.recursive: - self.ctes_recursive = True - text = self.preparer.format_alias(cte, cte_name) - if cte.recursive: - col_source = cte.element - - # TODO: can we get at the .columns_plus_names collection - # that is already (or will be?) generated for the SELECT - # rather than calling twice? - recur_cols = [ - # TODO: proxy_name is not technically safe, - # see test_cte-> - # test_with_recursive_no_name_currently_buggy. not - # clear what should be done with such a case - fallback_label_name or proxy_name - for ( - _, - proxy_name, - fallback_label_name, - c, - repeated, - ) in (col_source._generate_columns_plus_names(True)) - if not repeated - ] - - text += "(%s)" % ( - ", ".join( - self.preparer.format_label_name( - ident, anon_map=self.anon_map - ) - for ident in recur_cols - ) - ) - - assert kwargs.get("subquery", False) is False - - if not self.stack: - # toplevel, this is a stringify of the - # cte directly. just compile the inner - # the way alias() does. - return cte.element._compiler_dispatch( - self, asfrom=asfrom, **kwargs - ) - else: - prefixes = self._generate_prefixes( - cte, cte._prefixes, **kwargs - ) - inner = cte.element._compiler_dispatch( - self, asfrom=True, **kwargs - ) - - text += " AS %s\n(%s)" % (prefixes, inner) - - if cte._suffixes: - text += " " + self._generate_prefixes( - cte, cte._suffixes, **kwargs - ) - - self_ctes[cte] = text - - if asfrom: - if from_linter: - from_linter.froms[cte._de_clone()] = cte_name - - if not is_new_cte and embedded_in_current_named_cte: - return self.preparer.format_alias(cte, cte_name) - - if cte_pre_alias_name: - text = self.preparer.format_alias(cte, cte_pre_alias_name) - if self.preparer._requires_quotes(cte_name): - cte_name = self.preparer.quote(cte_name) - text += self.get_render_as_alias_suffix(cte_name) - return text - else: - return self.preparer.format_alias(cte, cte_name) - - return None - - def visit_table_valued_alias(self, element, **kw): - if element.joins_implicitly: - kw["from_linter"] = None - if element._is_lateral: - return self.visit_lateral(element, **kw) - else: - return self.visit_alias(element, **kw) - - def visit_table_valued_column(self, element, **kw): - return self.visit_column(element, **kw) - - def visit_alias( - self, - alias, - asfrom=False, - ashint=False, - iscrud=False, - fromhints=None, - subquery=False, - lateral=False, - enclosing_alias=None, - from_linter=None, - **kwargs, - ): - if lateral: - if "enclosing_lateral" not in kwargs: - # if lateral is set and enclosing_lateral is not - # present, we assume we are being called directly - # from visit_lateral() and we need to set enclosing_lateral. - assert alias._is_lateral - kwargs["enclosing_lateral"] = alias - - # for lateral objects, we track a second from_linter that is... - # lateral! to the level above us. - if ( - from_linter - and "lateral_from_linter" not in kwargs - and "enclosing_lateral" in kwargs - ): - kwargs["lateral_from_linter"] = from_linter - - if enclosing_alias is not None and enclosing_alias.element is alias: - inner = alias.element._compiler_dispatch( - self, - asfrom=asfrom, - ashint=ashint, - iscrud=iscrud, - fromhints=fromhints, - lateral=lateral, - enclosing_alias=alias, - **kwargs, - ) - if subquery and (asfrom or lateral): - inner = "(%s)" % (inner,) - return inner - else: - enclosing_alias = kwargs["enclosing_alias"] = alias - - if asfrom or ashint: - if isinstance(alias.name, elements._truncated_label): - alias_name = self._truncated_identifier("alias", alias.name) - else: - alias_name = alias.name - - if ashint: - return self.preparer.format_alias(alias, alias_name) - elif asfrom: - if from_linter: - from_linter.froms[alias._de_clone()] = alias_name - - inner = alias.element._compiler_dispatch( - self, asfrom=True, lateral=lateral, **kwargs - ) - if subquery: - inner = "(%s)" % (inner,) - - ret = inner + self.get_render_as_alias_suffix( - self.preparer.format_alias(alias, alias_name) - ) - - if alias._supports_derived_columns and alias._render_derived: - ret += "(%s)" % ( - ", ".join( - "%s%s" - % ( - self.preparer.quote(col.name), - ( - " %s" - % self.dialect.type_compiler_instance.process( - col.type, **kwargs - ) - if alias._render_derived_w_types - else "" - ), - ) - for col in alias.c - ) - ) - - if fromhints and alias in fromhints: - ret = self.format_from_hint_text( - ret, alias, fromhints[alias], iscrud - ) - - return ret - else: - # note we cancel the "subquery" flag here as well - return alias.element._compiler_dispatch( - self, lateral=lateral, **kwargs - ) - - def visit_subquery(self, subquery, **kw): - kw["subquery"] = True - return self.visit_alias(subquery, **kw) - - def visit_lateral(self, lateral_, **kw): - kw["lateral"] = True - return "LATERAL %s" % self.visit_alias(lateral_, **kw) - - def visit_tablesample(self, tablesample, asfrom=False, **kw): - text = "%s TABLESAMPLE %s" % ( - self.visit_alias(tablesample, asfrom=True, **kw), - tablesample._get_method()._compiler_dispatch(self, **kw), - ) - - if tablesample.seed is not None: - text += " REPEATABLE (%s)" % ( - tablesample.seed._compiler_dispatch(self, **kw) - ) - - return text - - def _render_values(self, element, **kw): - kw.setdefault("literal_binds", element.literal_binds) - tuples = ", ".join( - self.process( - elements.Tuple( - types=element._column_types, *elem - ).self_group(), - **kw, - ) - for chunk in element._data - for elem in chunk - ) - return f"VALUES {tuples}" - - def visit_values(self, element, asfrom=False, from_linter=None, **kw): - v = self._render_values(element, **kw) - - if element._unnamed: - name = None - elif isinstance(element.name, elements._truncated_label): - name = self._truncated_identifier("values", element.name) - else: - name = element.name - - if element._is_lateral: - lateral = "LATERAL " - else: - lateral = "" - - if asfrom: - if from_linter: - from_linter.froms[element._de_clone()] = ( - name if name is not None else "(unnamed VALUES element)" - ) - - if name: - kw["include_table"] = False - v = "%s(%s)%s (%s)" % ( - lateral, - v, - self.get_render_as_alias_suffix(self.preparer.quote(name)), - ( - ", ".join( - c._compiler_dispatch(self, **kw) - for c in element.columns - ) - ), - ) - else: - v = "%s(%s)" % (lateral, v) - return v - - def visit_scalar_values(self, element, **kw): - return f"({self._render_values(element, **kw)})" - - def get_render_as_alias_suffix(self, alias_name_text): - return " AS " + alias_name_text - - def _add_to_result_map( - self, - keyname: str, - name: str, - objects: Tuple[Any, ...], - type_: TypeEngine[Any], - ) -> None: - if keyname is None or keyname == "*": - self._ordered_columns = False - self._ad_hoc_textual = True - if type_._is_tuple_type: - raise exc.CompileError( - "Most backends don't support SELECTing " - "from a tuple() object. If this is an ORM query, " - "consider using the Bundle object." - ) - self._result_columns.append( - ResultColumnsEntry(keyname, name, objects, type_) - ) - - def _label_returning_column( - self, stmt, column, populate_result_map, column_clause_args=None, **kw - ): - """Render a column with necessary labels inside of a RETURNING clause. - - This method is provided for individual dialects in place of calling - the _label_select_column method directly, so that the two use cases - of RETURNING vs. SELECT can be disambiguated going forward. - - .. versionadded:: 1.4.21 - - """ - return self._label_select_column( - None, - column, - populate_result_map, - False, - {} if column_clause_args is None else column_clause_args, - **kw, - ) - - def _label_select_column( - self, - select, - column, - populate_result_map, - asfrom, - column_clause_args, - name=None, - proxy_name=None, - fallback_label_name=None, - within_columns_clause=True, - column_is_repeated=False, - need_column_expressions=False, - include_table=True, - ): - """produce labeled columns present in a select().""" - impl = column.type.dialect_impl(self.dialect) - - if impl._has_column_expression and ( - need_column_expressions or populate_result_map - ): - col_expr = impl.column_expression(column) - else: - col_expr = column - - if populate_result_map: - # pass an "add_to_result_map" callable into the compilation - # of embedded columns. this collects information about the - # column as it will be fetched in the result and is coordinated - # with cursor.description when the query is executed. - add_to_result_map = self._add_to_result_map - - # if the SELECT statement told us this column is a repeat, - # wrap the callable with one that prevents the addition of the - # targets - if column_is_repeated: - _add_to_result_map = add_to_result_map - - def add_to_result_map(keyname, name, objects, type_): - _add_to_result_map(keyname, name, (), type_) - - # if we redefined col_expr for type expressions, wrap the - # callable with one that adds the original column to the targets - elif col_expr is not column: - _add_to_result_map = add_to_result_map - - def add_to_result_map(keyname, name, objects, type_): - _add_to_result_map( - keyname, name, (column,) + objects, type_ - ) - - else: - add_to_result_map = None - - # this method is used by some of the dialects for RETURNING, - # which has different inputs. _label_returning_column was added - # as the better target for this now however for 1.4 we will keep - # _label_select_column directly compatible with this use case. - # these assertions right now set up the current expected inputs - assert within_columns_clause, ( - "_label_select_column is only relevant within " - "the columns clause of a SELECT or RETURNING" - ) - if isinstance(column, elements.Label): - if col_expr is not column: - result_expr = _CompileLabel( - col_expr, column.name, alt_names=(column.element,) - ) - else: - result_expr = col_expr - - elif name: - # here, _columns_plus_names has determined there's an explicit - # label name we need to use. this is the default for - # tablenames_plus_columnnames as well as when columns are being - # deduplicated on name - - assert ( - proxy_name is not None - ), "proxy_name is required if 'name' is passed" - - result_expr = _CompileLabel( - col_expr, - name, - alt_names=( - proxy_name, - # this is a hack to allow legacy result column lookups - # to work as they did before; this goes away in 2.0. - # TODO: this only seems to be tested indirectly - # via test/orm/test_deprecations.py. should be a - # resultset test for this - column._tq_label, - ), - ) - else: - # determine here whether this column should be rendered in - # a labelled context or not, as we were given no required label - # name from the caller. Here we apply heuristics based on the kind - # of SQL expression involved. - - if col_expr is not column: - # type-specific expression wrapping the given column, - # so we render a label - render_with_label = True - elif isinstance(column, elements.ColumnClause): - # table-bound column, we render its name as a label if we are - # inside of a subquery only - render_with_label = ( - asfrom - and not column.is_literal - and column.table is not None - ) - elif isinstance(column, elements.TextClause): - render_with_label = False - elif isinstance(column, elements.UnaryExpression): - render_with_label = column.wraps_column_expression or asfrom - elif ( - # general class of expressions that don't have a SQL-column - # addressible name. includes scalar selects, bind parameters, - # SQL functions, others - not isinstance(column, elements.NamedColumn) - # deeper check that indicates there's no natural "name" to - # this element, which accommodates for custom SQL constructs - # that might have a ".name" attribute (but aren't SQL - # functions) but are not implementing this more recently added - # base class. in theory the "NamedColumn" check should be - # enough, however here we seek to maintain legacy behaviors - # as well. - and column._non_anon_label is None - ): - render_with_label = True - else: - render_with_label = False - - if render_with_label: - if not fallback_label_name: - # used by the RETURNING case right now. we generate it - # here as 3rd party dialects may be referring to - # _label_select_column method directly instead of the - # just-added _label_returning_column method - assert not column_is_repeated - fallback_label_name = column._anon_name_label - - fallback_label_name = ( - elements._truncated_label(fallback_label_name) - if not isinstance( - fallback_label_name, elements._truncated_label - ) - else fallback_label_name - ) - - result_expr = _CompileLabel( - col_expr, fallback_label_name, alt_names=(proxy_name,) - ) - else: - result_expr = col_expr - - column_clause_args.update( - within_columns_clause=within_columns_clause, - add_to_result_map=add_to_result_map, - include_table=include_table, - ) - return result_expr._compiler_dispatch(self, **column_clause_args) - - def format_from_hint_text(self, sqltext, table, hint, iscrud): - hinttext = self.get_from_hint_text(table, hint) - if hinttext: - sqltext += " " + hinttext - return sqltext - - def get_select_hint_text(self, byfroms): - return None - - def get_from_hint_text(self, table, text): - return None - - def get_crud_hint_text(self, table, text): - return None - - def get_statement_hint_text(self, hint_texts): - return " ".join(hint_texts) - - _default_stack_entry: _CompilerStackEntry - - if not typing.TYPE_CHECKING: - _default_stack_entry = util.immutabledict( - [("correlate_froms", frozenset()), ("asfrom_froms", frozenset())] - ) - - def _display_froms_for_select( - self, select_stmt, asfrom, lateral=False, **kw - ): - # utility method to help external dialects - # get the correct from list for a select. - # specifically the oracle dialect needs this feature - # right now. - toplevel = not self.stack - entry = self._default_stack_entry if toplevel else self.stack[-1] - - compile_state = select_stmt._compile_state_factory(select_stmt, self) - - correlate_froms = entry["correlate_froms"] - asfrom_froms = entry["asfrom_froms"] - - if asfrom and not lateral: - froms = compile_state._get_display_froms( - explicit_correlate_froms=correlate_froms.difference( - asfrom_froms - ), - implicit_correlate_froms=(), - ) - else: - froms = compile_state._get_display_froms( - explicit_correlate_froms=correlate_froms, - implicit_correlate_froms=asfrom_froms, - ) - return froms - - translate_select_structure: Any = None - """if not ``None``, should be a callable which accepts ``(select_stmt, - **kw)`` and returns a select object. this is used for structural changes - mostly to accommodate for LIMIT/OFFSET schemes - - """ - - def visit_select( - self, - select_stmt, - asfrom=False, - insert_into=False, - fromhints=None, - compound_index=None, - select_wraps_for=None, - lateral=False, - from_linter=None, - **kwargs, - ): - assert select_wraps_for is None, ( - "SQLAlchemy 1.4 requires use of " - "the translate_select_structure hook for structural " - "translations of SELECT objects" - ) - - # initial setup of SELECT. the compile_state_factory may now - # be creating a totally different SELECT from the one that was - # passed in. for ORM use this will convert from an ORM-state - # SELECT to a regular "Core" SELECT. other composed operations - # such as computation of joins will be performed. - - kwargs["within_columns_clause"] = False - - compile_state = select_stmt._compile_state_factory( - select_stmt, self, **kwargs - ) - kwargs["ambiguous_table_name_map"] = ( - compile_state._ambiguous_table_name_map - ) - - select_stmt = compile_state.statement - - toplevel = not self.stack - - if toplevel and not self.compile_state: - self.compile_state = compile_state - - is_embedded_select = compound_index is not None or insert_into - - # translate step for Oracle, SQL Server which often need to - # restructure the SELECT to allow for LIMIT/OFFSET and possibly - # other conditions - if self.translate_select_structure: - new_select_stmt = self.translate_select_structure( - select_stmt, asfrom=asfrom, **kwargs - ) - - # if SELECT was restructured, maintain a link to the originals - # and assemble a new compile state - if new_select_stmt is not select_stmt: - compile_state_wraps_for = compile_state - select_wraps_for = select_stmt - select_stmt = new_select_stmt - - compile_state = select_stmt._compile_state_factory( - select_stmt, self, **kwargs - ) - select_stmt = compile_state.statement - - entry = self._default_stack_entry if toplevel else self.stack[-1] - - populate_result_map = need_column_expressions = ( - toplevel - or entry.get("need_result_map_for_compound", False) - or entry.get("need_result_map_for_nested", False) - ) - - # indicates there is a CompoundSelect in play and we are not the - # first select - if compound_index: - populate_result_map = False - - # this was first proposed as part of #3372; however, it is not - # reached in current tests and could possibly be an assertion - # instead. - if not populate_result_map and "add_to_result_map" in kwargs: - del kwargs["add_to_result_map"] - - froms = self._setup_select_stack( - select_stmt, compile_state, entry, asfrom, lateral, compound_index - ) - - column_clause_args = kwargs.copy() - column_clause_args.update( - {"within_label_clause": False, "within_columns_clause": False} - ) - - text = "SELECT " # we're off to a good start ! - - if select_stmt._hints: - hint_text, byfrom = self._setup_select_hints(select_stmt) - if hint_text: - text += hint_text + " " - else: - byfrom = None - - if select_stmt._independent_ctes: - self._dispatch_independent_ctes(select_stmt, kwargs) - - if select_stmt._prefixes: - text += self._generate_prefixes( - select_stmt, select_stmt._prefixes, **kwargs - ) - - text += self.get_select_precolumns(select_stmt, **kwargs) - # the actual list of columns to print in the SELECT column list. - inner_columns = [ - c - for c in [ - self._label_select_column( - select_stmt, - column, - populate_result_map, - asfrom, - column_clause_args, - name=name, - proxy_name=proxy_name, - fallback_label_name=fallback_label_name, - column_is_repeated=repeated, - need_column_expressions=need_column_expressions, - ) - for ( - name, - proxy_name, - fallback_label_name, - column, - repeated, - ) in compile_state.columns_plus_names - ] - if c is not None - ] - - if populate_result_map and select_wraps_for is not None: - # if this select was generated from translate_select, - # rewrite the targeted columns in the result map - - translate = dict( - zip( - [ - name - for ( - key, - proxy_name, - fallback_label_name, - name, - repeated, - ) in compile_state.columns_plus_names - ], - [ - name - for ( - key, - proxy_name, - fallback_label_name, - name, - repeated, - ) in compile_state_wraps_for.columns_plus_names - ], - ) - ) - - self._result_columns = [ - ResultColumnsEntry( - key, name, tuple(translate.get(o, o) for o in obj), type_ - ) - for key, name, obj, type_ in self._result_columns - ] - - text = self._compose_select_body( - text, - select_stmt, - compile_state, - inner_columns, - froms, - byfrom, - toplevel, - kwargs, - ) - - if select_stmt._statement_hints: - per_dialect = [ - ht - for (dialect_name, ht) in select_stmt._statement_hints - if dialect_name in ("*", self.dialect.name) - ] - if per_dialect: - text += " " + self.get_statement_hint_text(per_dialect) - - # In compound query, CTEs are shared at the compound level - if self.ctes and (not is_embedded_select or toplevel): - nesting_level = len(self.stack) if not toplevel else None - text = self._render_cte_clause(nesting_level=nesting_level) + text - - if select_stmt._suffixes: - text += " " + self._generate_prefixes( - select_stmt, select_stmt._suffixes, **kwargs - ) - - self.stack.pop(-1) - - return text - - def _setup_select_hints( - self, select: Select[Any] - ) -> Tuple[str, _FromHintsType]: - byfrom = { - from_: hinttext - % {"name": from_._compiler_dispatch(self, ashint=True)} - for (from_, dialect), hinttext in select._hints.items() - if dialect in ("*", self.dialect.name) - } - hint_text = self.get_select_hint_text(byfrom) - return hint_text, byfrom - - def _setup_select_stack( - self, select, compile_state, entry, asfrom, lateral, compound_index - ): - correlate_froms = entry["correlate_froms"] - asfrom_froms = entry["asfrom_froms"] - - if compound_index == 0: - entry["select_0"] = select - elif compound_index: - select_0 = entry["select_0"] - numcols = len(select_0._all_selected_columns) - - if len(compile_state.columns_plus_names) != numcols: - raise exc.CompileError( - "All selectables passed to " - "CompoundSelect must have identical numbers of " - "columns; select #%d has %d columns, select " - "#%d has %d" - % ( - 1, - numcols, - compound_index + 1, - len(select._all_selected_columns), - ) - ) - - if asfrom and not lateral: - froms = compile_state._get_display_froms( - explicit_correlate_froms=correlate_froms.difference( - asfrom_froms - ), - implicit_correlate_froms=(), - ) - else: - froms = compile_state._get_display_froms( - explicit_correlate_froms=correlate_froms, - implicit_correlate_froms=asfrom_froms, - ) - - new_correlate_froms = set(_from_objects(*froms)) - all_correlate_froms = new_correlate_froms.union(correlate_froms) - - new_entry: _CompilerStackEntry = { - "asfrom_froms": new_correlate_froms, - "correlate_froms": all_correlate_froms, - "selectable": select, - "compile_state": compile_state, - } - self.stack.append(new_entry) - - return froms - - def _compose_select_body( - self, - text, - select, - compile_state, - inner_columns, - froms, - byfrom, - toplevel, - kwargs, - ): - text += ", ".join(inner_columns) - - if self.linting & COLLECT_CARTESIAN_PRODUCTS: - from_linter = FromLinter({}, set()) - warn_linting = self.linting & WARN_LINTING - if toplevel: - self.from_linter = from_linter - else: - from_linter = None - warn_linting = False - - # adjust the whitespace for no inner columns, part of #9440, - # so that a no-col SELECT comes out as "SELECT WHERE..." or - # "SELECT FROM ...". - # while it would be better to have built the SELECT starting string - # without trailing whitespace first, then add whitespace only if inner - # cols were present, this breaks compatibility with various custom - # compilation schemes that are currently being tested. - if not inner_columns: - text = text.rstrip() - - if froms: - text += " \nFROM " - - if select._hints: - text += ", ".join( - [ - f._compiler_dispatch( - self, - asfrom=True, - fromhints=byfrom, - from_linter=from_linter, - **kwargs, - ) - for f in froms - ] - ) - else: - text += ", ".join( - [ - f._compiler_dispatch( - self, - asfrom=True, - from_linter=from_linter, - **kwargs, - ) - for f in froms - ] - ) - else: - text += self.default_from() - - if select._where_criteria: - t = self._generate_delimited_and_list( - select._where_criteria, from_linter=from_linter, **kwargs - ) - if t: - text += " \nWHERE " + t - - if warn_linting: - assert from_linter is not None - from_linter.warn() - - if select._group_by_clauses: - text += self.group_by_clause(select, **kwargs) - - if select._having_criteria: - t = self._generate_delimited_and_list( - select._having_criteria, **kwargs - ) - if t: - text += " \nHAVING " + t - - if select._order_by_clauses: - text += self.order_by_clause(select, **kwargs) - - if select._has_row_limiting_clause: - text += self._row_limit_clause(select, **kwargs) - - if select._for_update_arg is not None: - text += self.for_update_clause(select, **kwargs) - - return text - - def _generate_prefixes(self, stmt, prefixes, **kw): - clause = " ".join( - prefix._compiler_dispatch(self, **kw) - for prefix, dialect_name in prefixes - if dialect_name in (None, "*") or dialect_name == self.dialect.name - ) - if clause: - clause += " " - return clause - - def _render_cte_clause( - self, - nesting_level=None, - include_following_stack=False, - ): - """ - include_following_stack - Also render the nesting CTEs on the next stack. Useful for - SQL structures like UNION or INSERT that can wrap SELECT - statements containing nesting CTEs. - """ - if not self.ctes: - return "" - - ctes: MutableMapping[CTE, str] - - if nesting_level and nesting_level > 1: - ctes = util.OrderedDict() - for cte in list(self.ctes.keys()): - cte_level, cte_name, cte_opts = self.level_name_by_cte[ - cte._get_reference_cte() - ] - nesting = cte.nesting or cte_opts.nesting - is_rendered_level = cte_level == nesting_level or ( - include_following_stack and cte_level == nesting_level + 1 - ) - if not (nesting and is_rendered_level): - continue - - ctes[cte] = self.ctes[cte] - - else: - ctes = self.ctes - - if not ctes: - return "" - ctes_recursive = any([cte.recursive for cte in ctes]) - - cte_text = self.get_cte_preamble(ctes_recursive) + " " - cte_text += ", \n".join([txt for txt in ctes.values()]) - cte_text += "\n " - - if nesting_level and nesting_level > 1: - for cte in list(ctes.keys()): - cte_level, cte_name, cte_opts = self.level_name_by_cte[ - cte._get_reference_cte() - ] - del self.ctes[cte] - del self.ctes_by_level_name[(cte_level, cte_name)] - del self.level_name_by_cte[cte._get_reference_cte()] - - return cte_text - - def get_cte_preamble(self, recursive): - if recursive: - return "WITH RECURSIVE" - else: - return "WITH" - - def get_select_precolumns(self, select, **kw): - """Called when building a ``SELECT`` statement, position is just - before column list. - - """ - if select._distinct_on: - util.warn_deprecated( - "DISTINCT ON is currently supported only by the PostgreSQL " - "dialect. Use of DISTINCT ON for other backends is currently " - "silently ignored, however this usage is deprecated, and will " - "raise CompileError in a future release for all backends " - "that do not support this syntax.", - version="1.4", - ) - return "DISTINCT " if select._distinct else "" - - def group_by_clause(self, select, **kw): - """allow dialects to customize how GROUP BY is rendered.""" - - group_by = self._generate_delimited_list( - select._group_by_clauses, OPERATORS[operators.comma_op], **kw - ) - if group_by: - return " GROUP BY " + group_by - else: - return "" - - def order_by_clause(self, select, **kw): - """allow dialects to customize how ORDER BY is rendered.""" - - order_by = self._generate_delimited_list( - select._order_by_clauses, OPERATORS[operators.comma_op], **kw - ) - - if order_by: - return " ORDER BY " + order_by - else: - return "" - - def for_update_clause(self, select, **kw): - return " FOR UPDATE" - - def returning_clause( - self, - stmt: UpdateBase, - returning_cols: Sequence[ColumnElement[Any]], - *, - populate_result_map: bool, - **kw: Any, - ) -> str: - columns = [ - self._label_returning_column( - stmt, - column, - populate_result_map, - fallback_label_name=fallback_label_name, - column_is_repeated=repeated, - name=name, - proxy_name=proxy_name, - **kw, - ) - for ( - name, - proxy_name, - fallback_label_name, - column, - repeated, - ) in stmt._generate_columns_plus_names( - True, cols=base._select_iterables(returning_cols) - ) - ] - - return "RETURNING " + ", ".join(columns) - - def limit_clause(self, select, **kw): - text = "" - if select._limit_clause is not None: - text += "\n LIMIT " + self.process(select._limit_clause, **kw) - if select._offset_clause is not None: - if select._limit_clause is None: - text += "\n LIMIT -1" - text += " OFFSET " + self.process(select._offset_clause, **kw) - return text - - def fetch_clause( - self, - select, - fetch_clause=None, - require_offset=False, - use_literal_execute_for_simple_int=False, - **kw, - ): - if fetch_clause is None: - fetch_clause = select._fetch_clause - fetch_clause_options = select._fetch_clause_options - else: - fetch_clause_options = {"percent": False, "with_ties": False} - - text = "" - - if select._offset_clause is not None: - offset_clause = select._offset_clause - if ( - use_literal_execute_for_simple_int - and select._simple_int_clause(offset_clause) - ): - offset_clause = offset_clause.render_literal_execute() - offset_str = self.process(offset_clause, **kw) - text += "\n OFFSET %s ROWS" % offset_str - elif require_offset: - text += "\n OFFSET 0 ROWS" - - if fetch_clause is not None: - if ( - use_literal_execute_for_simple_int - and select._simple_int_clause(fetch_clause) - ): - fetch_clause = fetch_clause.render_literal_execute() - text += "\n FETCH FIRST %s%s ROWS %s" % ( - self.process(fetch_clause, **kw), - " PERCENT" if fetch_clause_options["percent"] else "", - "WITH TIES" if fetch_clause_options["with_ties"] else "ONLY", - ) - return text - - def visit_table( - self, - table, - asfrom=False, - iscrud=False, - ashint=False, - fromhints=None, - use_schema=True, - from_linter=None, - ambiguous_table_name_map=None, - **kwargs, - ): - if from_linter: - from_linter.froms[table] = table.fullname - - if asfrom or ashint: - effective_schema = self.preparer.schema_for_object(table) - - if use_schema and effective_schema: - ret = ( - self.preparer.quote_schema(effective_schema) - + "." - + self.preparer.quote(table.name) - ) - else: - ret = self.preparer.quote(table.name) - - if ( - not effective_schema - and ambiguous_table_name_map - and table.name in ambiguous_table_name_map - ): - anon_name = self._truncated_identifier( - "alias", ambiguous_table_name_map[table.name] - ) - - ret = ret + self.get_render_as_alias_suffix( - self.preparer.format_alias(None, anon_name) - ) - - if fromhints and table in fromhints: - ret = self.format_from_hint_text( - ret, table, fromhints[table], iscrud - ) - return ret - else: - return "" - - def visit_join(self, join, asfrom=False, from_linter=None, **kwargs): - if from_linter: - from_linter.edges.update( - itertools.product( - _de_clone(join.left._from_objects), - _de_clone(join.right._from_objects), - ) - ) - - if join.full: - join_type = " FULL OUTER JOIN " - elif join.isouter: - join_type = " LEFT OUTER JOIN " - else: - join_type = " JOIN " - return ( - join.left._compiler_dispatch( - self, asfrom=True, from_linter=from_linter, **kwargs - ) - + join_type - + join.right._compiler_dispatch( - self, asfrom=True, from_linter=from_linter, **kwargs - ) - + " ON " - # TODO: likely need asfrom=True here? - + join.onclause._compiler_dispatch( - self, from_linter=from_linter, **kwargs - ) - ) - - def _setup_crud_hints(self, stmt, table_text): - dialect_hints = { - table: hint_text - for (table, dialect), hint_text in stmt._hints.items() - if dialect in ("*", self.dialect.name) - } - if stmt.table in dialect_hints: - table_text = self.format_from_hint_text( - table_text, stmt.table, dialect_hints[stmt.table], True - ) - return dialect_hints, table_text - - # within the realm of "insertmanyvalues sentinel columns", - # these lookups match different kinds of Column() configurations - # to specific backend capabilities. they are broken into two - # lookups, one for autoincrement columns and the other for non - # autoincrement columns - _sentinel_col_non_autoinc_lookup = util.immutabledict( - { - _SentinelDefaultCharacterization.CLIENTSIDE: ( - InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT - ), - _SentinelDefaultCharacterization.SENTINEL_DEFAULT: ( - InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT - ), - _SentinelDefaultCharacterization.NONE: ( - InsertmanyvaluesSentinelOpts._SUPPORTED_OR_NOT - ), - _SentinelDefaultCharacterization.IDENTITY: ( - InsertmanyvaluesSentinelOpts.IDENTITY - ), - _SentinelDefaultCharacterization.SEQUENCE: ( - InsertmanyvaluesSentinelOpts.SEQUENCE - ), - } - ) - _sentinel_col_autoinc_lookup = _sentinel_col_non_autoinc_lookup.union( - { - _SentinelDefaultCharacterization.NONE: ( - InsertmanyvaluesSentinelOpts.AUTOINCREMENT - ), - } - ) - - def _get_sentinel_column_for_table( - self, table: Table - ) -> Optional[Sequence[Column[Any]]]: - """given a :class:`.Table`, return a usable sentinel column or - columns for this dialect if any. - - Return None if no sentinel columns could be identified, or raise an - error if a column was marked as a sentinel explicitly but isn't - compatible with this dialect. - - """ - - sentinel_opts = self.dialect.insertmanyvalues_implicit_sentinel - sentinel_characteristics = table._sentinel_column_characteristics - - sent_cols = sentinel_characteristics.columns - - if sent_cols is None: - return None - - if sentinel_characteristics.is_autoinc: - bitmask = self._sentinel_col_autoinc_lookup.get( - sentinel_characteristics.default_characterization, 0 - ) - else: - bitmask = self._sentinel_col_non_autoinc_lookup.get( - sentinel_characteristics.default_characterization, 0 - ) - - if sentinel_opts & bitmask: - return sent_cols - - if sentinel_characteristics.is_explicit: - # a column was explicitly marked as insert_sentinel=True, - # however it is not compatible with this dialect. they should - # not indicate this column as a sentinel if they need to include - # this dialect. - - # TODO: do we want non-primary key explicit sentinel cols - # that can gracefully degrade for some backends? - # insert_sentinel="degrade" perhaps. not for the initial release. - # I am hoping people are generally not dealing with this sentinel - # business at all. - - # if is_explicit is True, there will be only one sentinel column. - - raise exc.InvalidRequestError( - f"Column {sent_cols[0]} can't be explicitly " - "marked as a sentinel column when using the " - f"{self.dialect.name} dialect, as the " - "particular type of default generation on this column is " - "not currently compatible with this dialect's specific " - f"INSERT..RETURNING syntax which can receive the " - "server-generated value in " - "a deterministic way. To remove this error, remove " - "insert_sentinel=True from primary key autoincrement " - "columns; these columns are automatically used as " - "sentinels for supported dialects in any case." - ) - - return None - - def _deliver_insertmanyvalues_batches( - self, - statement: str, - parameters: _DBAPIMultiExecuteParams, - compiled_parameters: List[_MutableCoreSingleExecuteParams], - generic_setinputsizes: Optional[_GenericSetInputSizesType], - batch_size: int, - sort_by_parameter_order: bool, - schema_translate_map: Optional[SchemaTranslateMapType], - ) -> Iterator[_InsertManyValuesBatch]: - imv = self._insertmanyvalues - assert imv is not None - - if not imv.sentinel_param_keys: - _sentinel_from_params = None - else: - _sentinel_from_params = operator.itemgetter( - *imv.sentinel_param_keys - ) - - lenparams = len(parameters) - if imv.is_default_expr and not self.dialect.supports_default_metavalue: - # backend doesn't support - # INSERT INTO table (pk_col) VALUES (DEFAULT), (DEFAULT), ... - # at the moment this is basically SQL Server due to - # not being able to use DEFAULT for identity column - # just yield out that many single statements! still - # faster than a whole connection.execute() call ;) - # - # note we still are taking advantage of the fact that we know - # we are using RETURNING. The generalized approach of fetching - # cursor.lastrowid etc. still goes through the more heavyweight - # "ExecutionContext per statement" system as it isn't usable - # as a generic "RETURNING" approach - use_row_at_a_time = True - downgraded = False - elif not self.dialect.supports_multivalues_insert or ( - sort_by_parameter_order - and self._result_columns - and (imv.sentinel_columns is None or imv.includes_upsert_behaviors) - ): - # deterministic order was requested and the compiler could - # not organize sentinel columns for this dialect/statement. - # use row at a time - use_row_at_a_time = True - downgraded = True - else: - use_row_at_a_time = False - downgraded = False - - if use_row_at_a_time: - for batchnum, (param, compiled_param) in enumerate( - cast( - "Sequence[Tuple[_DBAPISingleExecuteParams, _MutableCoreSingleExecuteParams]]", # noqa: E501 - zip(parameters, compiled_parameters), - ), - 1, - ): - yield _InsertManyValuesBatch( - statement, - param, - generic_setinputsizes, - [param], - ( - [_sentinel_from_params(compiled_param)] - if _sentinel_from_params - else [] - ), - 1, - batchnum, - lenparams, - sort_by_parameter_order, - downgraded, - ) - return - - if schema_translate_map: - rst = functools.partial( - self.preparer._render_schema_translates, - schema_translate_map=schema_translate_map, - ) - else: - rst = None - - imv_single_values_expr = imv.single_values_expr - if rst: - imv_single_values_expr = rst(imv_single_values_expr) - - executemany_values = f"({imv_single_values_expr})" - statement = statement.replace(executemany_values, "__EXECMANY_TOKEN__") - - # Use optional insertmanyvalues_max_parameters - # to further shrink the batch size so that there are no more than - # insertmanyvalues_max_parameters params. - # Currently used by SQL Server, which limits statements to 2100 bound - # parameters (actually 2099). - max_params = self.dialect.insertmanyvalues_max_parameters - if max_params: - total_num_of_params = len(self.bind_names) - num_params_per_batch = len(imv.insert_crud_params) - num_params_outside_of_batch = ( - total_num_of_params - num_params_per_batch - ) - batch_size = min( - batch_size, - ( - (max_params - num_params_outside_of_batch) - // num_params_per_batch - ), - ) - - batches = cast("List[Sequence[Any]]", list(parameters)) - compiled_batches = cast( - "List[Sequence[Any]]", list(compiled_parameters) - ) - - processed_setinputsizes: Optional[_GenericSetInputSizesType] = None - batchnum = 1 - total_batches = lenparams // batch_size + ( - 1 if lenparams % batch_size else 0 - ) - - insert_crud_params = imv.insert_crud_params - assert insert_crud_params is not None - - if rst: - insert_crud_params = [ - (col, key, rst(expr), st) - for col, key, expr, st in insert_crud_params - ] - - escaped_bind_names: Mapping[str, str] - expand_pos_lower_index = expand_pos_upper_index = 0 - - if not self.positional: - if self.escaped_bind_names: - escaped_bind_names = self.escaped_bind_names - else: - escaped_bind_names = {} - - all_keys = set(parameters[0]) - - def apply_placeholders(keys, formatted): - for key in keys: - key = escaped_bind_names.get(key, key) - formatted = formatted.replace( - self.bindtemplate % {"name": key}, - self.bindtemplate - % {"name": f"{key}__EXECMANY_INDEX__"}, - ) - return formatted - - if imv.embed_values_counter: - imv_values_counter = ", _IMV_VALUES_COUNTER" - else: - imv_values_counter = "" - formatted_values_clause = f"""({', '.join( - apply_placeholders(bind_keys, formatted) - for _, _, formatted, bind_keys in insert_crud_params - )}{imv_values_counter})""" - - keys_to_replace = all_keys.intersection( - escaped_bind_names.get(key, key) - for _, _, _, bind_keys in insert_crud_params - for key in bind_keys - ) - base_parameters = { - key: parameters[0][key] - for key in all_keys.difference(keys_to_replace) - } - executemany_values_w_comma = "" - else: - formatted_values_clause = "" - keys_to_replace = set() - base_parameters = {} - - if imv.embed_values_counter: - executemany_values_w_comma = ( - f"({imv_single_values_expr}, _IMV_VALUES_COUNTER), " - ) - else: - executemany_values_w_comma = f"({imv_single_values_expr}), " - - all_names_we_will_expand: Set[str] = set() - for elem in imv.insert_crud_params: - all_names_we_will_expand.update(elem[3]) - - # get the start and end position in a particular list - # of parameters where we will be doing the "expanding". - # statements can have params on either side or both sides, - # given RETURNING and CTEs - if all_names_we_will_expand: - positiontup = self.positiontup - assert positiontup is not None - - all_expand_positions = { - idx - for idx, name in enumerate(positiontup) - if name in all_names_we_will_expand - } - expand_pos_lower_index = min(all_expand_positions) - expand_pos_upper_index = max(all_expand_positions) + 1 - assert ( - len(all_expand_positions) - == expand_pos_upper_index - expand_pos_lower_index - ) - - if self._numeric_binds: - escaped = re.escape(self._numeric_binds_identifier_char) - executemany_values_w_comma = re.sub( - rf"{escaped}\d+", "%s", executemany_values_w_comma - ) - - while batches: - batch = batches[0:batch_size] - compiled_batch = compiled_batches[0:batch_size] - - batches[0:batch_size] = [] - compiled_batches[0:batch_size] = [] - - if batches: - current_batch_size = batch_size - else: - current_batch_size = len(batch) - - if generic_setinputsizes: - # if setinputsizes is present, expand this collection to - # suit the batch length as well - # currently this will be mssql+pyodbc for internal dialects - processed_setinputsizes = [ - (new_key, len_, typ) - for new_key, len_, typ in ( - (f"{key}_{index}", len_, typ) - for index in range(current_batch_size) - for key, len_, typ in generic_setinputsizes - ) - ] - - replaced_parameters: Any - if self.positional: - num_ins_params = imv.num_positional_params_counted - - batch_iterator: Iterable[Sequence[Any]] - extra_params_left: Sequence[Any] - extra_params_right: Sequence[Any] - - if num_ins_params == len(batch[0]): - extra_params_left = extra_params_right = () - batch_iterator = batch - else: - extra_params_left = batch[0][:expand_pos_lower_index] - extra_params_right = batch[0][expand_pos_upper_index:] - batch_iterator = ( - b[expand_pos_lower_index:expand_pos_upper_index] - for b in batch - ) - - if imv.embed_values_counter: - expanded_values_string = ( - "".join( - executemany_values_w_comma.replace( - "_IMV_VALUES_COUNTER", str(i) - ) - for i, _ in enumerate(batch) - ) - )[:-2] - else: - expanded_values_string = ( - (executemany_values_w_comma * current_batch_size) - )[:-2] - - if self._numeric_binds and num_ins_params > 0: - # numeric will always number the parameters inside of - # VALUES (and thus order self.positiontup) to be higher - # than non-VALUES parameters, no matter where in the - # statement those non-VALUES parameters appear (this is - # ensured in _process_numeric by numbering first all - # params that are not in _values_bindparam) - # therefore all extra params are always - # on the left side and numbered lower than the VALUES - # parameters - assert not extra_params_right - - start = expand_pos_lower_index + 1 - end = num_ins_params * (current_batch_size) + start - - # need to format here, since statement may contain - # unescaped %, while values_string contains just (%s, %s) - positions = tuple( - f"{self._numeric_binds_identifier_char}{i}" - for i in range(start, end) - ) - expanded_values_string = expanded_values_string % positions - - replaced_statement = statement.replace( - "__EXECMANY_TOKEN__", expanded_values_string - ) - - replaced_parameters = tuple( - itertools.chain.from_iterable(batch_iterator) - ) - - replaced_parameters = ( - extra_params_left - + replaced_parameters - + extra_params_right - ) - - else: - replaced_values_clauses = [] - replaced_parameters = base_parameters.copy() - - for i, param in enumerate(batch): - fmv = formatted_values_clause.replace( - "EXECMANY_INDEX__", str(i) - ) - if imv.embed_values_counter: - fmv = fmv.replace("_IMV_VALUES_COUNTER", str(i)) - - replaced_values_clauses.append(fmv) - replaced_parameters.update( - {f"{key}__{i}": param[key] for key in keys_to_replace} - ) - - replaced_statement = statement.replace( - "__EXECMANY_TOKEN__", - ", ".join(replaced_values_clauses), - ) - - yield _InsertManyValuesBatch( - replaced_statement, - replaced_parameters, - processed_setinputsizes, - batch, - ( - [_sentinel_from_params(cb) for cb in compiled_batch] - if _sentinel_from_params - else [] - ), - current_batch_size, - batchnum, - total_batches, - sort_by_parameter_order, - False, - ) - batchnum += 1 - - def visit_insert( - self, insert_stmt, visited_bindparam=None, visiting_cte=None, **kw - ): - compile_state = insert_stmt._compile_state_factory( - insert_stmt, self, **kw - ) - insert_stmt = compile_state.statement - - if visiting_cte is not None: - kw["visiting_cte"] = visiting_cte - toplevel = False - else: - toplevel = not self.stack - - if toplevel: - self.isinsert = True - if not self.dml_compile_state: - self.dml_compile_state = compile_state - if not self.compile_state: - self.compile_state = compile_state - - self.stack.append( - { - "correlate_froms": set(), - "asfrom_froms": set(), - "selectable": insert_stmt, - } - ) - - counted_bindparam = 0 - - # reset any incoming "visited_bindparam" collection - visited_bindparam = None - - # for positional, insertmanyvalues needs to know how many - # bound parameters are in the VALUES sequence; there's no simple - # rule because default expressions etc. can have zero or more - # params inside them. After multiple attempts to figure this out, - # this very simplistic "count after" works and is - # likely the least amount of callcounts, though looks clumsy - if self.positional and visiting_cte is None: - # if we are inside a CTE, don't count parameters - # here since they wont be for insertmanyvalues. keep - # visited_bindparam at None so no counting happens. - # see #9173 - visited_bindparam = [] - - crud_params_struct = crud._get_crud_params( - self, - insert_stmt, - compile_state, - toplevel, - visited_bindparam=visited_bindparam, - **kw, - ) - - if self.positional and visited_bindparam is not None: - counted_bindparam = len(visited_bindparam) - if self._numeric_binds: - if self._values_bindparam is not None: - self._values_bindparam += visited_bindparam - else: - self._values_bindparam = visited_bindparam - - crud_params_single = crud_params_struct.single_params - - if ( - not crud_params_single - and not self.dialect.supports_default_values - and not self.dialect.supports_default_metavalue - and not self.dialect.supports_empty_insert - ): - raise exc.CompileError( - "The '%s' dialect with current database " - "version settings does not support empty " - "inserts." % self.dialect.name - ) - - if compile_state._has_multi_parameters: - if not self.dialect.supports_multivalues_insert: - raise exc.CompileError( - "The '%s' dialect with current database " - "version settings does not support " - "in-place multirow inserts." % self.dialect.name - ) - elif ( - self.implicit_returning or insert_stmt._returning - ) and insert_stmt._sort_by_parameter_order: - raise exc.CompileError( - "RETURNING cannot be determinstically sorted when " - "using an INSERT which includes multi-row values()." - ) - crud_params_single = crud_params_struct.single_params - else: - crud_params_single = crud_params_struct.single_params - - preparer = self.preparer - supports_default_values = self.dialect.supports_default_values - - text = "INSERT " - - if insert_stmt._prefixes: - text += self._generate_prefixes( - insert_stmt, insert_stmt._prefixes, **kw - ) - - text += "INTO " - table_text = preparer.format_table(insert_stmt.table) - - if insert_stmt._hints: - _, table_text = self._setup_crud_hints(insert_stmt, table_text) - - if insert_stmt._independent_ctes: - self._dispatch_independent_ctes(insert_stmt, kw) - - text += table_text - - if crud_params_single or not supports_default_values: - text += " (%s)" % ", ".join( - [expr for _, expr, _, _ in crud_params_single] - ) - - # look for insertmanyvalues attributes that would have been configured - # by crud.py as it scanned through the columns to be part of the - # INSERT - use_insertmanyvalues = crud_params_struct.use_insertmanyvalues - named_sentinel_params: Optional[Sequence[str]] = None - add_sentinel_cols = None - implicit_sentinel = False - - returning_cols = self.implicit_returning or insert_stmt._returning - if returning_cols: - add_sentinel_cols = crud_params_struct.use_sentinel_columns - if add_sentinel_cols is not None: - assert use_insertmanyvalues - - # search for the sentinel column explicitly present - # in the INSERT columns list, and additionally check that - # this column has a bound parameter name set up that's in the - # parameter list. If both of these cases are present, it means - # we will have a client side value for the sentinel in each - # parameter set. - - _params_by_col = { - col: param_names - for col, _, _, param_names in crud_params_single - } - named_sentinel_params = [] - for _add_sentinel_col in add_sentinel_cols: - if _add_sentinel_col not in _params_by_col: - named_sentinel_params = None - break - param_name = self._within_exec_param_key_getter( - _add_sentinel_col - ) - if param_name not in _params_by_col[_add_sentinel_col]: - named_sentinel_params = None - break - named_sentinel_params.append(param_name) - - if named_sentinel_params is None: - # if we are not going to have a client side value for - # the sentinel in the parameter set, that means it's - # an autoincrement, an IDENTITY, or a server-side SQL - # expression like nextval('seqname'). So this is - # an "implicit" sentinel; we will look for it in - # RETURNING - # only, and then sort on it. For this case on PG, - # SQL Server we have to use a special INSERT form - # that guarantees the server side function lines up with - # the entries in the VALUES. - if ( - self.dialect.insertmanyvalues_implicit_sentinel - & InsertmanyvaluesSentinelOpts.ANY_AUTOINCREMENT - ): - implicit_sentinel = True - else: - # here, we are not using a sentinel at all - # and we are likely the SQLite dialect. - # The first add_sentinel_col that we have should not - # be marked as "insert_sentinel=True". if it was, - # an error should have been raised in - # _get_sentinel_column_for_table. - assert not add_sentinel_cols[0]._insert_sentinel, ( - "sentinel selection rules should have prevented " - "us from getting here for this dialect" - ) - - # always put the sentinel columns last. even if they are - # in the returning list already, they will be there twice - # then. - returning_cols = list(returning_cols) + list(add_sentinel_cols) - - returning_clause = self.returning_clause( - insert_stmt, - returning_cols, - populate_result_map=toplevel, - ) - - if self.returning_precedes_values: - text += " " + returning_clause - - else: - returning_clause = None - - if insert_stmt.select is not None: - # placed here by crud.py - select_text = self.process( - self.stack[-1]["insert_from_select"], insert_into=True, **kw - ) - - if self.ctes and self.dialect.cte_follows_insert: - nesting_level = len(self.stack) if not toplevel else None - text += " %s%s" % ( - self._render_cte_clause( - nesting_level=nesting_level, - include_following_stack=True, - ), - select_text, - ) - else: - text += " %s" % select_text - elif not crud_params_single and supports_default_values: - text += " DEFAULT VALUES" - if use_insertmanyvalues: - self._insertmanyvalues = _InsertManyValues( - True, - self.dialect.default_metavalue_token, - cast( - "List[crud._CrudParamElementStr]", crud_params_single - ), - counted_bindparam, - sort_by_parameter_order=( - insert_stmt._sort_by_parameter_order - ), - includes_upsert_behaviors=( - insert_stmt._post_values_clause is not None - ), - sentinel_columns=add_sentinel_cols, - num_sentinel_columns=( - len(add_sentinel_cols) if add_sentinel_cols else 0 - ), - implicit_sentinel=implicit_sentinel, - ) - elif compile_state._has_multi_parameters: - text += " VALUES %s" % ( - ", ".join( - "(%s)" - % (", ".join(value for _, _, value, _ in crud_param_set)) - for crud_param_set in crud_params_struct.all_multi_params - ), - ) - else: - insert_single_values_expr = ", ".join( - [ - value - for _, _, value, _ in cast( - "List[crud._CrudParamElementStr]", - crud_params_single, - ) - ] - ) - - if use_insertmanyvalues: - if ( - implicit_sentinel - and ( - self.dialect.insertmanyvalues_implicit_sentinel - & InsertmanyvaluesSentinelOpts.USE_INSERT_FROM_SELECT - ) - # this is checking if we have - # INSERT INTO table (id) VALUES (DEFAULT). - and not (crud_params_struct.is_default_metavalue_only) - ): - # if we have a sentinel column that is server generated, - # then for selected backends render the VALUES list as a - # subquery. This is the orderable form supported by - # PostgreSQL and SQL Server. - embed_sentinel_value = True - - render_bind_casts = ( - self.dialect.insertmanyvalues_implicit_sentinel - & InsertmanyvaluesSentinelOpts.RENDER_SELECT_COL_CASTS - ) - - colnames = ", ".join( - f"p{i}" for i, _ in enumerate(crud_params_single) - ) - - if render_bind_casts: - # render casts for the SELECT list. For PG, we are - # already rendering bind casts in the parameter list, - # selectively for the more "tricky" types like ARRAY. - # however, even for the "easy" types, if the parameter - # is NULL for every entry, PG gives up and says - # "it must be TEXT", which fails for other easy types - # like ints. So we cast on this side too. - colnames_w_cast = ", ".join( - self.render_bind_cast( - col.type, - col.type._unwrapped_dialect_impl(self.dialect), - f"p{i}", - ) - for i, (col, *_) in enumerate(crud_params_single) - ) - else: - colnames_w_cast = colnames - - text += ( - f" SELECT {colnames_w_cast} FROM " - f"(VALUES ({insert_single_values_expr})) " - f"AS imp_sen({colnames}, sen_counter) " - "ORDER BY sen_counter" - ) - else: - # otherwise, if no sentinel or backend doesn't support - # orderable subquery form, use a plain VALUES list - embed_sentinel_value = False - text += f" VALUES ({insert_single_values_expr})" - - self._insertmanyvalues = _InsertManyValues( - is_default_expr=False, - single_values_expr=insert_single_values_expr, - insert_crud_params=cast( - "List[crud._CrudParamElementStr]", - crud_params_single, - ), - num_positional_params_counted=counted_bindparam, - sort_by_parameter_order=( - insert_stmt._sort_by_parameter_order - ), - includes_upsert_behaviors=( - insert_stmt._post_values_clause is not None - ), - sentinel_columns=add_sentinel_cols, - num_sentinel_columns=( - len(add_sentinel_cols) if add_sentinel_cols else 0 - ), - sentinel_param_keys=named_sentinel_params, - implicit_sentinel=implicit_sentinel, - embed_values_counter=embed_sentinel_value, - ) - - else: - text += f" VALUES ({insert_single_values_expr})" - - if insert_stmt._post_values_clause is not None: - post_values_clause = self.process( - insert_stmt._post_values_clause, **kw - ) - if post_values_clause: - text += " " + post_values_clause - - if returning_clause and not self.returning_precedes_values: - text += " " + returning_clause - - if self.ctes and not self.dialect.cte_follows_insert: - nesting_level = len(self.stack) if not toplevel else None - text = ( - self._render_cte_clause( - nesting_level=nesting_level, - include_following_stack=True, - ) - + text - ) - - self.stack.pop(-1) - - return text - - def update_limit_clause(self, update_stmt): - """Provide a hook for MySQL to add LIMIT to the UPDATE""" - return None - - def update_tables_clause(self, update_stmt, from_table, extra_froms, **kw): - """Provide a hook to override the initial table clause - in an UPDATE statement. - - MySQL overrides this. - - """ - kw["asfrom"] = True - return from_table._compiler_dispatch(self, iscrud=True, **kw) - - def update_from_clause( - self, update_stmt, from_table, extra_froms, from_hints, **kw - ): - """Provide a hook to override the generation of an - UPDATE..FROM clause. - - MySQL and MSSQL override this. - - """ - raise NotImplementedError( - "This backend does not support multiple-table " - "criteria within UPDATE" - ) - - def visit_update(self, update_stmt, visiting_cte=None, **kw): - compile_state = update_stmt._compile_state_factory( - update_stmt, self, **kw - ) - update_stmt = compile_state.statement - - if visiting_cte is not None: - kw["visiting_cte"] = visiting_cte - toplevel = False - else: - toplevel = not self.stack - - if toplevel: - self.isupdate = True - if not self.dml_compile_state: - self.dml_compile_state = compile_state - if not self.compile_state: - self.compile_state = compile_state - - if self.linting & COLLECT_CARTESIAN_PRODUCTS: - from_linter = FromLinter({}, set()) - warn_linting = self.linting & WARN_LINTING - if toplevel: - self.from_linter = from_linter - else: - from_linter = None - warn_linting = False - - extra_froms = compile_state._extra_froms - is_multitable = bool(extra_froms) - - if is_multitable: - # main table might be a JOIN - main_froms = set(_from_objects(update_stmt.table)) - render_extra_froms = [ - f for f in extra_froms if f not in main_froms - ] - correlate_froms = main_froms.union(extra_froms) - else: - render_extra_froms = [] - correlate_froms = {update_stmt.table} - - self.stack.append( - { - "correlate_froms": correlate_froms, - "asfrom_froms": correlate_froms, - "selectable": update_stmt, - } - ) - - text = "UPDATE " - - if update_stmt._prefixes: - text += self._generate_prefixes( - update_stmt, update_stmt._prefixes, **kw - ) - - table_text = self.update_tables_clause( - update_stmt, - update_stmt.table, - render_extra_froms, - from_linter=from_linter, - **kw, - ) - crud_params_struct = crud._get_crud_params( - self, update_stmt, compile_state, toplevel, **kw - ) - crud_params = crud_params_struct.single_params - - if update_stmt._hints: - dialect_hints, table_text = self._setup_crud_hints( - update_stmt, table_text - ) - else: - dialect_hints = None - - if update_stmt._independent_ctes: - self._dispatch_independent_ctes(update_stmt, kw) - - text += table_text - - text += " SET " - text += ", ".join( - expr + "=" + value - for _, expr, value, _ in cast( - "List[Tuple[Any, str, str, Any]]", crud_params - ) - ) - - if self.implicit_returning or update_stmt._returning: - if self.returning_precedes_values: - text += " " + self.returning_clause( - update_stmt, - self.implicit_returning or update_stmt._returning, - populate_result_map=toplevel, - ) - - if extra_froms: - extra_from_text = self.update_from_clause( - update_stmt, - update_stmt.table, - render_extra_froms, - dialect_hints, - from_linter=from_linter, - **kw, - ) - if extra_from_text: - text += " " + extra_from_text - - if update_stmt._where_criteria: - t = self._generate_delimited_and_list( - update_stmt._where_criteria, from_linter=from_linter, **kw - ) - if t: - text += " WHERE " + t - - limit_clause = self.update_limit_clause(update_stmt) - if limit_clause: - text += " " + limit_clause - - if ( - self.implicit_returning or update_stmt._returning - ) and not self.returning_precedes_values: - text += " " + self.returning_clause( - update_stmt, - self.implicit_returning or update_stmt._returning, - populate_result_map=toplevel, - ) - - if self.ctes: - nesting_level = len(self.stack) if not toplevel else None - text = self._render_cte_clause(nesting_level=nesting_level) + text - - if warn_linting: - assert from_linter is not None - from_linter.warn(stmt_type="UPDATE") - - self.stack.pop(-1) - - return text - - def delete_extra_from_clause( - self, update_stmt, from_table, extra_froms, from_hints, **kw - ): - """Provide a hook to override the generation of an - DELETE..FROM clause. - - This can be used to implement DELETE..USING for example. - - MySQL and MSSQL override this. - - """ - raise NotImplementedError( - "This backend does not support multiple-table " - "criteria within DELETE" - ) - - def delete_table_clause(self, delete_stmt, from_table, extra_froms, **kw): - return from_table._compiler_dispatch( - self, asfrom=True, iscrud=True, **kw - ) - - def visit_delete(self, delete_stmt, visiting_cte=None, **kw): - compile_state = delete_stmt._compile_state_factory( - delete_stmt, self, **kw - ) - delete_stmt = compile_state.statement - - if visiting_cte is not None: - kw["visiting_cte"] = visiting_cte - toplevel = False - else: - toplevel = not self.stack - - if toplevel: - self.isdelete = True - if not self.dml_compile_state: - self.dml_compile_state = compile_state - if not self.compile_state: - self.compile_state = compile_state - - if self.linting & COLLECT_CARTESIAN_PRODUCTS: - from_linter = FromLinter({}, set()) - warn_linting = self.linting & WARN_LINTING - if toplevel: - self.from_linter = from_linter - else: - from_linter = None - warn_linting = False - - extra_froms = compile_state._extra_froms - - correlate_froms = {delete_stmt.table}.union(extra_froms) - self.stack.append( - { - "correlate_froms": correlate_froms, - "asfrom_froms": correlate_froms, - "selectable": delete_stmt, - } - ) - - text = "DELETE " - - if delete_stmt._prefixes: - text += self._generate_prefixes( - delete_stmt, delete_stmt._prefixes, **kw - ) - - text += "FROM " - - try: - table_text = self.delete_table_clause( - delete_stmt, - delete_stmt.table, - extra_froms, - from_linter=from_linter, - ) - except TypeError: - # anticipate 3rd party dialects that don't include **kw - # TODO: remove in 2.1 - table_text = self.delete_table_clause( - delete_stmt, delete_stmt.table, extra_froms - ) - if from_linter: - _ = self.process(delete_stmt.table, from_linter=from_linter) - - crud._get_crud_params(self, delete_stmt, compile_state, toplevel, **kw) - - if delete_stmt._hints: - dialect_hints, table_text = self._setup_crud_hints( - delete_stmt, table_text - ) - else: - dialect_hints = None - - if delete_stmt._independent_ctes: - self._dispatch_independent_ctes(delete_stmt, kw) - - text += table_text - - if ( - self.implicit_returning or delete_stmt._returning - ) and self.returning_precedes_values: - text += " " + self.returning_clause( - delete_stmt, - self.implicit_returning or delete_stmt._returning, - populate_result_map=toplevel, - ) - - if extra_froms: - extra_from_text = self.delete_extra_from_clause( - delete_stmt, - delete_stmt.table, - extra_froms, - dialect_hints, - from_linter=from_linter, - **kw, - ) - if extra_from_text: - text += " " + extra_from_text - - if delete_stmt._where_criteria: - t = self._generate_delimited_and_list( - delete_stmt._where_criteria, from_linter=from_linter, **kw - ) - if t: - text += " WHERE " + t - - if ( - self.implicit_returning or delete_stmt._returning - ) and not self.returning_precedes_values: - text += " " + self.returning_clause( - delete_stmt, - self.implicit_returning or delete_stmt._returning, - populate_result_map=toplevel, - ) - - if self.ctes: - nesting_level = len(self.stack) if not toplevel else None - text = self._render_cte_clause(nesting_level=nesting_level) + text - - if warn_linting: - assert from_linter is not None - from_linter.warn(stmt_type="DELETE") - - self.stack.pop(-1) - - return text - - def visit_savepoint(self, savepoint_stmt, **kw): - return "SAVEPOINT %s" % self.preparer.format_savepoint(savepoint_stmt) - - def visit_rollback_to_savepoint(self, savepoint_stmt, **kw): - return "ROLLBACK TO SAVEPOINT %s" % self.preparer.format_savepoint( - savepoint_stmt - ) - - def visit_release_savepoint(self, savepoint_stmt, **kw): - return "RELEASE SAVEPOINT %s" % self.preparer.format_savepoint( - savepoint_stmt - ) - - -class StrSQLCompiler(SQLCompiler): - """A :class:`.SQLCompiler` subclass which allows a small selection - of non-standard SQL features to render into a string value. - - The :class:`.StrSQLCompiler` is invoked whenever a Core expression - element is directly stringified without calling upon the - :meth:`_expression.ClauseElement.compile` method. - It can render a limited set - of non-standard SQL constructs to assist in basic stringification, - however for more substantial custom or dialect-specific SQL constructs, - it will be necessary to make use of - :meth:`_expression.ClauseElement.compile` - directly. - - .. seealso:: - - :ref:`faq_sql_expression_string` - - """ - - def _fallback_column_name(self, column): - return "<name unknown>" - - @util.preload_module("sqlalchemy.engine.url") - def visit_unsupported_compilation(self, element, err, **kw): - if element.stringify_dialect != "default": - url = util.preloaded.engine_url - dialect = url.URL.create(element.stringify_dialect).get_dialect()() - - compiler = dialect.statement_compiler( - dialect, None, _supporting_against=self - ) - if not isinstance(compiler, StrSQLCompiler): - return compiler.process(element, **kw) - - return super().visit_unsupported_compilation(element, err) - - def visit_getitem_binary(self, binary, operator, **kw): - return "%s[%s]" % ( - self.process(binary.left, **kw), - self.process(binary.right, **kw), - ) - - def visit_json_getitem_op_binary(self, binary, operator, **kw): - return self.visit_getitem_binary(binary, operator, **kw) - - def visit_json_path_getitem_op_binary(self, binary, operator, **kw): - return self.visit_getitem_binary(binary, operator, **kw) - - def visit_sequence(self, seq, **kw): - return "<next sequence value: %s>" % self.preparer.format_sequence(seq) - - def returning_clause( - self, - stmt: UpdateBase, - returning_cols: Sequence[ColumnElement[Any]], - *, - populate_result_map: bool, - **kw: Any, - ) -> str: - columns = [ - self._label_select_column(None, c, True, False, {}) - for c in base._select_iterables(returning_cols) - ] - return "RETURNING " + ", ".join(columns) - - def update_from_clause( - self, update_stmt, from_table, extra_froms, from_hints, **kw - ): - kw["asfrom"] = True - return "FROM " + ", ".join( - t._compiler_dispatch(self, fromhints=from_hints, **kw) - for t in extra_froms - ) - - def delete_extra_from_clause( - self, update_stmt, from_table, extra_froms, from_hints, **kw - ): - kw["asfrom"] = True - return ", " + ", ".join( - t._compiler_dispatch(self, fromhints=from_hints, **kw) - for t in extra_froms - ) - - def visit_empty_set_expr(self, type_, **kw): - return "SELECT 1 WHERE 1!=1" - - def get_from_hint_text(self, table, text): - return "[%s]" % text - - def visit_regexp_match_op_binary(self, binary, operator, **kw): - return self._generate_generic_binary(binary, " <regexp> ", **kw) - - def visit_not_regexp_match_op_binary(self, binary, operator, **kw): - return self._generate_generic_binary(binary, " <not regexp> ", **kw) - - def visit_regexp_replace_op_binary(self, binary, operator, **kw): - return "<regexp replace>(%s, %s)" % ( - binary.left._compiler_dispatch(self, **kw), - binary.right._compiler_dispatch(self, **kw), - ) - - def visit_try_cast(self, cast, **kwargs): - return "TRY_CAST(%s AS %s)" % ( - cast.clause._compiler_dispatch(self, **kwargs), - cast.typeclause._compiler_dispatch(self, **kwargs), - ) - - -class DDLCompiler(Compiled): - is_ddl = True - - if TYPE_CHECKING: - - def __init__( - self, - dialect: Dialect, - statement: ExecutableDDLElement, - schema_translate_map: Optional[SchemaTranslateMapType] = ..., - render_schema_translate: bool = ..., - compile_kwargs: Mapping[str, Any] = ..., - ): ... - - @util.memoized_property - def sql_compiler(self): - return self.dialect.statement_compiler( - self.dialect, None, schema_translate_map=self.schema_translate_map - ) - - @util.memoized_property - def type_compiler(self): - return self.dialect.type_compiler_instance - - def construct_params( - self, - params: Optional[_CoreSingleExecuteParams] = None, - extracted_parameters: Optional[Sequence[BindParameter[Any]]] = None, - escape_names: bool = True, - ) -> Optional[_MutableCoreSingleExecuteParams]: - return None - - def visit_ddl(self, ddl, **kwargs): - # table events can substitute table and schema name - context = ddl.context - if isinstance(ddl.target, schema.Table): - context = context.copy() - - preparer = self.preparer - path = preparer.format_table_seq(ddl.target) - if len(path) == 1: - table, sch = path[0], "" - else: - table, sch = path[-1], path[0] - - context.setdefault("table", table) - context.setdefault("schema", sch) - context.setdefault("fullname", preparer.format_table(ddl.target)) - - return self.sql_compiler.post_process_text(ddl.statement % context) - - def visit_create_schema(self, create, **kw): - text = "CREATE SCHEMA " - if create.if_not_exists: - text += "IF NOT EXISTS " - return text + self.preparer.format_schema(create.element) - - def visit_drop_schema(self, drop, **kw): - text = "DROP SCHEMA " - if drop.if_exists: - text += "IF EXISTS " - text += self.preparer.format_schema(drop.element) - if drop.cascade: - text += " CASCADE" - return text - - def visit_create_table(self, create, **kw): - table = create.element - preparer = self.preparer - - text = "\nCREATE " - if table._prefixes: - text += " ".join(table._prefixes) + " " - - text += "TABLE " - if create.if_not_exists: - text += "IF NOT EXISTS " - - text += preparer.format_table(table) + " " - - create_table_suffix = self.create_table_suffix(table) - if create_table_suffix: - text += create_table_suffix + " " - - text += "(" - - separator = "\n" - - # if only one primary key, specify it along with the column - first_pk = False - for create_column in create.columns: - column = create_column.element - try: - processed = self.process( - create_column, first_pk=column.primary_key and not first_pk - ) - if processed is not None: - text += separator - separator = ", \n" - text += "\t" + processed - if column.primary_key: - first_pk = True - except exc.CompileError as ce: - raise exc.CompileError( - "(in table '%s', column '%s'): %s" - % (table.description, column.name, ce.args[0]) - ) from ce - - const = self.create_table_constraints( - table, - _include_foreign_key_constraints=create.include_foreign_key_constraints, # noqa - ) - if const: - text += separator + "\t" + const - - text += "\n)%s\n\n" % self.post_create_table(table) - return text - - def visit_create_column(self, create, first_pk=False, **kw): - column = create.element - - if column.system: - return None - - text = self.get_column_specification(column, first_pk=first_pk) - const = " ".join( - self.process(constraint) for constraint in column.constraints - ) - if const: - text += " " + const - - return text - - def create_table_constraints( - self, table, _include_foreign_key_constraints=None, **kw - ): - # On some DB order is significant: visit PK first, then the - # other constraints (engine.ReflectionTest.testbasic failed on FB2) - constraints = [] - if table.primary_key: - constraints.append(table.primary_key) - - all_fkcs = table.foreign_key_constraints - if _include_foreign_key_constraints is not None: - omit_fkcs = all_fkcs.difference(_include_foreign_key_constraints) - else: - omit_fkcs = set() - - constraints.extend( - [ - c - for c in table._sorted_constraints - if c is not table.primary_key and c not in omit_fkcs - ] - ) - - return ", \n\t".join( - p - for p in ( - self.process(constraint) - for constraint in constraints - if (constraint._should_create_for_compiler(self)) - and ( - not self.dialect.supports_alter - or not getattr(constraint, "use_alter", False) - ) - ) - if p is not None - ) - - def visit_drop_table(self, drop, **kw): - text = "\nDROP TABLE " - if drop.if_exists: - text += "IF EXISTS " - return text + self.preparer.format_table(drop.element) - - def visit_drop_view(self, drop, **kw): - return "\nDROP VIEW " + self.preparer.format_table(drop.element) - - def _verify_index_table(self, index): - if index.table is None: - raise exc.CompileError( - "Index '%s' is not associated with any table." % index.name - ) - - def visit_create_index( - self, create, include_schema=False, include_table_schema=True, **kw - ): - index = create.element - self._verify_index_table(index) - preparer = self.preparer - text = "CREATE " - if index.unique: - text += "UNIQUE " - if index.name is None: - raise exc.CompileError( - "CREATE INDEX requires that the index have a name" - ) - - text += "INDEX " - if create.if_not_exists: - text += "IF NOT EXISTS " - - text += "%s ON %s (%s)" % ( - self._prepared_index_name(index, include_schema=include_schema), - preparer.format_table( - index.table, use_schema=include_table_schema - ), - ", ".join( - self.sql_compiler.process( - expr, include_table=False, literal_binds=True - ) - for expr in index.expressions - ), - ) - return text - - def visit_drop_index(self, drop, **kw): - index = drop.element - - if index.name is None: - raise exc.CompileError( - "DROP INDEX requires that the index have a name" - ) - text = "\nDROP INDEX " - if drop.if_exists: - text += "IF EXISTS " - - return text + self._prepared_index_name(index, include_schema=True) - - def _prepared_index_name(self, index, include_schema=False): - if index.table is not None: - effective_schema = self.preparer.schema_for_object(index.table) - else: - effective_schema = None - if include_schema and effective_schema: - schema_name = self.preparer.quote_schema(effective_schema) - else: - schema_name = None - - index_name = self.preparer.format_index(index) - - if schema_name: - index_name = schema_name + "." + index_name - return index_name - - def visit_add_constraint(self, create, **kw): - return "ALTER TABLE %s ADD %s" % ( - self.preparer.format_table(create.element.table), - self.process(create.element), - ) - - def visit_set_table_comment(self, create, **kw): - return "COMMENT ON TABLE %s IS %s" % ( - self.preparer.format_table(create.element), - self.sql_compiler.render_literal_value( - create.element.comment, sqltypes.String() - ), - ) - - def visit_drop_table_comment(self, drop, **kw): - return "COMMENT ON TABLE %s IS NULL" % self.preparer.format_table( - drop.element - ) - - def visit_set_column_comment(self, create, **kw): - return "COMMENT ON COLUMN %s IS %s" % ( - self.preparer.format_column( - create.element, use_table=True, use_schema=True - ), - self.sql_compiler.render_literal_value( - create.element.comment, sqltypes.String() - ), - ) - - def visit_drop_column_comment(self, drop, **kw): - return "COMMENT ON COLUMN %s IS NULL" % self.preparer.format_column( - drop.element, use_table=True - ) - - def visit_set_constraint_comment(self, create, **kw): - raise exc.UnsupportedCompilationError(self, type(create)) - - def visit_drop_constraint_comment(self, drop, **kw): - raise exc.UnsupportedCompilationError(self, type(drop)) - - def get_identity_options(self, identity_options): - text = [] - if identity_options.increment is not None: - text.append("INCREMENT BY %d" % identity_options.increment) - if identity_options.start is not None: - text.append("START WITH %d" % identity_options.start) - if identity_options.minvalue is not None: - text.append("MINVALUE %d" % identity_options.minvalue) - if identity_options.maxvalue is not None: - text.append("MAXVALUE %d" % identity_options.maxvalue) - if identity_options.nominvalue is not None: - text.append("NO MINVALUE") - if identity_options.nomaxvalue is not None: - text.append("NO MAXVALUE") - if identity_options.cache is not None: - text.append("CACHE %d" % identity_options.cache) - if identity_options.cycle is not None: - text.append("CYCLE" if identity_options.cycle else "NO CYCLE") - return " ".join(text) - - def visit_create_sequence(self, create, prefix=None, **kw): - text = "CREATE SEQUENCE " - if create.if_not_exists: - text += "IF NOT EXISTS " - text += self.preparer.format_sequence(create.element) - - if prefix: - text += prefix - options = self.get_identity_options(create.element) - if options: - text += " " + options - return text - - def visit_drop_sequence(self, drop, **kw): - text = "DROP SEQUENCE " - if drop.if_exists: - text += "IF EXISTS " - return text + self.preparer.format_sequence(drop.element) - - def visit_drop_constraint(self, drop, **kw): - constraint = drop.element - if constraint.name is not None: - formatted_name = self.preparer.format_constraint(constraint) - else: - formatted_name = None - - if formatted_name is None: - raise exc.CompileError( - "Can't emit DROP CONSTRAINT for constraint %r; " - "it has no name" % drop.element - ) - return "ALTER TABLE %s DROP CONSTRAINT %s%s%s" % ( - self.preparer.format_table(drop.element.table), - "IF EXISTS " if drop.if_exists else "", - formatted_name, - " CASCADE" if drop.cascade else "", - ) - - def get_column_specification(self, column, **kwargs): - colspec = ( - self.preparer.format_column(column) - + " " - + self.dialect.type_compiler_instance.process( - column.type, type_expression=column - ) - ) - default = self.get_column_default_string(column) - if default is not None: - colspec += " DEFAULT " + default - - if column.computed is not None: - colspec += " " + self.process(column.computed) - - if ( - column.identity is not None - and self.dialect.supports_identity_columns - ): - colspec += " " + self.process(column.identity) - - if not column.nullable and ( - not column.identity or not self.dialect.supports_identity_columns - ): - colspec += " NOT NULL" - return colspec - - def create_table_suffix(self, table): - return "" - - def post_create_table(self, table): - return "" - - def get_column_default_string(self, column): - if isinstance(column.server_default, schema.DefaultClause): - return self.render_default_string(column.server_default.arg) - else: - return None - - def render_default_string(self, default): - if isinstance(default, str): - return self.sql_compiler.render_literal_value( - default, sqltypes.STRINGTYPE - ) - else: - return self.sql_compiler.process(default, literal_binds=True) - - def visit_table_or_column_check_constraint(self, constraint, **kw): - if constraint.is_column_level: - return self.visit_column_check_constraint(constraint) - else: - return self.visit_check_constraint(constraint) - - def visit_check_constraint(self, constraint, **kw): - text = "" - if constraint.name is not None: - formatted_name = self.preparer.format_constraint(constraint) - if formatted_name is not None: - text += "CONSTRAINT %s " % formatted_name - text += "CHECK (%s)" % self.sql_compiler.process( - constraint.sqltext, include_table=False, literal_binds=True - ) - text += self.define_constraint_deferrability(constraint) - return text - - def visit_column_check_constraint(self, constraint, **kw): - text = "" - if constraint.name is not None: - formatted_name = self.preparer.format_constraint(constraint) - if formatted_name is not None: - text += "CONSTRAINT %s " % formatted_name - text += "CHECK (%s)" % self.sql_compiler.process( - constraint.sqltext, include_table=False, literal_binds=True - ) - text += self.define_constraint_deferrability(constraint) - return text - - def visit_primary_key_constraint(self, constraint, **kw): - if len(constraint) == 0: - return "" - text = "" - if constraint.name is not None: - formatted_name = self.preparer.format_constraint(constraint) - if formatted_name is not None: - text += "CONSTRAINT %s " % formatted_name - text += "PRIMARY KEY " - text += "(%s)" % ", ".join( - self.preparer.quote(c.name) - for c in ( - constraint.columns_autoinc_first - if constraint._implicit_generated - else constraint.columns - ) - ) - text += self.define_constraint_deferrability(constraint) - return text - - def visit_foreign_key_constraint(self, constraint, **kw): - preparer = self.preparer - text = "" - if constraint.name is not None: - formatted_name = self.preparer.format_constraint(constraint) - if formatted_name is not None: - text += "CONSTRAINT %s " % formatted_name - remote_table = list(constraint.elements)[0].column.table - text += "FOREIGN KEY(%s) REFERENCES %s (%s)" % ( - ", ".join( - preparer.quote(f.parent.name) for f in constraint.elements - ), - self.define_constraint_remote_table( - constraint, remote_table, preparer - ), - ", ".join( - preparer.quote(f.column.name) for f in constraint.elements - ), - ) - text += self.define_constraint_match(constraint) - text += self.define_constraint_cascades(constraint) - text += self.define_constraint_deferrability(constraint) - return text - - def define_constraint_remote_table(self, constraint, table, preparer): - """Format the remote table clause of a CREATE CONSTRAINT clause.""" - - return preparer.format_table(table) - - def visit_unique_constraint(self, constraint, **kw): - if len(constraint) == 0: - return "" - text = "" - if constraint.name is not None: - formatted_name = self.preparer.format_constraint(constraint) - if formatted_name is not None: - text += "CONSTRAINT %s " % formatted_name - text += "UNIQUE %s(%s)" % ( - self.define_unique_constraint_distinct(constraint, **kw), - ", ".join(self.preparer.quote(c.name) for c in constraint), - ) - text += self.define_constraint_deferrability(constraint) - return text - - def define_unique_constraint_distinct(self, constraint, **kw): - return "" - - def define_constraint_cascades(self, constraint): - text = "" - if constraint.ondelete is not None: - text += " ON DELETE %s" % self.preparer.validate_sql_phrase( - constraint.ondelete, FK_ON_DELETE - ) - if constraint.onupdate is not None: - text += " ON UPDATE %s" % self.preparer.validate_sql_phrase( - constraint.onupdate, FK_ON_UPDATE - ) - return text - - def define_constraint_deferrability(self, constraint): - text = "" - if constraint.deferrable is not None: - if constraint.deferrable: - text += " DEFERRABLE" - else: - text += " NOT DEFERRABLE" - if constraint.initially is not None: - text += " INITIALLY %s" % self.preparer.validate_sql_phrase( - constraint.initially, FK_INITIALLY - ) - return text - - def define_constraint_match(self, constraint): - text = "" - if constraint.match is not None: - text += " MATCH %s" % constraint.match - return text - - def visit_computed_column(self, generated, **kw): - text = "GENERATED ALWAYS AS (%s)" % self.sql_compiler.process( - generated.sqltext, include_table=False, literal_binds=True - ) - if generated.persisted is True: - text += " STORED" - elif generated.persisted is False: - text += " VIRTUAL" - return text - - def visit_identity_column(self, identity, **kw): - text = "GENERATED %s AS IDENTITY" % ( - "ALWAYS" if identity.always else "BY DEFAULT", - ) - options = self.get_identity_options(identity) - if options: - text += " (%s)" % options - return text - - -class GenericTypeCompiler(TypeCompiler): - def visit_FLOAT(self, type_, **kw): - return "FLOAT" - - def visit_DOUBLE(self, type_, **kw): - return "DOUBLE" - - def visit_DOUBLE_PRECISION(self, type_, **kw): - return "DOUBLE PRECISION" - - def visit_REAL(self, type_, **kw): - return "REAL" - - def visit_NUMERIC(self, type_, **kw): - if type_.precision is None: - return "NUMERIC" - elif type_.scale is None: - return "NUMERIC(%(precision)s)" % {"precision": type_.precision} - else: - return "NUMERIC(%(precision)s, %(scale)s)" % { - "precision": type_.precision, - "scale": type_.scale, - } - - def visit_DECIMAL(self, type_, **kw): - if type_.precision is None: - return "DECIMAL" - elif type_.scale is None: - return "DECIMAL(%(precision)s)" % {"precision": type_.precision} - else: - return "DECIMAL(%(precision)s, %(scale)s)" % { - "precision": type_.precision, - "scale": type_.scale, - } - - def visit_INTEGER(self, type_, **kw): - return "INTEGER" - - def visit_SMALLINT(self, type_, **kw): - return "SMALLINT" - - def visit_BIGINT(self, type_, **kw): - return "BIGINT" - - def visit_TIMESTAMP(self, type_, **kw): - return "TIMESTAMP" - - def visit_DATETIME(self, type_, **kw): - return "DATETIME" - - def visit_DATE(self, type_, **kw): - return "DATE" - - def visit_TIME(self, type_, **kw): - return "TIME" - - def visit_CLOB(self, type_, **kw): - return "CLOB" - - def visit_NCLOB(self, type_, **kw): - return "NCLOB" - - def _render_string_type(self, type_, name, length_override=None): - text = name - if length_override: - text += "(%d)" % length_override - elif type_.length: - text += "(%d)" % type_.length - if type_.collation: - text += ' COLLATE "%s"' % type_.collation - return text - - def visit_CHAR(self, type_, **kw): - return self._render_string_type(type_, "CHAR") - - def visit_NCHAR(self, type_, **kw): - return self._render_string_type(type_, "NCHAR") - - def visit_VARCHAR(self, type_, **kw): - return self._render_string_type(type_, "VARCHAR") - - def visit_NVARCHAR(self, type_, **kw): - return self._render_string_type(type_, "NVARCHAR") - - def visit_TEXT(self, type_, **kw): - return self._render_string_type(type_, "TEXT") - - def visit_UUID(self, type_, **kw): - return "UUID" - - def visit_BLOB(self, type_, **kw): - return "BLOB" - - def visit_BINARY(self, type_, **kw): - return "BINARY" + (type_.length and "(%d)" % type_.length or "") - - def visit_VARBINARY(self, type_, **kw): - return "VARBINARY" + (type_.length and "(%d)" % type_.length or "") - - def visit_BOOLEAN(self, type_, **kw): - return "BOOLEAN" - - def visit_uuid(self, type_, **kw): - if not type_.native_uuid or not self.dialect.supports_native_uuid: - return self._render_string_type(type_, "CHAR", length_override=32) - else: - return self.visit_UUID(type_, **kw) - - def visit_large_binary(self, type_, **kw): - return self.visit_BLOB(type_, **kw) - - def visit_boolean(self, type_, **kw): - return self.visit_BOOLEAN(type_, **kw) - - def visit_time(self, type_, **kw): - return self.visit_TIME(type_, **kw) - - def visit_datetime(self, type_, **kw): - return self.visit_DATETIME(type_, **kw) - - def visit_date(self, type_, **kw): - return self.visit_DATE(type_, **kw) - - def visit_big_integer(self, type_, **kw): - return self.visit_BIGINT(type_, **kw) - - def visit_small_integer(self, type_, **kw): - return self.visit_SMALLINT(type_, **kw) - - def visit_integer(self, type_, **kw): - return self.visit_INTEGER(type_, **kw) - - def visit_real(self, type_, **kw): - return self.visit_REAL(type_, **kw) - - def visit_float(self, type_, **kw): - return self.visit_FLOAT(type_, **kw) - - def visit_double(self, type_, **kw): - return self.visit_DOUBLE(type_, **kw) - - def visit_numeric(self, type_, **kw): - return self.visit_NUMERIC(type_, **kw) - - def visit_string(self, type_, **kw): - return self.visit_VARCHAR(type_, **kw) - - def visit_unicode(self, type_, **kw): - return self.visit_VARCHAR(type_, **kw) - - def visit_text(self, type_, **kw): - return self.visit_TEXT(type_, **kw) - - def visit_unicode_text(self, type_, **kw): - return self.visit_TEXT(type_, **kw) - - def visit_enum(self, type_, **kw): - return self.visit_VARCHAR(type_, **kw) - - def visit_null(self, type_, **kw): - raise exc.CompileError( - "Can't generate DDL for %r; " - "did you forget to specify a " - "type on this Column?" % type_ - ) - - def visit_type_decorator(self, type_, **kw): - return self.process(type_.type_engine(self.dialect), **kw) - - def visit_user_defined(self, type_, **kw): - return type_.get_col_spec(**kw) - - -class StrSQLTypeCompiler(GenericTypeCompiler): - def process(self, type_, **kw): - try: - _compiler_dispatch = type_._compiler_dispatch - except AttributeError: - return self._visit_unknown(type_, **kw) - else: - return _compiler_dispatch(self, **kw) - - def __getattr__(self, key): - if key.startswith("visit_"): - return self._visit_unknown - else: - raise AttributeError(key) - - def _visit_unknown(self, type_, **kw): - if type_.__class__.__name__ == type_.__class__.__name__.upper(): - return type_.__class__.__name__ - else: - return repr(type_) - - def visit_null(self, type_, **kw): - return "NULL" - - def visit_user_defined(self, type_, **kw): - try: - get_col_spec = type_.get_col_spec - except AttributeError: - return repr(type_) - else: - return get_col_spec(**kw) - - -class _SchemaForObjectCallable(Protocol): - def __call__(self, obj: Any) -> str: ... - - -class _BindNameForColProtocol(Protocol): - def __call__(self, col: ColumnClause[Any]) -> str: ... - - -class IdentifierPreparer: - """Handle quoting and case-folding of identifiers based on options.""" - - reserved_words = RESERVED_WORDS - - legal_characters = LEGAL_CHARACTERS - - illegal_initial_characters = ILLEGAL_INITIAL_CHARACTERS - - initial_quote: str - - final_quote: str - - _strings: MutableMapping[str, str] - - schema_for_object: _SchemaForObjectCallable = operator.attrgetter("schema") - """Return the .schema attribute for an object. - - For the default IdentifierPreparer, the schema for an object is always - the value of the ".schema" attribute. if the preparer is replaced - with one that has a non-empty schema_translate_map, the value of the - ".schema" attribute is rendered a symbol that will be converted to a - real schema name from the mapping post-compile. - - """ - - _includes_none_schema_translate: bool = False - - def __init__( - self, - dialect, - initial_quote='"', - final_quote=None, - escape_quote='"', - quote_case_sensitive_collations=True, - omit_schema=False, - ): - """Construct a new ``IdentifierPreparer`` object. - - initial_quote - Character that begins a delimited identifier. - - final_quote - Character that ends a delimited identifier. Defaults to - `initial_quote`. - - omit_schema - Prevent prepending schema name. Useful for databases that do - not support schemae. - """ - - self.dialect = dialect - self.initial_quote = initial_quote - self.final_quote = final_quote or self.initial_quote - self.escape_quote = escape_quote - self.escape_to_quote = self.escape_quote * 2 - self.omit_schema = omit_schema - self.quote_case_sensitive_collations = quote_case_sensitive_collations - self._strings = {} - self._double_percents = self.dialect.paramstyle in ( - "format", - "pyformat", - ) - - def _with_schema_translate(self, schema_translate_map): - prep = self.__class__.__new__(self.__class__) - prep.__dict__.update(self.__dict__) - - includes_none = None in schema_translate_map - - def symbol_getter(obj): - name = obj.schema - if obj._use_schema_map and (name is not None or includes_none): - if name is not None and ("[" in name or "]" in name): - raise exc.CompileError( - "Square bracket characters ([]) not supported " - "in schema translate name '%s'" % name - ) - return quoted_name( - "__[SCHEMA_%s]" % (name or "_none"), quote=False - ) - else: - return obj.schema - - prep.schema_for_object = symbol_getter - prep._includes_none_schema_translate = includes_none - return prep - - def _render_schema_translates(self, statement, schema_translate_map): - d = schema_translate_map - if None in d: - if not self._includes_none_schema_translate: - raise exc.InvalidRequestError( - "schema translate map which previously did not have " - "`None` present as a key now has `None` present; compiled " - "statement may lack adequate placeholders. Please use " - "consistent keys in successive " - "schema_translate_map dictionaries." - ) - - d["_none"] = d[None] - - def replace(m): - name = m.group(2) - if name in d: - effective_schema = d[name] - else: - if name in (None, "_none"): - raise exc.InvalidRequestError( - "schema translate map which previously had `None` " - "present as a key now no longer has it present; don't " - "know how to apply schema for compiled statement. " - "Please use consistent keys in successive " - "schema_translate_map dictionaries." - ) - effective_schema = name - - if not effective_schema: - effective_schema = self.dialect.default_schema_name - if not effective_schema: - # TODO: no coverage here - raise exc.CompileError( - "Dialect has no default schema name; can't " - "use None as dynamic schema target." - ) - return self.quote_schema(effective_schema) - - return re.sub(r"(__\[SCHEMA_([^\]]+)\])", replace, statement) - - def _escape_identifier(self, value: str) -> str: - """Escape an identifier. - - Subclasses should override this to provide database-dependent - escaping behavior. - """ - - value = value.replace(self.escape_quote, self.escape_to_quote) - if self._double_percents: - value = value.replace("%", "%%") - return value - - def _unescape_identifier(self, value: str) -> str: - """Canonicalize an escaped identifier. - - Subclasses should override this to provide database-dependent - unescaping behavior that reverses _escape_identifier. - """ - - return value.replace(self.escape_to_quote, self.escape_quote) - - def validate_sql_phrase(self, element, reg): - """keyword sequence filter. - - a filter for elements that are intended to represent keyword sequences, - such as "INITIALLY", "INITIALLY DEFERRED", etc. no special characters - should be present. - - .. versionadded:: 1.3 - - """ - - if element is not None and not reg.match(element): - raise exc.CompileError( - "Unexpected SQL phrase: %r (matching against %r)" - % (element, reg.pattern) - ) - return element - - def quote_identifier(self, value: str) -> str: - """Quote an identifier. - - Subclasses should override this to provide database-dependent - quoting behavior. - """ - - return ( - self.initial_quote - + self._escape_identifier(value) - + self.final_quote - ) - - def _requires_quotes(self, value: str) -> bool: - """Return True if the given identifier requires quoting.""" - lc_value = value.lower() - return ( - lc_value in self.reserved_words - or value[0] in self.illegal_initial_characters - or not self.legal_characters.match(str(value)) - or (lc_value != value) - ) - - def _requires_quotes_illegal_chars(self, value): - """Return True if the given identifier requires quoting, but - not taking case convention into account.""" - return not self.legal_characters.match(str(value)) - - def quote_schema(self, schema: str, force: Any = None) -> str: - """Conditionally quote a schema name. - - - The name is quoted if it is a reserved word, contains quote-necessary - characters, or is an instance of :class:`.quoted_name` which includes - ``quote`` set to ``True``. - - Subclasses can override this to provide database-dependent - quoting behavior for schema names. - - :param schema: string schema name - :param force: unused - - .. deprecated:: 0.9 - - The :paramref:`.IdentifierPreparer.quote_schema.force` - parameter is deprecated and will be removed in a future - release. This flag has no effect on the behavior of the - :meth:`.IdentifierPreparer.quote` method; please refer to - :class:`.quoted_name`. - - """ - if force is not None: - # not using the util.deprecated_params() decorator in this - # case because of the additional function call overhead on this - # very performance-critical spot. - util.warn_deprecated( - "The IdentifierPreparer.quote_schema.force parameter is " - "deprecated and will be removed in a future release. This " - "flag has no effect on the behavior of the " - "IdentifierPreparer.quote method; please refer to " - "quoted_name().", - # deprecated 0.9. warning from 1.3 - version="0.9", - ) - - return self.quote(schema) - - def quote(self, ident: str, force: Any = None) -> str: - """Conditionally quote an identifier. - - The identifier is quoted if it is a reserved word, contains - quote-necessary characters, or is an instance of - :class:`.quoted_name` which includes ``quote`` set to ``True``. - - Subclasses can override this to provide database-dependent - quoting behavior for identifier names. - - :param ident: string identifier - :param force: unused - - .. deprecated:: 0.9 - - The :paramref:`.IdentifierPreparer.quote.force` - parameter is deprecated and will be removed in a future - release. This flag has no effect on the behavior of the - :meth:`.IdentifierPreparer.quote` method; please refer to - :class:`.quoted_name`. - - """ - if force is not None: - # not using the util.deprecated_params() decorator in this - # case because of the additional function call overhead on this - # very performance-critical spot. - util.warn_deprecated( - "The IdentifierPreparer.quote.force parameter is " - "deprecated and will be removed in a future release. This " - "flag has no effect on the behavior of the " - "IdentifierPreparer.quote method; please refer to " - "quoted_name().", - # deprecated 0.9. warning from 1.3 - version="0.9", - ) - - force = getattr(ident, "quote", None) - - if force is None: - if ident in self._strings: - return self._strings[ident] - else: - if self._requires_quotes(ident): - self._strings[ident] = self.quote_identifier(ident) - else: - self._strings[ident] = ident - return self._strings[ident] - elif force: - return self.quote_identifier(ident) - else: - return ident - - def format_collation(self, collation_name): - if self.quote_case_sensitive_collations: - return self.quote(collation_name) - else: - return collation_name - - def format_sequence(self, sequence, use_schema=True): - name = self.quote(sequence.name) - - effective_schema = self.schema_for_object(sequence) - - if ( - not self.omit_schema - and use_schema - and effective_schema is not None - ): - name = self.quote_schema(effective_schema) + "." + name - return name - - def format_label( - self, label: Label[Any], name: Optional[str] = None - ) -> str: - return self.quote(name or label.name) - - def format_alias( - self, alias: Optional[AliasedReturnsRows], name: Optional[str] = None - ) -> str: - if name is None: - assert alias is not None - return self.quote(alias.name) - else: - return self.quote(name) - - def format_savepoint(self, savepoint, name=None): - # Running the savepoint name through quoting is unnecessary - # for all known dialects. This is here to support potential - # third party use cases - ident = name or savepoint.ident - if self._requires_quotes(ident): - ident = self.quote_identifier(ident) - return ident - - @util.preload_module("sqlalchemy.sql.naming") - def format_constraint(self, constraint, _alembic_quote=True): - naming = util.preloaded.sql_naming - - if constraint.name is _NONE_NAME: - name = naming._constraint_name_for_table( - constraint, constraint.table - ) - - if name is None: - return None - else: - name = constraint.name - - if constraint.__visit_name__ == "index": - return self.truncate_and_render_index_name( - name, _alembic_quote=_alembic_quote - ) - else: - return self.truncate_and_render_constraint_name( - name, _alembic_quote=_alembic_quote - ) - - def truncate_and_render_index_name(self, name, _alembic_quote=True): - # calculate these at format time so that ad-hoc changes - # to dialect.max_identifier_length etc. can be reflected - # as IdentifierPreparer is long lived - max_ = ( - self.dialect.max_index_name_length - or self.dialect.max_identifier_length - ) - return self._truncate_and_render_maxlen_name( - name, max_, _alembic_quote - ) - - def truncate_and_render_constraint_name(self, name, _alembic_quote=True): - # calculate these at format time so that ad-hoc changes - # to dialect.max_identifier_length etc. can be reflected - # as IdentifierPreparer is long lived - max_ = ( - self.dialect.max_constraint_name_length - or self.dialect.max_identifier_length - ) - return self._truncate_and_render_maxlen_name( - name, max_, _alembic_quote - ) - - def _truncate_and_render_maxlen_name(self, name, max_, _alembic_quote): - if isinstance(name, elements._truncated_label): - if len(name) > max_: - name = name[0 : max_ - 8] + "_" + util.md5_hex(name)[-4:] - else: - self.dialect.validate_identifier(name) - - if not _alembic_quote: - return name - else: - return self.quote(name) - - def format_index(self, index): - return self.format_constraint(index) - - def format_table(self, table, use_schema=True, name=None): - """Prepare a quoted table and schema name.""" - - if name is None: - name = table.name - - result = self.quote(name) - - effective_schema = self.schema_for_object(table) - - if not self.omit_schema and use_schema and effective_schema: - result = self.quote_schema(effective_schema) + "." + result - return result - - def format_schema(self, name): - """Prepare a quoted schema name.""" - - return self.quote(name) - - def format_label_name( - self, - name, - anon_map=None, - ): - """Prepare a quoted column name.""" - - if anon_map is not None and isinstance( - name, elements._truncated_label - ): - name = name.apply_map(anon_map) - - return self.quote(name) - - def format_column( - self, - column, - use_table=False, - name=None, - table_name=None, - use_schema=False, - anon_map=None, - ): - """Prepare a quoted column name.""" - - if name is None: - name = column.name - - if anon_map is not None and isinstance( - name, elements._truncated_label - ): - name = name.apply_map(anon_map) - - if not getattr(column, "is_literal", False): - if use_table: - return ( - self.format_table( - column.table, use_schema=use_schema, name=table_name - ) - + "." - + self.quote(name) - ) - else: - return self.quote(name) - else: - # literal textual elements get stuck into ColumnClause a lot, - # which shouldn't get quoted - - if use_table: - return ( - self.format_table( - column.table, use_schema=use_schema, name=table_name - ) - + "." - + name - ) - else: - return name - - def format_table_seq(self, table, use_schema=True): - """Format table name and schema as a tuple.""" - - # Dialects with more levels in their fully qualified references - # ('database', 'owner', etc.) could override this and return - # a longer sequence. - - effective_schema = self.schema_for_object(table) - - if not self.omit_schema and use_schema and effective_schema: - return ( - self.quote_schema(effective_schema), - self.format_table(table, use_schema=False), - ) - else: - return (self.format_table(table, use_schema=False),) - - @util.memoized_property - def _r_identifiers(self): - initial, final, escaped_final = ( - re.escape(s) - for s in ( - self.initial_quote, - self.final_quote, - self._escape_identifier(self.final_quote), - ) - ) - r = re.compile( - r"(?:" - r"(?:%(initial)s((?:%(escaped)s|[^%(final)s])+)%(final)s" - r"|([^\.]+))(?=\.|$))+" - % {"initial": initial, "final": final, "escaped": escaped_final} - ) - return r - - def unformat_identifiers(self, identifiers): - """Unpack 'schema.table.column'-like strings into components.""" - - r = self._r_identifiers - return [ - self._unescape_identifier(i) - for i in [a or b for a, b in r.findall(identifiers)] - ] |