#!/usr/bin/env python import json import argparse import os.path def escape_c_string(s): """Escape a string for safe inclusion in C code as a string literal.""" if s is None: return "" # Replace backslashes first to avoid double-escaping s = s.replace('\\', '\\\\') # Replace double quotes s = s.replace('"', '\\"') # Replace newlines, tabs, and other common escape sequences s = s.replace('\n', '\\n') s = s.replace('\r', '\\r') s = s.replace('\t', '\\t') # Replace null bytes (just in case) s = s.replace('\0', '\\0') return s class Scope: def __init__(self, start, end, file): self.file = file self.start = start self.end = end def __enter__(self): self.write(self.start + '{\n') Scope.indent += 1 return self def __exit__(self, exc_type, exc_val, exc_tb): Scope.indent -= 1 self.write(f'}}{self.end}\n') def write(self, line): indent = ' ' * Scope.indent self.file.write(indent + line) Scope.indent = 0 COMMANDS_WITH_MINIMUM_ARITY = { # These handlers intentionally process or ignore extra arguments instead of returning WrongArity. 'FT.CONFIG GET', 'FT.CONFIG SET', 'FT.CONFIG HELP', 'FT.CURSOR DEL', } COMMAND_ARITY_OVERRIDES = { # FT.CONFIG SET accepts SET for deprecated flag-style configs. 'FT.CONFIG SET': -3, } def get_function_signature(name): tokens = [token.title() for token in name.replace('.', ' ').split(' ')] value = ''.join(tokens) return f'int Set{value}Info(RedisModuleCommand *cmd)' license = '/*\n* Copyright Redis Ltd. 2016 - present\n' \ '* Licensed under your choice of the Redis Source Available License 2.0 (RSALv2) or' \ '\n* the Server Side Public License v1 (SSPLv1).\n*/\n' def generate_header_file(file_name, command_names): with open(file_name + '.h', 'w') as header: header.write(license) header.write('\n// This file is generated by gen_command_info.py\n') header.write('#pragma once\n') header.write('#include "redismodule.h"\n\n') for name in command_names: header.write(f'{get_function_signature(name)};\n') def generate_history(file, changes): with Scope('.history = (RedisModuleCommandHistoryEntry[])', ',', file) as history: for change in changes: version, what = change history.write(f'{{"{escape_c_string(version)}", "{escape_c_string(what)}"}},\n') history.write('{0}\n') def get_arg_arity(arg): if 'optional' in arg: return 0, False arg_type = arg.get('type') token_arity = len(arg['token'].split()) if 'token' in arg else 0 is_exact_arity = 'multiple' not in arg and 'multiple-token' not in arg if arg_type == 'function': # Keep arity calculation aligned with generate_arguments(): functions with # arguments are emitted as blocks, functions without arguments as pure tokens. arg_type = 'block' if arg.get('arguments') else 'pure-token' if arg_type == 'oneof': subargs = arg.get('arguments', []) subarg_arities = [get_arg_arity(subarg) for subarg in subargs] if not subarg_arities: return token_arity, is_exact_arity min_subarg_arity = min(subarg_arity for subarg_arity, _ in subarg_arities) subargs_exact_arity = all(subarg_exact_arity for _, subarg_exact_arity in subarg_arities) subargs_same_arity = all(subarg_arity == min_subarg_arity for subarg_arity, _ in subarg_arities) return token_arity + min_subarg_arity, is_exact_arity and subargs_exact_arity and subargs_same_arity if 'arguments' in arg: subargs_arity, subargs_exact_arity = get_arguments_arity(arg['arguments']) return token_arity + subargs_arity, is_exact_arity and subargs_exact_arity value_arity = 0 if arg_type == 'pure-token' else 1 return token_arity + value_arity, is_exact_arity def get_arguments_arity(arguments): min_arity = 0 is_exact_arity = True for arg in arguments: arg_arity, arg_exact_arity = get_arg_arity(arg) min_arity += arg_arity is_exact_arity = is_exact_arity and arg_exact_arity return min_arity, is_exact_arity def generate_arguments(file, member, arguments): with Scope(f'.{member} = (RedisModuleCommandArg[])', ',', file) as args_scope: for arg in arguments: with Scope('', ',', file) as arg_scope: for string_arg in ['name', 'token', 'summary', 'since', 'deprecated_since', 'display_text']: if string_arg in arg: arg_scope.write(f'.{string_arg} = "{escape_c_string(arg[string_arg])}",\n') if 'type' in arg: type_text = arg['type'] if type_text == 'function': # Functions with arguments should be treated as blocks, # functions without arguments as pure tokens if 'arguments' in arg and len(arg['arguments']) > 0: type_text = 'block' else: type_text = 'pure_token' type_text = type_text.replace('-', '_').upper() arg_scope.write(f'.type = REDISMODULE_ARG_TYPE_{type_text},\n') flags = [] for flag in ['optional', 'multiple', 'multiple-token']: if flag in arg: flag_text = flag.replace('-', '_').upper() flags.append(f'REDISMODULE_CMD_ARG_{flag_text}') if len(flags) > 0: flags_text = ' | '.join(flags) arg_scope.write(f'.flags = {flags_text},\n') if 'arguments' in arg: generate_arguments(file, 'subargs', arg['arguments']) args_scope.write('{0}\n') def generate_redis_module_command_info(name, cmd_info, file): with Scope('const RedisModuleCommandInfo info = ', ';', file) as info: info.write('.version = REDISMODULE_COMMAND_INFO_VERSION,\n') for key, value in cmd_info.items(): if key == 'group' or key == 'deprecated_since' or key == 'replaced_by': continue if type(value) == str: info.write(f'.{key} = "{escape_c_string(value)}",\n') elif key == 'command_tips': # Handle command_tips array - convert to space-delimited string for tips field # Keep the colon format as used in Redis (e.g., "request_policy:special") if type(value) == list and len(value) > 0: # Convert to lowercase and join with spaces tips_string = ' '.join([tip.lower() for tip in value]) info.write(f'.tips = "{escape_c_string(tips_string)}",\n') elif key == 'history': generate_history(file, value) elif key == 'arguments': generate_arguments(file, 'args', value) arity, is_exact_arity = get_arguments_arity(value) # arity includes the command name itself, including subcommand tokens. arity += len(name.split()) if not is_exact_arity or name in COMMANDS_WITH_MINIMUM_ARITY: arity = -arity arity = COMMAND_ARITY_OVERRIDES.get(name, arity) info.write(f'.arity = {arity},\n') def generate_command_info_definition(name, info, file): signature = get_function_signature(name) file.write(f'// Info for {name}\n') with Scope(signature + ' ', '\n', file) as function: generate_redis_module_command_info(name, info, file) function.write('return RedisModule_SetCommandInfo(cmd, &info);\n') def generate_c_file(file_name, include_path, commands): full_include_path = os.path.join(include_path, os.path.basename(file_name)) with open(file_name + '.c', 'w') as c_file: c_file.write(license) c_file.write('// This file is generated by gen_command_info.py\n') c_file.write('#include "redismodule.h"\n') c_file.write(f'#include "{escape_c_string(full_include_path)}.h"\n\n') for name, info in commands.items(): generate_command_info_definition(name, info, c_file) def main(): parser = argparse.ArgumentParser() parser.add_argument('-j', '--json', help='JSON commands input file', required=True) parser.add_argument('-f', '--file', help='Output file path, will output .h and .c files with this name, ensure all folders in the path exist', default='command_info', required=True) parser.add_argument('-i', '--include', help='Include path to use when generating the c file, default is the same as the output file', required=False) args = parser.parse_args() if args.include is None: args.include = os.path.dirname(args.file) print (f'Generating command info files from {args.json} to {args.file}.h and {args.file}.c') with open(args.json, 'r') as f: data = json.load(f) generate_header_file(args.file, data.keys()) generate_c_file(args.file, args.include, data) if __name__ == '__main__': main()