# sql/compiler.py

# Copyright (C) 2005-2020 the SQLAlchemy authors and contributors

# <see AUTHORS file>

#

# This module is part of SQLAlchemy and is released under

# the MIT License: http://www.opensource.org/licenses/mit-license.php

"""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`.

"""

import collections

import contextlib

import itertools

import operator

import re

import time

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 .base import NO_ARG

from .base import prefix_anon_map

from .elements import quoted_name

from .. import exc

from .. import util

RESERVED_WORDS = set(

[

"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)

BIND_TEMPLATES = {

"pyformat": "%%(%(name)s)s",

"qmark": "?",

"format": "%%s",

"numeric": ":[_POSITION]",

"named": ":%(name)s",

}

OPERATORS = {

# binary

operators.and_: " AND ",

operators.or_: " OR ",

operators.add: " + ",

operators.mul: " * ",

operators.sub: " - ",

operators.div: " / ",

operators.mod: " % ",

operators.truediv: " / ",

operators.neg: "-",

operators.lt: " < ",

operators.le: " <= ",

operators.ne: " != ",

operators.gt: " > ",

operators.ge: " >= ",

operators.eq: " = ",

operators.is_distinct_from: " IS DISTINCT FROM ",

operators.isnot_distinct_from: " IS NOT DISTINCT FROM ",

operators.concat_op: " || ",

operators.match_op: " MATCH ",

operators.notmatch_op: " NOT MATCH ",

operators.in_op: " IN ",

operators.notin_op: " NOT IN ",

operators.comma_op: ", ",

operators.from_: " FROM ",

operators.as_: " AS ",

operators.is_: " IS ",

operators.isnot: " 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.nullsfirst_op: " NULLS FIRST",

operators.nullslast_op: " NULLS LAST",

}

FUNCTIONS = {

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.CompoundSelect.UNION: "UNION",

selectable.CompoundSelect.UNION_ALL: "UNION ALL",

selectable.CompoundSelect.EXCEPT: "EXCEPT",

selectable.CompoundSelect.EXCEPT_ALL: "EXCEPT ALL",

selectable.CompoundSelect.INTERSECT: "INTERSECT",

selectable.CompoundSelect.INTERSECT_ALL: "INTERSECT ALL",

}

RM_RENDERED_NAME = 0

RM_NAME = 1

RM_OBJECTS = 2

RM_TYPE = 3

ExpandedState = collections.namedtuple(

"ExpandedState",

[

"statement",

"additional_parameters",

"processors",

"positiontup",

"parameter_expansion",

],

)

NO_LINTING = util.symbol("NO_LINTING", "Disable all linting.", canonical=0)

COLLECT_CARTESIAN_PRODUCTS = util.symbol(

"COLLECT_CARTESIAN_PRODUCTS",

"Collect data on FROMs and cartesian products and gather "

"into 'self.from_linter'",

canonical=1,

)

WARN_LINTING = util.symbol(

"WARN_LINTING", "Emit warnings for linters that find problems", canonical=2

)

FROM_LINTING = util.symbol(

"FROM_LINTING",

"Warn for cartesian products; "

"combines COLLECT_CARTESIAN_PRODUCTS and WARN_LINTING",

canonical=COLLECT_CARTESIAN_PRODUCTS | WARN_LINTING,

)

class FromLinter(collections.namedtuple("FromLinter", ["froms", "edges"])):

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):

the_rest, start_with = self.lint()

# FROMS left over? boom

if the_rest:

froms = the_rest

if froms:

template = (

"SELECT 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(

'"{elem}"'.format(elem=self.froms[from_])

for from_ in froms

)

message = template.format(

froms=froms_str, start=self.froms[start_with]

)

util.warn(message)

class Compiled(object):

"""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.

"""

_cached_metadata = None

schema_translate_map = None

execution_options = util.immutabledict()

"""

Execution options propagated from the statement. In some cases,

sub-elements of the statement can modify these.

"""

compile_state = 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

"""

_rewrites_selected_columns = False

"""if True, indicates the compile_state object rewrites an incoming

ReturnsRows (like a Select) so that the columns we compile against in the

result set are not what were expressed on the outside. this is a hint to

the execution context to not link the statement.selected_columns to the

columns mapped in the result object.

That is, when this flag is False::

stmt = some_statement()

result = conn.execute(stmt)

row = result.first()

# selected_columns are in a 1-1 relationship with the

# columns in the result, and are targetable in mapping

for col in stmt.selected_columns:

assert col in row._mapping

When True::

# selected columns are not what are in the rows. the context

# rewrote the statement for some other set of selected_columns.

for col in stmt.selected_columns:

assert col not in row._mapping

"""

cache_key = None

_gen_time = None

def __init__(

self,

dialect,

statement,

bind=None,

schema_translate_map=None,

render_schema_translate=False,

compile_kwargs=util.immutabledict(),

):

"""Construct a new :class:`.Compiled` object.

:param dialect: :class:`.Dialect` to compile against.

:param statement: :class:`_expression.ClauseElement` to be compiled.

:param bind: Optional Engine or Connection to compile this

statement against.

:param schema_translate_map: dictionary of schema names to be

translated when forming the resultant SQL

.. versionadded:: 1.1

.. 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.bind = bind

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.statement = statement

self.can_execute = statement.supports_execution

if self.can_execute:

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._gen_time = time.time()

def _execute_on_connection(

self, connection, multiparams, params, execution_options

):

if self.can_execute:

return connection._execute_compiled(

self, multiparams, params, execution_options

)

else:

raise exc.ObjectNotExecutableError(self.statement)

@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, **kwargs):

