python-3.x - netdev lib 协程异常
问题描述
我已经尝试 netdev lib 一段时间了,下面的程序从未通过以下异常运行它:
回溯(最后一次调用):文件“D:/Code/async_npa/async_npa.py”,第 93 行,在 r = asyncio.run(main(dev_data())) 文件“C:\Users\omera\AppData\ Local\Programs\Python\Python38-32\lib\asyncio\runners.py”,第 43 行,在运行中返回 loop.run_until_complete(main) 文件“C:\Users\omera\AppData\Local\Programs\Python\Python38- 32\lib\asyncio\base_events.py",第 612 行,在 run_until_complete 返回 future.result() 文件 "D:/Code/async_npa/async_npa.py",第 88 行,在主要结果中 = await asyncio.gather(task for任务中的任务)文件“D:/Code/async_npa/async_npa.py”,第 88 行,结果 = 等待 asyncio.gather(任务中任务的任务)RuntimeError:任务产生错误产量:sys:1:RuntimeWarning:协程'从未等待 device_connection'
我也尝试过使用 asyncio 的旧语法创建事件循环和任务,但仍然没有运气
代码块:
from jinja2 import Environment, FileSystemLoader
import yaml
import asyncio
import netdev
def j2_command(file_name: dict = 'script.j2', directory: str = '.') -> dict:
env = Environment(loader=FileSystemLoader(directory))
temp = env.get_template(file_name)
temp_1 = temp.render()
temp_1 = temp_1.split('\n')
return temp_1
def get_host_name(open_connection) -> str:
hostname = open_connection.base_prompt()
hostname = hostname.split('#')[0]
return hostname
def write_to_file(data, dev_conn):
with open(f'./output/config_{get_host_name(dev_conn)}.txt', 'w') as conf:
conf.write(data)
def load_yml(yaml_file='inventory.yml'):
with open(yaml_file) as f:
host_obj = yaml.safe_load(f)
return host_obj
async def device_connection(connect_param):
dev_connect = netdev.create(**connect_param)
await dev_connect.connect()
commands = j2_command()
output = [f'\n\n\n\n\n########################## 1'
f' ##########################\n\n\n\n\n']
for command in commands:
breaker = f'\n\n\n\n\n########################## {command} ##########################\n\n\n\n\n'
command_result = await dev_connect.send_command(command)
output.append(breaker + command_result)
await dev_connect.disconnect()
output_result_string = "\n\n".join(output)
return output_result_string
def dev_data():
device_data = []
# devices_names = []
host_obj = load_yml()
generic_data = host_obj[0]['generic_data']
generic_username = generic_data['username']
generic_password = generic_data['password']
devices = host_obj[0]['devices']
device_type = generic_data['device_type']
device_secret = generic_data['secret']
for device in devices:
device_ip = device['ip_address']
try:
if device["username"]: generic_username = device['username']
if device['password']: generic_password = device['password']
if device["device_type"]: device_type = device['device_type']
if device['secret']: device_secret = device['secret']
except:
pass
dev = {
'device_type': device_type,
'host': device_ip,
'username': generic_username,
'password': generic_password,
'secret': device_secret
}
print(dev)
device_data.append(dev)
return device_data
async def main(device_data):
tasks = [device_connection(dev) for dev in device_data]
result = await asyncio.gather(task for task in tasks)
return result
if __name__ == '__main__':
r = asyncio.run(main(dev_data()))
print(r)
任何帮助,将不胜感激
解决方案
很抱歉我的回复晚了,但我希望它会对你有所帮助。似乎您在运行任务时遇到问题。
device_connection()
您可以定义一个全局output_result_string
变量并将其附加到每个任务中,而不是返回结果。通过这种方式,您不必收集任何东西main()
然后更改main()
如下run()
:
async def run(device_data):
tasks = [device_connection(dev) for dev in device_data]
await asyncio.wait(tasks)
并在您的主块中启动它:
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
这是文档链接: netdev 示例链接
推荐阅读
- android - 根据子值对firebase数据库进行排序
- javascript - IIFE 和常规函数的区别(示例)
- python - 为什么 0.0 == False 在 Python 中返回 True?
- heroku - 使用什么服务来部署我的烧瓶 + dash 应用程序
- ios - PropertyList 解码器解码返回 nil
- apache-spark - 如何在数据块中使用 Spark sql 连接 Spark 数据框列
- javascript - React.js TypeError: Object(...) is not a function
- haskell - 理解简单 Reader monad 的 do 表示法:a <- (*2), b <- (+10), return (a+b)
- dialogflow-es - 谷歌助理每 5-10 分钟就有新用户
- javascript - 在没有空格的php中获取完整的URL