Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlparse/sql.py: 36%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#
2# Copyright (C) 2009-2020 the sqlparse authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of python-sqlparse and is released under
6# the BSD License: https://opensource.org/licenses/BSD-3-Clause
8"""This module contains classes representing syntactical elements of SQL."""
10import re
12from sqlparse import tokens as T
13from sqlparse.utils import imt, remove_quotes
16class NameAliasMixin:
17 """Implements get_real_name and get_alias."""
19 def get_real_name(self):
20 """Returns the real name (object name) of this identifier."""
21 # a.b.c -> real name is the component after the *last* dot
22 dot_idx = None
23 for idx, tok in enumerate(self.tokens):
24 if tok.match(T.Punctuation, '.'):
25 dot_idx = idx
26 return self._get_first_name(dot_idx, real_name=True)
28 def get_alias(self):
29 """Returns the alias for this identifier or ``None``."""
31 # "name AS alias"
32 kw_idx, kw = self.token_next_by(m=(T.Keyword, 'AS'))
33 if kw is not None:
34 return self._get_first_name(kw_idx + 1, keywords=True)
36 # "name alias" or "complicated column expression alias"
37 _, ws = self.token_next_by(t=T.Whitespace)
38 if len(self.tokens) > 2 and ws is not None:
39 return self._get_first_name(reverse=True)
42class Token:
43 """Base class for all other classes in this module.
45 It represents a single token and has two instance attributes:
46 ``value`` is the unchanged value of the token and ``ttype`` is
47 the type of the token.
48 """
50 __slots__ = (
51 'is_group',
52 'is_keyword',
53 'is_newline',
54 'is_whitespace',
55 'normalized',
56 'parent',
57 'ttype',
58 'value',
59 )
61 def __init__(self, ttype, value):
62 value = str(value)
63 self.value = value
64 self.ttype = ttype
65 self.parent = None
66 self.is_group = False
67 self.is_keyword = ttype in T.Keyword
68 self.is_whitespace = self.ttype in T.Whitespace
69 self.is_newline = self.ttype in T.Newline
70 self.normalized = value.upper() if self.is_keyword else value
72 def __str__(self):
73 return self.value
75 # Pending tokenlist __len__ bug fix
76 # def __len__(self):
77 # return len(self.value)
79 def __repr__(self):
80 cls = self._get_repr_name()
81 value = self._get_repr_value()
83 q = '"' if value.startswith("'") and value.endswith("'") else "'"
84 return "<{cls} {q}{value}{q} at 0x{id:2X}>".format(
85 id=id(self), **locals())
87 def _get_repr_name(self):
88 return str(self.ttype).split('.')[-1]
90 def _get_repr_value(self):
91 raw = str(self)
92 if len(raw) > 7:
93 raw = raw[:6] + '...'
94 return re.sub(r'\s+', ' ', raw)
96 def flatten(self):
97 """Resolve subgroups."""
98 yield self
100 def match(self, ttype, values, regex=False):
101 """Checks whether the token matches the given arguments.
103 *ttype* is a token type as defined in `sqlparse.tokens`. If it does
104 not match, ``False`` is returned.
105 *values* is a list of possible values for this token. For match to be
106 considered valid, the token value needs to be in this list. For tokens
107 of type ``Keyword`` the comparison is case-insensitive. For
108 convenience, a single value can be given passed as a string.
109 If *regex* is ``True``, the given values are treated as regular
110 expressions. Partial matches are allowed. Defaults to ``False``.
111 """
112 type_matched = self.ttype is ttype
113 if not type_matched or values is None:
114 return type_matched
116 if isinstance(values, str):
117 values = (values,)
119 if regex:
120 # TODO: Add test for regex with is_keyword = false
121 flag = re.IGNORECASE if self.is_keyword else 0
122 values = (re.compile(v, flag) for v in values)
124 return any(pattern.search(self.normalized) for pattern in values)
126 if self.is_keyword:
127 values = (v.upper() for v in values)
129 return self.normalized in values
131 def within(self, group_cls):
132 """Returns ``True`` if this token is within *group_cls*.
134 Use this method for example to check if an identifier is within
135 a function: ``t.within(sql.Function)``.
136 """
137 parent = self.parent
138 while parent:
139 if isinstance(parent, group_cls):
140 return True
141 parent = parent.parent
142 return False
144 def is_child_of(self, other):
145 """Returns ``True`` if this token is a direct child of *other*."""
146 return self.parent == other
148 def has_ancestor(self, other):
149 """Returns ``True`` if *other* is in this tokens ancestry."""
150 parent = self.parent
151 while parent:
152 if parent == other:
153 return True
154 parent = parent.parent
155 return False
158class TokenList(Token):
159 """A group of tokens.
161 It has an additional instance attribute ``tokens`` which holds a
162 list of child-tokens.
163 """
165 __slots__ = 'tokens'
167 def __init__(self, tokens=None):
168 self.tokens = tokens or []
169 [setattr(token, 'parent', self) for token in self.tokens]
170 super().__init__(None, ''.join(token.value for token in self.tokens))
171 self.is_group = True
173 def __str__(self):
174 return ''.join(token.value for token in self.flatten())
176 # weird bug
177 # def __len__(self):
178 # return len(self.tokens)
180 def __iter__(self):
181 return iter(self.tokens)
183 def __getitem__(self, item):
184 return self.tokens[item]
186 def _get_repr_name(self):
187 return type(self).__name__
189 def _pprint_tree(self, max_depth=None, depth=0, f=None, _pre=''):
190 """Pretty-print the object tree."""
191 token_count = len(self.tokens)
192 for idx, token in enumerate(self.tokens):
193 cls = token._get_repr_name()
194 value = token._get_repr_value()
196 last = idx == (token_count - 1)
197 pre = '`- ' if last else '|- '
199 q = '"' if value.startswith("'") and value.endswith("'") else "'"
200 print(f"{_pre}{pre}{idx} {cls} {q}{value}{q}", file=f)
202 if token.is_group and (max_depth is None or depth < max_depth):
203 parent_pre = ' ' if last else '| '
204 token._pprint_tree(max_depth, depth + 1, f, _pre + parent_pre)
206 def get_token_at_offset(self, offset):
207 """Returns the token that is on position offset."""
208 idx = 0
209 for token in self.flatten():
210 end = idx + len(token.value)
211 if idx <= offset < end:
212 return token
213 idx = end
215 def flatten(self):
216 """Generator yielding ungrouped tokens.
218 This method is recursively called for all child tokens.
219 """
220 for token in self.tokens:
221 if token.is_group:
222 yield from token.flatten()
223 else:
224 yield token
226 def get_sublists(self):
227 for token in self.tokens:
228 if token.is_group:
229 yield token
231 @property
232 def _groupable_tokens(self):
233 return self.tokens
235 def _token_matching(self, funcs, start=0, end=None, reverse=False):
236 """next token that match functions"""
237 if start is None:
238 return None
240 if not isinstance(funcs, (list, tuple)):
241 funcs = (funcs,)
243 if reverse:
244 assert end is None
245 indexes = range(start - 2, -1, -1)
246 else:
247 if end is None:
248 end = len(self.tokens)
249 indexes = range(start, end)
250 for idx in indexes:
251 token = self.tokens[idx]
252 for func in funcs:
253 if func(token):
254 return idx, token
255 return None, None
257 def token_first(self, skip_ws=True, skip_cm=False):
258 """Returns the first child token.
260 If *skip_ws* is ``True`` (the default), whitespace
261 tokens are ignored.
263 if *skip_cm* is ``True`` (default: ``False``), comments are
264 ignored too.
265 """
266 # this on is inconsistent, using Comment instead of T.Comment...
267 def matcher(tk):
268 return not ((skip_ws and tk.is_whitespace)
269 or (skip_cm and imt(tk, t=T.Comment, i=Comment)))
270 return self._token_matching(matcher)[1]
272 def token_next_by(self, i=None, m=None, t=None, idx=-1, end=None):
273 idx += 1
274 return self._token_matching(lambda tk: imt(tk, i, m, t), idx, end)
276 def token_not_matching(self, funcs, idx):
277 funcs = (funcs,) if not isinstance(funcs, (list, tuple)) else funcs
278 funcs = [lambda tk, func=func: not func(tk) for func in funcs]
279 return self._token_matching(funcs, idx)
281 def token_matching(self, funcs, idx):
282 return self._token_matching(funcs, idx)[1]
284 def token_prev(self, idx, skip_ws=True, skip_cm=False):
285 """Returns the previous token relative to *idx*.
287 If *skip_ws* is ``True`` (the default) whitespace tokens are ignored.
288 If *skip_cm* is ``True`` comments are ignored.
289 ``None`` is returned if there's no previous token.
290 """
291 return self.token_next(idx, skip_ws, skip_cm, _reverse=True)
293 # TODO: May need to re-add default value to idx
294 def token_next(self, idx, skip_ws=True, skip_cm=False, _reverse=False):
295 """Returns the next token relative to *idx*.
297 If *skip_ws* is ``True`` (the default) whitespace tokens are ignored.
298 If *skip_cm* is ``True`` comments are ignored.
299 ``None`` is returned if there's no next token.
300 """
301 if idx is None:
302 return None, None
303 idx += 1 # alot of code usage current pre-compensates for this
305 def matcher(tk):
306 return not ((skip_ws and tk.is_whitespace)
307 or (skip_cm and imt(tk, t=T.Comment, i=Comment)))
308 return self._token_matching(matcher, idx, reverse=_reverse)
310 def token_index(self, token, start=0):
311 """Return list index of token."""
312 start = start if isinstance(start, int) else self.token_index(start)
313 return start + self.tokens[start:].index(token)
315 def group_tokens(self, grp_cls, start, end, include_end=True,
316 extend=False):
317 """Replace tokens by an instance of *grp_cls*."""
318 start_idx = start
319 start = self.tokens[start_idx]
321 end_idx = end + include_end
323 # will be needed later for new group_clauses
324 # while skip_ws and tokens and tokens[-1].is_whitespace:
325 # tokens = tokens[:-1]
327 if extend and isinstance(start, grp_cls):
328 subtokens = self.tokens[start_idx + 1:end_idx]
330 grp = start
331 grp.tokens.extend(subtokens)
332 del self.tokens[start_idx + 1:end_idx]
333 grp.value += ''.join(token.value for token in subtokens)
334 else:
335 subtokens = self.tokens[start_idx:end_idx]
336 grp = grp_cls(subtokens)
337 self.tokens[start_idx:end_idx] = [grp]
338 grp.parent = self
340 for token in subtokens:
341 token.parent = grp
343 return grp
345 def insert_before(self, where, token):
346 """Inserts *token* before *where*."""
347 if not isinstance(where, int):
348 where = self.token_index(where)
349 token.parent = self
350 self.tokens.insert(where, token)
352 def insert_after(self, where, token, skip_ws=True):
353 """Inserts *token* after *where*."""
354 if not isinstance(where, int):
355 where = self.token_index(where)
356 nidx, next_ = self.token_next(where, skip_ws=skip_ws)
357 token.parent = self
358 if next_ is None:
359 self.tokens.append(token)
360 else:
361 self.tokens.insert(nidx, token)
363 def has_alias(self):
364 """Returns ``True`` if an alias is present."""
365 return self.get_alias() is not None
367 def get_alias(self):
368 """Returns the alias for this identifier or ``None``."""
369 return None
371 def get_name(self):
372 """Returns the name of this identifier.
374 This is either it's alias or it's real name. The returned valued can
375 be considered as the name under which the object corresponding to
376 this identifier is known within the current statement.
377 """
378 return self.get_alias() or self.get_real_name()
380 def get_real_name(self):
381 """Returns the real name (object name) of this identifier."""
382 return None
384 def get_parent_name(self):
385 """Return name of the parent object if any.
387 A parent object is identified by the first occurring dot.
388 """
389 dot_idx, _ = self.token_next_by(m=(T.Punctuation, '.'))
390 _, prev_ = self.token_prev(dot_idx)
391 return remove_quotes(prev_.value) if prev_ is not None else None
393 def _get_first_name(self, idx=None, reverse=False, keywords=False,
394 real_name=False):
395 """Returns the name of the first token with a name"""
397 tokens = self.tokens[idx:] if idx else self.tokens
398 tokens = reversed(tokens) if reverse else tokens
399 types = [T.Name, T.Wildcard, T.String.Symbol]
401 if keywords:
402 types.append(T.Keyword)
404 for token in tokens:
405 if token.ttype in types:
406 return remove_quotes(token.value)
407 elif isinstance(token, (Identifier, Function)):
408 return token.get_real_name() if real_name else token.get_name()
411class Statement(TokenList):
412 """Represents a SQL statement."""
414 def get_type(self):
415 """Returns the type of a statement.
417 The returned value is a string holding an upper-cased reprint of
418 the first DML or DDL keyword. If the first token in this group
419 isn't a DML or DDL keyword "UNKNOWN" is returned.
421 Whitespaces and comments at the beginning of the statement
422 are ignored.
423 """
424 token = self.token_first(skip_cm=True)
425 if token is None:
426 # An "empty" statement that either has not tokens at all
427 # or only whitespace tokens.
428 return 'UNKNOWN'
430 elif token.ttype in (T.Keyword.DML, T.Keyword.DDL):
431 return token.normalized
433 elif token.ttype == T.Keyword.CTE:
434 # The WITH keyword should be followed by either an Identifier or
435 # an IdentifierList containing the CTE definitions; the actual
436 # DML keyword (e.g. SELECT, INSERT) will follow next.
437 tidx = self.token_index(token)
438 while tidx is not None:
439 tidx, token = self.token_next(tidx, skip_ws=True)
440 if isinstance(token, (Identifier, IdentifierList)):
441 tidx, token = self.token_next(tidx, skip_ws=True)
443 if token is not None \
444 and token.ttype == T.Keyword.DML:
445 return token.normalized
447 # Hmm, probably invalid syntax, so return unknown.
448 return 'UNKNOWN'
451class Identifier(NameAliasMixin, TokenList):
452 """Represents an identifier.
454 Identifiers may have aliases or typecasts.
455 """
457 def is_wildcard(self):
458 """Return ``True`` if this identifier contains a wildcard."""
459 _, token = self.token_next_by(t=T.Wildcard)
460 return token is not None
462 def get_typecast(self):
463 """Returns the typecast or ``None`` of this object as a string."""
464 midx, marker = self.token_next_by(m=(T.Punctuation, '::'))
465 nidx, next_ = self.token_next(midx, skip_ws=False)
466 return next_.value if next_ else None
468 def get_ordering(self):
469 """Returns the ordering or ``None`` as uppercase string."""
470 _, ordering = self.token_next_by(t=T.Keyword.Order)
471 return ordering.normalized if ordering else None
473 def get_array_indices(self):
474 """Returns an iterator of index token lists"""
476 for token in self.tokens:
477 if isinstance(token, SquareBrackets):
478 # Use [1:-1] index to discard the square brackets
479 yield token.tokens[1:-1]
482class IdentifierList(TokenList):
483 """A list of :class:`~sqlparse.sql.Identifier`\'s."""
485 def get_identifiers(self):
486 """Returns the identifiers.
488 Whitespaces and punctuations are not included in this generator.
489 """
490 for token in self.tokens:
491 if not (token.is_whitespace or token.match(T.Punctuation, ',')):
492 yield token
495class TypedLiteral(TokenList):
496 """A typed literal, such as "date '2001-09-28'" or "interval '2 hours'"."""
497 M_OPEN = [(T.Name.Builtin, None), (T.Keyword, "TIMESTAMP")]
498 M_CLOSE = T.String.Single, None
499 M_EXTEND = T.Keyword, ("DAY", "HOUR", "MINUTE", "MONTH", "SECOND", "YEAR")
502class Parenthesis(TokenList):
503 """Tokens between parenthesis."""
504 M_OPEN = T.Punctuation, '('
505 M_CLOSE = T.Punctuation, ')'
507 @property
508 def _groupable_tokens(self):
509 return self.tokens[1:-1]
512class SquareBrackets(TokenList):
513 """Tokens between square brackets"""
514 M_OPEN = T.Punctuation, '['
515 M_CLOSE = T.Punctuation, ']'
517 @property
518 def _groupable_tokens(self):
519 return self.tokens[1:-1]
522class Assignment(TokenList):
523 """An assignment like 'var := val;'"""
526class If(TokenList):
527 """An 'if' clause with possible 'else if' or 'else' parts."""
528 M_OPEN = T.Keyword, 'IF'
529 M_CLOSE = T.Keyword, 'END IF'
532class For(TokenList):
533 """A 'FOR' loop."""
534 M_OPEN = T.Keyword, ('FOR', 'FOREACH')
535 M_CLOSE = T.Keyword, 'END LOOP'
538class Comparison(TokenList):
539 """A comparison used for example in WHERE clauses."""
541 @property
542 def left(self):
543 return self.tokens[0]
545 @property
546 def right(self):
547 return self.tokens[-1]
550class Comment(TokenList):
551 """A comment."""
553 def is_multiline(self):
554 return self.tokens and self.tokens[0].ttype == T.Comment.Multiline
557class Where(TokenList):
558 """A WHERE clause."""
559 M_OPEN = T.Keyword, 'WHERE'
560 M_CLOSE = T.Keyword, (
561 'ORDER BY', 'GROUP BY', 'LIMIT', 'UNION', 'UNION ALL', 'EXCEPT',
562 'INTERSECT', 'HAVING', 'RETURNING', 'INTO')
565class Over(TokenList):
566 """An OVER clause."""
567 M_OPEN = T.Keyword, 'OVER'
570class Having(TokenList):
571 """A HAVING clause."""
572 M_OPEN = T.Keyword, 'HAVING'
573 M_CLOSE = T.Keyword, ('ORDER BY', 'LIMIT')
576class Case(TokenList):
577 """A CASE statement with one or more WHEN and possibly an ELSE part."""
578 M_OPEN = T.Keyword, 'CASE'
579 M_CLOSE = T.Keyword, 'END'
581 def get_cases(self, skip_ws=False):
582 """Returns a list of 2-tuples (condition, value).
584 If an ELSE exists condition is None.
585 """
586 CONDITION = 1
587 VALUE = 2
589 ret = []
590 mode = CONDITION
592 for token in self.tokens:
593 # Set mode from the current statement
594 if token.match(T.Keyword, 'CASE') or (skip_ws and token.ttype in T.Whitespace):
595 continue
597 elif token.match(T.Keyword, 'WHEN'):
598 ret.append(([], []))
599 mode = CONDITION
601 elif token.match(T.Keyword, 'THEN'):
602 mode = VALUE
604 elif token.match(T.Keyword, 'ELSE'):
605 ret.append((None, []))
606 mode = VALUE
608 elif token.match(T.Keyword, 'END'):
609 mode = None
611 # First condition without preceding WHEN
612 if mode and not ret:
613 ret.append(([], []))
615 # Append token depending of the current mode
616 if mode == CONDITION:
617 ret[-1][0].append(token)
619 elif mode == VALUE:
620 ret[-1][1].append(token)
622 # Return cases list
623 return ret
626class Function(NameAliasMixin, TokenList):
627 """A function or procedure call."""
629 def get_parameters(self):
630 """Return a list of parameters."""
631 parenthesis = self.token_next_by(i=Parenthesis)[1]
632 result = []
633 for token in parenthesis.tokens:
634 if isinstance(token, IdentifierList):
635 return token.get_identifiers()
636 elif imt(token, i=(Function, Identifier, TypedLiteral),
637 t=T.Literal):
638 result.append(token)
639 return result
641 def get_window(self):
642 """Return the window if it exists."""
643 over_clause = self.token_next_by(i=Over)
644 if not over_clause:
645 return None
646 return over_clause[1].tokens[-1]
649class Begin(TokenList):
650 """A BEGIN/END block."""
651 M_OPEN = T.Keyword, 'BEGIN'
652 M_CLOSE = T.Keyword, 'END'
655class Operation(TokenList):
656 """Grouping of operations"""
659class Values(TokenList):
660 """Grouping of values"""
663class Command(TokenList):
664 """Grouping of CLI commands."""