OwlCyberSecurity - MANAGER

Edit File: default_comparator.py

# sql/default_comparator.py
# Copyright (C) 2005-2021 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

"""Default implementation of SQL comparison operations.
"""

from . import operators
from . import type_api
from .elements import _clause_element_as_expr
from .elements import _const_expr
from .elements import _is_literal
from .elements import _literal_as_text
from .elements import and_
from .elements import BinaryExpression
from .elements import BindParameter
from .elements import ClauseElement
from .elements import ClauseList
from .elements import collate
from .elements import CollectionAggregate
from .elements import ColumnElement
from .elements import False_
from .elements import Null
from .elements import or_
from .elements import TextClause
from .elements import True_
from .elements import Tuple
from .elements import UnaryExpression
from .elements import Visitable
from .selectable import Alias
from .selectable import ScalarSelect
from .selectable import Selectable
from .selectable import SelectBase
from .. import exc
from .. import util

def _boolean_compare(
    expr,
    op,
    obj,
    negate=None,
    reverse=False,
    _python_is_types=(util.NoneType, bool),
    result_type=None,
    **kwargs
):

if result_type is None:
        result_type = type_api.BOOLEANTYPE

if isinstance(obj, _python_is_types + (Null, True_, False_)):

# allow x ==/!= True/False to be treated as a literal.
        # this comes out to "== / != true/false" or "1/0" if those
        # constants aren't supported and works on all platforms
        if op in (operators.eq, operators.ne) and isinstance(
            obj, (bool, True_, False_)
        ):
            return BinaryExpression(
                expr,
                _literal_as_text(obj),
                op,
                type_=result_type,
                negate=negate,
                modifiers=kwargs,
            )
        elif op in (operators.is_distinct_from, operators.isnot_distinct_from):
            return BinaryExpression(
                expr,
                _literal_as_text(obj),
                op,
                type_=result_type,
                negate=negate,
                modifiers=kwargs,
            )
        else:
            # all other None/True/False uses IS, IS NOT
            if op in (operators.eq, operators.is_):
                return BinaryExpression(
                    expr,
                    _const_expr(obj),
                    operators.is_,
                    negate=operators.isnot,
                    type_=result_type,
                )
            elif op in (operators.ne, operators.isnot):
                return BinaryExpression(
                    expr,
                    _const_expr(obj),
                    operators.isnot,
                    negate=operators.is_,
                    type_=result_type,
                )
            else:
                raise exc.ArgumentError(
                    "Only '=', '!=', 'is_()', 'isnot()', "
                    "'is_distinct_from()', 'isnot_distinct_from()' "
                    "operators can be used with None/True/False"
                )
    else:
        obj = _check_literal(expr, op, obj)

if reverse:
        return BinaryExpression(
            obj, expr, op, type_=result_type, negate=negate, modifiers=kwargs
        )
    else:
        return BinaryExpression(
            expr, obj, op, type_=result_type, negate=negate, modifiers=kwargs
        )

def _custom_op_operate(expr, op, obj, reverse=False, result_type=None, **kw):
    if result_type is None:
        if op.return_type:
            result_type = op.return_type
        elif op.is_comparison:
            result_type = type_api.BOOLEANTYPE

return _binary_operate(
        expr, op, obj, reverse=reverse, result_type=result_type, **kw
    )

def _binary_operate(expr, op, obj, reverse=False, result_type=None, **kw):
    obj = _check_literal(expr, op, obj)

if reverse:
        left, right = obj, expr
    else:
        left, right = expr, obj

if result_type is None:
        op, result_type = left.comparator._adapt_expression(
            op, right.comparator
        )

return BinaryExpression(left, right, op, type_=result_type, modifiers=kw)

def _conjunction_operate(expr, op, other, **kw):
    if op is operators.and_:
        return and_(expr, other)
    elif op is operators.or_:
        return or_(expr, other)
    else:
        raise NotImplementedError()

