Method execute

backend/app/executors/database.py:29–75

(self, config: dict, context: ExecutionContext)

        return "db_connect"
    
    async def execute(self, config: dict, context: ExecutionContext) -> ModuleResult:
        import pymysql
        from pymysql.cursors import DictCursor
        host = context.resolve_value(config.get('host', 'localhost'))
        port = to_int(config.get('port', 3306), 3306, context)
        user = context.resolve_value(config.get('user', 'root'))
        password = context.resolve_value(config.get('password', ''))
        database = context.resolve_value(config.get('database', ''))
        charset = context.resolve_value(config.get('charset', 'utf8mb4'))
        connection_name = context.resolve_value(config.get('connectionName', 'default'))
        
        connections = get_db_connections(context)
        
        try:
            # 如果已有同名连接，先关闭
            if connection_name in connections:
                try:
                    connections[connection_name].close()
                except Exception:
                    pass
                del connections[connection_name]
            
            # 创建新连接
            conn = pymysql.connect(
                host=host,
                port=port,
                user=user,
                password=password,
                database=database if database else None,
                charset=charset,
                cursorclass=DictCursor,
                autocommit=True
            )
            
            connections[connection_name] = conn
            
            db_info = f"{host}:{port}"
            if database:
                db_info += f"/{database}"
            
            return ModuleResult(
                success=True,
                message=f"成功连接到数据库 {db_info}",
                data={"connectionName": connection_name}
            )
        except Exception as e:
            return ModuleResult(success=False, error=f"连接数据库失败: {str(e)}")


@register_executor