def _invoke()

in tools/hologres_text2data.py [0:0]

• 53 lines of code
• 3 McCabe index (conditional complexity)

    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
        model_info= tool_parameters.get('model')
        meta_data = get_db_schema(
            db_type=tool_parameters['db_type'],
            host=tool_parameters['host'],
            port=tool_parameters['port'],
            database=tool_parameters['db_name'],
            username=tool_parameters['username'],
            password=tool_parameters['password'],
            table_names=tool_parameters.get('table_names', None)  # Use get method with default value None
        )
        with_comment = tool_parameters.get('with_comment', False)
        dsl_text = format_schema_dsl(meta_data, with_type=True, with_comment=with_comment)
        print(dsl_text)
        # Initialize template loader
        prompt_loader = PromptLoader()
        # Build template context
        context = {
            'db_type': tool_parameters['db_type'].upper(),
            'meta_data': dsl_text
        }
        # Load dynamic prompt
        system_prompt = prompt_loader.get_prompt(
            db_type=tool_parameters['db_type'],
            context=context,
            limit=tool_parameters.get( 'limit', 100 ),
            user_custom_prompt=tool_parameters.get('custom_prompt', '')
        )
        print(f"System prompt:\n{system_prompt}")
        response = self.session.model.llm.invoke(
            model_config=LLMModelConfig(
                provider=model_info.get('provider'),
                model=model_info.get('model'),
                mode=model_info.get('mode'),
                completion_params=model_info.get('completion_params')
            ),
            prompt_messages=[
                SystemPromptMessage(content=system_prompt),
                UserPromptMessage(
                    content=f"Database type: {tool_parameters['db_type']}\n"
                            f"User requirement: {tool_parameters['query']}"
                )
            ],
            stream=False
        )
        print(response)
        excute_sql = response.message.content
        if (isinstance(excute_sql, str)):
            if (tool_parameters['result_format'] == 'json'):
                yield self.create_json_message({
                    "excute_sql": excute_sql
                })
            else:
                yield self.create_text_message(excute_sql)
        else:
            yield self.create_text_message("Generation failed, please check if the input parameters are correct")