return obj._compiler_dispatch(self, **kwargs)

def __str__(self):

"""Return the string text of the generated SQL or DDL."""

return self.string or ""

def construct_params(self, params=None, extracted_parameters=None):

"""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()

def execute(self, *multiparams, **params):

"""Execute this compiled object."""

e = self.bind

if e is None:

raise exc.UnboundExecutionError(

"This Compiled object is not bound to any Engine "

"or Connection.",

code="2afi",

)

return e._execute_compiled(self, multiparams, params)

def scalar(self, *multiparams, **params):

"""Execute this compiled object and return the result's

scalar value."""

return self.execute(*multiparams, **params).scalar()

class TypeCompiler(util.with_metaclass(util.EnsureKWArgType, object)):

"""Produces DDL specification for TypeEngine objects."""

ensure_kwarg = r"visit_\w+"

def __init__(self, dialect):

self.dialect = dialect

def process(self, type_, **kw):

return type_._compiler_dispatch(self, **kw)

# this was a Visitable, but to allow accurate detection of

# column elements this is actually a column element

class _CompileLabel(elements.ColumnElement):

"""lightweight label object which acts as an expression.Label."""

__visit_name__ = "label"

__slots__ = "element", "name"

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 SQLCompiler(Compiled):

"""Default implementation of :class:`.Compiled`.

Compiles :class:`_expression.ClauseElement` objects into SQL strings.

"""

extract_map = EXTRACT_MAP

compound_keywords = COMPOUND_KEYWORDS

isdelete = isinsert = isupdate = False

"""class-level defaults which can be set at the instance

level to define if this Compiled instance represents

INSERT/UPDATE/DELETE

"""

isplaintext = False

returning = None

"""holds the "returning" collection of columns if

the statement is CRUD and defines returning columns

either implicitly or explicitly

"""

returning_precedes_values = 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 = 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 = 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

"""

_textual_ordered_columns = 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.

"""

_ordered_columns = 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 = False

"""tell the result object that the SQL staement is textual, wants to match

up to Column objects, and may be using the ._label in the SELECT rather

than the base name.

"""

_numeric_binds = False

"""

True if paramstyle is "numeric". This paramstyle is trickier than

all the others.

"""

_render_postcompile = False

"""

whether to render out POSTCOMPILE params during the compile phase.

"""

insert_single_values_expr = None

"""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

"""

literal_execute_params = frozenset()

"""bindparameter objects that are rendered as literal values at statement

execution time.

"""

post_compile_params = frozenset()

"""bindparameter objects that are rendered as bound parameter placeholders

at statement execution time.

