diff --git a/sqlparse/engine/grouping.py b/sqlparse/engine/grouping.py index fae6983..d8cfa9e 100644 --- a/sqlparse/engine/grouping.py +++ b/sqlparse/engine/grouping.py @@ -309,34 +309,13 @@ def group_operator(tlist): def group_identifier_list(tlist): m_role = T.Keyword, ('null', 'role') - m_to = T.Keyword, 'TO' sqlcls = (sql.Function, sql.Case, sql.Identifier, sql.Comparison, sql.IdentifierList, sql.Operation) ttypes = (T_NUMERICAL + T_STRING + T_NAME + (T.Keyword, T.Comment, T.Wildcard)) - def is_rename_separator(token): - # In ``RENAME TABLE a TO b, c TO d`` the comma separates independent - # rename clauses, so the target of one clause (``b``) and the source - # of the next (``c``) must not be merged into a single IdentifierList. - # Detect that layout by the ``TO`` keywords flanking both sides of the - # comma and leave such a comma ungrouped (issue779). - parent = token.parent - if parent is None: - return False - cidx = parent.token_index(token) - pidx, prev_ = parent.token_prev(cidx) - nidx, next_ = parent.token_next(cidx) - if prev_ is None or next_ is None: - return False - _, before_prev = parent.token_prev(pidx) - _, after_next = parent.token_next(nidx) - return (before_prev is not None and before_prev.match(*m_to) - and after_next is not None and after_next.match(*m_to)) - def match(token): - return token.match(T.Punctuation, ',') \ - and not is_rename_separator(token) + return token.match(T.Punctuation, ',') def valid(token): return imt(token, i=sqlcls, m=m_role, t=ttypes) diff --git a/sqlparse/filters/others.py b/sqlparse/filters/others.py new file mode 100644 index 0000000..95bc436 --- /dev/null +++ b/sqlparse/filters/others.py @@ -0,0 +1,174 @@ +# +# Copyright (C) 2009-2020 the sqlparse authors and contributors +# +# +# This module is part of python-sqlparse and is released under +# the BSD License: https://opensource.org/licenses/BSD-3-Clause + +import re + +from sqlparse import sql +from sqlparse import tokens as T +from sqlparse.utils import split_unquoted_newlines + + +class StripCommentsFilter: + + @staticmethod + def _process(tlist): + def get_next_comment(idx=-1): + # TODO(andi) Comment types should be unified, see related issue38 + return tlist.token_next_by(i=sql.Comment, t=T.Comment, idx=idx) + + def _get_insert_token(token): + """Returns either a whitespace or the line breaks from token.""" + # See issue484 why line breaks should be preserved. + # Note: The actual value for a line break is replaced by \n + # in SerializerUnicode which will be executed in the + # postprocessing state. + m = re.search(r'([\r\n]+) *$', token.value) + if m is not None: + return sql.Token(T.Whitespace.Newline, m.groups()[0]) + else: + return sql.Token(T.Whitespace, ' ') + + sql_hints = (T.Comment.Multiline.Hint, T.Comment.Single.Hint) + tidx, token = get_next_comment() + while token: + # skipping token remove if token is a SQL-Hint. issue262 + is_sql_hint = False + if token.ttype in sql_hints: + is_sql_hint = True + elif isinstance(token, sql.Comment): + comment_tokens = token.tokens + if len(comment_tokens) > 0: + if comment_tokens[0].ttype in sql_hints: + is_sql_hint = True + + if is_sql_hint: + # using current index as start index to search next token for + # preventing infinite loop in cases when token type is a + # "SQL-Hint" and has to be skipped + tidx, token = get_next_comment(idx=tidx) + continue + + pidx, prev_ = tlist.token_prev(tidx, skip_ws=False) + nidx, next_ = tlist.token_next(tidx, skip_ws=False) + # Replace by whitespace if prev and next exist and if they're not + # whitespaces. This doesn't apply if prev or next is a parenthesis. + if ( + prev_ is None or next_ is None + or prev_.is_whitespace or prev_.match(T.Punctuation, '(') + or next_.is_whitespace or next_.match(T.Punctuation, ')') + ): + # Insert a whitespace to ensure the following SQL produces + # a valid SQL (see #425). + if prev_ is not None and not prev_.match(T.Punctuation, '('): + tlist.tokens.insert(tidx, _get_insert_token(token)) + tlist.tokens.remove(token) + tidx -= 1 + else: + tlist.tokens[tidx] = _get_insert_token(token) + + # using current index as start index to search next token for + # preventing infinite loop in cases when token type is a + # "SQL-Hint" and has to be skipped + tidx, token = get_next_comment(idx=tidx) + + def process(self, stmt): + [self.process(sgroup) for sgroup in stmt.get_sublists()] + StripCommentsFilter._process(stmt) + return stmt + + +class StripWhitespaceFilter: + def _stripws(self, tlist): + func_name = f'_stripws_{type(tlist).__name__}' + func = getattr(self, func_name.lower(), self._stripws_default) + func(tlist) + + @staticmethod + def _stripws_default(tlist): + last_was_ws = False + is_first_char = True + for token in tlist.tokens: + if token.is_whitespace: + token.value = '' if last_was_ws or is_first_char else ' ' + last_was_ws = token.is_whitespace + is_first_char = False + + def _stripws_identifierlist(self, tlist): + # Removes newlines before commas, see issue140 + last_nl = None + for token in list(tlist.tokens): + if last_nl and token.ttype is T.Punctuation and token.value == ',': + tlist.tokens.remove(last_nl) + last_nl = token if token.is_whitespace else None + + # next_ = tlist.token_next(token, skip_ws=False) + # if (next_ and not next_.is_whitespace and + # token.ttype is T.Punctuation and token.value == ','): + # tlist.insert_after(token, sql.Token(T.Whitespace, ' ')) + return self._stripws_default(tlist) + + def _stripws_parenthesis(self, tlist): + while tlist.tokens[1].is_whitespace: + tlist.tokens.pop(1) + while tlist.tokens[-2].is_whitespace: + tlist.tokens.pop(-2) + if tlist.tokens[-2].is_group: + # save to remove the last whitespace + while tlist.tokens[-2].tokens[-1].is_whitespace: + tlist.tokens[-2].tokens.pop(-1) + self._stripws_default(tlist) + + def process(self, stmt, depth=0): + [self.process(sgroup, depth + 1) for sgroup in stmt.get_sublists()] + self._stripws(stmt) + if depth == 0 and stmt.tokens and stmt.tokens[-1].is_whitespace: + stmt.tokens.pop(-1) + return stmt + + +class SpacesAroundOperatorsFilter: + @staticmethod + def _process(tlist): + + ttypes = (T.Operator, T.Comparison) + tidx, token = tlist.token_next_by(t=ttypes) + while token: + nidx, next_ = tlist.token_next(tidx, skip_ws=False) + if next_ and next_.ttype != T.Whitespace: + tlist.insert_after(tidx, sql.Token(T.Whitespace, ' ')) + + pidx, prev_ = tlist.token_prev(tidx, skip_ws=False) + if prev_ and prev_.ttype != T.Whitespace: + tlist.insert_before(tidx, sql.Token(T.Whitespace, ' ')) + tidx += 1 # has to shift since token inserted before it + + # assert tlist.token_index(token) == tidx + tidx, token = tlist.token_next_by(t=ttypes, idx=tidx) + + def process(self, stmt): + [self.process(sgroup) for sgroup in stmt.get_sublists()] + SpacesAroundOperatorsFilter._process(stmt) + return stmt + + +class StripTrailingSemicolonFilter: + + def process(self, stmt): + while stmt.tokens and (stmt.tokens[-1].is_whitespace + or stmt.tokens[-1].value == ';'): + stmt.tokens.pop() + return stmt + + +# --------------------------- +# postprocess + +class SerializerUnicode: + @staticmethod + def process(stmt): + lines = split_unquoted_newlines(stmt) + return '\n'.join(line.rstrip() for line in lines) diff --git a/sqlparse/sql.py b/sqlparse/sql.py new file mode 100644 index 0000000..ec44a6d --- /dev/null +++ b/sqlparse/sql.py @@ -0,0 +1,664 @@ +# +# Copyright (C) 2009-2020 the sqlparse authors and contributors +# +# +# This module is part of python-sqlparse and is released under +# the BSD License: https://opensource.org/licenses/BSD-3-Clause + +"""This module contains classes representing syntactical elements of SQL.""" + +import re + +from sqlparse import tokens as T +from sqlparse.utils import imt, remove_quotes + + +class NameAliasMixin: + """Implements get_real_name and get_alias.""" + + def get_real_name(self): + """Returns the real name (object name) of this identifier.""" + # a.b.c -> real name is the component after the *last* dot + dot_idx = None + for idx, tok in enumerate(self.tokens): + if tok.match(T.Punctuation, '.'): + dot_idx = idx + return self._get_first_name(dot_idx, real_name=True) + + def get_alias(self): + """Returns the alias for this identifier or ``None``.""" + + # "name AS alias" + kw_idx, kw = self.token_next_by(m=(T.Keyword, 'AS')) + if kw is not None: + return self._get_first_name(kw_idx + 1, keywords=True) + + # "name alias" or "complicated column expression alias" + _, ws = self.token_next_by(t=T.Whitespace) + if len(self.tokens) > 2 and ws is not None: + return self._get_first_name(reverse=True) + + +class Token: + """Base class for all other classes in this module. + + It represents a single token and has two instance attributes: + ``value`` is the unchanged value of the token and ``ttype`` is + the type of the token. + """ + + __slots__ = ( + 'is_group', + 'is_keyword', + 'is_newline', + 'is_whitespace', + 'normalized', + 'parent', + 'ttype', + 'value', + ) + + def __init__(self, ttype, value): + value = str(value) + self.value = value + self.ttype = ttype + self.parent = None + self.is_group = False + self.is_keyword = ttype in T.Keyword + self.is_whitespace = self.ttype in T.Whitespace + self.is_newline = self.ttype in T.Newline + self.normalized = value.upper() if self.is_keyword else value + + def __str__(self): + return self.value + + # Pending tokenlist __len__ bug fix + # def __len__(self): + # return len(self.value) + + def __repr__(self): + cls = self._get_repr_name() + value = self._get_repr_value() + + q = '"' if value.startswith("'") and value.endswith("'") else "'" + return "<{cls} {q}{value}{q} at 0x{id:2X}>".format( + id=id(self), **locals()) + + def _get_repr_name(self): + return str(self.ttype).split('.')[-1] + + def _get_repr_value(self): + raw = str(self) + if len(raw) > 7: + raw = raw[:6] + '...' + return re.sub(r'\s+', ' ', raw) + + def flatten(self): + """Resolve subgroups.""" + yield self + + def match(self, ttype, values, regex=False): + """Checks whether the token matches the given arguments. + + *ttype* is a token type as defined in `sqlparse.tokens`. If it does + not match, ``False`` is returned. + *values* is a list of possible values for this token. For match to be + considered valid, the token value needs to be in this list. For tokens + of type ``Keyword`` the comparison is case-insensitive. For + convenience, a single value can be given passed as a string. + If *regex* is ``True``, the given values are treated as regular + expressions. Partial matches are allowed. Defaults to ``False``. + """ + type_matched = self.ttype is ttype + if not type_matched or values is None: + return type_matched + + if isinstance(values, str): + values = (values,) + + if regex: + # TODO: Add test for regex with is_keyword = false + flag = re.IGNORECASE if self.is_keyword else 0 + values = (re.compile(v, flag) for v in values) + + return any(pattern.search(self.normalized) for pattern in values) + + if self.is_keyword: + values = (v.upper() for v in values) + + return self.normalized in values + + def within(self, group_cls): + """Returns ``True`` if this token is within *group_cls*. + + Use this method for example to check if an identifier is within + a function: ``t.within(sql.Function)``. + """ + parent = self.parent + while parent: + if isinstance(parent, group_cls): + return True + parent = parent.parent + return False + + def is_child_of(self, other): + """Returns ``True`` if this token is a direct child of *other*.""" + return self.parent == other + + def has_ancestor(self, other): + """Returns ``True`` if *other* is in this tokens ancestry.""" + parent = self.parent + while parent: + if parent == other: + return True + parent = parent.parent + return False + + +class TokenList(Token): + """A group of tokens. + + It has an additional instance attribute ``tokens`` which holds a + list of child-tokens. + """ + + __slots__ = 'tokens' + + def __init__(self, tokens=None): + self.tokens = tokens or [] + [setattr(token, 'parent', self) for token in self.tokens] + super().__init__(None, ''.join(token.value for token in self.tokens)) + self.is_group = True + + def __str__(self): + return ''.join(token.value for token in self.flatten()) + + # weird bug + # def __len__(self): + # return len(self.tokens) + + def __iter__(self): + return iter(self.tokens) + + def __getitem__(self, item): + return self.tokens[item] + + def _get_repr_name(self): + return type(self).__name__ + + def _pprint_tree(self, max_depth=None, depth=0, f=None, _pre=''): + """Pretty-print the object tree.""" + token_count = len(self.tokens) + for idx, token in enumerate(self.tokens): + cls = token._get_repr_name() + value = token._get_repr_value() + + last = idx == (token_count - 1) + pre = '`- ' if last else '|- ' + + q = '"' if value.startswith("'") and value.endswith("'") else "'" + print(f"{_pre}{pre}{idx} {cls} {q}{value}{q}", file=f) + + if token.is_group and (max_depth is None or depth < max_depth): + parent_pre = ' ' if last else '| ' + token._pprint_tree(max_depth, depth + 1, f, _pre + parent_pre) + + def get_token_at_offset(self, offset): + """Returns the token that is on position offset.""" + idx = 0 + for token in self.flatten(): + end = idx + len(token.value) + if idx <= offset < end: + return token + idx = end + + def flatten(self): + """Generator yielding ungrouped tokens. + + This method is recursively called for all child tokens. + """ + for token in self.tokens: + if token.is_group: + yield from token.flatten() + else: + yield token + + def get_sublists(self): + for token in self.tokens: + if token.is_group: + yield token + + @property + def _groupable_tokens(self): + return self.tokens + + def _token_matching(self, funcs, start=0, end=None, reverse=False): + """next token that match functions""" + if start is None: + return None + + if not isinstance(funcs, (list, tuple)): + funcs = (funcs,) + + if reverse: + assert end is None + indexes = range(start - 2, -1, -1) + else: + if end is None: + end = len(self.tokens) + indexes = range(start, end) + for idx in indexes: + token = self.tokens[idx] + for func in funcs: + if func(token): + return idx, token + return None, None + + def token_first(self, skip_ws=True, skip_cm=False): + """Returns the first child token. + + If *skip_ws* is ``True`` (the default), whitespace + tokens are ignored. + + if *skip_cm* is ``True`` (default: ``False``), comments are + ignored too. + """ + # this on is inconsistent, using Comment instead of T.Comment... + def matcher(tk): + return not ((skip_ws and tk.is_whitespace) + or (skip_cm and imt(tk, t=T.Comment, i=Comment))) + return self._token_matching(matcher)[1] + + def token_next_by(self, i=None, m=None, t=None, idx=-1, end=None): + idx += 1 + return self._token_matching(lambda tk: imt(tk, i, m, t), idx, end) + + def token_not_matching(self, funcs, idx): + funcs = (funcs,) if not isinstance(funcs, (list, tuple)) else funcs + funcs = [lambda tk, func=func: not func(tk) for func in funcs] + return self._token_matching(funcs, idx) + + def token_matching(self, funcs, idx): + return self._token_matching(funcs, idx)[1] + + def token_prev(self, idx, skip_ws=True, skip_cm=False): + """Returns the previous token relative to *idx*. + + If *skip_ws* is ``True`` (the default) whitespace tokens are ignored. + If *skip_cm* is ``True`` comments are ignored. + ``None`` is returned if there's no previous token. + """ + return self.token_next(idx, skip_ws, skip_cm, _reverse=True) + + # TODO: May need to re-add default value to idx + def token_next(self, idx, skip_ws=True, skip_cm=False, _reverse=False): + """Returns the next token relative to *idx*. + + If *skip_ws* is ``True`` (the default) whitespace tokens are ignored. + If *skip_cm* is ``True`` comments are ignored. + ``None`` is returned if there's no next token. + """ + if idx is None: + return None, None + idx += 1 # alot of code usage current pre-compensates for this + + def matcher(tk): + return not ((skip_ws and tk.is_whitespace) + or (skip_cm and imt(tk, t=T.Comment, i=Comment))) + return self._token_matching(matcher, idx, reverse=_reverse) + + def token_index(self, token, start=0): + """Return list index of token.""" + start = start if isinstance(start, int) else self.token_index(start) + return start + self.tokens[start:].index(token) + + def group_tokens(self, grp_cls, start, end, include_end=True, + extend=False): + """Replace tokens by an instance of *grp_cls*.""" + start_idx = start + start = self.tokens[start_idx] + + end_idx = end + include_end + + # will be needed later for new group_clauses + # while skip_ws and tokens and tokens[-1].is_whitespace: + # tokens = tokens[:-1] + + if extend and isinstance(start, grp_cls): + subtokens = self.tokens[start_idx + 1:end_idx] + + grp = start + grp.tokens.extend(subtokens) + del self.tokens[start_idx + 1:end_idx] + grp.value += ''.join(token.value for token in subtokens) + else: + subtokens = self.tokens[start_idx:end_idx] + grp = grp_cls(subtokens) + self.tokens[start_idx:end_idx] = [grp] + grp.parent = self + + for token in subtokens: + token.parent = grp + + return grp + + def insert_before(self, where, token): + """Inserts *token* before *where*.""" + if not isinstance(where, int): + where = self.token_index(where) + token.parent = self + self.tokens.insert(where, token) + + def insert_after(self, where, token, skip_ws=True): + """Inserts *token* after *where*.""" + if not isinstance(where, int): + where = self.token_index(where) + nidx, next_ = self.token_next(where, skip_ws=skip_ws) + token.parent = self + if next_ is None: + self.tokens.append(token) + else: + self.tokens.insert(nidx, token) + + def has_alias(self): + """Returns ``True`` if an alias is present.""" + return self.get_alias() is not None + + def get_alias(self): + """Returns the alias for this identifier or ``None``.""" + return None + + def get_name(self): + """Returns the name of this identifier. + + This is either it's alias or it's real name. The returned valued can + be considered as the name under which the object corresponding to + this identifier is known within the current statement. + """ + return self.get_alias() or self.get_real_name() + + def get_real_name(self): + """Returns the real name (object name) of this identifier.""" + return None + + def get_parent_name(self): + """Return name of the parent object if any. + + A parent object is identified by the first occurring dot. + """ + dot_idx, _ = self.token_next_by(m=(T.Punctuation, '.')) + _, prev_ = self.token_prev(dot_idx) + return remove_quotes(prev_.value) if prev_ is not None else None + + def _get_first_name(self, idx=None, reverse=False, keywords=False, + real_name=False): + """Returns the name of the first token with a name""" + + tokens = self.tokens[idx:] if idx else self.tokens + tokens = reversed(tokens) if reverse else tokens + types = [T.Name, T.Wildcard, T.String.Symbol] + + if keywords: + types.append(T.Keyword) + + for token in tokens: + if token.ttype in types: + return remove_quotes(token.value) + elif isinstance(token, (Identifier, Function)): + return token.get_real_name() if real_name else token.get_name() + + +class Statement(TokenList): + """Represents a SQL statement.""" + + def get_type(self): + """Returns the type of a statement. + + The returned value is a string holding an upper-cased reprint of + the first DML or DDL keyword. If the first token in this group + isn't a DML or DDL keyword "UNKNOWN" is returned. + + Whitespaces and comments at the beginning of the statement + are ignored. + """ + token = self.token_first(skip_cm=True) + if token is None: + # An "empty" statement that either has not tokens at all + # or only whitespace tokens. + return 'UNKNOWN' + + elif token.ttype in (T.Keyword.DML, T.Keyword.DDL): + return token.normalized + + elif token.ttype == T.Keyword.CTE: + # The WITH keyword should be followed by either an Identifier or + # an IdentifierList containing the CTE definitions; the actual + # DML keyword (e.g. SELECT, INSERT) will follow next. + tidx = self.token_index(token) + while tidx is not None: + tidx, token = self.token_next(tidx, skip_ws=True) + if isinstance(token, (Identifier, IdentifierList)): + tidx, token = self.token_next(tidx, skip_ws=True) + + if token is not None \ + and token.ttype == T.Keyword.DML: + return token.normalized + + # Hmm, probably invalid syntax, so return unknown. + return 'UNKNOWN' + + +class Identifier(NameAliasMixin, TokenList): + """Represents an identifier. + + Identifiers may have aliases or typecasts. + """ + + def is_wildcard(self): + """Return ``True`` if this identifier contains a wildcard.""" + _, token = self.token_next_by(t=T.Wildcard) + return token is not None + + def get_typecast(self): + """Returns the typecast or ``None`` of this object as a string.""" + midx, marker = self.token_next_by(m=(T.Punctuation, '::')) + nidx, next_ = self.token_next(midx, skip_ws=False) + return next_.value if next_ else None + + def get_ordering(self): + """Returns the ordering or ``None`` as uppercase string.""" + _, ordering = self.token_next_by(t=T.Keyword.Order) + return ordering.normalized if ordering else None + + def get_array_indices(self): + """Returns an iterator of index token lists""" + + for token in self.tokens: + if isinstance(token, SquareBrackets): + # Use [1:-1] index to discard the square brackets + yield token.tokens[1:-1] + + +class IdentifierList(TokenList): + """A list of :class:`~sqlparse.sql.Identifier`\'s.""" + + def get_identifiers(self): + """Returns the identifiers. + + Whitespaces and punctuations are not included in this generator. + """ + for token in self.tokens: + if not (token.is_whitespace or token.match(T.Punctuation, ',')): + yield token + + +class TypedLiteral(TokenList): + """A typed literal, such as "date '2001-09-28'" or "interval '2 hours'".""" + M_OPEN = [(T.Name.Builtin, None), (T.Keyword, "TIMESTAMP")] + M_CLOSE = T.String.Single, None + M_EXTEND = T.Keyword, ("DAY", "HOUR", "MINUTE", "MONTH", "SECOND", "YEAR") + + +class Parenthesis(TokenList): + """Tokens between parenthesis.""" + M_OPEN = T.Punctuation, '(' + M_CLOSE = T.Punctuation, ')' + + @property + def _groupable_tokens(self): + return self.tokens[1:-1] + + +class SquareBrackets(TokenList): + """Tokens between square brackets""" + M_OPEN = T.Punctuation, '[' + M_CLOSE = T.Punctuation, ']' + + @property + def _groupable_tokens(self): + return self.tokens[1:-1] + + +class Assignment(TokenList): + """An assignment like 'var := val;'""" + + +class If(TokenList): + """An 'if' clause with possible 'else if' or 'else' parts.""" + M_OPEN = T.Keyword, 'IF' + M_CLOSE = T.Keyword, 'END IF' + + +class For(TokenList): + """A 'FOR' loop.""" + M_OPEN = T.Keyword, ('FOR', 'FOREACH') + M_CLOSE = T.Keyword, 'END LOOP' + + +class Comparison(TokenList): + """A comparison used for example in WHERE clauses.""" + + @property + def left(self): + return self.tokens[0] + + @property + def right(self): + return self.tokens[-1] + + +class Comment(TokenList): + """A comment.""" + + def is_multiline(self): + return self.tokens and self.tokens[0].ttype == T.Comment.Multiline + + +class Where(TokenList): + """A WHERE clause.""" + M_OPEN = T.Keyword, 'WHERE' + M_CLOSE = T.Keyword, ( + 'ORDER BY', 'GROUP BY', 'LIMIT', 'UNION', 'UNION ALL', 'EXCEPT', + 'INTERSECT', 'HAVING', 'RETURNING', 'INTO') + + +class Over(TokenList): + """An OVER clause.""" + M_OPEN = T.Keyword, 'OVER' + + +class Having(TokenList): + """A HAVING clause.""" + M_OPEN = T.Keyword, 'HAVING' + M_CLOSE = T.Keyword, ('ORDER BY', 'LIMIT') + + +class Case(TokenList): + """A CASE statement with one or more WHEN and possibly an ELSE part.""" + M_OPEN = T.Keyword, 'CASE' + M_CLOSE = T.Keyword, 'END' + + def get_cases(self, skip_ws=False): + """Returns a list of 2-tuples (condition, value). + + If an ELSE exists condition is None. + """ + CONDITION = 1 + VALUE = 2 + + ret = [] + mode = CONDITION + + for token in self.tokens: + # Set mode from the current statement + if token.match(T.Keyword, 'CASE') or (skip_ws and token.ttype in T.Whitespace): + continue + + elif token.match(T.Keyword, 'WHEN'): + ret.append(([], [])) + mode = CONDITION + + elif token.match(T.Keyword, 'THEN'): + mode = VALUE + + elif token.match(T.Keyword, 'ELSE'): + ret.append((None, [])) + mode = VALUE + + elif token.match(T.Keyword, 'END'): + mode = None + + # First condition without preceding WHEN + if mode and not ret: + ret.append(([], [])) + + # Append token depending of the current mode + if mode == CONDITION: + ret[-1][0].append(token) + + elif mode == VALUE: + ret[-1][1].append(token) + + # Return cases list + return ret + + +class Function(NameAliasMixin, TokenList): + """A function or procedure call.""" + + def get_parameters(self): + """Return a list of parameters.""" + parenthesis = self.token_next_by(i=Parenthesis)[1] + result = [] + for token in parenthesis.tokens: + if isinstance(token, IdentifierList): + return token.get_identifiers() + elif imt(token, i=(Function, Identifier, TypedLiteral), + t=T.Literal): + result.append(token) + return result + + def get_window(self): + """Return the window if it exists.""" + over_clause = self.token_next_by(i=Over) + if not over_clause: + return None + return over_clause[1].tokens[-1] + + +class Begin(TokenList): + """A BEGIN/END block.""" + M_OPEN = T.Keyword, 'BEGIN' + M_CLOSE = T.Keyword, 'END' + + +class Operation(TokenList): + """Grouping of operations""" + + +class Values(TokenList): + """Grouping of values""" + + +class Command(TokenList): + """Grouping of CLI commands.""" diff --git a/tests/test_grouping.py b/tests/test_grouping.py new file mode 100644 index 0000000..20fab9b --- /dev/null +++ b/tests/test_grouping.py @@ -0,0 +1,687 @@ +import pytest + +import sqlparse +from sqlparse import sql, tokens as T + + +def test_grouping_parenthesis(): + s = 'select (select (x3) x2) and (y2) bar' + parsed = sqlparse.parse(s)[0] + assert str(parsed) == s + assert len(parsed.tokens) == 7 + assert isinstance(parsed.tokens[2], sql.Parenthesis) + assert isinstance(parsed.tokens[-1], sql.Identifier) + assert len(parsed.tokens[2].tokens) == 5 + assert isinstance(parsed.tokens[2].tokens[3], sql.Identifier) + assert isinstance(parsed.tokens[2].tokens[3].tokens[0], sql.Parenthesis) + assert len(parsed.tokens[2].tokens[3].tokens) == 3 + + +@pytest.mark.parametrize('s', ['foo := 1;', 'foo := 1']) +def test_grouping_assignment(s): + parsed = sqlparse.parse(s)[0] + assert len(parsed.tokens) == 1 + assert isinstance(parsed.tokens[0], sql.Assignment) + + +@pytest.mark.parametrize('s', ["x > DATE '2020-01-01'", "x > TIMESTAMP '2020-01-01 00:00:00'"]) +def test_grouping_typed_literal(s): + parsed = sqlparse.parse(s)[0] + assert isinstance(parsed[0][4], sql.TypedLiteral) + + +@pytest.mark.parametrize('s, a, b', [ + ('select a from b where c < d + e', sql.Identifier, sql.Identifier), + ('select a from b where c < d + interval \'1 day\'', sql.Identifier, sql.TypedLiteral), + ('select a from b where c < d + interval \'6\' month', sql.Identifier, sql.TypedLiteral), + ('select a from b where c < current_timestamp - interval \'1 day\'', sql.Token, sql.TypedLiteral), +]) +def test_compare_expr(s, a, b): + parsed = sqlparse.parse(s)[0] + assert str(parsed) == s + assert isinstance(parsed.tokens[2], sql.Identifier) + assert isinstance(parsed.tokens[6], sql.Identifier) + assert isinstance(parsed.tokens[8], sql.Where) + assert len(parsed.tokens) == 9 + where = parsed.tokens[8] + assert isinstance(where.tokens[2], sql.Comparison) + assert len(where.tokens) == 3 + comparison = where.tokens[2] + assert isinstance(comparison.tokens[0], sql.Identifier) + assert comparison.tokens[2].ttype is T.Operator.Comparison + assert isinstance(comparison.tokens[4], sql.Operation) + assert len(comparison.tokens) == 5 + operation = comparison.tokens[4] + assert isinstance(operation.tokens[0], a) + assert operation.tokens[2].ttype is T.Operator + assert isinstance(operation.tokens[4], b) + assert len(operation.tokens) == 5 + + +def test_grouping_identifiers(): + s = 'select foo.bar from "myscheme"."table" where fail. order' + parsed = sqlparse.parse(s)[0] + assert str(parsed) == s + assert isinstance(parsed.tokens[2], sql.Identifier) + assert isinstance(parsed.tokens[6], sql.Identifier) + assert isinstance(parsed.tokens[8], sql.Where) + s = 'select * from foo where foo.id = 1' + parsed = sqlparse.parse(s)[0] + assert str(parsed) == s + assert isinstance(parsed.tokens[-1].tokens[-1].tokens[0], sql.Identifier) + s = 'select * from (select "foo"."id" from foo)' + parsed = sqlparse.parse(s)[0] + assert str(parsed) == s + assert isinstance(parsed.tokens[-1].tokens[3], sql.Identifier) + + for s in ["INSERT INTO `test` VALUES('foo', 'bar');", + "INSERT INTO `test` VALUES(1, 2), (3, 4), (5, 6);", + "INSERT INTO `test(a, b)` VALUES(1, 2), (3, 4), (5, 6);"]: + parsed = sqlparse.parse(s)[0] + types = [l.ttype for l in parsed.tokens if not l.is_whitespace] + assert types == [T.DML, T.Keyword, None, None, T.Punctuation] + assert isinstance(parsed.tokens[6], sql.Values) + + s = "select 1.0*(a+b) as col, sum(c)/sum(d) from myschema.mytable" + parsed = sqlparse.parse(s)[0] + assert len(parsed.tokens) == 7 + assert isinstance(parsed.tokens[2], sql.IdentifierList) + assert len(parsed.tokens[2].tokens) == 4 + identifiers = list(parsed.tokens[2].get_identifiers()) + assert len(identifiers) == 2 + assert identifiers[0].get_alias() == "col" + + +@pytest.mark.parametrize('s', [ + '1 as f', + 'foo as f', + 'foo f', + '1/2 as f', + '1/2 f', + '1<2 as f', # issue327 + '1<2 f', +]) +def test_simple_identifiers(s): + parsed = sqlparse.parse(s)[0] + assert isinstance(parsed.tokens[0], sql.Identifier) + + +@pytest.mark.parametrize('s', [ + 'foo, bar', + 'sum(a), sum(b)', + 'sum(a) as x, b as y', + 'sum(a)::integer, b', + 'sum(a)/count(b) as x, y', + 'sum(a)::integer as x, y', + 'sum(a)::integer/count(b) as x, y', # issue297 +]) +def test_group_identifier_list(s): + parsed = sqlparse.parse(s)[0] + assert isinstance(parsed.tokens[0], sql.IdentifierList) + + +def test_grouping_identifier_wildcard(): + p = sqlparse.parse('a.*, b.id')[0] + assert isinstance(p.tokens[0], sql.IdentifierList) + assert isinstance(p.tokens[0].tokens[0], sql.Identifier) + assert isinstance(p.tokens[0].tokens[-1], sql.Identifier) + + +def test_grouping_identifier_name_wildcard(): + p = sqlparse.parse('a.*')[0] + t = p.tokens[0] + assert t.get_name() == '*' + assert t.is_wildcard() is True + + +def test_grouping_identifier_invalid(): + p = sqlparse.parse('a.')[0] + assert isinstance(p.tokens[0], sql.Identifier) + assert p.tokens[0].has_alias() is False + assert p.tokens[0].get_name() is None + assert p.tokens[0].get_real_name() is None + assert p.tokens[0].get_parent_name() == 'a' + + +def test_grouping_identifier_invalid_in_middle(): + # issue261 + s = 'SELECT foo. FROM foo' + p = sqlparse.parse(s)[0] + assert isinstance(p[2], sql.Identifier) + assert p[2][1].ttype == T.Punctuation + assert p[3].ttype == T.Whitespace + assert str(p[2]) == 'foo.' + +@pytest.mark.parametrize('s', ['foo as (select *)', 'foo as(select *)']) +def test_grouping_identifer_as(s): + # issue507 + p = sqlparse.parse(s)[0] + assert isinstance(p.tokens[0], sql.Identifier) + token = p.tokens[0].tokens[2] + assert token.ttype == T.Keyword + assert token.normalized == 'AS' + +def test_grouping_identifier_as_invalid(): + # issue8 + p = sqlparse.parse('foo as select *')[0] + assert len(p.tokens), 5 + assert isinstance(p.tokens[0], sql.Identifier) + assert len(p.tokens[0].tokens) == 1 + assert p.tokens[2].ttype == T.Keyword + + +def test_grouping_identifier_function(): + p = sqlparse.parse('foo() as bar')[0] + assert isinstance(p.tokens[0], sql.Identifier) + assert isinstance(p.tokens[0].tokens[0], sql.Function) + p = sqlparse.parse('foo()||col2 bar')[0] + assert isinstance(p.tokens[0], sql.Identifier) + assert isinstance(p.tokens[0].tokens[0], sql.Operation) + assert isinstance(p.tokens[0].tokens[0].tokens[0], sql.Function) + p = sqlparse.parse('foo(c1) over win1 as bar')[0] + assert isinstance(p.tokens[0], sql.Identifier) + assert isinstance(p.tokens[0].tokens[0], sql.Function) + assert len(p.tokens[0].tokens[0].tokens) == 4 + assert isinstance(p.tokens[0].tokens[0].tokens[3], sql.Over) + assert isinstance(p.tokens[0].tokens[0].tokens[3].tokens[2], + sql.Identifier) + p = sqlparse.parse('foo(c1) over (partition by c2 order by c3) as bar')[0] + assert isinstance(p.tokens[0], sql.Identifier) + assert isinstance(p.tokens[0].tokens[0], sql.Function) + assert len(p.tokens[0].tokens[0].tokens) == 4 + assert isinstance(p.tokens[0].tokens[0].tokens[3], sql.Over) + assert isinstance(p.tokens[0].tokens[0].tokens[3].tokens[2], + sql.Parenthesis) + + +@pytest.mark.parametrize('s', ['foo+100', 'foo + 100', 'foo*100']) +def test_grouping_operation(s): + p = sqlparse.parse(s)[0] + assert isinstance(p.tokens[0], sql.Operation) + + +def test_grouping_identifier_list(): + p = sqlparse.parse('a, b, c')[0] + assert isinstance(p.tokens[0], sql.IdentifierList) + p = sqlparse.parse('(a, b, c)')[0] + assert isinstance(p.tokens[0].tokens[1], sql.IdentifierList) + + +def test_grouping_identifier_list_subquery(): + """identifier lists should still work in subqueries with aliases""" + p = sqlparse.parse("select * from (" + "select a, b + c as d from table) sub")[0] + subquery = p.tokens[-1].tokens[0] + idx, iden_list = subquery.token_next_by(i=sql.IdentifierList) + assert iden_list is not None + # all the identifiers should be within the IdentifierList + _, ilist = subquery.token_next_by(i=sql.Identifier, idx=idx) + assert ilist is None + + +def test_grouping_identifier_list_case(): + p = sqlparse.parse('a, case when 1 then 2 else 3 end as b, c')[0] + assert isinstance(p.tokens[0], sql.IdentifierList) + p = sqlparse.parse('(a, case when 1 then 2 else 3 end as b, c)')[0] + assert isinstance(p.tokens[0].tokens[1], sql.IdentifierList) + + +def test_grouping_identifier_list_other(): + # issue2 + p = sqlparse.parse("select *, null, 1, 'foo', bar from mytable, x")[0] + assert isinstance(p.tokens[2], sql.IdentifierList) + assert len(p.tokens[2].tokens) == 13 + + +def test_grouping_identifier_list_with_inline_comments(): + # issue163 + p = sqlparse.parse('foo /* a comment */, bar')[0] + assert isinstance(p.tokens[0], sql.IdentifierList) + assert isinstance(p.tokens[0].tokens[0], sql.Identifier) + assert isinstance(p.tokens[0].tokens[3], sql.Identifier) + + +def test_grouping_identifiers_with_operators(): + p = sqlparse.parse('a+b as c from table where (d-e)%2= 1')[0] + assert len([x for x in p.flatten() if x.ttype == T.Name]) == 5 + + +def test_grouping_identifier_list_with_order(): + # issue101 + p = sqlparse.parse('1, 2 desc, 3')[0] + assert isinstance(p.tokens[0], sql.IdentifierList) + assert isinstance(p.tokens[0].tokens[3], sql.Identifier) + assert str(p.tokens[0].tokens[3]) == '2 desc' + + +def test_grouping_nested_identifier_with_order(): + # issue745 + p = sqlparse.parse('(a desc)')[0] + assert isinstance(p.tokens[0], sql.Parenthesis) + assert isinstance(p.tokens[0].tokens[1], sql.Identifier) + assert str(p.tokens[0].tokens[1]) == 'a desc' + + +def test_grouping_where(): + s = 'select * from foo where bar = 1 order by id desc' + p = sqlparse.parse(s)[0] + assert str(p) == s + assert len(p.tokens) == 12 + + s = 'select x from (select y from foo where bar = 1) z' + p = sqlparse.parse(s)[0] + assert str(p) == s + assert isinstance(p.tokens[-1].tokens[0].tokens[-2], sql.Where) + + +@pytest.mark.parametrize('s', ( + 'select 1 where 1 = 2 union select 2', + 'select 1 where 1 = 2 union all select 2', +)) +def test_grouping_where_union(s): + p = sqlparse.parse(s)[0] + assert p.tokens[5].value.startswith('union') + + +def test_returning_kw_ends_where_clause(): + s = 'delete from foo where x > y returning z' + p = sqlparse.parse(s)[0] + assert isinstance(p.tokens[6], sql.Where) + assert p.tokens[7].ttype == T.Keyword + assert p.tokens[7].value == 'returning' + + +def test_into_kw_ends_where_clause(): # issue324 + s = 'select * from foo where a = 1 into baz' + p = sqlparse.parse(s)[0] + assert isinstance(p.tokens[8], sql.Where) + assert p.tokens[9].ttype == T.Keyword + assert p.tokens[9].value == 'into' + + +@pytest.mark.parametrize('sql, expected', [ + # note: typecast needs to be 2nd token for this test + ('select foo::integer from bar', 'integer'), + ('select (current_database())::information_schema.sql_identifier', + 'information_schema.sql_identifier'), +]) +def test_grouping_typecast(sql, expected): + p = sqlparse.parse(sql)[0] + assert p.tokens[2].get_typecast() == expected + + +def test_grouping_alias(): + s = 'select foo as bar from mytable' + p = sqlparse.parse(s)[0] + assert str(p) == s + assert p.tokens[2].get_real_name() == 'foo' + assert p.tokens[2].get_alias() == 'bar' + s = 'select foo from mytable t1' + p = sqlparse.parse(s)[0] + assert str(p) == s + assert p.tokens[6].get_real_name() == 'mytable' + assert p.tokens[6].get_alias() == 't1' + s = 'select foo::integer as bar from mytable' + p = sqlparse.parse(s)[0] + assert str(p) == s + assert p.tokens[2].get_alias() == 'bar' + s = ('SELECT DISTINCT ' + '(current_database())::information_schema.sql_identifier AS view') + p = sqlparse.parse(s)[0] + assert str(p) == s + assert p.tokens[4].get_alias() == 'view' + + +def test_grouping_alias_case(): + # see issue46 + p = sqlparse.parse('CASE WHEN 1 THEN 2 ELSE 3 END foo')[0] + assert len(p.tokens) == 1 + assert p.tokens[0].get_alias() == 'foo' + + +def test_grouping_alias_ctas(): + p = sqlparse.parse('CREATE TABLE tbl1 AS SELECT coalesce(t1.col1, 0) AS col1 FROM t1')[0] + assert p.tokens[10].get_alias() == 'col1' + assert isinstance(p.tokens[10].tokens[0], sql.Function) + + +def test_grouping_alias_ctas_lowercase_as(): + # 'as' is case-insensitive; a lowercase keyword in CREATE TABLE ... AS + # SELECT must not disable function grouping in the SELECT body. + p = sqlparse.parse('create table tbl1 as select coalesce(t1.col1, 0) as col1 from t1')[0] + assert p.tokens[10].get_alias() == 'col1' + assert isinstance(p.tokens[10].tokens[0], sql.Function) + +def test_grouping_subquery_no_parens(): + # Not totally sure if this is the right approach... + # When a THEN clause contains a subquery w/o parenthesis around it *and* + # a WHERE condition, the WHERE grouper consumes END too. + # This takes makes sure that it doesn't fail. + p = sqlparse.parse('CASE WHEN 1 THEN select 2 where foo = 1 end')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Case) + + +@pytest.mark.parametrize('s', ['foo.bar', 'x, y', 'x > y', 'x / y']) +def test_grouping_alias_returns_none(s): + # see issue185 and issue445 + p = sqlparse.parse(s)[0] + assert len(p.tokens) == 1 + assert p.tokens[0].get_alias() is None + + +def test_grouping_idlist_function(): + # see issue10 too + p = sqlparse.parse('foo(1) x, bar')[0] + assert isinstance(p.tokens[0], sql.IdentifierList) + + +def test_grouping_comparison_exclude(): + # make sure operators are not handled too lazy + p = sqlparse.parse('(=)')[0] + assert isinstance(p.tokens[0], sql.Parenthesis) + assert not isinstance(p.tokens[0].tokens[1], sql.Comparison) + p = sqlparse.parse('(a=1)')[0] + assert isinstance(p.tokens[0].tokens[1], sql.Comparison) + p = sqlparse.parse('(a>=1)')[0] + assert isinstance(p.tokens[0].tokens[1], sql.Comparison) + + +def test_grouping_function(): + p = sqlparse.parse('foo()')[0] + assert isinstance(p.tokens[0], sql.Function) + p = sqlparse.parse('foo(null, bar)')[0] + assert isinstance(p.tokens[0], sql.Function) + assert len(list(p.tokens[0].get_parameters())) == 2 + p = sqlparse.parse('foo(5) over win1')[0] + assert isinstance(p.tokens[0], sql.Function) + assert len(list(p.tokens[0].get_parameters())) == 1 + assert isinstance(p.tokens[0].get_window(), sql.Identifier) + p = sqlparse.parse('foo(5) over (PARTITION BY c1)')[0] + assert isinstance(p.tokens[0], sql.Function) + assert len(list(p.tokens[0].get_parameters())) == 1 + assert isinstance(p.tokens[0].get_window(), sql.Parenthesis) + + +def test_grouping_function_not_in(): + # issue183 + p = sqlparse.parse('in(1, 2)')[0] + assert len(p.tokens) == 2 + assert p.tokens[0].ttype == T.Keyword + assert isinstance(p.tokens[1], sql.Parenthesis) + + +def test_grouping_varchar(): + p = sqlparse.parse('"text" Varchar(50) NOT NULL')[0] + assert isinstance(p.tokens[2], sql.Function) + + +def test_statement_get_type(): + def f(sql): + return sqlparse.parse(sql)[0] + + assert f('select * from foo').get_type() == 'SELECT' + assert f('update foo').get_type() == 'UPDATE' + assert f(' update foo').get_type() == 'UPDATE' + assert f('\nupdate foo').get_type() == 'UPDATE' + assert f('foo').get_type() == 'UNKNOWN' + + +def test_identifier_with_operators(): + # issue 53 + p = sqlparse.parse('foo||bar')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Operation) + # again with whitespaces + p = sqlparse.parse('foo || bar')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Operation) + + +def test_identifier_with_op_trailing_ws(): + # make sure trailing whitespace isn't grouped with identifier + p = sqlparse.parse('foo || bar ')[0] + assert len(p.tokens) == 2 + assert isinstance(p.tokens[0], sql.Operation) + assert p.tokens[1].ttype is T.Whitespace + + +def test_identifier_with_string_literals(): + p = sqlparse.parse("foo + 'bar'")[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Operation) + + +# This test seems to be wrong. It was introduced when fixing #53, but #111 +# showed that this shouldn't be an identifier at all. I'm leaving this +# commented in the source for a while. +# def test_identifier_string_concat(): +# p = sqlparse.parse("'foo' || bar")[0] +# assert len(p.tokens) == 1 +# assert isinstance(p.tokens[0], sql.Identifier) + + +def test_identifier_consumes_ordering(): + # issue89 + p = sqlparse.parse('select * from foo order by c1 desc, c2, c3')[0] + assert isinstance(p.tokens[-1], sql.IdentifierList) + ids = list(p.tokens[-1].get_identifiers()) + assert len(ids) == 3 + assert ids[0].get_name() == 'c1' + assert ids[0].get_ordering() == 'DESC' + assert ids[1].get_name() == 'c2' + assert ids[1].get_ordering() is None + + +def test_comparison_with_keywords(): + # issue90 + # in fact these are assignments, but for now we don't distinguish them + p = sqlparse.parse('foo = NULL')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Comparison) + assert len(p.tokens[0].tokens) == 5 + assert p.tokens[0].left.value == 'foo' + assert p.tokens[0].right.value == 'NULL' + # make sure it's case-insensitive + p = sqlparse.parse('foo = null')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Comparison) + + +def test_comparison_with_floats(): + # issue145 + p = sqlparse.parse('foo = 25.5')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Comparison) + assert len(p.tokens[0].tokens) == 5 + assert p.tokens[0].left.value == 'foo' + assert p.tokens[0].right.value == '25.5' + + +def test_comparison_with_parenthesis(): + # issue23 + p = sqlparse.parse('(3 + 4) = 7')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Comparison) + comp = p.tokens[0] + assert isinstance(comp.left, sql.Parenthesis) + assert comp.right.ttype is T.Number.Integer + + +@pytest.mark.parametrize('operator', ( + '=', '!=', '>', '<', '<=', '>=', '~', '~~', '!~~', + 'LIKE', 'NOT LIKE', 'ILIKE', 'NOT ILIKE', +)) +def test_comparison_with_strings(operator): + # issue148 + p = sqlparse.parse(f"foo {operator} 'bar'")[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Comparison) + assert p.tokens[0].right.value == "'bar'" + assert p.tokens[0].right.ttype == T.String.Single + + +def test_like_and_ilike_comparison(): + def validate_where_clause(where_clause, expected_tokens): + assert len(where_clause.tokens) == len(expected_tokens) + for where_token, expected_token in zip(where_clause, expected_tokens): + expected_ttype, expected_value = expected_token + if where_token.ttype is not None: + assert where_token.match(expected_ttype, expected_value, regex=True) + else: + # Certain tokens, such as comparison tokens, do not define a ttype that can be + # matched against. For these tokens, we ensure that the token instance is of + # the expected type and has a value conforming to specified regular expression + import re + assert (isinstance(where_token, expected_ttype) + and re.match(expected_value, where_token.value)) + + [p1] = sqlparse.parse("select * from mytable where mytable.mycolumn LIKE 'expr%' limit 5;") + [p1_where] = [token for token in p1 if isinstance(token, sql.Where)] + validate_where_clause(p1_where, [ + (T.Keyword, "where"), + (T.Whitespace, None), + (sql.Comparison, r"mytable.mycolumn LIKE.*"), + (T.Whitespace, None), + ]) + + [p2] = sqlparse.parse( + "select * from mytable where mycolumn NOT ILIKE '-expr' group by othercolumn;") + [p2_where] = [token for token in p2 if isinstance(token, sql.Where)] + validate_where_clause(p2_where, [ + (T.Keyword, "where"), + (T.Whitespace, None), + (sql.Comparison, r"mycolumn NOT ILIKE.*"), + (T.Whitespace, None), + ]) + + +def test_comparison_with_functions(): + # issue230 + p = sqlparse.parse('foo = DATE(bar.baz)')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Comparison) + assert len(p.tokens[0].tokens) == 5 + assert p.tokens[0].left.value == 'foo' + assert p.tokens[0].right.value == 'DATE(bar.baz)' + + p = sqlparse.parse('DATE(foo.bar) = DATE(bar.baz)')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Comparison) + assert len(p.tokens[0].tokens) == 5 + assert p.tokens[0].left.value == 'DATE(foo.bar)' + assert p.tokens[0].right.value == 'DATE(bar.baz)' + + p = sqlparse.parse('DATE(foo.bar) = bar.baz')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Comparison) + assert len(p.tokens[0].tokens) == 5 + assert p.tokens[0].left.value == 'DATE(foo.bar)' + assert p.tokens[0].right.value == 'bar.baz' + + +def test_comparison_with_typed_literal(): + p = sqlparse.parse("foo = DATE 'bar.baz'")[0] + assert len(p.tokens) == 1 + comp = p.tokens[0] + assert isinstance(comp, sql.Comparison) + assert len(comp.tokens) == 5 + assert comp.left.value == 'foo' + assert isinstance(comp.right, sql.TypedLiteral) + assert comp.right.value == "DATE 'bar.baz'" + + +@pytest.mark.parametrize('start', ['FOR', 'FOREACH']) +def test_forloops(start): + p = sqlparse.parse(f'{start} foo in bar LOOP foobar END LOOP')[0] + assert (len(p.tokens)) == 1 + assert isinstance(p.tokens[0], sql.For) + + +def test_nested_for(): + p = sqlparse.parse('FOR foo LOOP FOR bar LOOP END LOOP END LOOP')[0] + assert len(p.tokens) == 1 + for1 = p.tokens[0] + assert for1.tokens[0].value == 'FOR' + assert for1.tokens[-1].value == 'END LOOP' + for2 = for1.tokens[6] + assert isinstance(for2, sql.For) + assert for2.tokens[0].value == 'FOR' + assert for2.tokens[-1].value == 'END LOOP' + + +def test_begin(): + p = sqlparse.parse('BEGIN foo END')[0] + assert len(p.tokens) == 1 + assert isinstance(p.tokens[0], sql.Begin) + + +def test_keyword_followed_by_parenthesis(): + p = sqlparse.parse('USING(somecol')[0] + assert len(p.tokens) == 3 + assert p.tokens[0].ttype == T.Keyword + assert p.tokens[1].ttype == T.Punctuation + + +def test_nested_begin(): + p = sqlparse.parse('BEGIN foo BEGIN bar END END')[0] + assert len(p.tokens) == 1 + outer = p.tokens[0] + assert outer.tokens[0].value == 'BEGIN' + assert outer.tokens[-1].value == 'END' + inner = outer.tokens[4] + assert inner.tokens[0].value == 'BEGIN' + assert inner.tokens[-1].value == 'END' + assert isinstance(inner, sql.Begin) + + +def test_aliased_column_without_as(): + p = sqlparse.parse('foo bar')[0].tokens + assert len(p) == 1 + assert p[0].get_real_name() == 'foo' + assert p[0].get_alias() == 'bar' + + p = sqlparse.parse('foo.bar baz')[0].tokens[0] + assert p.get_parent_name() == 'foo' + assert p.get_real_name() == 'bar' + assert p.get_alias() == 'baz' + + +def test_qualified_function(): + p = sqlparse.parse('foo()')[0].tokens[0] + assert p.get_parent_name() is None + assert p.get_real_name() == 'foo' + + p = sqlparse.parse('foo.bar()')[0].tokens[0] + assert p.get_parent_name() == 'foo' + assert p.get_real_name() == 'bar' + + +def test_aliased_function_without_as(): + p = sqlparse.parse('foo() bar')[0].tokens[0] + assert p.get_parent_name() is None + assert p.get_real_name() == 'foo' + assert p.get_alias() == 'bar' + + p = sqlparse.parse('foo.bar() baz')[0].tokens[0] + assert p.get_parent_name() == 'foo' + assert p.get_real_name() == 'bar' + assert p.get_alias() == 'baz' + + +def test_aliased_literal_without_as(): + p = sqlparse.parse('1 foo')[0].tokens + assert len(p) == 1 + assert p[0].get_alias() == 'foo' + + +def test_grouping_as_cte(): + p = sqlparse.parse('foo AS WITH apple AS 1, banana AS 2')[0].tokens + assert len(p) > 4 + assert p[0].get_alias() is None + assert p[2].value == 'AS' + assert p[4].value == 'WITH' + +def test_grouping_create_table(): + p = sqlparse.parse("create table db.tbl (a string)")[0].tokens + assert p[4].value == "db.tbl"