Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_split_commands(command, expected):
assert split_command_args(command, all_commands)[0] == expected
def send_command(self, raw_command, completer):
"""
Send raw_command to redis-server, return parsed response.
:param raw_command: text raw_command, not parsed
:param completer: RedisGrammarCompleter will update completer
based on redis response. eg: update key completer after ``keys``
raw_command
"""
command_name = ""
try:
command_name, args = split_command_args(raw_command, all_commands)
# if raw_command is not supposed to send to server
if command_name.upper() in CLIENT_COMMANDS:
redis_resp = self.client_execute_command(command_name, *args)
return redis_resp
self.pre_hook(raw_command, command_name, args, completer)
redis_resp = self.execute_command_and_read_response(
completer, command_name, *args
)
self.after_hook(raw_command, command_name, args, completer)
if command_name.upper() == "MONITOR":
logger.info("monitor")
print_formatted_text(redis_resp, style=STYLE)
try:
self.monitor()
except KeyboardInterrupt:
return
def apply_transformation(
self, transformation_input: TransformationInput
) -> Transformation:
input_text = transformation_input.document.text
if input_text != self.last_text:
try:
command, _ = split_command_args(input_text, all_commands)
except InvalidArguments:
self.command_holder.command = None
else:
self.command_holder.command = command.upper()
self.last_text = input_text
return Transformation(transformation_input.fragments)
completer_mapping.update(
{
# all key related completers share the same completer
"keys": key_completer,
"key": key_completer,
"destination": key_completer,
"newkey": key_completer,
# member
"member": member_completer,
"members": member_completer,
# hash fields
"field": field_completer,
"fields": field_completer,
}
)
completer_mapping["commandname"] = WordCompleter(all_commands, ignore_case=True)
completer = RedisGrammarCompleter(redis_grammar, completer_mapping)
return completer