REPL-CLI boilerplate for simple DSLs with built-in help, autocomplete (like Redis-CLI)

cli.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import argparse
import readline
import rlcompleter
import shlex
import os
import re
try:
  import tabulate
except ImportError:
  tabulate = None
import sys
from collections import namedtuple

Command = namedtuple('Command', ['type', 'parsed', 'raw'])
Response = namedtuple('Response', ['data', 'response', 'error'])

class CLI(object):
  __NAME__           = "Sample CLI"
  __VERSION__        = "1.0"
  __HISTORY__        = os.path.join(os.path.expanduser('~'), '.samplecli')
  __HISTORY_LENGTH__ = 100
  Parameters         = None
  Commands           = []
  CurrentResponse    = None
  CurrentCommand     = None
  PROMPT             = '>> '
  COMMAND_LIST = {
    "HELP" : {
      "help"      : "Display command help",
      "syntax"    : "HELP [command]",
      "arguments" : 0,
    },
  }
  def __init__(self, **kwargs):
    self.parameterize(kwargs['parameters']) 

  def parameterize(self, parameters):
    optparser = argparse.ArgumentParser(self.__NAME__)
    for parameter in parameters:
      switches = parameter['switches']
      del parameter['switches']
      optparser.add_argument(*switches, **parameter)
    self.Parameters = optparser.parse_args()

  def add_command(self, name, processor, help_message, help_syntax):
    if not name in self.COMMAND_LIST:
      self.COMMAND_LIST[name] = {}    
    ''' argument count detection needs improvement to allow for optional arguments '''
    setup = {
      "help"      : help_message,
      "syntax"    : help_syntax,
      "processor" : processor,
      "arguments" : len(shlex.split(help_syntax)) - 1,
    }
    self.COMMAND_LIST[name].update(setup)

  def setup(self):
    if not hasattr(self.Parameters, 'history'):
      self.Parameters.history = self.__HISTORY__
    history = self.Parameters.history
    if not os.path.exists(history):
      f = open(history, 'w')
      f.close()
    readline.parse_and_bind("tab: complete")
    readline.set_completer(self.completer)
    readline.read_history_file(history)
    readline.set_completer_delims('`~!@#$%^&*()-=+[{]}\|;:\'",<>/?')
    readline.set_history_length(self.__HISTORY_LENGTH__)
    self.Commands = self.COMMAND_LIST.keys()
    for history_index in xrange(readline.get_current_history_length()):
      item = readline.get_history_item(history_index)
      if not item is None and not item in self.Commands:
        if not isinstance(item, unicode):
          item = item.decode('utf-8')
        self.Commands.append(item)
    self.Commands.sort()
  
  def completer(self, text, state):
    options = [ c for c in self.Commands if re.match('\s*' + text + '\s*', c, re.I) ]
    options_icase = []
    for option in options:
      try:
        if not option.upper() in options_icase:
          options_icase.append(option)
      except:
        options_icase.append(option)
    options_icase.sort()
    options = options_icase
    if state < len(options):
      return options[state]
    else:
      return None

  def respond(self):
    if not self.CurrentCommand is None:
      if self.CurrentCommand.type == "HELP":
        try:
          if len(self.CurrentCommand.parsed) > 1:
            self.helper(self.CurrentCommand.parsed[1].upper())
          else:
            self.helper()
        except:
          self.helper()
      else:
        response = self.COMMAND_LIST[self.CurrentCommand.type]['processor'](*self.CurrentCommand.parsed[1:]) 
        self.response(error=False, response=response, data=response)
      self.save_history()
    self.printer()    

  def printer(self):
    if self.CurrentResponse is None:
      return
    if self.CurrentResponse.error:
      print "ERROR: %s" % self.CurrentResponse.response
      return
    print self.CurrentResponse.response

  def save_history(self):
    if self.CurrentCommand.type == "HELP":
      return
    try:
      if not self.CurrentCommand.raw in self.Commands:
        self.Commands.append(self.CurrentCommand.raw)
      readline.write_history_file(self.Parameters.history)
    except:
      print 'WARNING: COULD NOT SAVE COMMAND IN HISTORY'
      pass

  def analyze(self):
    self.CurrentResponse = None
    try:
      raw_command = raw_input(self.PROMPT).strip()
      if raw_command == "":
        return      
      parsed_command = shlex.split(raw_command)
      command = parsed_command[0].upper()      
      if command == "?":
        command = "HELP"
      self.CurrentCommand = Command(raw=raw_command, parsed=parsed_command, type=command)
      if not self.CurrentCommand.type in self.COMMAND_LIST:
        self.response(error=True, response="UNKNOWN COMMAND")
        return
      if self.COMMAND_LIST[self.CurrentCommand.type]['arguments'] < len(parsed_command) - 1:
        self.response(error=True, response="%s REQUIRES %d ARGUMENTS" % (self.CurrentCommand.type, len(parsed_command)))
        return
    except ValueError:
      self.CurrentResponse = self.response(error=True, response="WRONG SYNTAX")
    except KeyboardInterrupt:
      self.post_repl()      
    except EOFError:
      self.post_repl()
 
  def helper(self, command=None):
    commands = sorted(self.COMMAND_LIST.keys())
    for c in commands:
      if not command is None:
        if not c.startswith(command.upper()):
          continue
      print "%s" % self.COMMAND_LIST[c]['syntax']
      print '- ' + self.COMMAND_LIST[c]['help']
  
  def response(self, **kwargs):
    defaults = {
      "error"    : False,
      "response" : "",
      "data"     : None,
    }
    defaults.update(kwargs)
    self.CurrentResponse = Response(**defaults)

  def generic_error(self):
    self.response(error=True, response="REQUEST COULD NOT BE COMPLETED")

  def pre_repl(self):
    welcome = "| %s (v%s) |" % (self.__NAME__, self.__VERSION__)
    print "-"*len(welcome)
    print welcome
    print "-"*len(welcome)

  def repl(self):
    self.setup()
    self.pre_repl()
    while True:
      try:
        self.analyze()
        self.respond()
      except EOFError:
        self.post_repl()
      except KeyboardInterrupt:
        self.post_repl()

  def post_repl(self):
    ''' cleanup & terminate '''
    print 'Nice seeing you. Bye!'
    sys.exit()

if __name__ == "__main__":  
  ''' Building a simple CLI for Redis as an example '''
  from redis import StrictRedis as Redis
  parameters = [
    {
      'switches' : [ '--host', '-H' ],
      'help'     : 'Host for the connection',
      'default'  : 'localhost',
    },
    {
      'switches' : [ '--port', '-P' ],
      'help'     : 'Port for the connection',
      'default'  : 6379,
    }
  ]
  
  cli = CLI(parameters=parameters)
  R = Redis(host=cli.Parameters.host, port=cli.Parameters.port)
  cli.add_command('GET', R.get, "Get a key from the database", "GET key")
  cli.add_command('SET', R.set, "Store a key in the database", "SET key value")
  cli.repl()