Filter a leading comment in the SQL statement. This function returns a tuple containing (leading comment, line without the leading comment).
(sql)
| 1836 | |
| 1837 | @staticmethod |
| 1838 | def strip_leading_comment(sql): |
| 1839 | """ |
| 1840 | Filter a leading comment in the SQL statement. This function returns a tuple |
| 1841 | containing (leading comment, line without the leading comment). |
| 1842 | """ |
| 1843 | class StripLeadingCommentFilter(object): |
| 1844 | def __init__(self): |
| 1845 | self.comment = None |
| 1846 | |
| 1847 | def _process(self, tlist): |
| 1848 | """ |
| 1849 | Iterate through the list of tokens, appending each leading commment |
| 1850 | to self.comment, and then popping that element off the list. When we |
| 1851 | hit the first non-comment and non-whitespace token, then we're done -- |
| 1852 | the remainder after that point is the SQL statement. |
| 1853 | """ |
| 1854 | token = tlist.token_first() |
| 1855 | while token: |
| 1856 | if self._is_comment(token) or self._is_whitespace(token): |
| 1857 | if self.comment is None: |
| 1858 | self.comment = token.value |
| 1859 | else: |
| 1860 | self.comment += token.value |
| 1861 | tlist.tokens.pop(0) |
| 1862 | |
| 1863 | # skip_ws=False treats white space characters as tokens also |
| 1864 | token = tlist.token_first(skip_ws=False) |
| 1865 | else: |
| 1866 | break |
| 1867 | |
| 1868 | def _is_comment(self, token): |
| 1869 | return isinstance(token, sqlparse.sql.Comment) or \ |
| 1870 | token.ttype == sqlparse.tokens.Comment.Single or \ |
| 1871 | token.ttype == sqlparse.tokens.Comment.Multiline |
| 1872 | |
| 1873 | def _is_whitespace(self, token): |
| 1874 | return token.ttype == sqlparse.tokens.Whitespace or \ |
| 1875 | token.ttype == sqlparse.tokens.Newline |
| 1876 | |
| 1877 | def process(self, stmt): |
| 1878 | [self.process(sgroup) for sgroup in stmt.get_sublists()] |
| 1879 | self._process(stmt) |
| 1880 | |
| 1881 | stack = sqlparse.engine.FilterStack() |
| 1882 | strip_leading_comment_filter = StripLeadingCommentFilter() |
| 1883 | stack.stmtprocess.append(strip_leading_comment_filter) |
| 1884 | stack.postprocess.append(sqlparse.filters.SerializerUnicode()) |
| 1885 | stripped_line = ''.join(stack.run(sql, 'utf-8')) |
| 1886 | return strip_leading_comment_filter.comment, stripped_line |
| 1887 | |
| 1888 | def _replace_history_delimiters(self, src_delim, tgt_delim): |
| 1889 | """Replaces source_delim with target_delim for all items in history. |