database functions that do comparisons or type conversions from django db import notsupportederror from django db models expressions import func value from django db models fields import textfield from django db models fields json import jsonfield from django utils regexhelper import lazyrecompile class castfunc use strftime as datetimetime don t keep fractional seconds mysql doesn t support explicit cast to float mariadb doesn t support explicit cast to json cast would be valid too but the shortcut syntax is more readable expressions is wrapped in parentheses in case it s a complex expression oracle doesn t support explicit cast to json return from left to right the first nonnull expression function coalesce def initself expressions extra if lenexpressions 2 raise valueerrorcoalesce must take at least two expressions super initexpressions extra property def emptyresultsetvalueself for expression in self getsourceexpressions result expression emptyresultsetvalue if result is notimplemented or result is not none return result return none def asoracleself compiler connection extracontext oracle prohibits mixing textfield nclob and charfield nvarchar2 so convert all fields to nclob when that type is expected if self outputfield getinternaltype textfield clone self copy clone setsourceexpressions funcexpression functiontonclob for expression in self getsourceexpressions return supercoalesce clone assqlcompiler connection extracontext return self assqlcompiler connection extracontext class collatefunc function collate template expressionss functions collations alloweddefault false inspired from https www postgresql orgdocscurrentsqlsyntaxlexical htmlsqlsyntaxidentifiers collationre lazyrecompilerw def initself expression collation if not collation and self collationre matchcollation raise valueerrorinvalid collation name r collation self collation collation super initexpression def assqlself compiler connection extracontext extracontext setdefaultcollation connection ops quotenameself collation return super assqlcompiler connection extracontext class greatestfunc function greatest def initself expressions extra if lenexpressions 2 raise valueerrorgreatest must take at least two expressions super initexpressions extra def assqliteself compiler connection extracontext return the minimum expression if any expression is null the return value is databasespecific on postgresql return the minimum notnull expression on mysql oracle and sqlite if any expression is null return null use the min function on sqlite return super assqlitecompiler connection functionmin extracontext class nulliffunc function nullif arity 2 def asoracleself compiler connection extracontext expression1 self getsourceexpressions0 if isinstanceexpression1 value and expression1 value is none raise valueerrororacle does not allow valuenone for expression1 return super assqlcompiler connection extracontext coerce an expression to a new field type use strftime as datetime time don t keep fractional seconds mysql doesn t support explicit cast to float mariadb doesn t support explicit cast to json cast would be valid too but the shortcut syntax is more readable expressions is wrapped in parentheses in case it s a complex expression oracle doesn t support explicit cast to json return from left to right the first non null expression oracle prohibits mixing textfield nclob and charfield nvarchar2 so convert all fields to nclob when that type is expected inspired from https www postgresql org docs current sql syntax lexical html sql syntax identifiers return the maximum expression if any expression is null the return value is database specific on postgresql the maximum not null expression is returned on mysql oracle and sqlite if any expression is null null is returned use the max function on sqlite return the minimum expression if any expression is null the return value is database specific on postgresql return the minimum not null expression on mysql oracle and sqlite if any expression is null return null use the min function on sqlite
from django.db import NotSupportedError
from django.db.models.expressions import Func, Value
from django.db.models.fields import TextField
from django.db.models.fields.json import JSONField
from django.utils.regex_helper import _lazy_re_compile


class Cast(Func):
    

    function = "CAST"
    template = "%(function)s(%(expressions)s AS %(db_type)s)"

    def __init__(self, expression, output_field):
        super().__init__(expression, output_field=output_field)

    def as_sql(self, compiler, connection, **extra_context):
        extra_context["db_type"] = self.output_field.cast_db_type(connection)
        return super().as_sql(compiler, connection, **extra_context)

    def as_sqlite(self, compiler, connection, **extra_context):
        db_type = self.output_field.db_type(connection)
        if db_type in {"datetime", "time"}:
            
            template = "strftime(%%s, %(expressions)s)"
            sql, params = super().as_sql(
                compiler, connection, template=template, **extra_context
            )
            format_string = "%H:%M:%f" if db_type == "time" else "%Y-%m-%d %H:%M:%f"
            params.insert(0, format_string)
            return sql, params
        elif db_type == "date":
            template = "date(%(expressions)s)"
            return super().as_sql(
                compiler, connection, template=template, **extra_context
            )
        return self.as_sql(compiler, connection, **extra_context)

    def as_mysql(self, compiler, connection, **extra_context):
        template = None
        output_type = self.output_field.get_internal_type()
        
        if output_type == "FloatField":
            template = "(%(expressions)s + 0.0)"
        
        elif output_type == "JSONField" and connection.mysql_is_mariadb:
            template = "JSON_EXTRACT(%(expressions)s, '$')"
        return self.as_sql(compiler, connection, template=template, **extra_context)

    def as_postgresql(self, compiler, connection, **extra_context):
        
        
        
        return self.as_sql(
            compiler,
            connection,
            template="(%(expressions)s)::%(db_type)s",
            **extra_context,
        )

    def as_oracle(self, compiler, connection, **extra_context):
        if self.output_field.get_internal_type() == "JSONField":
            
            template = "JSON_QUERY(%(expressions)s, '$')"
            return super().as_sql(
                compiler, connection, template=template, **extra_context
            )
        return self.as_sql(compiler, connection, **extra_context)


