def sql_type()

in src/pydolphinscheduler/tasks/sql.py [0:0]

• 19 lines of code
• 4 McCabe index (conditional complexity)

    def sql_type(self) -> str:
        """Judgement sql type, it will return the SQL type for type `SELECT` or `NOT_SELECT`.

        If `param_sql_type` dot not specific, will use regexp to check
        which type of the SQL is. But if `param_sql_type` is specific
        will use the parameter overwrites the regexp way
        """
        if (
            self.param_sql_type == SqlType.SELECT
            or self.param_sql_type == SqlType.NOT_SELECT
        ):
            log.info(
                "The sql type is specified by a parameter, with value %s",
                self.param_sql_type,
            )
            return self.param_sql_type
        pattern_select_str = (
            "^(?!(.* |)insert |(.* |)delete |(.* |)drop "
            "|(.* |)update |(.* |)truncate |(.* |)alter |(.* |)create ).*"
        )
        pattern_select = re.compile(pattern_select_str, re.IGNORECASE)
        if pattern_select.match(self._sql) is None:
            return SqlType.NOT_SELECT
        else:
            return SqlType.SELECT