"""

has_out_parameters = False

"""if True, there are bindparam() objects that have the isoutparam

flag set."""

insert_prefetch = update_prefetch = ()

def __init__(

self,

dialect,

statement,

cache_key=None,

column_keys=None,

inline=False,

linting=NO_LINTING,

**kwargs

):

"""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 inline: whether to generate INSERT statements as "inline", e.g.

not formatted to return any generated defaults

:param kwargs: additional keyword arguments to be consumed by the

superclass.

"""

self.column_keys = column_keys

self.cache_key = cache_key

# compile INSERT/UPDATE defaults/sequences inlined (no pre-

# execute)

self.inline = inline or getattr(statement, "_inline", False)

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 = []

# 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

self._result_columns = []

# true if the paramstyle is positional

self.positional = dialect.positional

if self.positional:

self.positiontup = []

self._numeric_binds = dialect.paramstyle == "numeric"

self.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 = {}

Compiled.__init__(self, dialect, statement, **kwargs)

if (

self.isinsert or self.isupdate or self.isdelete

) and statement._returning:

self.returning = statement._returning

if self.positional and self._numeric_binds:

self._apply_numbered_params()

if self._render_postcompile:

self._process_parameters_for_postcompile(_populate_self=True)

@property

def prefetch(self):

return list(self.insert_prefetch + self.update_prefetch)

@util.memoized_instancemethod

def _init_cte_state(self):

"""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

self.ctes = util.OrderedDict()

self.ctes_by_name = {}

self.ctes_recursive = False

if self.positional:

self.cte_positional = {}

@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 _apply_numbered_params(self):

poscount = itertools.count(1)

self.string = re.sub(

r"\[_POSITION\]", lambda m: str(util.next(poscount)), self.string

)

@util.memoized_property

def _bind_processors(self):

return dict(

(key, value)

for key, value in (

(

self.bind_names[bindparam],

bindparam.type._cached_bind_processor(self.dialect)

if not bindparam._expanding_in_types

else tuple(

elem_type._cached_bind_processor(self.dialect)

for elem_type in bindparam._expanding_in_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_params(

self,

params=None,

_group_number=None,

_check=True,

extracted_parameters=None,

):

"""return a dictionary of bind parameter keys and values"""

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.

try:

orig_extracted = self.cache_key[1]

except TypeError as err:

util.raise_(

exc.CompileError(

"This compiled object has no original cache key; "

"can't pass extracted_parameters to construct_params"

),

replace_context=err,

)

resolved_extracted = {

b.key: extracted

for b, extracted in zip(orig_extracted, extracted_parameters)

}

else:

resolved_extracted = None

if params:

pd = {}

for bindparam, name in self.bind_names.items():

if bindparam.key in params:

pd[name] = params[bindparam.key]

elif name in params:

pd[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.key, bindparam

)

else:

value_param = bindparam

if bindparam.callable:

pd[name] = value_param.effective_value

else:

pd[name] = value_param.value

return pd

else:

pd = {}

for bindparam, name in self.bind_names.items():

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.key, bindparam

)

else:

value_param = bindparam

if bindparam.callable:

pd[name] = value_param.effective_value

else:

pd[name] = value_param.value

return pd

@property

def params(self):

"""Return the bind param dictionary embedded into this

compiled object, for those values that are present."""

return self.construct_params(_check=False)

def _process_parameters_for_postcompile(

self, parameters=None, _populate_self=False

):

"""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.

"""

if parameters is None:

parameters = self.construct_params()

expanded_parameters = {}

if self.positional:

positiontup = []

else:

positiontup = None

processors = self._bind_processors

new_processors = {}

if self.positional and self._numeric_binds:

# I'm not familiar with any DBAPI that uses 'numeric'.

# strategy would likely be to make use of numbers greater than

# the highest number present; then for expanding parameters,

# append them to the end of the parameter list. that way

# we avoid having to renumber all the existing parameters.

raise NotImplementedError(

"'post-compile' bind parameters are not supported with "

"the 'numeric' paramstyle at this time."

)

replacement_expressions = {}

to_update_sets = {}

for name in (