class Coalesce(Func):
    

    function = "COALESCE"

    def __init__(self, *expressions, **extra):
        if len(expressions) < 2:
            raise ValueError("Coalesce must take at least two expressions")
        super().__init__(*expressions, **extra)

    @property
    def empty_result_set_value(self):
        for expression in self.get_source_expressions():
            result = expression.empty_result_set_value
            if result is NotImplemented or result is not None:
                return result
        return None

    def as_oracle(self, compiler, connection, **extra_context):
        
        
        if self.output_field.get_internal_type() == "TextField":
            clone = self.copy()
            clone.set_source_expressions(
                [
                    Func(expression, function="TO_NCLOB")
                    for expression in self.get_source_expressions()
                ]
            )
            return super(Coalesce, clone).as_sql(compiler, connection, **extra_context)
        return self.as_sql(compiler, connection, **extra_context)


class Collate(Func):
    function = "COLLATE"
    template = "%(expressions)s %(function)s %(collation)s"
    allowed_default = False
    
    
    collation_re = _lazy_re_compile(r"^[\w-]+$")

    def __init__(self, expression, collation):
        if not (collation and self.collation_re.match(collation)):
            raise ValueError("Invalid collation name: %r." % collation)
        self.collation = collation
        super().__init__(expression)

    def as_sql(self, compiler, connection, **extra_context):
        extra_context.setdefault("collation", connection.ops.quote_name(self.collation))
        return super().as_sql(compiler, connection, **extra_context)


class Greatest(Func):
    

    function = "GREATEST"

    def __init__(self, *expressions, **extra):
        if len(expressions) < 2:
            raise ValueError("Greatest must take at least two expressions")
        super().__init__(*expressions, **extra)

    def as_sqlite(self, compiler, connection, **extra_context):
        
        return super().as_sqlite(compiler, connection, function="MAX", **extra_context)


class JSONObject(Func):
    function = "JSON_OBJECT"
    output_field = JSONField()

    def __init__(self, **fields):
        expressions = []
        for key, value in fields.items():
            expressions.extend((Value(key), value))
        super().__init__(*expressions)

    def as_sql(self, compiler, connection, **extra_context):
        if not connection.features.has_json_object_function:
            raise NotSupportedError(
                "JSONObject() is not supported on this database backend."
            )
        return super().as_sql(compiler, connection, **extra_context)

    def as_postgresql(self, compiler, connection, **extra_context):
        copy = self.copy()
        copy.set_source_expressions(
            [
                Cast(expression, TextField()) if index % 2 == 0 else expression
                for index, expression in enumerate(copy.get_source_expressions())
            ]
        )
        return super(JSONObject, copy).as_sql(
            compiler,
            connection,
            function="JSONB_BUILD_OBJECT",
            **extra_context,
        )

    def as_oracle(self, compiler, connection, **extra_context):
        class ArgJoiner:
            def join(self, args):
                args = [" VALUE ".join(arg) for arg in zip(args[::2], args[1::2])]
                return ", ".join(args)

        return self.as_sql(
            compiler,
            connection,
            arg_joiner=ArgJoiner(),
            template="%(function)s(%(expressions)s RETURNING CLOB)",
            **extra_context,
        )


class Least(Func):
    

    function = "LEAST"

    def __init__(self, *expressions, **extra):
        if len(expressions) < 2:
            raise ValueError("Least must take at least two expressions")
        super().__init__(*expressions, **extra)

    def as_sqlite(self, compiler, connection, **extra_context):
        
        return super().as_sqlite(compiler, connection, function="MIN", **extra_context)


class NullIf(Func):
    function = "NULLIF"
    arity = 2

    def as_oracle(self, compiler, connection, **extra_context):
        expression1 = self.get_source_expressions()[0]
        if isinstance(expression1, Value) and expression1.value is None:
            raise ValueError("Oracle does not allow Value(None) for expression1.")
        return super().as_sql(compiler, connection, **extra_context)