def _scalar(expr, op, fn, **kw):
    return fn(expr)

def _in_impl(expr, op, seq_or_selectable, negate_op, **kw):
    seq_or_selectable = _clause_element_as_expr(seq_or_selectable)

if isinstance(seq_or_selectable, ScalarSelect):
        return _boolean_compare(expr, op, seq_or_selectable, negate=negate_op)
    elif isinstance(seq_or_selectable, SelectBase):

# TODO: if we ever want to support (x, y, z) IN (select x,
        # y, z from table), we would need a multi-column version of
        # as_scalar() to produce a multi- column selectable that
        # does not export itself as a FROM clause

return _boolean_compare(
            expr, op, seq_or_selectable.as_scalar(), negate=negate_op, **kw
        )
    elif isinstance(seq_or_selectable, (Selectable, TextClause)):
        return _boolean_compare(
            expr, op, seq_or_selectable, negate=negate_op, **kw
        )
    elif isinstance(seq_or_selectable, ClauseElement):
        if (
            isinstance(seq_or_selectable, BindParameter)
            and seq_or_selectable.expanding
        ):

if isinstance(expr, Tuple):
                seq_or_selectable = seq_or_selectable._with_expanding_in_types(
                    [elem.type for elem in expr]
                )

return _boolean_compare(
                expr, op, seq_or_selectable, negate=negate_op
            )
        else:
            raise exc.InvalidRequestError(
                "in_() accepts"
                " either a list of expressions, "
                'a selectable, or an "expanding" bound parameter: %r'
                % seq_or_selectable
            )

# Handle non selectable arguments as sequences
    args = []
    for o in seq_or_selectable:
        if not _is_literal(o):
            if not isinstance(o, operators.ColumnOperators):
                raise exc.InvalidRequestError(
                    "in_() accepts"
                    " either a list of expressions, "
                    'a selectable, or an "expanding" bound parameter: %r' % o
                )
        elif o is None:
            o = Null()
        else:
            o = expr._bind_param(op, o)
        args.append(o)

if len(args) == 0:
        op, negate_op = (
            (operators.empty_in_op, operators.empty_notin_op)
            if op is operators.in_op
            else (operators.empty_notin_op, operators.empty_in_op)
        )

return _boolean_compare(
        expr,
        op,
        ClauseList(_tuple_values=isinstance(expr, Tuple), *args).self_group(
            against=op
        ),
        negate=negate_op,
    )

def _getitem_impl(expr, op, other, **kw):
    if isinstance(expr.type, type_api.INDEXABLE):
        other = _check_literal(expr, op, other)
        return _binary_operate(expr, op, other, **kw)
    else:
        _unsupported_impl(expr, op, other, **kw)

def _unsupported_impl(expr, op, *arg, **kw):
    raise NotImplementedError(
        "Operator '%s' is not supported on " "this expression" % op.__name__
    )

def _inv_impl(expr, op, **kw):
    """See :meth:`.ColumnOperators.__inv__`."""
    if hasattr(expr, "negation_clause"):
        return expr.negation_clause
    else:
        return expr._negate()

def _neg_impl(expr, op, **kw):
    """See :meth:`.ColumnOperators.__neg__`."""
    return UnaryExpression(expr, operator=operators.neg, type_=expr.type)

def _match_impl(expr, op, other, **kw):
    """See :meth:`.ColumnOperators.match`."""

return _boolean_compare(
        expr,
        operators.match_op,
        _check_literal(expr, operators.match_op, other),
        result_type=type_api.MATCHTYPE,
        negate=operators.notmatch_op
        if op is operators.match_op
        else operators.match_op,
        **kw
    )

def _distinct_impl(expr, op, **kw):
    """See :meth:`.ColumnOperators.distinct`."""
    return UnaryExpression(
        expr, operator=operators.distinct_op, type_=expr.type
    )

