[shell] provide autocompletion for rql queries in cubicweb-ctl shell (closes #3560541)
authorAdrien Di Mascio <Adrien.DiMascio@logilab.fr>
Mon, 12 May 2014 12:37:19 +0200
changeset 9740 c0239d8ae742
parent 9737 c6f47e635845
child 9741 b6cc9da206df
[shell] provide autocompletion for rql queries in cubicweb-ctl shell (closes #3560541) Use the ``rql.suggestions`` component to complete user's input - readline calls the ``complete`` method with user's input, - the ``complete`` method asks for each known completers if it can suggest completions for user's input.
migration.py
test/unittest_toolsutils.py
toolsutils.py
--- a/migration.py	Tue May 06 15:59:39 2014 +0200
+++ b/migration.py	Mon May 12 12:37:19 2014 +0200
@@ -247,12 +247,13 @@
         local_ctx = self._create_context()
         try:
             import readline
-            from rlcompleter import Completer
+            from cubicweb.toolsutils import CWShellCompleter
         except ImportError:
             # readline not available
             pass
         else:
-            readline.set_completer(Completer(local_ctx).complete)
+            rql_completer = CWShellCompleter(local_ctx)
+            readline.set_completer(rql_completer.complete)
             readline.parse_and_bind('tab: complete')
             home_key = 'HOME'
             if sys.platform == 'win32':
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/test/unittest_toolsutils.py	Mon May 12 12:37:19 2014 +0200
@@ -0,0 +1,57 @@
+# copyright 2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
+#
+# This file is part of CubicWeb.
+#
+# CubicWeb is free software: you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation, either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# CubicWeb is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+# FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
+
+
+from logilab.common.testlib import TestCase, unittest_main
+
+from cubicweb.toolsutils import RQLExecuteMatcher
+
+
+class RQLExecuteMatcherTests(TestCase):
+    def matched_query(self, text):
+        match = RQLExecuteMatcher.match(text)
+        if match is None:
+            return None
+        return match['rql_query']
+
+    def test_unknown_function_dont_match(self):
+        self.assertIsNone(self.matched_query('foo'))
+        self.assertIsNone(self.matched_query('rql('))
+        self.assertIsNone(self.matched_query('hell("")'))
+        self.assertIsNone(self.matched_query('eval("rql(\'bla\''))
+
+    def test_rql_other_parameters_dont_match(self):
+        self.assertIsNone(self.matched_query('rql("Any X WHERE X eid %(x)s")'))
+        self.assertIsNone(self.matched_query('rql("Any X WHERE X eid %(x)s", {'))
+        self.assertIsNone(self.matched_query('session.execute("Any X WHERE X eid %(x)s")'))
+        self.assertIsNone(self.matched_query('session.execute("Any X WHERE X eid %(x)s", {'))
+
+    def test_rql_function_match(self):
+        for func_expr in ('rql', 'session.execute'):
+            query = self.matched_query('%s("Any X WHERE X is ' % func_expr)
+            self.assertEqual(query, 'Any X WHERE X is ')
+
+    def test_offseted_rql_function_match(self):
+        """check indentation is allowed"""
+        for func_expr in ('  rql', '  session.execute'):
+            query = self.matched_query('%s("Any X WHERE X is ' % func_expr)
+            self.assertEqual(query, 'Any X WHERE X is ')
+
+
+if __name__ == '__main__':
+    unittest_main()
--- a/toolsutils.py	Tue May 06 15:59:39 2014 +0200
+++ b/toolsutils.py	Mon May 12 12:37:19 2014 +0200
@@ -25,7 +25,12 @@
 import subprocess
 from os import listdir, makedirs, environ, chmod, walk, remove
 from os.path import exists, join, abspath, normpath
-
+import re
+from rlcompleter import Completer
+try:
+    import readline
+except ImportError: # readline not available, no completion
+    pass
 try:
     from os import symlink
 except ImportError:
@@ -264,3 +269,155 @@
         password = getpass('password: ')
     return connect(login=user, password=password, host=optconfig.host, database=appid)
 
+
+## cwshell helpers #############################################################
+
+class AbstractMatcher(object):
+    """Abstract class for CWShellCompleter's matchers.
+
+    A matcher should implement a ``possible_matches`` method. This
+    method has to return the list of possible completions for user's input.
+    Because of the python / readline interaction, each completion should
+    be a superset of the user's input.
+
+    NOTE: readline tokenizes user's input and only passes last token to
+    completers.
+    """
+
+    def possible_matches(self, text):
+        """return possible completions for user's input.
+
+        Parameters:
+            text: the user's input
+
+        Return:
+            a list of completions. Each completion includes the original input.
+        """
+        raise NotImplementedError()
+
+
+class RQLExecuteMatcher(AbstractMatcher):
+    """Custom matcher for rql queries.
+
+    If user's input starts with ``rql(`` or ``session.execute(`` and
+    the corresponding rql query is incomplete, suggest some valid completions.
+    """
+    query_match_rgx = re.compile(
+        r'(?P<func_prefix>\s*(?:rql)'  # match rql, possibly indented
+        r'|'                           # or
+        r'\s*(?:\w+\.execute))'        # match .execute, possibly indented
+        # end of <func_prefix>
+        r'\('                          # followed by a parenthesis
+        r'(?P<quote_delim>["\'])'      # a quote or double quote
+        r'(?P<parameters>.*)')         # and some content
+
+    def __init__(self, local_ctx, req):
+        self.local_ctx = local_ctx
+        self.req = req
+        self.schema = req.vreg.schema
+        self.rsb = req.vreg['components'].select('rql.suggestions', req)
+
+    @staticmethod
+    def match(text):
+        """check if ``text`` looks like a call to ``rql`` or ``session.execute``
+
+        Parameters:
+            text: the user's input
+
+        Returns:
+            None if it doesn't match, the query structure otherwise.
+        """
+        query_match = RQLExecuteMatcher.query_match_rgx.match(text)
+        if query_match is None:
+            return None
+        parameters_text = query_match.group('parameters')
+        quote_delim = query_match.group('quote_delim')
+        # first parameter is fully specified, no completion needed
+        if re.match(r"(.*?)%s" % quote_delim, parameters_text) is not None:
+            return None
+        func_prefix = query_match.group('func_prefix')
+        return {
+            # user's input
+            'text': text,
+            # rql( or session.execute(
+            'func_prefix': func_prefix,
+            # offset of rql query
+            'rql_offset': len(func_prefix) + 2,
+            # incomplete rql query
+            'rql_query': parameters_text,
+            }
+
+    def possible_matches(self, text):
+        """call ``rql.suggestions`` component to complete user's input.
+        """
+        # readline will only send last token, but we need the entire user's input
+        user_input = readline.get_line_buffer()
+        query_struct = self.match(user_input)
+        if query_struct is None:
+            return []
+        else:
+            # we must only send completions of the last token => compute where it
+            # starts relatively to the rql query itself.
+            completion_offset = readline.get_begidx() - query_struct['rql_offset']
+            rql_query = query_struct['rql_query']
+            return [suggestion[completion_offset:]
+                    for suggestion in self.rsb.build_suggestions(rql_query)]
+
+
+class DefaultMatcher(AbstractMatcher):
+    """Default matcher: delegate to standard's `rlcompleter.Completer`` class
+    """
+    def __init__(self, local_ctx):
+        self.completer = Completer(local_ctx)
+
+    def possible_matches(self, text):
+        if "." in text:
+            return self.completer.attr_matches(text)
+        else:
+            return self.completer.global_matches(text)
+
+
+class CWShellCompleter(object):
+    """Custom auto-completion helper for cubicweb-ctl shell.
+
+    ``CWShellCompleter`` provides a ``complete`` method suitable for
+    ``readline.set_completer``.
+
+    Attributes:
+        matchers: the list of ``AbstractMatcher`` instances that will suggest
+                  possible completions
+
+    The completion process is the following:
+
+    - readline calls the ``complete`` method with user's input,
+    - the ``complete`` method asks for each known matchers if
+      it can suggest completions for user's input.
+    """
+
+    def __init__(self, local_ctx):
+        # list of matchers to ask for possible matches on completion
+        self.matchers = [DefaultMatcher(local_ctx)]
+        self.matchers.insert(0, RQLExecuteMatcher(local_ctx, local_ctx['session']))
+
+    def complete(self, text, state):
+        """readline's completer method
+
+        cf http://docs.python.org/2/library/readline.html#readline.set_completer
+        for more details.
+
+        Implementation inspired by `rlcompleter.Completer`
+        """
+        if state == 0:
+            # reset self.matches
+            self.matches = []
+            for matcher in self.matchers:
+                matches = matcher.possible_matches(text)
+                if matches:
+                    self.matches = matches
+                    break
+            else:
+                return None # no matcher able to handle `text`
+        try:
+            return self.matches[state]
+        except IndexError:
+            return None