gateway_cmd.py
1 import click 2 from pathlib import Path 3 4 from config_portal.backend.common import GATEWAY_DEFAULTS, USE_DEFAULT_SHARED_ARTIFACT 5 from ...utils import ( 6 get_formatted_names, 7 load_template, 8 error_exit, 9 ask_if_not_provided, 10 indent_multiline_string, 11 ) 12 from .web_add_gateway_step import launch_add_gateway_web_portal 13 14 15 def create_gateway_files( 16 gateway_name_input: str, 17 cli_options: dict, 18 project_root: Path, 19 skip_interactive: bool, 20 ): 21 """ 22 Generates the gateway skeleton files based on templates and collected options. 23 """ 24 collected_options = cli_options.copy() 25 formatted_names = get_formatted_names(gateway_name_input) 26 gateway_name_snake_case = formatted_names["SNAKE_CASE_NAME"] 27 gateway_name_pascal_case = formatted_names["PASCAL_CASE_NAME"] 28 gateway_name_upper_case = formatted_names["SNAKE_UPPER_CASE_NAME"] 29 gateway_name_kebab_case = formatted_names["KEBAB_CASE_NAME"] 30 31 collected_options["namespace"] = ask_if_not_provided( 32 collected_options, 33 "namespace", 34 "Enter namespace for the gateway (e.g., myorg/dev, or leave for ${NAMESPACE})", 35 GATEWAY_DEFAULTS["namespace"], 36 skip_interactive, 37 ) 38 default_gateway_id = ( 39 f"{gateway_name_kebab_case}{GATEWAY_DEFAULTS['gateway_id_suffix']}" 40 ) 41 collected_options["gateway_id"] = ask_if_not_provided( 42 collected_options, 43 "gateway_id", 44 f"Enter Gateway ID (default: {default_gateway_id})", 45 default_gateway_id, 46 skip_interactive, 47 ) 48 49 collected_options["artifact_service_type"] = ask_if_not_provided( 50 collected_options, 51 "artifact_service_type", 52 "Artifact service type for the gateway", 53 GATEWAY_DEFAULTS["artifact_service_type"], 54 skip_interactive, 55 choices=[USE_DEFAULT_SHARED_ARTIFACT, "memory", "filesystem", "gcs"], 56 ) 57 58 if collected_options["artifact_service_type"] != USE_DEFAULT_SHARED_ARTIFACT: 59 if collected_options.get("artifact_service_type") == "filesystem": 60 default_artifact_base_path = GATEWAY_DEFAULTS[ 61 "artifact_service_base_path" 62 ].replace("__GATEWAY_NAME_SNAKE_CASE__", gateway_name_snake_case) 63 collected_options["artifact_service_base_path"] = ask_if_not_provided( 64 collected_options, 65 "artifact_service_base_path", 66 f"Artifact service base path (default: {default_artifact_base_path})", 67 default_artifact_base_path, 68 skip_interactive, 69 ) 70 collected_options["artifact_service_scope"] = ask_if_not_provided( 71 collected_options, 72 "artifact_service_scope", 73 "Artifact service scope", 74 GATEWAY_DEFAULTS["artifact_service_scope"], 75 skip_interactive, 76 choices=["namespace", "app", "custom"], 77 ) 78 79 if ( 80 "system_purpose" not in collected_options 81 or collected_options["system_purpose"] is None 82 ): 83 if skip_interactive: 84 collected_options["system_purpose"] = GATEWAY_DEFAULTS["system_purpose"] 85 else: 86 click.echo("Define system purpose for the gateway (opens editor):") 87 edited_purpose = click.edit(text=GATEWAY_DEFAULTS["system_purpose"]) 88 collected_options["system_purpose"] = ( 89 edited_purpose 90 if edited_purpose is not None 91 else GATEWAY_DEFAULTS["system_purpose"] 92 ) 93 94 if ( 95 "response_format" not in collected_options 96 or collected_options["response_format"] is None 97 ): 98 if skip_interactive: 99 collected_options["response_format"] = GATEWAY_DEFAULTS["response_format"] 100 else: 101 click.echo("Define response format for the gateway (opens editor):") 102 edited_format = click.edit(text=GATEWAY_DEFAULTS["response_format"]) 103 collected_options["response_format"] = ( 104 edited_format 105 if edited_format is not None 106 else GATEWAY_DEFAULTS["response_format"] 107 ) 108 109 configs_gateway_dir = project_root / "configs" / "gateways" 110 src_gateway_dir = project_root / "src" / gateway_name_snake_case 111 112 if ( 113 src_gateway_dir.exists() 114 or (configs_gateway_dir / f"{gateway_name_snake_case}_config.yaml").exists() 115 ): 116 if not skip_interactive: 117 if not click.confirm( 118 click.style( 119 f"Warning: Gateway '{gateway_name_snake_case}' already exists or has conflicting files. Overwrite?", 120 fg="yellow", 121 ) 122 ): 123 click.echo("Operation cancelled by user.") 124 return False, "Operation cancelled." 125 126 try: 127 configs_gateway_dir.mkdir(parents=True, exist_ok=True) 128 src_gateway_dir.mkdir(parents=True, exist_ok=True) 129 130 artifact_service_type_opt = collected_options.get("artifact_service_type") 131 if ( 132 artifact_service_type_opt 133 and artifact_service_type_opt != USE_DEFAULT_SHARED_ARTIFACT 134 ): 135 type_val = artifact_service_type_opt 136 scope_val = collected_options.get( 137 "artifact_service_scope", GATEWAY_DEFAULTS["artifact_service_scope"] 138 ) 139 custom_artifact_lines = [f'type: "{type_val}"'] 140 if type_val == "filesystem": 141 base_path_val = collected_options.get("artifact_service_base_path") 142 if ( 143 "ARTIFACT_BASE_PATH" not in base_path_val 144 and "${" not in base_path_val 145 ): 146 base_path_val_processed = ( 147 f"${{ARTIFACT_BASE_PATH, {base_path_val}}}" 148 ) 149 else: 150 base_path_val_processed = f'"{base_path_val}"' 151 152 custom_artifact_lines.append(f"base_path: {base_path_val_processed}") 153 154 custom_artifact_lines.append(f"artifact_scope: {scope_val}") 155 artifact_service_block = "\n" + "\n".join( 156 [f" {line}" for line in custom_artifact_lines] 157 ) 158 else: 159 artifact_service_block = "*default_artifact_service" 160 161 system_purpose_value = collected_options.get("system_purpose", "") 162 if system_purpose_value is None: 163 system_purpose_value = "" 164 formatted_system_purpose = indent_multiline_string(system_purpose_value, 8) 165 166 response_format_value = collected_options.get("response_format", "") 167 if response_format_value is None: 168 response_format_value = "" 169 formatted_response_format = indent_multiline_string(response_format_value, 8) 170 171 placeholders = { 172 "__GATEWAY_NAME_SNAKE_CASE__": gateway_name_snake_case, 173 "__GATEWAY_NAME_PASCAL_CASE__": gateway_name_pascal_case, 174 "__GATEWAY_NAME_UPPER_CASE__": gateway_name_upper_case, 175 "__GATEWAY_NAME_KEBAB_CASE__": gateway_name_kebab_case, 176 "__APP_CONFIG_NAMESPACE__": collected_options["namespace"], 177 "__GATEWAY_ID__": collected_options["gateway_id"], 178 "__ARTIFACT_SERVICE__": artifact_service_block, 179 "__SYSTEM_PURPOSE__": formatted_system_purpose, 180 "__RESPONSE_FORMAT__": formatted_response_format, 181 } 182 183 template_files_map = { 184 "gateway_config_template.yaml": configs_gateway_dir 185 / f"{gateway_name_snake_case}_config.yaml", 186 "gateway_app_template.py": src_gateway_dir / "app.py", 187 "gateway_component_template.py": src_gateway_dir / "component.py", 188 } 189 190 generated_files_relative_paths = [] 191 192 for template_name, target_path in template_files_map.items(): 193 template_content = load_template(template_name) 194 processed_content = template_content 195 for placeholder, value in placeholders.items(): 196 processed_content = processed_content.replace(placeholder, value) 197 198 with open(target_path, "w", encoding="utf-8") as f: 199 f.write(processed_content) 200 generated_files_relative_paths.append( 201 str(target_path.relative_to(project_root)) 202 ) 203 204 py_init_file = src_gateway_dir / "__init__.py" 205 with open(py_init_file, "w", encoding="utf-8") as f: 206 f.write("") 207 generated_files_relative_paths.append( 208 str(py_init_file.relative_to(project_root)) 209 ) 210 211 success_message = f"Gateway '{gateway_name_snake_case}' skeleton created successfully.\nGenerated files:\n" 212 for rel_path in generated_files_relative_paths: 213 success_message += f" - {rel_path}\n" 214 success_message += "\nNext steps:\n" 215 success_message += f"1. Review and customize the specific parameters in '{generated_files_relative_paths[0]}'.\n" 216 success_message += f"2. Define your gateway's specific schema in '{generated_files_relative_paths[2]}'.\n" 217 success_message += ( 218 f"3. Implement the core logic in '{generated_files_relative_paths[3]}'." 219 ) 220 221 return True, success_message 222 223 except FileNotFoundError as fnf_error: 224 error_message = ( 225 f"Error creating gateway files: Template file not found. {fnf_error}" 226 ) 227 click.echo(click.style(error_message, fg="red"), err=True) 228 return False, error_message 229 except Exception as e: 230 import traceback 231 232 error_message = f"An unexpected error occurred: {e}\n{traceback.format_exc()}" 233 click.echo(click.style(error_message, fg="red"), err=True) 234 return False, error_message 235 236 237 @click.command(name="gateway") 238 @click.argument("name", required=False) 239 @click.option("--namespace", help="namespace for the gateway (e.g., myorg/dev).") 240 @click.option("--gateway-id", help="Custom Gateway ID for the gateway.") 241 @click.option( 242 "--artifact-service-type", 243 type=click.Choice([USE_DEFAULT_SHARED_ARTIFACT, "memory", "filesystem", "gcs"]), 244 help="Artifact service type for the gateway.", 245 ) 246 @click.option( 247 "--artifact-service-base-path", 248 help="Base path for filesystem artifact service (if type is 'filesystem').", 249 ) 250 @click.option( 251 "--artifact-service-scope", 252 type=click.Choice(["namespace", "app", "custom"]), 253 help="Artifact service scope (if not using default shared artifact service).", 254 ) 255 @click.option( 256 "--system-purpose", help="System purpose for the gateway (can be multi-line)." 257 ) 258 @click.option( 259 "--response-format", help="Response format for the gateway (can be multi-line)." 260 ) 261 @click.option( 262 "--skip", 263 is_flag=True, 264 help="Skip interactive prompts and use defaults (CLI mode only).", 265 ) 266 @click.option("--gui", is_flag=True, help="Launch the web UI to configure the gateway.") 267 def add_gateway(name: str | None, gui: bool = False, **kwargs): 268 """ 269 Creates a new gateway skeleton structure via CLI or Web UI. 270 271 NAME: Name of the gateway component to create (e.g., my-new-gateway). 272 Required if not using --gui. 273 """ 274 project_root = Path.cwd() 275 gui_initial_options = { 276 k: v for k, v in kwargs.items() if v is not None and k not in ["gui", "skip"] 277 } 278 if name: 279 gui_initial_options["name"] = name 280 281 if gui: 282 click.echo("Launching Add Gateway GUI...") 283 gui_result = launch_add_gateway_web_portal(cli_options=gui_initial_options) 284 285 if gui_result: 286 gateway_name_from_gui, collected_options_from_gui, project_root_from_gui = ( 287 gui_result 288 ) 289 success, message = create_gateway_files( 290 gateway_name_input=gateway_name_from_gui, 291 cli_options=collected_options_from_gui, 292 project_root=project_root_from_gui, 293 skip_interactive=True, 294 ) 295 if success: 296 click.echo(click.style(message, fg="green")) 297 else: 298 click.echo( 299 click.style( 300 "Gateway creation via GUI was cancelled or failed.", fg="yellow" 301 ), 302 err=True, 303 ) 304 return 305 306 if not name: 307 ctx = click.get_current_context() 308 click.echo(ctx.get_help()) 309 error_exit("Error: Gateway NAME is required when not using the --gui option.") 310 311 cli_options_for_creation = { 312 k: v for k, v in kwargs.items() if v is not None and k not in ["gui", "skip"] 313 } 314 skip_interactive_cli = kwargs.get("skip", False) 315 316 click.echo(f"Creating gateway skeleton for '{name}' via CLI...") 317 success, message = create_gateway_files( 318 name, cli_options_for_creation, project_root, skip_interactive_cli 319 ) 320 321 if success: 322 click.echo(click.style(message, fg="green"))