def _between_impl(expr, op, cleft, cright, **kw):
    """See :meth:`.ColumnOperators.between`."""
    return BinaryExpression(
        expr,
        ClauseList(
            _check_literal(expr, operators.and_, cleft),
            _check_literal(expr, operators.and_, cright),
            operator=operators.and_,
            group=False,
            group_contents=False,
        ),
        op,
        negate=operators.notbetween_op
        if op is operators.between_op
        else operators.between_op,
        modifiers=kw,
    )

def _collate_impl(expr, op, other, **kw):
    return collate(expr, other)

# a mapping of operators with the method they use, along with
# their negated operator for comparison operators
operator_lookup = {
    "and_": (_conjunction_operate,),
    "or_": (_conjunction_operate,),
    "inv": (_inv_impl,),
    "add": (_binary_operate,),
    "mul": (_binary_operate,),
    "sub": (_binary_operate,),
    "div": (_binary_operate,),
    "mod": (_binary_operate,),
    "truediv": (_binary_operate,),
    "custom_op": (_custom_op_operate,),
    "json_path_getitem_op": (_binary_operate,),
    "json_getitem_op": (_binary_operate,),
    "concat_op": (_binary_operate,),
    "any_op": (_scalar, CollectionAggregate._create_any),
    "all_op": (_scalar, CollectionAggregate._create_all),
    "lt": (_boolean_compare, operators.ge),
    "le": (_boolean_compare, operators.gt),
    "ne": (_boolean_compare, operators.eq),
    "gt": (_boolean_compare, operators.le),
    "ge": (_boolean_compare, operators.lt),
    "eq": (_boolean_compare, operators.ne),
    "is_distinct_from": (_boolean_compare, operators.isnot_distinct_from),
    "isnot_distinct_from": (_boolean_compare, operators.is_distinct_from),
    "like_op": (_boolean_compare, operators.notlike_op),
    "ilike_op": (_boolean_compare, operators.notilike_op),
    "notlike_op": (_boolean_compare, operators.like_op),
    "notilike_op": (_boolean_compare, operators.ilike_op),
    "contains_op": (_boolean_compare, operators.notcontains_op),
    "startswith_op": (_boolean_compare, operators.notstartswith_op),
    "endswith_op": (_boolean_compare, operators.notendswith_op),
    "desc_op": (_scalar, UnaryExpression._create_desc),
    "asc_op": (_scalar, UnaryExpression._create_asc),
    "nullsfirst_op": (_scalar, UnaryExpression._create_nullsfirst),
    "nullslast_op": (_scalar, UnaryExpression._create_nullslast),
    "in_op": (_in_impl, operators.notin_op),
    "notin_op": (_in_impl, operators.in_op),
    "is_": (_boolean_compare, operators.is_),
    "isnot": (_boolean_compare, operators.isnot),
    "collate": (_collate_impl,),
    "match_op": (_match_impl,),
    "notmatch_op": (_match_impl,),
    "distinct_op": (_distinct_impl,),
    "between_op": (_between_impl,),
    "notbetween_op": (_between_impl,),
    "neg": (_neg_impl,),
    "getitem": (_getitem_impl,),
    "lshift": (_unsupported_impl,),
    "rshift": (_unsupported_impl,),
    "contains": (_unsupported_impl,),
}

def _check_literal(expr, operator, other, bindparam_type=None):
    if isinstance(other, (ColumnElement, TextClause)):
        if isinstance(other, BindParameter) and other.type._isnull:
            other = other._clone()
            other.type = expr.type
        return other
    elif hasattr(other, "__clause_element__"):
        other = other.__clause_element__()
    elif isinstance(other, type_api.TypeEngine.Comparator):
        other = other.expr

if isinstance(other, (SelectBase, Alias)):
        return other.as_scalar()
    elif not isinstance(other, Visitable):
        return expr._bind_param(operator, other, type_=bindparam_type)
    else:
        return other