Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/sql/coercions.py: 50%
493 statements
« prev ^ index » next coverage.py v7.0.1, created at 2022-12-25 06:11 +0000
« prev ^ index » next coverage.py v7.0.1, created at 2022-12-25 06:11 +0000
1# sql/coercions.py
2# Copyright (C) 2005-2022 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
8import numbers
9import re
11from . import operators
12from . import roles
13from . import visitors
14from .base import ExecutableOption
15from .base import Options
16from .traversals import HasCacheKey
17from .visitors import Visitable
18from .. import exc
19from .. import inspection
20from .. import util
21from ..util import collections_abc
24elements = None
25lambdas = None
26schema = None
27selectable = None
28sqltypes = None
29traversals = None
32def _is_literal(element):
33 """Return whether or not the element is a "literal" in the context
34 of a SQL expression construct.
36 """
38 return (
39 not isinstance(
40 element,
41 (Visitable, schema.SchemaEventTarget),
42 )
43 and not hasattr(element, "__clause_element__")
44 )
47def _deep_is_literal(element):
48 """Return whether or not the element is a "literal" in the context
49 of a SQL expression construct.
51 does a deeper more esoteric check than _is_literal. is used
52 for lambda elements that have to distinguish values that would
53 be bound vs. not without any context.
55 """
57 if isinstance(element, collections_abc.Sequence) and not isinstance(
58 element, str
59 ):
60 for elem in element:
61 if not _deep_is_literal(elem):
62 return False
63 else:
64 return True
66 return (
67 not isinstance(
68 element,
69 (
70 Visitable,
71 schema.SchemaEventTarget,
72 HasCacheKey,
73 Options,
74 util.langhelpers._symbol,
75 ),
76 )
77 and not hasattr(element, "__clause_element__")
78 and (
79 not isinstance(element, type)
80 or not issubclass(element, HasCacheKey)
81 )
82 )
85def _document_text_coercion(paramname, meth_rst, param_rst):
86 return util.add_parameter_text(
87 paramname,
88 (
89 ".. warning:: "
90 "The %s argument to %s can be passed as a Python string argument, "
91 "which will be treated "
92 "as **trusted SQL text** and rendered as given. **DO NOT PASS "
93 "UNTRUSTED INPUT TO THIS PARAMETER**."
94 )
95 % (param_rst, meth_rst),
96 )
99def _expression_collection_was_a_list(attrname, fnname, args):
100 if args and isinstance(args[0], (list, set, dict)) and len(args) == 1:
101 if isinstance(args[0], list):
102 util.warn_deprecated_20(
103 'The "%s" argument to %s(), when referring to a sequence '
104 "of items, is now passed as a series of positional "
105 "elements, rather than as a list. " % (attrname, fnname)
106 )
107 return args[0]
108 else:
109 return args
112def expect(
113 role,
114 element,
115 apply_propagate_attrs=None,
116 argname=None,
117 post_inspect=False,
118 **kw
119):
120 if (
121 role.allows_lambda
122 # note callable() will not invoke a __getattr__() method, whereas
123 # hasattr(obj, "__call__") will. by keeping the callable() check here
124 # we prevent most needless calls to hasattr() and therefore
125 # __getattr__(), which is present on ColumnElement.
126 and callable(element)
127 and hasattr(element, "__code__")
128 ):
129 return lambdas.LambdaElement(
130 element,
131 role,
132 lambdas.LambdaOptions(**kw),
133 apply_propagate_attrs=apply_propagate_attrs,
134 )
136 # major case is that we are given a ClauseElement already, skip more
137 # elaborate logic up front if possible
138 impl = _impl_lookup[role]
140 original_element = element
142 if not isinstance(
143 element,
144 (elements.ClauseElement, schema.SchemaItem, schema.FetchedValue),
145 ):
146 resolved = None
148 if impl._resolve_literal_only:
149 resolved = impl._literal_coercion(element, **kw)
150 else:
152 original_element = element
154 is_clause_element = False
156 # this is a special performance optimization for ORM
157 # joins used by JoinTargetImpl that we don't go through the
158 # work of creating __clause_element__() when we only need the
159 # original QueryableAttribute, as the former will do clause
160 # adaption and all that which is just thrown away here.
161 if (
162 impl._skip_clauseelement_for_target_match
163 and isinstance(element, role)
164 and hasattr(element, "__clause_element__")
165 ):
166 is_clause_element = True
167 else:
168 while hasattr(element, "__clause_element__"):
169 is_clause_element = True
171 if not getattr(element, "is_clause_element", False):
172 element = element.__clause_element__()
173 else:
174 break
176 if not is_clause_element:
177 if impl._use_inspection:
178 insp = inspection.inspect(element, raiseerr=False)
179 if insp is not None:
180 if post_inspect:
181 insp._post_inspect
182 try:
183 resolved = insp.__clause_element__()
184 except AttributeError:
185 impl._raise_for_expected(original_element, argname)
187 if resolved is None:
188 resolved = impl._literal_coercion(
189 element, argname=argname, **kw
190 )
191 else:
192 resolved = element
193 else:
194 resolved = element
195 if (
196 apply_propagate_attrs is not None
197 and not apply_propagate_attrs._propagate_attrs
198 and resolved._propagate_attrs
199 ):
200 apply_propagate_attrs._propagate_attrs = resolved._propagate_attrs
202 if impl._role_class in resolved.__class__.__mro__:
203 if impl._post_coercion:
204 resolved = impl._post_coercion(
205 resolved,
206 argname=argname,
207 original_element=original_element,
208 **kw
209 )
210 return resolved
211 else:
212 return impl._implicit_coercions(
213 original_element, resolved, argname=argname, **kw
214 )
217def expect_as_key(role, element, **kw):
218 kw["as_key"] = True
219 return expect(role, element, **kw)
222def expect_col_expression_collection(role, expressions):
223 for expr in expressions:
224 strname = None
225 column = None
227 resolved = expect(role, expr)
228 if isinstance(resolved, util.string_types):
229 strname = resolved = expr
230 else:
231 cols = []
232 visitors.traverse(resolved, {}, {"column": cols.append})
233 if cols:
234 column = cols[0]
235 add_element = column if column is not None else strname
236 yield resolved, column, strname, add_element
239class RoleImpl(object):
240 __slots__ = ("_role_class", "name", "_use_inspection")
242 def _literal_coercion(self, element, **kw):
243 raise NotImplementedError()
245 _post_coercion = None
246 _resolve_literal_only = False
247 _skip_clauseelement_for_target_match = False
249 def __init__(self, role_class):
250 self._role_class = role_class
251 self.name = role_class._role_name
252 self._use_inspection = issubclass(role_class, roles.UsesInspection)
254 def _implicit_coercions(self, element, resolved, argname=None, **kw):
255 self._raise_for_expected(element, argname, resolved)
257 def _raise_for_expected(
258 self,
259 element,
260 argname=None,
261 resolved=None,
262 advice=None,
263 code=None,
264 err=None,
265 ):
266 if resolved is not None and resolved is not element:
267 got = "%r object resolved from %r object" % (resolved, element)
268 else:
269 got = repr(element)
271 if argname:
272 msg = "%s expected for argument %r; got %s." % (
273 self.name,
274 argname,
275 got,
276 )
277 else:
278 msg = "%s expected, got %s." % (self.name, got)
280 if advice:
281 msg += " " + advice
283 util.raise_(exc.ArgumentError(msg, code=code), replace_context=err)
286class _Deannotate(object):
287 __slots__ = ()
289 def _post_coercion(self, resolved, **kw):
290 from .util import _deep_deannotate
292 return _deep_deannotate(resolved)
295class _StringOnly(object):
296 __slots__ = ()
298 _resolve_literal_only = True
301class _ReturnsStringKey(object):
302 __slots__ = ()
304 def _implicit_coercions(
305 self, original_element, resolved, argname=None, **kw
306 ):
307 if isinstance(original_element, util.string_types):
308 return original_element
309 else:
310 self._raise_for_expected(original_element, argname, resolved)
312 def _literal_coercion(self, element, **kw):
313 return element
316class _ColumnCoercions(object):
317 __slots__ = ()
319 def _warn_for_scalar_subquery_coercion(self):
320 util.warn(
321 "implicitly coercing SELECT object to scalar subquery; "
322 "please use the .scalar_subquery() method to produce a scalar "
323 "subquery.",
324 )
326 def _implicit_coercions(
327 self, original_element, resolved, argname=None, **kw
328 ):
329 if not getattr(resolved, "is_clause_element", False):
330 self._raise_for_expected(original_element, argname, resolved)
331 elif resolved._is_select_statement:
332 self._warn_for_scalar_subquery_coercion()
333 return resolved.scalar_subquery()
334 elif resolved._is_from_clause and isinstance(
335 resolved, selectable.Subquery
336 ):
337 self._warn_for_scalar_subquery_coercion()
338 return resolved.element.scalar_subquery()
339 elif self._role_class.allows_lambda and resolved._is_lambda_element:
340 return resolved
341 else:
342 self._raise_for_expected(original_element, argname, resolved)
345def _no_text_coercion(
346 element, argname=None, exc_cls=exc.ArgumentError, extra=None, err=None
347):
348 util.raise_(
349 exc_cls(
350 "%(extra)sTextual SQL expression %(expr)r %(argname)sshould be "
351 "explicitly declared as text(%(expr)r)"
352 % {
353 "expr": util.ellipses_string(element),
354 "argname": "for argument %s" % (argname,) if argname else "",
355 "extra": "%s " % extra if extra else "",
356 }
357 ),
358 replace_context=err,
359 )
362class _NoTextCoercion(object):
363 __slots__ = ()
365 def _literal_coercion(self, element, argname=None, **kw):
366 if isinstance(element, util.string_types) and issubclass(
367 elements.TextClause, self._role_class
368 ):
369 _no_text_coercion(element, argname)
370 else:
371 self._raise_for_expected(element, argname)
374class _CoerceLiterals(object):
375 __slots__ = ()
376 _coerce_consts = False
377 _coerce_star = False
378 _coerce_numerics = False
380 def _text_coercion(self, element, argname=None):
381 return _no_text_coercion(element, argname)
383 def _literal_coercion(self, element, argname=None, **kw):
384 if isinstance(element, util.string_types):
385 if self._coerce_star and element == "*":
386 return elements.ColumnClause("*", is_literal=True)
387 else:
388 return self._text_coercion(element, argname, **kw)
390 if self._coerce_consts:
391 if element is None:
392 return elements.Null()
393 elif element is False:
394 return elements.False_()
395 elif element is True:
396 return elements.True_()
398 if self._coerce_numerics and isinstance(element, (numbers.Number)):
399 return elements.ColumnClause(str(element), is_literal=True)
401 self._raise_for_expected(element, argname)
404class LiteralValueImpl(RoleImpl):
405 _resolve_literal_only = True
407 def _implicit_coercions(
408 self, element, resolved, argname, type_=None, **kw
409 ):
410 if not _is_literal(resolved):
411 self._raise_for_expected(
412 element, resolved=resolved, argname=argname, **kw
413 )
415 return elements.BindParameter(None, element, type_=type_, unique=True)
417 def _literal_coercion(self, element, argname=None, type_=None, **kw):
418 return element
421class _SelectIsNotFrom(object):
422 __slots__ = ()
424 def _raise_for_expected(self, element, argname=None, resolved=None, **kw):
425 if isinstance(element, roles.SelectStatementRole) or isinstance(
426 resolved, roles.SelectStatementRole
427 ):
428 advice = (
429 "To create a "
430 "FROM clause from a %s object, use the .subquery() method."
431 % (resolved.__class__ if resolved is not None else element,)
432 )
433 code = "89ve"
434 else:
435 advice = code = None
437 return super(_SelectIsNotFrom, self)._raise_for_expected(
438 element,
439 argname=argname,
440 resolved=resolved,
441 advice=advice,
442 code=code,
443 **kw
444 )
447class HasCacheKeyImpl(RoleImpl):
448 __slots__ = ()
450 def _implicit_coercions(
451 self, original_element, resolved, argname=None, **kw
452 ):
453 if isinstance(original_element, traversals.HasCacheKey):
454 return original_element
455 else:
456 self._raise_for_expected(original_element, argname, resolved)
458 def _literal_coercion(self, element, **kw):
459 return element
462class ExecutableOptionImpl(RoleImpl):
463 __slots__ = ()
465 def _implicit_coercions(
466 self, original_element, resolved, argname=None, **kw
467 ):
468 if isinstance(original_element, ExecutableOption):
469 return original_element
470 else:
471 self._raise_for_expected(original_element, argname, resolved)
473 def _literal_coercion(self, element, **kw):
474 return element
477class ExpressionElementImpl(_ColumnCoercions, RoleImpl):
478 __slots__ = ()
480 def _literal_coercion(
481 self, element, name=None, type_=None, argname=None, is_crud=False, **kw
482 ):
483 if (
484 element is None
485 and not is_crud
486 and (type_ is None or not type_.should_evaluate_none)
487 ):
488 # TODO: there's no test coverage now for the
489 # "should_evaluate_none" part of this, as outside of "crud" this
490 # codepath is not normally used except in some special cases
491 return elements.Null()
492 else:
493 try:
494 return elements.BindParameter(
495 name, element, type_, unique=True, _is_crud=is_crud
496 )
497 except exc.ArgumentError as err:
498 self._raise_for_expected(element, err=err)
500 def _raise_for_expected(self, element, argname=None, resolved=None, **kw):
501 if isinstance(element, roles.AnonymizedFromClauseRole):
502 advice = (
503 "To create a "
504 "column expression from a FROM clause row "
505 "as a whole, use the .table_valued() method."
506 )
507 else:
508 advice = None
510 return super(ExpressionElementImpl, self)._raise_for_expected(
511 element, argname=argname, resolved=resolved, advice=advice, **kw
512 )
515class BinaryElementImpl(ExpressionElementImpl, RoleImpl):
517 __slots__ = ()
519 def _literal_coercion(
520 self, element, expr, operator, bindparam_type=None, argname=None, **kw
521 ):
522 try:
523 return expr._bind_param(operator, element, type_=bindparam_type)
524 except exc.ArgumentError as err:
525 self._raise_for_expected(element, err=err)
527 def _post_coercion(self, resolved, expr, bindparam_type=None, **kw):
528 if resolved.type._isnull and not expr.type._isnull:
529 resolved = resolved._with_binary_element_type(
530 bindparam_type if bindparam_type is not None else expr.type
531 )
532 return resolved
535class InElementImpl(RoleImpl):
536 __slots__ = ()
538 def _implicit_coercions(
539 self, original_element, resolved, argname=None, **kw
540 ):
541 if resolved._is_from_clause:
542 if (
543 isinstance(resolved, selectable.Alias)
544 and resolved.element._is_select_statement
545 ):
546 self._warn_for_implicit_coercion(resolved)
547 return self._post_coercion(resolved.element, **kw)
548 else:
549 self._warn_for_implicit_coercion(resolved)
550 return self._post_coercion(resolved.select(), **kw)
551 else:
552 self._raise_for_expected(original_element, argname, resolved)
554 def _warn_for_implicit_coercion(self, elem):
555 util.warn(
556 "Coercing %s object into a select() for use in IN(); "
557 "please pass a select() construct explicitly"
558 % (elem.__class__.__name__)
559 )
561 def _literal_coercion(self, element, expr, operator, **kw):
562 if isinstance(element, collections_abc.Iterable) and not isinstance(
563 element, util.string_types
564 ):
565 non_literal_expressions = {}
566 element = list(element)
567 for o in element:
568 if not _is_literal(o):
569 if not isinstance(o, operators.ColumnOperators):
570 self._raise_for_expected(element, **kw)
571 else:
572 non_literal_expressions[o] = o
573 elif o is None:
574 non_literal_expressions[o] = elements.Null()
576 if non_literal_expressions:
577 return elements.ClauseList(
578 *[
579 non_literal_expressions[o]
580 if o in non_literal_expressions
581 else expr._bind_param(operator, o)
582 for o in element
583 ]
584 )
585 else:
586 return expr._bind_param(operator, element, expanding=True)
588 else:
589 self._raise_for_expected(element, **kw)
591 def _post_coercion(self, element, expr, operator, **kw):
592 if element._is_select_statement:
593 # for IN, we are doing scalar_subquery() coercion without
594 # a warning
595 return element.scalar_subquery()
596 elif isinstance(element, elements.ClauseList):
597 assert not len(element.clauses) == 0
598 return element.self_group(against=operator)
600 elif isinstance(element, elements.BindParameter):
601 element = element._clone(maintain_key=True)
602 element.expanding = True
603 element.expand_op = operator
605 return element
606 else:
607 return element
610class OnClauseImpl(_CoerceLiterals, _ColumnCoercions, RoleImpl):
611 __slots__ = ()
613 _coerce_consts = True
615 def _implicit_coercions(
616 self, original_element, resolved, argname=None, legacy=False, **kw
617 ):
618 if legacy and isinstance(resolved, str):
619 return resolved
620 else:
621 return super(OnClauseImpl, self)._implicit_coercions(
622 original_element,
623 resolved,
624 argname=argname,
625 legacy=legacy,
626 **kw
627 )
629 def _text_coercion(self, element, argname=None, legacy=False):
630 if legacy and isinstance(element, str):
631 util.warn_deprecated_20(
632 "Using strings to indicate relationship names in "
633 "Query.join() is deprecated and will be removed in "
634 "SQLAlchemy 2.0. Please use the class-bound attribute "
635 "directly."
636 )
637 return element
639 return super(OnClauseImpl, self)._text_coercion(element, argname)
641 def _post_coercion(self, resolved, original_element=None, **kw):
642 # this is a hack right now as we want to use coercion on an
643 # ORM InstrumentedAttribute, but we want to return the object
644 # itself if it is one, not its clause element.
645 # ORM context _join and _legacy_join() would need to be improved
646 # to look for annotations in a clause element form.
647 if isinstance(original_element, roles.JoinTargetRole):
648 return original_element
649 return resolved
652class WhereHavingImpl(_CoerceLiterals, _ColumnCoercions, RoleImpl):
653 __slots__ = ()
655 _coerce_consts = True
657 def _text_coercion(self, element, argname=None):
658 return _no_text_coercion(element, argname)
661class StatementOptionImpl(_CoerceLiterals, RoleImpl):
662 __slots__ = ()
664 _coerce_consts = True
666 def _text_coercion(self, element, argname=None):
667 return elements.TextClause(element)
670class ColumnArgumentImpl(_NoTextCoercion, RoleImpl):
671 __slots__ = ()
674class ColumnArgumentOrKeyImpl(_ReturnsStringKey, RoleImpl):
675 __slots__ = ()
678class StrAsPlainColumnImpl(_CoerceLiterals, RoleImpl):
679 __slots__ = ()
681 def _text_coercion(self, element, argname=None):
682 return elements.ColumnClause(element)
685class ByOfImpl(_CoerceLiterals, _ColumnCoercions, RoleImpl, roles.ByOfRole):
687 __slots__ = ()
689 _coerce_consts = True
691 def _text_coercion(self, element, argname=None):
692 return elements._textual_label_reference(element)
695class OrderByImpl(ByOfImpl, RoleImpl):
696 __slots__ = ()
698 def _post_coercion(self, resolved, **kw):
699 if (
700 isinstance(resolved, self._role_class)
701 and resolved._order_by_label_element is not None
702 ):
703 return elements._label_reference(resolved)
704 else:
705 return resolved
708class GroupByImpl(ByOfImpl, RoleImpl):
709 __slots__ = ()
711 def _implicit_coercions(
712 self, original_element, resolved, argname=None, **kw
713 ):
714 if isinstance(resolved, roles.StrictFromClauseRole):
715 return elements.ClauseList(*resolved.c)
716 else:
717 return resolved
720class DMLColumnImpl(_ReturnsStringKey, RoleImpl):
721 __slots__ = ()
723 def _post_coercion(self, element, as_key=False, **kw):
724 if as_key:
725 return element.key
726 else:
727 return element
730class ConstExprImpl(RoleImpl):
731 __slots__ = ()
733 def _literal_coercion(self, element, argname=None, **kw):
734 if element is None:
735 return elements.Null()
736 elif element is False:
737 return elements.False_()
738 elif element is True:
739 return elements.True_()
740 else:
741 self._raise_for_expected(element, argname)
744class TruncatedLabelImpl(_StringOnly, RoleImpl):
745 __slots__ = ()
747 def _implicit_coercions(
748 self, original_element, resolved, argname=None, **kw
749 ):
750 if isinstance(original_element, util.string_types):
751 return resolved
752 else:
753 self._raise_for_expected(original_element, argname, resolved)
755 def _literal_coercion(self, element, argname=None, **kw):
756 """coerce the given value to :class:`._truncated_label`.
758 Existing :class:`._truncated_label` and
759 :class:`._anonymous_label` objects are passed
760 unchanged.
761 """
763 if isinstance(element, elements._truncated_label):
764 return element
765 else:
766 return elements._truncated_label(element)
769class DDLExpressionImpl(_Deannotate, _CoerceLiterals, RoleImpl):
771 __slots__ = ()
773 _coerce_consts = True
775 def _text_coercion(self, element, argname=None):
776 # see #5754 for why we can't easily deprecate this coercion.
777 # essentially expressions like postgresql_where would have to be
778 # text() as they come back from reflection and we don't want to
779 # have text() elements wired into the inspection dictionaries.
780 return elements.TextClause(element)
783class DDLConstraintColumnImpl(_Deannotate, _ReturnsStringKey, RoleImpl):
784 __slots__ = ()
787class DDLReferredColumnImpl(DDLConstraintColumnImpl):
788 __slots__ = ()
791class LimitOffsetImpl(RoleImpl):
792 __slots__ = ()
794 def _implicit_coercions(self, element, resolved, argname=None, **kw):
795 if resolved is None:
796 return None
797 else:
798 self._raise_for_expected(element, argname, resolved)
800 def _literal_coercion(self, element, name, type_, **kw):
801 if element is None:
802 return None
803 else:
804 value = util.asint(element)
805 return selectable._OffsetLimitParam(
806 name, value, type_=type_, unique=True
807 )
810class LabeledColumnExprImpl(ExpressionElementImpl):
811 __slots__ = ()
813 def _implicit_coercions(
814 self, original_element, resolved, argname=None, **kw
815 ):
816 if isinstance(resolved, roles.ExpressionElementRole):
817 return resolved.label(None)
818 else:
819 new = super(LabeledColumnExprImpl, self)._implicit_coercions(
820 original_element, resolved, argname=argname, **kw
821 )
822 if isinstance(new, roles.ExpressionElementRole):
823 return new.label(None)
824 else:
825 self._raise_for_expected(original_element, argname, resolved)
828class ColumnsClauseImpl(_SelectIsNotFrom, _CoerceLiterals, RoleImpl):
829 __slots__ = ()
831 _coerce_consts = True
832 _coerce_numerics = True
833 _coerce_star = True
835 _guess_straight_column = re.compile(r"^\w\S*$", re.I)
837 def _text_coercion(self, element, argname=None):
838 element = str(element)
840 guess_is_literal = not self._guess_straight_column.match(element)
841 raise exc.ArgumentError(
842 "Textual column expression %(column)r %(argname)sshould be "
843 "explicitly declared with text(%(column)r), "
844 "or use %(literal_column)s(%(column)r) "
845 "for more specificity"
846 % {
847 "column": util.ellipses_string(element),
848 "argname": "for argument %s" % (argname,) if argname else "",
849 "literal_column": "literal_column"
850 if guess_is_literal
851 else "column",
852 }
853 )
856class ReturnsRowsImpl(RoleImpl):
857 __slots__ = ()
860class StatementImpl(_CoerceLiterals, RoleImpl):
861 __slots__ = ()
863 def _post_coercion(self, resolved, original_element, argname=None, **kw):
864 if resolved is not original_element and not isinstance(
865 original_element, util.string_types
866 ):
867 # use same method as Connection uses; this will later raise
868 # ObjectNotExecutableError
869 try:
870 original_element._execute_on_connection
871 except AttributeError:
872 util.warn_deprecated(
873 "Object %r should not be used directly in a SQL statement "
874 "context, such as passing to methods such as "
875 "session.execute(). This usage will be disallowed in a "
876 "future release. "
877 "Please use Core select() / update() / delete() etc. "
878 "with Session.execute() and other statement execution "
879 "methods." % original_element,
880 "1.4",
881 )
883 return resolved
885 def _implicit_coercions(
886 self, original_element, resolved, argname=None, **kw
887 ):
888 if resolved._is_lambda_element:
889 return resolved
890 else:
891 return super(StatementImpl, self)._implicit_coercions(
892 original_element, resolved, argname=argname, **kw
893 )
895 def _text_coercion(self, element, argname=None):
896 util.warn_deprecated_20(
897 "Using plain strings to indicate SQL statements without using "
898 "the text() construct is "
899 "deprecated and will be removed in version 2.0. Ensure plain "
900 "SQL statements are passed using the text() construct."
901 )
902 return elements.TextClause(element)
905class SelectStatementImpl(_NoTextCoercion, RoleImpl):
906 __slots__ = ()
908 def _implicit_coercions(
909 self, original_element, resolved, argname=None, **kw
910 ):
911 if resolved._is_text_clause:
912 return resolved.columns()
913 else:
914 self._raise_for_expected(original_element, argname, resolved)
917class HasCTEImpl(ReturnsRowsImpl):
918 __slots__ = ()
921class IsCTEImpl(RoleImpl):
922 __slots__ = ()
925class JoinTargetImpl(RoleImpl):
926 __slots__ = ()
928 _skip_clauseelement_for_target_match = True
930 def _literal_coercion(self, element, legacy=False, **kw):
931 if isinstance(element, str):
932 return element
934 def _implicit_coercions(
935 self, original_element, resolved, argname=None, legacy=False, **kw
936 ):
937 if isinstance(original_element, roles.JoinTargetRole):
938 # note that this codepath no longer occurs as of
939 # #6550, unless JoinTargetImpl._skip_clauseelement_for_target_match
940 # were set to False.
941 return original_element
942 elif legacy and isinstance(resolved, str):
943 util.warn_deprecated_20(
944 "Using strings to indicate relationship names in "
945 "Query.join() is deprecated and will be removed in "
946 "SQLAlchemy 2.0. Please use the class-bound attribute "
947 "directly."
948 )
949 return resolved
950 elif legacy and isinstance(resolved, roles.WhereHavingRole):
951 return resolved
952 elif legacy and resolved._is_select_statement:
953 util.warn_deprecated(
954 "Implicit coercion of SELECT and textual SELECT "
955 "constructs into FROM clauses is deprecated; please call "
956 ".subquery() on any Core select or ORM Query object in "
957 "order to produce a subquery object.",
958 version="1.4",
959 )
960 # TODO: doing _implicit_subquery here causes tests to fail,
961 # how was this working before? probably that ORM
962 # join logic treated it as a select and subquery would happen
963 # in _ORMJoin->Join
964 return resolved
965 else:
966 self._raise_for_expected(original_element, argname, resolved)
969class FromClauseImpl(_SelectIsNotFrom, _NoTextCoercion, RoleImpl):
970 __slots__ = ()
972 def _implicit_coercions(
973 self,
974 original_element,
975 resolved,
976 argname=None,
977 explicit_subquery=False,
978 allow_select=True,
979 **kw
980 ):
981 if resolved._is_select_statement:
982 if explicit_subquery:
983 return resolved.subquery()
984 elif allow_select:
985 util.warn_deprecated(
986 "Implicit coercion of SELECT and textual SELECT "
987 "constructs into FROM clauses is deprecated; please call "
988 ".subquery() on any Core select or ORM Query object in "
989 "order to produce a subquery object.",
990 version="1.4",
991 )
992 return resolved._implicit_subquery
993 elif resolved._is_text_clause:
994 return resolved
995 else:
996 self._raise_for_expected(original_element, argname, resolved)
998 def _post_coercion(self, element, deannotate=False, **kw):
999 if deannotate:
1000 return element._deannotate()
1001 else:
1002 return element
1005class StrictFromClauseImpl(FromClauseImpl):
1006 __slots__ = ()
1008 def _implicit_coercions(
1009 self,
1010 original_element,
1011 resolved,
1012 argname=None,
1013 allow_select=False,
1014 **kw
1015 ):
1016 if resolved._is_select_statement and allow_select:
1017 util.warn_deprecated(
1018 "Implicit coercion of SELECT and textual SELECT constructs "
1019 "into FROM clauses is deprecated; please call .subquery() "
1020 "on any Core select or ORM Query object in order to produce a "
1021 "subquery object.",
1022 version="1.4",
1023 )
1024 return resolved._implicit_subquery
1025 else:
1026 self._raise_for_expected(original_element, argname, resolved)
1029class AnonymizedFromClauseImpl(StrictFromClauseImpl):
1030 __slots__ = ()
1032 def _post_coercion(self, element, flat=False, name=None, **kw):
1033 assert name is None
1035 return element._anonymous_fromclause(flat=flat)
1038class DMLTableImpl(_SelectIsNotFrom, _NoTextCoercion, RoleImpl):
1039 __slots__ = ()
1041 def _post_coercion(self, element, **kw):
1042 if "dml_table" in element._annotations:
1043 return element._annotations["dml_table"]
1044 else:
1045 return element
1048class DMLSelectImpl(_NoTextCoercion, RoleImpl):
1049 __slots__ = ()
1051 def _implicit_coercions(
1052 self, original_element, resolved, argname=None, **kw
1053 ):
1054 if resolved._is_from_clause:
1055 if (
1056 isinstance(resolved, selectable.Alias)
1057 and resolved.element._is_select_statement
1058 ):
1059 return resolved.element
1060 else:
1061 return resolved.select()
1062 else:
1063 self._raise_for_expected(original_element, argname, resolved)
1066class CompoundElementImpl(_NoTextCoercion, RoleImpl):
1067 __slots__ = ()
1069 def _raise_for_expected(self, element, argname=None, resolved=None, **kw):
1070 if isinstance(element, roles.FromClauseRole):
1071 if element._is_subquery:
1072 advice = (
1073 "Use the plain select() object without "
1074 "calling .subquery() or .alias()."
1075 )
1076 else:
1077 advice = (
1078 "To SELECT from any FROM clause, use the .select() method."
1079 )
1080 else:
1081 advice = None
1082 return super(CompoundElementImpl, self)._raise_for_expected(
1083 element, argname=argname, resolved=resolved, advice=advice, **kw
1084 )
1087_impl_lookup = {}
1090for name in dir(roles):
1091 cls = getattr(roles, name)
1092 if name.endswith("Role"):
1093 name = name.replace("Role", "Impl")
1094 if name in globals():
1095 impl = globals()[name](cls)
1096 _impl_lookup[cls] = impl