return response
+def get_ietf_network_node_request(network: str, node: str, content: str):
+ url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}',
+ 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'}
+ if RESTCONF_VERSION == 'rfc8040':
+ format_args = ('{}', network, node, content)
+ elif content == 'config':
+ format_args = ('{}', content, network, node)
+ else:
+ format_args = ('{}', 'operational', network, node)
+ response = get_request(url[RESTCONF_VERSION].format(*format_args))
+ res = response.json()
+ return_key = {'rfc8040': 'ietf-network:node',
+ 'draft-bierman02': 'node'}
+ node = res[return_key[RESTCONF_VERSION]][0]
+ return {'status_code': response.status_code,
+ 'node': node}
+
+
def del_ietf_network_node_request(network: str, node: str, content: str):
url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}',
'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'}
response = delete_request(url[RESTCONF_VERSION].format(*format_args))
return response
+
#
-# TransportPCE internal API RPCs
+# Service list operations
#
-def prepend_dict_keys(input_dict: dict, prefix: str):
- return_dict = {}
- for key, value in input_dict.items():
- newkey = prefix + key
- if isinstance(value, dict):
- return_dict[newkey] = prepend_dict_keys(value, prefix)
- # TODO: perhaps some recursion depth limit or another solution has to be considered here
- # even if recursion depth is given by the input_dict argument
- # direct (self-)recursive functions may carry unwanted side-effects such as ressource consumptions
- else:
- return_dict[newkey] = value
- return return_dict
-
-
-def connect_xpdr_to_rdm_request(payload: dict):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links"
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': {'links-input': payload}}, 'networkutils:')
- else:
- data = {'input': {'links-input': payload}}
- return post_request(url, data)
-
-
-def connect_rdm_to_xpdr_request(payload: dict):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': {'links-input': payload}}, 'networkutils:')
- else:
- data = {'input': {'links-input': payload}}
- return post_request(url, data)
-
-
-def device_renderer_create_ots_oms_request(nodeid: str, lcp: str):
- url = "{}/operations/transportpce-device-renderer:create-ots-oms"
- payload = {
- 'node-id': nodeid,
- 'logical-connection-point': lcp}
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': payload}, 'transportpce-device-renderer:')
- else:
- data = {'input': payload}
- return post_request(url, data)
-
-
-def device_renderer_service_path_request(payload: dict):
- url = "{}/operations/transportpce-device-renderer:service-path"
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': payload}, 'transportpce-device-renderer:')
- else:
- data = {'input': payload}
- response = post_request(url, data)
+def get_ordm_serv_list_request():
+ url = {'rfc8040': '{}/data/org-openroadm-service:service-list?content=nonconfig',
+ 'draft-bierman02': '{}/operational/org-openroadm-service:service-list/'}
+ response = get_request(url[RESTCONF_VERSION])
res = response.json()
- return_key = {'rfc8040': 'transportpce-device-renderer:output',
- 'draft-bierman02': 'output'}
- return_output = res[return_key[RESTCONF_VERSION]]
- return {'status_code': response.status_code,
- 'output': return_output}
-
-
-def device_renderer_otn_service_path_request(payload: dict):
- url = "{}/operations/transportpce-device-renderer:otn-service-path"
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': payload}, 'transportpce-device-renderer:')
+ return_key = {'rfc8040': 'org-openroadm-service:service-list',
+ 'draft-bierman02': 'service-list'}
+ if return_key[RESTCONF_VERSION] in res.keys():
+ response_attribute = res[return_key[RESTCONF_VERSION]]
else:
- data = {'input': payload}
- response = post_request(url, data)
- res = response.json()
- return_key = {'rfc8040': 'transportpce-device-renderer:output',
- 'draft-bierman02': 'output'}
- return_output = res[return_key[RESTCONF_VERSION]]
+ response_attribute = res['errors']['error'][0]
return {'status_code': response.status_code,
- 'output': return_output}
+ 'service-list': response_attribute}
-def renderer_service_implementation_request(payload: dict):
- url = "{}/operations/transportpce-renderer:service-implementation-request"
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': payload}, 'transportpce-renderer:')
+def get_ordm_serv_list_attr_request(attribute: str, value: str):
+ url = {'rfc8040': '{}/data/org-openroadm-service:service-list/{}={}?content=nonconfig',
+ 'draft-bierman02': '{}/operational/org-openroadm-service:service-list/{}/{}'}
+ if RESTCONF_VERSION == 'rfc8040':
+ format_args = ('{}', attribute, value)
else:
- data = {'input': payload}
- response = post_request(url, data)
+ format_args = ('{}', attribute, value)
+ response = get_request(url[RESTCONF_VERSION].format(*format_args))
res = response.json()
- return_key = {'rfc8040': 'transportpce-renderer:output',
- 'draft-bierman02': 'output'}
- return_output = res[return_key[RESTCONF_VERSION]]
- return {'status_code': response.status_code,
- 'output': return_output}
-
-
-def olm_get_pm_request(payload: dict):
- url = "{}/operations/transportpce-olm:get-pm"
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': payload}, 'transportpce-olm:')
+ return_key = {'rfc8040': 'org-openroadm-service:' + attribute,
+ 'draft-bierman02': attribute}
+ if return_key[RESTCONF_VERSION] in res.keys():
+ response_attribute = res[return_key[RESTCONF_VERSION]]
else:
- data = {'input': payload}
- response = post_request(url, data)
- res = response.json()
- return_key = {'rfc8040': 'transportpce-olm:output',
- 'draft-bierman02': 'output'}
- return_output = res[return_key[RESTCONF_VERSION]]
+ response_attribute = res['errors']['error'][0]
return {'status_code': response.status_code,
- 'output': return_output}
+ attribute: response_attribute}
-def olm_calculate_spanloss_base_request(payload: dict):
- url = "{}/operations/transportpce-olm:calculate-spanloss-base"
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': payload}, 'transportpce-olm:')
- else:
- data = {'input': payload}
- response = post_request(url, data)
- res = response.json()
- return_key = {'rfc8040': 'transportpce-olm:output',
- 'draft-bierman02': 'output'}
- return_output = res[return_key[RESTCONF_VERSION]]
- return {'status_code': response.status_code,
- 'output': return_output}
+#
+# TransportPCE internal API RPCs
+#
-def olm_service_power_setup_request(payload: dict):
- url = "{}/operations/transportpce-olm:service-power-setup"
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': payload}, 'transportpce-olm:')
- else:
- data = {'input': payload}
- response = post_request(url, data)
- res = response.json()
- return_key = {'rfc8040': 'transportpce-olm:output',
- 'draft-bierman02': 'output'}
- return_output = res[return_key[RESTCONF_VERSION]]
- return {'status_code': response.status_code,
- 'output': return_output}
+def prepend_dict_keys(input_dict: dict, prefix: str):
+ return_dict = {}
+ for key, value in input_dict.items():
+ newkey = prefix + key
+ if isinstance(value, dict):
+ return_dict[newkey] = prepend_dict_keys(value, prefix)
+ # TODO: perhaps some recursion depth limit or another solution has to be considered here
+ # even if recursion depth is given by the input_dict argument
+ # direct (self-)recursive functions may carry unwanted side-effects such as ressource consumptions
+ else:
+ return_dict[newkey] = value
+ return return_dict
-def olm_service_power_turndown_request(payload: dict):
- url = "{}/operations/transportpce-olm:service-power-turndown"
- if RESTCONF_VERSION == 'draft-bierman02':
- data = prepend_dict_keys({'input': payload}, 'transportpce-olm:')
+def transportpce_api_rpc_request(api_module: str, rpc: str, payload: dict):
+ # pylint: disable=consider-using-f-string
+ url = "{}/operations/{}:{}".format('{}', api_module, rpc)
+ if payload is None:
+ data = None
+ elif RESTCONF_VERSION == 'draft-bierman02':
+ data = prepend_dict_keys({'input': payload}, api_module + ':')
else:
data = {'input': payload}
response = post_request(url, data)
res = response.json()
- return_key = {'rfc8040': 'transportpce-olm:output',
- 'draft-bierman02': 'output'}
- return_output = res[return_key[RESTCONF_VERSION]]
- return {'status_code': response.status_code,
- 'output': return_output}
-
-
-def olm_calculate_spanloss_current_request():
- url = "{}/operations/transportpce-olm:calculate-spanloss-current"
- response = post_request(url, None)
- res = response.json()
- return_key = {'rfc8040': 'transportpce-olm:output',
+ return_key = {'rfc8040': api_module + ':output',
'draft-bierman02': 'output'}
return_output = res[return_key[RESTCONF_VERSION]]
return {'status_code': response.status_code,
Code Review / transportpce.git / blobdiff