You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1390 lines
40 KiB
1390 lines
40 KiB
# sql/coercions.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
|
|
|
|
from __future__ import annotations
|
|
|
|
import collections.abc as collections_abc
|
|
import numbers
|
|
import re
|
|
import typing
|
|
from typing import Any
|
|
from typing import Callable
|
|
from typing import cast
|
|
from typing import Dict
|
|
from typing import Iterable
|
|
from typing import Iterator
|
|
from typing import List
|
|
from typing import NoReturn
|
|
from typing import Optional
|
|
from typing import overload
|
|
from typing import Sequence
|
|
from typing import Tuple
|
|
from typing import Type
|
|
from typing import TYPE_CHECKING
|
|
from typing import TypeVar
|
|
from typing import Union
|
|
|
|
from . import operators
|
|
from . import roles
|
|
from . import visitors
|
|
from ._typing import is_from_clause
|
|
from .base import ExecutableOption
|
|
from .base import Options
|
|
from .cache_key import HasCacheKey
|
|
from .visitors import Visitable
|
|
from .. import exc
|
|
from .. import inspection
|
|
from .. import util
|
|
from ..util.typing import Literal
|
|
|
|
if typing.TYPE_CHECKING:
|
|
# elements lambdas schema selectable are set by __init__
|
|
from . import elements
|
|
from . import lambdas
|
|
from . import schema
|
|
from . import selectable
|
|
from ._typing import _ColumnExpressionArgument
|
|
from ._typing import _ColumnsClauseArgument
|
|
from ._typing import _DDLColumnArgument
|
|
from ._typing import _DMLTableArgument
|
|
from ._typing import _FromClauseArgument
|
|
from .dml import _DMLTableElement
|
|
from .elements import BindParameter
|
|
from .elements import ClauseElement
|
|
from .elements import ColumnClause
|
|
from .elements import ColumnElement
|
|
from .elements import DQLDMLClauseElement
|
|
from .elements import NamedColumn
|
|
from .elements import SQLCoreOperations
|
|
from .schema import Column
|
|
from .selectable import _ColumnsClauseElement
|
|
from .selectable import _JoinTargetProtocol
|
|
from .selectable import FromClause
|
|
from .selectable import HasCTE
|
|
from .selectable import SelectBase
|
|
from .selectable import Subquery
|
|
from .visitors import _TraverseCallableType
|
|
|
|
_SR = TypeVar("_SR", bound=roles.SQLRole)
|
|
_F = TypeVar("_F", bound=Callable[..., Any])
|
|
_StringOnlyR = TypeVar("_StringOnlyR", bound=roles.StringRole)
|
|
_T = TypeVar("_T", bound=Any)
|
|
|
|
|
|
def _is_literal(element):
|
|
"""Return whether or not the element is a "literal" in the context
|
|
of a SQL expression construct.
|
|
|
|
"""
|
|
|
|
return not isinstance(
|
|
element,
|
|
(Visitable, schema.SchemaEventTarget),
|
|
) and not hasattr(element, "__clause_element__")
|
|
|
|
|
|
def _deep_is_literal(element):
|
|
"""Return whether or not the element is a "literal" in the context
|
|
of a SQL expression construct.
|
|
|
|
does a deeper more esoteric check than _is_literal. is used
|
|
for lambda elements that have to distinguish values that would
|
|
be bound vs. not without any context.
|
|
|
|
"""
|
|
|
|
if isinstance(element, collections_abc.Sequence) and not isinstance(
|
|
element, str
|
|
):
|
|
for elem in element:
|
|
if not _deep_is_literal(elem):
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
return (
|
|
not isinstance(
|
|
element,
|
|
(
|
|
Visitable,
|
|
schema.SchemaEventTarget,
|
|
HasCacheKey,
|
|
Options,
|
|
util.langhelpers.symbol,
|
|
),
|
|
)
|
|
and not hasattr(element, "__clause_element__")
|
|
and (
|
|
not isinstance(element, type)
|
|
or not issubclass(element, HasCacheKey)
|
|
)
|
|
)
|
|
|
|
|
|
def _document_text_coercion(
|
|
paramname: str, meth_rst: str, param_rst: str
|
|
) -> Callable[[_F], _F]:
|
|
return util.add_parameter_text(
|
|
paramname,
|
|
(
|
|
".. warning:: "
|
|
"The %s argument to %s can be passed as a Python string argument, "
|
|
"which will be treated "
|
|
"as **trusted SQL text** and rendered as given. **DO NOT PASS "
|
|
"UNTRUSTED INPUT TO THIS PARAMETER**."
|
|
)
|
|
% (param_rst, meth_rst),
|
|
)
|
|
|
|
|
|
def _expression_collection_was_a_list(
|
|
attrname: str,
|
|
fnname: str,
|
|
args: Union[Sequence[_T], Sequence[Sequence[_T]]],
|
|
) -> Sequence[_T]:
|
|
if args and isinstance(args[0], (list, set, dict)) and len(args) == 1:
|
|
if isinstance(args[0], list):
|
|
raise exc.ArgumentError(
|
|
f'The "{attrname}" argument to {fnname}(), when '
|
|
"referring to a sequence "
|
|
"of items, is now passed as a series of positional "
|
|
"elements, rather than as a list. "
|
|
)
|
|
return cast("Sequence[_T]", args[0])
|
|
|
|
return cast("Sequence[_T]", args)
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.TruncatedLabelRole],
|
|
element: Any,
|
|
**kw: Any,
|
|
) -> str: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.DMLColumnRole],
|
|
element: Any,
|
|
*,
|
|
as_key: Literal[True] = ...,
|
|
**kw: Any,
|
|
) -> str: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.LiteralValueRole],
|
|
element: Any,
|
|
**kw: Any,
|
|
) -> BindParameter[Any]: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.DDLReferredColumnRole],
|
|
element: Any,
|
|
**kw: Any,
|
|
) -> Column[Any]: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.DDLConstraintColumnRole],
|
|
element: Any,
|
|
**kw: Any,
|
|
) -> Union[Column[Any], str]: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.StatementOptionRole],
|
|
element: Any,
|
|
**kw: Any,
|
|
) -> DQLDMLClauseElement: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.LabeledColumnExprRole[Any]],
|
|
element: _ColumnExpressionArgument[_T],
|
|
**kw: Any,
|
|
) -> NamedColumn[_T]: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Union[
|
|
Type[roles.ExpressionElementRole[Any]],
|
|
Type[roles.LimitOffsetRole],
|
|
Type[roles.WhereHavingRole],
|
|
],
|
|
element: _ColumnExpressionArgument[_T],
|
|
**kw: Any,
|
|
) -> ColumnElement[_T]: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Union[
|
|
Type[roles.ExpressionElementRole[Any]],
|
|
Type[roles.LimitOffsetRole],
|
|
Type[roles.WhereHavingRole],
|
|
Type[roles.OnClauseRole],
|
|
Type[roles.ColumnArgumentRole],
|
|
],
|
|
element: Any,
|
|
**kw: Any,
|
|
) -> ColumnElement[Any]: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.DMLTableRole],
|
|
element: _DMLTableArgument,
|
|
**kw: Any,
|
|
) -> _DMLTableElement: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.HasCTERole],
|
|
element: HasCTE,
|
|
**kw: Any,
|
|
) -> HasCTE: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.SelectStatementRole],
|
|
element: SelectBase,
|
|
**kw: Any,
|
|
) -> SelectBase: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.FromClauseRole],
|
|
element: _FromClauseArgument,
|
|
**kw: Any,
|
|
) -> FromClause: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.FromClauseRole],
|
|
element: SelectBase,
|
|
*,
|
|
explicit_subquery: Literal[True] = ...,
|
|
**kw: Any,
|
|
) -> Subquery: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.ColumnsClauseRole],
|
|
element: _ColumnsClauseArgument[Any],
|
|
**kw: Any,
|
|
) -> _ColumnsClauseElement: ...
|
|
|
|
|
|
@overload
|
|
def expect(
|
|
role: Type[roles.JoinTargetRole],
|
|
element: _JoinTargetProtocol,
|
|
**kw: Any,
|
|
) -> _JoinTargetProtocol: ...
|
|
|
|
|
|
# catchall for not-yet-implemented overloads
|
|
@overload
|
|
def expect(
|
|
role: Type[_SR],
|
|
element: Any,
|
|
**kw: Any,
|
|
) -> Any: ...
|
|
|
|
|
|
def expect(
|
|
role: Type[_SR],
|
|
element: Any,
|
|
*,
|
|
apply_propagate_attrs: Optional[ClauseElement] = None,
|
|
argname: Optional[str] = None,
|
|
post_inspect: bool = False,
|
|
disable_inspection: bool = False,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if (
|
|
role.allows_lambda
|
|
# note callable() will not invoke a __getattr__() method, whereas
|
|
# hasattr(obj, "__call__") will. by keeping the callable() check here
|
|
# we prevent most needless calls to hasattr() and therefore
|
|
# __getattr__(), which is present on ColumnElement.
|
|
and callable(element)
|
|
and hasattr(element, "__code__")
|
|
):
|
|
return lambdas.LambdaElement(
|
|
element,
|
|
role,
|
|
lambdas.LambdaOptions(**kw),
|
|
apply_propagate_attrs=apply_propagate_attrs,
|
|
)
|
|
|
|
# major case is that we are given a ClauseElement already, skip more
|
|
# elaborate logic up front if possible
|
|
impl = _impl_lookup[role]
|
|
|
|
original_element = element
|
|
|
|
if not isinstance(
|
|
element,
|
|
(
|
|
elements.CompilerElement,
|
|
schema.SchemaItem,
|
|
schema.FetchedValue,
|
|
lambdas.PyWrapper,
|
|
),
|
|
):
|
|
resolved = None
|
|
|
|
if impl._resolve_literal_only:
|
|
resolved = impl._literal_coercion(element, **kw)
|
|
else:
|
|
original_element = element
|
|
|
|
is_clause_element = False
|
|
|
|
# this is a special performance optimization for ORM
|
|
# joins used by JoinTargetImpl that we don't go through the
|
|
# work of creating __clause_element__() when we only need the
|
|
# original QueryableAttribute, as the former will do clause
|
|
# adaption and all that which is just thrown away here.
|
|
if (
|
|
impl._skip_clauseelement_for_target_match
|
|
and isinstance(element, role)
|
|
and hasattr(element, "__clause_element__")
|
|
):
|
|
is_clause_element = True
|
|
else:
|
|
while hasattr(element, "__clause_element__"):
|
|
is_clause_element = True
|
|
|
|
if not getattr(element, "is_clause_element", False):
|
|
element = element.__clause_element__()
|
|
else:
|
|
break
|
|
|
|
if not is_clause_element:
|
|
if impl._use_inspection and not disable_inspection:
|
|
insp = inspection.inspect(element, raiseerr=False)
|
|
if insp is not None:
|
|
if post_inspect:
|
|
insp._post_inspect
|
|
try:
|
|
resolved = insp.__clause_element__()
|
|
except AttributeError:
|
|
impl._raise_for_expected(original_element, argname)
|
|
|
|
if resolved is None:
|
|
resolved = impl._literal_coercion(
|
|
element, argname=argname, **kw
|
|
)
|
|
else:
|
|
resolved = element
|
|
elif isinstance(element, lambdas.PyWrapper):
|
|
resolved = element._sa__py_wrapper_literal(**kw)
|
|
else:
|
|
resolved = element
|
|
|
|
if apply_propagate_attrs is not None:
|
|
if typing.TYPE_CHECKING:
|
|
assert isinstance(resolved, (SQLCoreOperations, ClauseElement))
|
|
|
|
if not apply_propagate_attrs._propagate_attrs and getattr(
|
|
resolved, "_propagate_attrs", None
|
|
):
|
|
apply_propagate_attrs._propagate_attrs = resolved._propagate_attrs
|
|
|
|
if impl._role_class in resolved.__class__.__mro__:
|
|
if impl._post_coercion:
|
|
resolved = impl._post_coercion(
|
|
resolved,
|
|
argname=argname,
|
|
original_element=original_element,
|
|
**kw,
|
|
)
|
|
return resolved
|
|
else:
|
|
return impl._implicit_coercions(
|
|
original_element, resolved, argname=argname, **kw
|
|
)
|
|
|
|
|
|
def expect_as_key(
|
|
role: Type[roles.DMLColumnRole], element: Any, **kw: Any
|
|
) -> str:
|
|
kw.pop("as_key", None)
|
|
return expect(role, element, as_key=True, **kw)
|
|
|
|
|
|
def expect_col_expression_collection(
|
|
role: Type[roles.DDLConstraintColumnRole],
|
|
expressions: Iterable[_DDLColumnArgument],
|
|
) -> Iterator[
|
|
Tuple[
|
|
Union[str, Column[Any]],
|
|
Optional[ColumnClause[Any]],
|
|
Optional[str],
|
|
Optional[Union[Column[Any], str]],
|
|
]
|
|
]:
|
|
for expr in expressions:
|
|
strname = None
|
|
column = None
|
|
|
|
resolved: Union[Column[Any], str] = expect(role, expr)
|
|
if isinstance(resolved, str):
|
|
assert isinstance(expr, str)
|
|
strname = resolved = expr
|
|
else:
|
|
cols: List[Column[Any]] = []
|
|
col_append: _TraverseCallableType[Column[Any]] = cols.append
|
|
visitors.traverse(resolved, {}, {"column": col_append})
|
|
if cols:
|
|
column = cols[0]
|
|
add_element = column if column is not None else strname
|
|
|
|
yield resolved, column, strname, add_element
|
|
|
|
|
|
class RoleImpl:
|
|
__slots__ = ("_role_class", "name", "_use_inspection")
|
|
|
|
def _literal_coercion(self, element, **kw):
|
|
raise NotImplementedError()
|
|
|
|
_post_coercion: Any = None
|
|
_resolve_literal_only = False
|
|
_skip_clauseelement_for_target_match = False
|
|
|
|
def __init__(self, role_class):
|
|
self._role_class = role_class
|
|
self.name = role_class._role_name
|
|
self._use_inspection = issubclass(role_class, roles.UsesInspection)
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
def _raise_for_expected(
|
|
self,
|
|
element: Any,
|
|
argname: Optional[str] = None,
|
|
resolved: Optional[Any] = None,
|
|
advice: Optional[str] = None,
|
|
code: Optional[str] = None,
|
|
err: Optional[Exception] = None,
|
|
**kw: Any,
|
|
) -> NoReturn:
|
|
if resolved is not None and resolved is not element:
|
|
got = "%r object resolved from %r object" % (resolved, element)
|
|
else:
|
|
got = repr(element)
|
|
|
|
if argname:
|
|
msg = "%s expected for argument %r; got %s." % (
|
|
self.name,
|
|
argname,
|
|
got,
|
|
)
|
|
else:
|
|
msg = "%s expected, got %s." % (self.name, got)
|
|
|
|
if advice:
|
|
msg += " " + advice
|
|
|
|
raise exc.ArgumentError(msg, code=code) from err
|
|
|
|
|
|
class _Deannotate:
|
|
__slots__ = ()
|
|
|
|
def _post_coercion(self, resolved, **kw):
|
|
from .util import _deep_deannotate
|
|
|
|
return _deep_deannotate(resolved)
|
|
|
|
|
|
class _StringOnly:
|
|
__slots__ = ()
|
|
|
|
_resolve_literal_only = True
|
|
|
|
|
|
class _ReturnsStringKey(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(self, element, resolved, argname=None, **kw):
|
|
if isinstance(element, str):
|
|
return element
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
def _literal_coercion(self, element, **kw):
|
|
return element
|
|
|
|
|
|
class _ColumnCoercions(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _warn_for_scalar_subquery_coercion(self):
|
|
util.warn(
|
|
"implicitly coercing SELECT object to scalar subquery; "
|
|
"please use the .scalar_subquery() method to produce a scalar "
|
|
"subquery.",
|
|
)
|
|
|
|
def _implicit_coercions(self, element, resolved, argname=None, **kw):
|
|
original_element = element
|
|
if not getattr(resolved, "is_clause_element", False):
|
|
self._raise_for_expected(original_element, argname, resolved)
|
|
elif resolved._is_select_base:
|
|
self._warn_for_scalar_subquery_coercion()
|
|
return resolved.scalar_subquery()
|
|
elif resolved._is_from_clause and isinstance(
|
|
resolved, selectable.Subquery
|
|
):
|
|
self._warn_for_scalar_subquery_coercion()
|
|
return resolved.element.scalar_subquery()
|
|
elif self._role_class.allows_lambda and resolved._is_lambda_element:
|
|
return resolved
|
|
else:
|
|
self._raise_for_expected(original_element, argname, resolved)
|
|
|
|
|
|
def _no_text_coercion(
|
|
element: Any,
|
|
argname: Optional[str] = None,
|
|
exc_cls: Type[exc.SQLAlchemyError] = exc.ArgumentError,
|
|
extra: Optional[str] = None,
|
|
err: Optional[Exception] = None,
|
|
) -> NoReturn:
|
|
raise exc_cls(
|
|
"%(extra)sTextual SQL expression %(expr)r %(argname)sshould be "
|
|
"explicitly declared as text(%(expr)r)"
|
|
% {
|
|
"expr": util.ellipses_string(element),
|
|
"argname": "for argument %s" % (argname,) if argname else "",
|
|
"extra": "%s " % extra if extra else "",
|
|
}
|
|
) from err
|
|
|
|
|
|
class _NoTextCoercion(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _literal_coercion(self, element, argname=None, **kw):
|
|
if isinstance(element, str) and issubclass(
|
|
elements.TextClause, self._role_class
|
|
):
|
|
_no_text_coercion(element, argname)
|
|
else:
|
|
self._raise_for_expected(element, argname)
|
|
|
|
|
|
class _CoerceLiterals(RoleImpl):
|
|
__slots__ = ()
|
|
_coerce_consts = False
|
|
_coerce_star = False
|
|
_coerce_numerics = False
|
|
|
|
def _text_coercion(self, element, argname=None):
|
|
return _no_text_coercion(element, argname)
|
|
|
|
def _literal_coercion(self, element, argname=None, **kw):
|
|
if isinstance(element, str):
|
|
if self._coerce_star and element == "*":
|
|
return elements.ColumnClause("*", is_literal=True)
|
|
else:
|
|
return self._text_coercion(element, argname, **kw)
|
|
|
|
if self._coerce_consts:
|
|
if element is None:
|
|
return elements.Null()
|
|
elif element is False:
|
|
return elements.False_()
|
|
elif element is True:
|
|
return elements.True_()
|
|
|
|
if self._coerce_numerics and isinstance(element, (numbers.Number)):
|
|
return elements.ColumnClause(str(element), is_literal=True)
|
|
|
|
self._raise_for_expected(element, argname)
|
|
|
|
|
|
class LiteralValueImpl(RoleImpl):
|
|
_resolve_literal_only = True
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element,
|
|
resolved,
|
|
argname,
|
|
type_=None,
|
|
literal_execute=False,
|
|
**kw,
|
|
):
|
|
if not _is_literal(resolved):
|
|
self._raise_for_expected(
|
|
element, resolved=resolved, argname=argname, **kw
|
|
)
|
|
|
|
return elements.BindParameter(
|
|
None,
|
|
element,
|
|
type_=type_,
|
|
unique=True,
|
|
literal_execute=literal_execute,
|
|
)
|
|
|
|
def _literal_coercion(self, element, argname=None, type_=None, **kw):
|
|
return element
|
|
|
|
|
|
class _SelectIsNotFrom(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _raise_for_expected(
|
|
self,
|
|
element: Any,
|
|
argname: Optional[str] = None,
|
|
resolved: Optional[Any] = None,
|
|
advice: Optional[str] = None,
|
|
code: Optional[str] = None,
|
|
err: Optional[Exception] = None,
|
|
**kw: Any,
|
|
) -> NoReturn:
|
|
if (
|
|
not advice
|
|
and isinstance(element, roles.SelectStatementRole)
|
|
or isinstance(resolved, roles.SelectStatementRole)
|
|
):
|
|
advice = (
|
|
"To create a "
|
|
"FROM clause from a %s object, use the .subquery() method."
|
|
% (resolved.__class__ if resolved is not None else element,)
|
|
)
|
|
code = "89ve"
|
|
else:
|
|
code = None
|
|
|
|
super()._raise_for_expected(
|
|
element,
|
|
argname=argname,
|
|
resolved=resolved,
|
|
advice=advice,
|
|
code=code,
|
|
err=err,
|
|
**kw,
|
|
)
|
|
# never reached
|
|
assert False
|
|
|
|
|
|
class HasCacheKeyImpl(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if isinstance(element, HasCacheKey):
|
|
return element
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
def _literal_coercion(self, element, **kw):
|
|
return element
|
|
|
|
|
|
class ExecutableOptionImpl(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if isinstance(element, ExecutableOption):
|
|
return element
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
def _literal_coercion(self, element, **kw):
|
|
return element
|
|
|
|
|
|
class ExpressionElementImpl(_ColumnCoercions, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _literal_coercion(
|
|
self, element, name=None, type_=None, argname=None, is_crud=False, **kw
|
|
):
|
|
if (
|
|
element is None
|
|
and not is_crud
|
|
and (type_ is None or not type_.should_evaluate_none)
|
|
):
|
|
# TODO: there's no test coverage now for the
|
|
# "should_evaluate_none" part of this, as outside of "crud" this
|
|
# codepath is not normally used except in some special cases
|
|
return elements.Null()
|
|
else:
|
|
try:
|
|
return elements.BindParameter(
|
|
name, element, type_, unique=True, _is_crud=is_crud
|
|
)
|
|
except exc.ArgumentError as err:
|
|
self._raise_for_expected(element, err=err)
|
|
|
|
def _raise_for_expected(self, element, argname=None, resolved=None, **kw):
|
|
# select uses implicit coercion with warning instead of raising
|
|
if isinstance(element, selectable.Values):
|
|
advice = (
|
|
"To create a column expression from a VALUES clause, "
|
|
"use the .scalar_values() method."
|
|
)
|
|
elif isinstance(element, roles.AnonymizedFromClauseRole):
|
|
advice = (
|
|
"To create a column expression from a FROM clause row "
|
|
"as a whole, use the .table_valued() method."
|
|
)
|
|
else:
|
|
advice = None
|
|
|
|
return super()._raise_for_expected(
|
|
element, argname=argname, resolved=resolved, advice=advice, **kw
|
|
)
|
|
|
|
|
|
class BinaryElementImpl(ExpressionElementImpl, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _literal_coercion(
|
|
self, element, expr, operator, bindparam_type=None, argname=None, **kw
|
|
):
|
|
try:
|
|
return expr._bind_param(operator, element, type_=bindparam_type)
|
|
except exc.ArgumentError as err:
|
|
self._raise_for_expected(element, err=err)
|
|
|
|
def _post_coercion(self, resolved, expr, bindparam_type=None, **kw):
|
|
if resolved.type._isnull and not expr.type._isnull:
|
|
resolved = resolved._with_binary_element_type(
|
|
bindparam_type if bindparam_type is not None else expr.type
|
|
)
|
|
return resolved
|
|
|
|
|
|
class InElementImpl(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if resolved._is_from_clause:
|
|
if (
|
|
isinstance(resolved, selectable.Alias)
|
|
and resolved.element._is_select_base
|
|
):
|
|
self._warn_for_implicit_coercion(resolved)
|
|
return self._post_coercion(resolved.element, **kw)
|
|
else:
|
|
self._warn_for_implicit_coercion(resolved)
|
|
return self._post_coercion(resolved.select(), **kw)
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
def _warn_for_implicit_coercion(self, elem):
|
|
util.warn(
|
|
"Coercing %s object into a select() for use in IN(); "
|
|
"please pass a select() construct explicitly"
|
|
% (elem.__class__.__name__)
|
|
)
|
|
|
|
def _literal_coercion(self, element, expr, operator, **kw):
|
|
if util.is_non_string_iterable(element):
|
|
non_literal_expressions: Dict[
|
|
Optional[operators.ColumnOperators],
|
|
operators.ColumnOperators,
|
|
] = {}
|
|
element = list(element)
|
|
for o in element:
|
|
if not _is_literal(o):
|
|
if not isinstance(o, operators.ColumnOperators):
|
|
self._raise_for_expected(element, **kw)
|
|
|
|
else:
|
|
non_literal_expressions[o] = o
|
|
elif o is None:
|
|
non_literal_expressions[o] = elements.Null()
|
|
|
|
if non_literal_expressions:
|
|
return elements.ClauseList(
|
|
*[
|
|
(
|
|
non_literal_expressions[o]
|
|
if o in non_literal_expressions
|
|
else expr._bind_param(operator, o)
|
|
)
|
|
for o in element
|
|
]
|
|
)
|
|
else:
|
|
return expr._bind_param(operator, element, expanding=True)
|
|
|
|
else:
|
|
self._raise_for_expected(element, **kw)
|
|
|
|
def _post_coercion(self, element, expr, operator, **kw):
|
|
if element._is_select_base:
|
|
# for IN, we are doing scalar_subquery() coercion without
|
|
# a warning
|
|
return element.scalar_subquery()
|
|
elif isinstance(element, elements.ClauseList):
|
|
assert not len(element.clauses) == 0
|
|
return element.self_group(against=operator)
|
|
|
|
elif isinstance(element, elements.BindParameter):
|
|
element = element._clone(maintain_key=True)
|
|
element.expanding = True
|
|
element.expand_op = operator
|
|
|
|
return element
|
|
elif isinstance(element, selectable.Values):
|
|
return element.scalar_values()
|
|
else:
|
|
return element
|
|
|
|
|
|
class OnClauseImpl(_ColumnCoercions, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
_coerce_consts = True
|
|
|
|
def _literal_coercion(
|
|
self, element, name=None, type_=None, argname=None, is_crud=False, **kw
|
|
):
|
|
self._raise_for_expected(element)
|
|
|
|
def _post_coercion(self, resolved, original_element=None, **kw):
|
|
# this is a hack right now as we want to use coercion on an
|
|
# ORM InstrumentedAttribute, but we want to return the object
|
|
# itself if it is one, not its clause element.
|
|
# ORM context _join and _legacy_join() would need to be improved
|
|
# to look for annotations in a clause element form.
|
|
if isinstance(original_element, roles.JoinTargetRole):
|
|
return original_element
|
|
return resolved
|
|
|
|
|
|
class WhereHavingImpl(_CoerceLiterals, _ColumnCoercions, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
_coerce_consts = True
|
|
|
|
def _text_coercion(self, element, argname=None):
|
|
return _no_text_coercion(element, argname)
|
|
|
|
|
|
class StatementOptionImpl(_CoerceLiterals, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
_coerce_consts = True
|
|
|
|
def _text_coercion(self, element, argname=None):
|
|
return elements.TextClause(element)
|
|
|
|
|
|
class ColumnArgumentImpl(_NoTextCoercion, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
|
|
class ColumnArgumentOrKeyImpl(_ReturnsStringKey, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
|
|
class StrAsPlainColumnImpl(_CoerceLiterals, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _text_coercion(self, element, argname=None):
|
|
return elements.ColumnClause(element)
|
|
|
|
|
|
class ByOfImpl(_CoerceLiterals, _ColumnCoercions, RoleImpl, roles.ByOfRole):
|
|
__slots__ = ()
|
|
|
|
_coerce_consts = True
|
|
|
|
def _text_coercion(self, element, argname=None):
|
|
return elements._textual_label_reference(element)
|
|
|
|
|
|
class OrderByImpl(ByOfImpl, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _post_coercion(self, resolved, **kw):
|
|
if (
|
|
isinstance(resolved, self._role_class)
|
|
and resolved._order_by_label_element is not None
|
|
):
|
|
return elements._label_reference(resolved)
|
|
else:
|
|
return resolved
|
|
|
|
|
|
class GroupByImpl(ByOfImpl, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if is_from_clause(resolved):
|
|
return elements.ClauseList(*resolved.c)
|
|
else:
|
|
return resolved
|
|
|
|
|
|
class DMLColumnImpl(_ReturnsStringKey, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _post_coercion(self, element, as_key=False, **kw):
|
|
if as_key:
|
|
return element.key
|
|
else:
|
|
return element
|
|
|
|
|
|
class ConstExprImpl(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _literal_coercion(self, element, argname=None, **kw):
|
|
if element is None:
|
|
return elements.Null()
|
|
elif element is False:
|
|
return elements.False_()
|
|
elif element is True:
|
|
return elements.True_()
|
|
else:
|
|
self._raise_for_expected(element, argname)
|
|
|
|
|
|
class TruncatedLabelImpl(_StringOnly, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if isinstance(element, str):
|
|
return resolved
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
def _literal_coercion(self, element, argname=None, **kw):
|
|
"""coerce the given value to :class:`._truncated_label`.
|
|
|
|
Existing :class:`._truncated_label` and
|
|
:class:`._anonymous_label` objects are passed
|
|
unchanged.
|
|
"""
|
|
|
|
if isinstance(element, elements._truncated_label):
|
|
return element
|
|
else:
|
|
return elements._truncated_label(element)
|
|
|
|
|
|
class DDLExpressionImpl(_Deannotate, _CoerceLiterals, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
_coerce_consts = True
|
|
|
|
def _text_coercion(self, element, argname=None):
|
|
# see #5754 for why we can't easily deprecate this coercion.
|
|
# essentially expressions like postgresql_where would have to be
|
|
# text() as they come back from reflection and we don't want to
|
|
# have text() elements wired into the inspection dictionaries.
|
|
return elements.TextClause(element)
|
|
|
|
|
|
class DDLConstraintColumnImpl(_Deannotate, _ReturnsStringKey, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
|
|
class DDLReferredColumnImpl(DDLConstraintColumnImpl):
|
|
__slots__ = ()
|
|
|
|
|
|
class LimitOffsetImpl(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if resolved is None:
|
|
return None
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
def _literal_coercion(self, element, name, type_, **kw):
|
|
if element is None:
|
|
return None
|
|
else:
|
|
value = util.asint(element)
|
|
return selectable._OffsetLimitParam(
|
|
name, value, type_=type_, unique=True
|
|
)
|
|
|
|
|
|
class LabeledColumnExprImpl(ExpressionElementImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if isinstance(resolved, roles.ExpressionElementRole):
|
|
return resolved.label(None)
|
|
else:
|
|
new = super()._implicit_coercions(
|
|
element, resolved, argname=argname, **kw
|
|
)
|
|
if isinstance(new, roles.ExpressionElementRole):
|
|
return new.label(None)
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
|
|
class ColumnsClauseImpl(_SelectIsNotFrom, _CoerceLiterals, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
_coerce_consts = True
|
|
_coerce_numerics = True
|
|
_coerce_star = True
|
|
|
|
_guess_straight_column = re.compile(r"^\w\S*$", re.I)
|
|
|
|
def _raise_for_expected(
|
|
self, element, argname=None, resolved=None, advice=None, **kw
|
|
):
|
|
if not advice and isinstance(element, list):
|
|
advice = (
|
|
f"Did you mean to say select("
|
|
f"{', '.join(repr(e) for e in element)})?"
|
|
)
|
|
|
|
return super()._raise_for_expected(
|
|
element, argname=argname, resolved=resolved, advice=advice, **kw
|
|
)
|
|
|
|
def _text_coercion(self, element, argname=None):
|
|
element = str(element)
|
|
|
|
guess_is_literal = not self._guess_straight_column.match(element)
|
|
raise exc.ArgumentError(
|
|
"Textual column expression %(column)r %(argname)sshould be "
|
|
"explicitly declared with text(%(column)r), "
|
|
"or use %(literal_column)s(%(column)r) "
|
|
"for more specificity"
|
|
% {
|
|
"column": util.ellipses_string(element),
|
|
"argname": "for argument %s" % (argname,) if argname else "",
|
|
"literal_column": (
|
|
"literal_column" if guess_is_literal else "column"
|
|
),
|
|
}
|
|
)
|
|
|
|
|
|
class ReturnsRowsImpl(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
|
|
class StatementImpl(_CoerceLiterals, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _post_coercion(self, resolved, original_element, argname=None, **kw):
|
|
if resolved is not original_element and not isinstance(
|
|
original_element, str
|
|
):
|
|
# use same method as Connection uses; this will later raise
|
|
# ObjectNotExecutableError
|
|
try:
|
|
original_element._execute_on_connection
|
|
except AttributeError:
|
|
util.warn_deprecated(
|
|
"Object %r should not be used directly in a SQL statement "
|
|
"context, such as passing to methods such as "
|
|
"session.execute(). This usage will be disallowed in a "
|
|
"future release. "
|
|
"Please use Core select() / update() / delete() etc. "
|
|
"with Session.execute() and other statement execution "
|
|
"methods." % original_element,
|
|
"1.4",
|
|
)
|
|
|
|
return resolved
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if resolved._is_lambda_element:
|
|
return resolved
|
|
else:
|
|
return super()._implicit_coercions(
|
|
element, resolved, argname=argname, **kw
|
|
)
|
|
|
|
|
|
class SelectStatementImpl(_NoTextCoercion, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if resolved._is_text_clause:
|
|
return resolved.columns()
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
|
|
class HasCTEImpl(ReturnsRowsImpl):
|
|
__slots__ = ()
|
|
|
|
|
|
class IsCTEImpl(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
|
|
class JoinTargetImpl(RoleImpl):
|
|
__slots__ = ()
|
|
|
|
_skip_clauseelement_for_target_match = True
|
|
|
|
def _literal_coercion(self, element, argname=None, **kw):
|
|
self._raise_for_expected(element, argname)
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
legacy: bool = False,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if isinstance(element, roles.JoinTargetRole):
|
|
# note that this codepath no longer occurs as of
|
|
# #6550, unless JoinTargetImpl._skip_clauseelement_for_target_match
|
|
# were set to False.
|
|
return element
|
|
elif legacy and resolved._is_select_base:
|
|
util.warn_deprecated(
|
|
"Implicit coercion of SELECT and textual SELECT "
|
|
"constructs into FROM clauses is deprecated; please call "
|
|
".subquery() on any Core select or ORM Query object in "
|
|
"order to produce a subquery object.",
|
|
version="1.4",
|
|
)
|
|
# TODO: doing _implicit_subquery here causes tests to fail,
|
|
# how was this working before? probably that ORM
|
|
# join logic treated it as a select and subquery would happen
|
|
# in _ORMJoin->Join
|
|
return resolved
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
|
|
class FromClauseImpl(_SelectIsNotFrom, _NoTextCoercion, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
explicit_subquery: bool = False,
|
|
allow_select: bool = True,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if resolved._is_select_base:
|
|
if explicit_subquery:
|
|
return resolved.subquery()
|
|
elif allow_select:
|
|
util.warn_deprecated(
|
|
"Implicit coercion of SELECT and textual SELECT "
|
|
"constructs into FROM clauses is deprecated; please call "
|
|
".subquery() on any Core select or ORM Query object in "
|
|
"order to produce a subquery object.",
|
|
version="1.4",
|
|
)
|
|
return resolved._implicit_subquery
|
|
elif resolved._is_text_clause:
|
|
return resolved
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
def _post_coercion(self, element, deannotate=False, **kw):
|
|
if deannotate:
|
|
return element._deannotate()
|
|
else:
|
|
return element
|
|
|
|
|
|
class StrictFromClauseImpl(FromClauseImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
explicit_subquery: bool = False,
|
|
allow_select: bool = False,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if resolved._is_select_base and allow_select:
|
|
util.warn_deprecated(
|
|
"Implicit coercion of SELECT and textual SELECT constructs "
|
|
"into FROM clauses is deprecated; please call .subquery() "
|
|
"on any Core select or ORM Query object in order to produce a "
|
|
"subquery object.",
|
|
version="1.4",
|
|
)
|
|
return resolved._implicit_subquery
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
|
|
class AnonymizedFromClauseImpl(StrictFromClauseImpl):
|
|
__slots__ = ()
|
|
|
|
def _post_coercion(self, element, flat=False, name=None, **kw):
|
|
assert name is None
|
|
|
|
return element._anonymous_fromclause(flat=flat)
|
|
|
|
|
|
class DMLTableImpl(_SelectIsNotFrom, _NoTextCoercion, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _post_coercion(self, element, **kw):
|
|
if "dml_table" in element._annotations:
|
|
return element._annotations["dml_table"]
|
|
else:
|
|
return element
|
|
|
|
|
|
class DMLSelectImpl(_NoTextCoercion, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _implicit_coercions(
|
|
self,
|
|
element: Any,
|
|
resolved: Any,
|
|
argname: Optional[str] = None,
|
|
**kw: Any,
|
|
) -> Any:
|
|
if resolved._is_from_clause:
|
|
if (
|
|
isinstance(resolved, selectable.Alias)
|
|
and resolved.element._is_select_base
|
|
):
|
|
return resolved.element
|
|
else:
|
|
return resolved.select()
|
|
else:
|
|
self._raise_for_expected(element, argname, resolved)
|
|
|
|
|
|
class CompoundElementImpl(_NoTextCoercion, RoleImpl):
|
|
__slots__ = ()
|
|
|
|
def _raise_for_expected(self, element, argname=None, resolved=None, **kw):
|
|
if isinstance(element, roles.FromClauseRole):
|
|
if element._is_subquery:
|
|
advice = (
|
|
"Use the plain select() object without "
|
|
"calling .subquery() or .alias()."
|
|
)
|
|
else:
|
|
advice = (
|
|
"To SELECT from any FROM clause, use the .select() method."
|
|
)
|
|
else:
|
|
advice = None
|
|
return super()._raise_for_expected(
|
|
element, argname=argname, resolved=resolved, advice=advice, **kw
|
|
)
|
|
|
|
|
|
_impl_lookup = {}
|
|
|
|
|
|
for name in dir(roles):
|
|
cls = getattr(roles, name)
|
|
if name.endswith("Role"):
|
|
name = name.replace("Role", "Impl")
|
|
if name in globals():
|
|
impl = globals()[name](cls)
|
|
_impl_lookup[cls] = impl
|
|
|
|
if not TYPE_CHECKING:
|
|
ee_impl = _impl_lookup[roles.ExpressionElementRole]
|
|
|
|
for py_type in (int, bool, str, float):
|
|
_impl_lookup[roles.ExpressionElementRole[py_type]] = ee_impl
|