# sql/elements.py # Copyright (C) 2005-2022 the SQLAlchemy authors and contributors # # # This module is part of SQLAlchemy and is released under # the MIT License: https://www.opensource.org/licenses/mit-license.php """Core SQL expression elements, including :class:`_expression.ClauseElement`, :class:`_expression.ColumnElement`, and derived classes. """ from __future__ import unicode_literals import itertools import operator import re from . import coercions from . import operators from . import roles from . import traversals from . import type_api from .annotation import Annotated from .annotation import SupportsWrappingAnnotations from .base import _clone from .base import _generative from .base import Executable from .base import HasMemoized from .base import Immutable from .base import NO_ARG from .base import PARSE_AUTOCOMMIT from .base import SingletonConstant from .coercions import _document_text_coercion from .traversals import HasCopyInternals from .traversals import MemoizedHasCacheKey from .traversals import NO_CACHE from .visitors import cloned_traverse from .visitors import InternalTraversal from .visitors import traverse from .visitors import Traversible from .. import exc from .. import inspection from .. import util def collate(expression, collation): """Return the clause ``expression COLLATE collation``. e.g.:: collate(mycolumn, 'utf8_bin') produces:: mycolumn COLLATE utf8_bin The collation expression is also quoted if it is a case sensitive identifier, e.g. contains uppercase characters. .. versionchanged:: 1.2 quoting is automatically applied to COLLATE expressions if they are case sensitive. """ expr = coercions.expect(roles.ExpressionElementRole, expression) return BinaryExpression( expr, CollationClause(collation), operators.collate, type_=expr.type ) def between(expr, lower_bound, upper_bound, symmetric=False): """Produce a ``BETWEEN`` predicate clause. E.g.:: from sqlalchemy import between stmt = select(users_table).where(between(users_table.c.id, 5, 7)) Would produce SQL resembling:: SELECT id, name FROM user WHERE id BETWEEN :id_1 AND :id_2 The :func:`.between` function is a standalone version of the :meth:`_expression.ColumnElement.between` method available on all SQL expressions, as in:: stmt = select(users_table).where(users_table.c.id.between(5, 7)) All arguments passed to :func:`.between`, including the left side column expression, are coerced from Python scalar values if a the value is not a :class:`_expression.ColumnElement` subclass. For example, three fixed values can be compared as in:: print(between(5, 3, 7)) Which would produce:: :param_1 BETWEEN :param_2 AND :param_3 :param expr: a column expression, typically a :class:`_expression.ColumnElement` instance or alternatively a Python scalar expression to be coerced into a column expression, serving as the left side of the ``BETWEEN`` expression. :param lower_bound: a column or Python scalar expression serving as the lower bound of the right side of the ``BETWEEN`` expression. :param upper_bound: a column or Python scalar expression serving as the upper bound of the right side of the ``BETWEEN`` expression. :param symmetric: if True, will render " BETWEEN SYMMETRIC ". Note that not all databases support this syntax. .. versionadded:: 0.9.5 .. seealso:: :meth:`_expression.ColumnElement.between` """ expr = coercions.expect(roles.ExpressionElementRole, expr) return expr.between(lower_bound, upper_bound, symmetric=symmetric) def literal(value, type_=None): r"""Return a literal clause, bound to a bind parameter. Literal clauses are created automatically when non- :class:`_expression.ClauseElement` objects (such as strings, ints, dates, etc.) are used in a comparison operation with a :class:`_expression.ColumnElement` subclass, such as a :class:`~sqlalchemy.schema.Column` object. Use this function to force the generation of a literal clause, which will be created as a :class:`BindParameter` with a bound value. :param value: the value to be bound. Can be any Python object supported by the underlying DB-API, or is translatable via the given type argument. :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will provide bind-parameter translation for this literal. """ return coercions.expect(roles.LiteralValueRole, value, type_=type_) def outparam(key, type_=None): """Create an 'OUT' parameter for usage in functions (stored procedures), for databases which support them. The ``outparam`` can be used like a regular function parameter. The "output" value will be available from the :class:`~sqlalchemy.engine.CursorResult` object via its ``out_parameters`` attribute, which returns a dictionary containing the values. """ return BindParameter(key, None, type_=type_, unique=False, isoutparam=True) def not_(clause): """Return a negation of the given clause, i.e. ``NOT(clause)``. The ``~`` operator is also overloaded on all :class:`_expression.ColumnElement` subclasses to produce the same result. """ return operators.inv(coercions.expect(roles.ExpressionElementRole, clause)) @inspection._self_inspects class ClauseElement( roles.SQLRole, SupportsWrappingAnnotations, MemoizedHasCacheKey, HasCopyInternals, Traversible, ): """Base class for elements of a programmatically constructed SQL expression. """ __visit_name__ = "clause" _propagate_attrs = util.immutabledict() """like annotations, however these propagate outwards liberally as SQL constructs are built, and are set up at construction time. """ supports_execution = False stringify_dialect = "default" _from_objects = [] bind = None description = None _is_clone_of = None is_clause_element = True is_selectable = False _is_textual = False _is_from_clause = False _is_returns_rows = False _is_text_clause = False _is_from_container = False _is_select_container = False _is_select_statement = False _is_bind_parameter = False _is_clause_list = False _is_lambda_element = False _is_singleton_constant = False _is_immutable = False _order_by_label_element = None _cache_key_traversal = None def _set_propagate_attrs(self, values): # usually, self._propagate_attrs is empty here. one case where it's # not is a subquery against ORM select, that is then pulled as a # property of an aliased class. should all be good # assert not self._propagate_attrs self._propagate_attrs = util.immutabledict(values) return self def _clone(self, **kw): """Create a shallow copy of this ClauseElement. This method may be used by a generative API. Its also used as part of the "deep" copy afforded by a traversal that combines the _copy_internals() method. """ skip = self._memoized_keys c = self.__class__.__new__(self.__class__) c.__dict__ = {k: v for k, v in self.__dict__.items() if k not in skip} # this is a marker that helps to "equate" clauses to each other # when a Select returns its list of FROM clauses. the cloning # process leaves around a lot of remnants of the previous clause # typically in the form of column expressions still attached to the # old table. c._is_clone_of = self return c def _negate_in_binary(self, negated_op, original_op): """a hook to allow the right side of a binary expression to respond to a negation of the binary expression. Used for the special case of expanding bind parameter with IN. """ return self def _with_binary_element_type(self, type_): """in the context of binary expression, convert the type of this object to the one given. applies only to :class:`_expression.ColumnElement` classes. """ return self @property def _constructor(self): """return the 'constructor' for this ClauseElement. This is for the purposes for creating a new object of this type. Usually, its just the element's __class__. However, the "Annotated" version of the object overrides to return the class of its proxied element. """ return self.__class__ @HasMemoized.memoized_attribute def _cloned_set(self): """Return the set consisting all cloned ancestors of this ClauseElement. Includes this ClauseElement. This accessor tends to be used for FromClause objects to identify 'equivalent' FROM clauses, regardless of transformative operations. """ s = util.column_set() f = self # note this creates a cycle, asserted in test_memusage. however, # turning this into a plain @property adds tends of thousands of method # calls to Core / ORM performance tests, so the small overhead # introduced by the relatively small amount of short term cycles # produced here is preferable while f is not None: s.add(f) f = f._is_clone_of return s @property def entity_namespace(self): raise AttributeError( "This SQL expression has no entity namespace " "with which to filter from." ) def __getstate__(self): d = self.__dict__.copy() d.pop("_is_clone_of", None) d.pop("_generate_cache_key", None) return d def _execute_on_connection( self, connection, multiparams, params, execution_options, _force=False ): if _force or self.supports_execution: return connection._execute_clauseelement( self, multiparams, params, execution_options ) else: raise exc.ObjectNotExecutableError(self) def unique_params(self, *optionaldict, **kwargs): """Return a copy with :func:`_expression.bindparam` elements replaced. Same functionality as :meth:`_expression.ClauseElement.params`, except adds `unique=True` to affected bind parameters so that multiple statements can be used. """ return self._replace_params(True, optionaldict, kwargs) def params(self, *optionaldict, **kwargs): """Return a copy with :func:`_expression.bindparam` elements replaced. Returns a copy of this ClauseElement with :func:`_expression.bindparam` elements replaced with values taken from the given dictionary:: >>> clause = column('x') + bindparam('foo') >>> print(clause.compile().params) {'foo':None} >>> print(clause.params({'foo':7}).compile().params) {'foo':7} """ return self._replace_params(False, optionaldict, kwargs) def _replace_params(self, unique, optionaldict, kwargs): if len(optionaldict) == 1: kwargs.update(optionaldict[0]) elif len(optionaldict) > 1: raise exc.ArgumentError( "params() takes zero or one positional dictionary argument" ) def visit_bindparam(bind): if bind.key in kwargs: bind.value = kwargs[bind.key] bind.required = False if unique: bind._convert_to_unique() return cloned_traverse( self, {"maintain_key": True, "detect_subquery_cols": True}, {"bindparam": visit_bindparam}, ) def compare(self, other, **kw): r"""Compare this :class:`_expression.ClauseElement` to the given :class:`_expression.ClauseElement`. Subclasses should override the default behavior, which is a straight identity comparison. \**kw are arguments consumed by subclass ``compare()`` methods and may be used to modify the criteria for comparison (see :class:`_expression.ColumnElement`). """ return traversals.compare(self, other, **kw) def self_group(self, against=None): """Apply a 'grouping' to this :class:`_expression.ClauseElement`. This method is overridden by subclasses to return a "grouping" construct, i.e. parenthesis. In particular it's used by "binary" expressions to provide a grouping around themselves when placed into a larger expression, as well as by :func:`_expression.select` constructs when placed into the FROM clause of another :func:`_expression.select`. (Note that subqueries should be normally created using the :meth:`_expression.Select.alias` method, as many platforms require nested SELECT statements to be named). As expressions are composed together, the application of :meth:`self_group` is automatic - end-user code should never need to use this method directly. Note that SQLAlchemy's clause constructs take operator precedence into account - so parenthesis might not be needed, for example, in an expression like ``x OR (y AND z)`` - AND takes precedence over OR. The base :meth:`self_group` method of :class:`_expression.ClauseElement` just returns self. """ return self def _ungroup(self): """Return this :class:`_expression.ClauseElement` without any groupings. """ return self @util.preload_module("sqlalchemy.engine.default") @util.preload_module("sqlalchemy.engine.url") def compile(self, bind=None, dialect=None, **kw): """Compile this SQL expression. The return value is a :class:`~.Compiled` object. Calling ``str()`` or ``unicode()`` on the returned value will yield a string representation of the result. The :class:`~.Compiled` object also can return a dictionary of bind parameter names and values using the ``params`` accessor. :param bind: An ``Engine`` or ``Connection`` from which a ``Compiled`` will be acquired. This argument takes precedence over this :class:`_expression.ClauseElement`'s bound engine, if any. :param column_keys: Used for INSERT and UPDATE statements, a list of column names which should be present in the VALUES clause of the compiled statement. If ``None``, all columns from the target table object are rendered. :param dialect: A ``Dialect`` instance from which a ``Compiled`` will be acquired. This argument takes precedence over the `bind` argument as well as this :class:`_expression.ClauseElement` 's bound engine, if any. :param compile_kwargs: optional dictionary of additional parameters that will be passed through to the compiler within all "visit" methods. This allows any custom flag to be passed through to a custom compilation construct, for example. It is also used for the case of passing the ``literal_binds`` flag through:: from sqlalchemy.sql import table, column, select t = table('t', column('x')) s = select(t).where(t.c.x == 5) print(s.compile(compile_kwargs={"literal_binds": True})) .. versionadded:: 0.9.0 .. seealso:: :ref:`faq_sql_expression_string` """ if not dialect: if bind: dialect = bind.dialect elif self.bind: dialect = self.bind.dialect else: if self.stringify_dialect == "default": default = util.preloaded.engine_default dialect = default.StrCompileDialect() else: url = util.preloaded.engine_url dialect = url.URL.create( self.stringify_dialect ).get_dialect()() return self._compiler(dialect, **kw) def _compile_w_cache( self, dialect, compiled_cache=None, column_keys=None, for_executemany=False, schema_translate_map=None, **kw ): if compiled_cache is not None and dialect._supports_statement_cache: elem_cache_key = self._generate_cache_key() else: elem_cache_key = None if elem_cache_key: cache_key, extracted_params = elem_cache_key key = ( dialect, cache_key, tuple(column_keys), bool(schema_translate_map), for_executemany, ) compiled_sql = compiled_cache.get(key) if compiled_sql is None: cache_hit = dialect.CACHE_MISS compiled_sql = self._compiler( dialect, cache_key=elem_cache_key, column_keys=column_keys, for_executemany=for_executemany, schema_translate_map=schema_translate_map, **kw ) compiled_cache[key] = compiled_sql else: cache_hit = dialect.CACHE_HIT else: extracted_params = None compiled_sql = self._compiler( dialect, cache_key=elem_cache_key, column_keys=column_keys, for_executemany=for_executemany, schema_translate_map=schema_translate_map, **kw ) if not dialect._supports_statement_cache: cache_hit = dialect.NO_DIALECT_SUPPORT elif compiled_cache is None: cache_hit = dialect.CACHING_DISABLED else: cache_hit = dialect.NO_CACHE_KEY return compiled_sql, extracted_params, cache_hit def _compiler(self, dialect, **kw): """Return a compiler appropriate for this ClauseElement, given a Dialect.""" return dialect.statement_compiler(dialect, self, **kw) def __str__(self): if util.py3k: return str(self.compile()) else: return unicode(self.compile()).encode( # noqa "ascii", "backslashreplace" ) # noqa def __invert__(self): # undocumented element currently used by the ORM for # relationship.contains() if hasattr(self, "negation_clause"): return self.negation_clause else: return self._negate() def _negate(self): return UnaryExpression( self.self_group(against=operators.inv), operator=operators.inv ) def __bool__(self): raise TypeError("Boolean value of this clause is not defined") __nonzero__ = __bool__ def __repr__(self): friendly = self.description if friendly is None: return object.__repr__(self) else: return "<%s.%s at 0x%x; %s>" % ( self.__module__, self.__class__.__name__, id(self), friendly, ) class ColumnElement( roles.ColumnArgumentOrKeyRole, roles.StatementOptionRole, roles.WhereHavingRole, roles.BinaryElementRole, roles.OrderByRole, roles.ColumnsClauseRole, roles.LimitOffsetRole, roles.DMLColumnRole, roles.DDLConstraintColumnRole, roles.DDLExpressionRole, operators.ColumnOperators, ClauseElement, ): """Represent a column-oriented SQL expression suitable for usage in the "columns" clause, WHERE clause etc. of a statement. While the most familiar kind of :class:`_expression.ColumnElement` is the :class:`_schema.Column` object, :class:`_expression.ColumnElement` serves as the basis for any unit that may be present in a SQL expression, including the expressions themselves, SQL functions, bound parameters, literal expressions, keywords such as ``NULL``, etc. :class:`_expression.ColumnElement` is the ultimate base class for all such elements. A wide variety of SQLAlchemy Core functions work at the SQL expression level, and are intended to accept instances of :class:`_expression.ColumnElement` as arguments. These functions will typically document that they accept a "SQL expression" as an argument. What this means in terms of SQLAlchemy usually refers to an input which is either already in the form of a :class:`_expression.ColumnElement` object, or a value which can be **coerced** into one. The coercion rules followed by most, but not all, SQLAlchemy Core functions with regards to SQL expressions are as follows: * a literal Python value, such as a string, integer or floating point value, boolean, datetime, ``Decimal`` object, or virtually any other Python object, will be coerced into a "literal bound value". This generally means that a :func:`.bindparam` will be produced featuring the given value embedded into the construct; the resulting :class:`.BindParameter` object is an instance of :class:`_expression.ColumnElement`. The Python value will ultimately be sent to the DBAPI at execution time as a parameterized argument to the ``execute()`` or ``executemany()`` methods, after SQLAlchemy type-specific converters (e.g. those provided by any associated :class:`.TypeEngine` objects) are applied to the value. * any special object value, typically ORM-level constructs, which feature an accessor called ``__clause_element__()``. The Core expression system looks for this method when an object of otherwise unknown type is passed to a function that is looking to coerce the argument into a :class:`_expression.ColumnElement` and sometimes a :class:`_expression.SelectBase` expression. It is used within the ORM to convert from ORM-specific objects like mapped classes and mapped attributes into Core expression objects. * The Python ``None`` value is typically interpreted as ``NULL``, which in SQLAlchemy Core produces an instance of :func:`.null`. A :class:`_expression.ColumnElement` provides the ability to generate new :class:`_expression.ColumnElement` objects using Python expressions. This means that Python operators such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations, and allow the instantiation of further :class:`_expression.ColumnElement` instances which are composed from other, more fundamental :class:`_expression.ColumnElement` objects. For example, two :class:`.ColumnClause` objects can be added together with the addition operator ``+`` to produce a :class:`.BinaryExpression`. Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses of :class:`_expression.ColumnElement`:: >>> from sqlalchemy.sql import column >>> column('a') + column('b') >>> print(column('a') + column('b')) a + b .. seealso:: :class:`_schema.Column` :func:`_expression.column` """ __visit_name__ = "column_element" primary_key = False foreign_keys = [] _proxies = () _tq_label = None """The named label that can be used to target this column in a result set in a "table qualified" context. This label is almost always the label used when rendering AS