Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
REPL-CLI boilerplate for simple DSLs with built-in help, autocomplete (like Redis-CLI)
#!/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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment