NetMiko面向对象改造遇KeyError: 'device_type'问题求助

错误原因 #

1. 类继承逻辑错误：class NetConnect(ConnectHandler()) 写法完全错误。ConnectHandler 是NetMiko提供的工厂函数，调用它必须传入device_type等设备参数；而且继承语法要求传入类对象，不是函数调用的返回值，这直接导致了KeyError: 'device_type'。
2. 实例方法调用错误：NetConnect.net_connection() 直接调用类的实例方法，但未先创建类的实例，即使继承问题修复，后续也会抛出TypeError。

修正方案一：封装ConnectHandler（推荐入门用） #

这种方式将连接逻辑封装到自定义类中，避免直接继承NetMiko底层类，更灵活易维护，同时支持线程化调用。

from netmiko import ConnectHandler
import threading

class NetConnect:
    def __init__(self, device_params):
        self.device_params = device_params
        self.connection = None

    def connect(self):
        # 建立设备连接
        self.connection = ConnectHandler(**self.device_params)
        self.connection.enable()

    def run_command(self, command):
        if not self.connection:
            self.connect()
        return self.connection.send_command(command)

    def disconnect(self):
        if self.connection:
            self.connection.disconnect()

# 设备配置参数
voice_ware = {
    'device_type': 'linux',
    'session_log': 'session.txt',
    'host': "127.0.0.1",
    'username': "*********",
    'password': "*********",
    'port': 51002,
    'verbose': True,
    'secret': "*********"
}

def main():
    # 创建连接实例并执行命令
    conn = NetConnect(voice_ware)
    output = conn.run_command("uname -r")
    print(output)
    conn.disconnect()

# 线程化调用示例
def thread_task(device_params):
    conn = NetConnect(device_params)
    try:
        output = conn.run_command("uname -r")
        print(f"线程执行结果: {output}")
    finally:
        conn.disconnect()

if __name__ == "__main__":
    main()

    # 测试多线程调用
    thread = threading.Thread(target=thread_task, args=(voice_ware,))
    thread.start()
    thread.join()

修正方案二：继承NetMiko设备连接类（进阶用法） #

如果需要深度定制连接逻辑，可以直接继承对应设备的连接类（如Linux设备的LinuxSSHConnection），复用NetMiko底层实现。

from netmiko.linux.linux_ssh import LinuxSSHConnection
import threading

class NetConnect(LinuxSSHConnection):
    def __init__(self, **kwargs):
        # 调用父类初始化方法传入设备参数
        super().__init__(**kwargs)

    def custom_exec(self, command):
        # 自定义命令执行方法，封装enable操作
        self.enable()
        return self.send_command(command)

# 设备配置参数
voice_ware = {
    'device_type': 'linux',
    'session_log': 'session.txt',
    'host': "127.0.0.1",
    'username': "*********",
    'password': "*********",
    'port': 51002,
    'verbose': True,
    'secret': "*********"
}

def main():
    conn = NetConnect(**voice_ware)
    conn.connect()
    output = conn.custom_exec("uname -r")
    print(output)
    conn.disconnect()

# 线程化调用示例
def thread_task(device_params):
    conn = NetConnect(**device_params)
    try:
        conn.connect()
        output = conn.custom_exec("uname -r")
        print(f"线程执行结果: {output}")
    finally:
        conn.disconnect()

if __name__ == "__main__":
    main()

    thread = threading.Thread(target=thread_task, args=(voice_ware,))
    thread.start()
    thread.join()

关键改进点 #

• 将设备参数作为类初始化参数，让类支持多设备复用。
• 拆分连接、命令执行、断开连接方法，职责更清晰。
• 线程化调用时，每个线程创建独立的连接实例，避免多线程共享连接导致的异常。

内容的提问来源于stack exchange，提问作者PolarityNotRequired