
# example: authenticate + obtain token (http-get)
headers = {'X-CSRF-Token': 'Fetch', 'Accept': 'application/json', 'Content-Type': 'application/json'}
response = client.get('http://' + host + ':' + port + '/sap/opu/odata/sap/ProcessManagement?sap-client=' + sap_client, auth=(user, passwd), headers=headers)
token = response.headers['x-csrf-token']
# example: retrieve and json-decode branch content (http-get)
headers = {'Accept': 'application/json', 'Content-Type': 'application/json'}
response = client.get('http://'+ host + ':' + port + '/sap/opu/odata/sap/ProcessManagement/BranchContentSet(BranchId=\'' + branch_id + '\',ScopeId=\'SAP_DEFAULT_SCOPE\',SiteId=\'\',SystemRole=\'D\')/$value?sap-client=' + sap_client ,auth=(user, passwd),headers=headers)
branch_content = response.json()
# example: loop through nodes
for section in branch_content['sections']:
if section['section-id'] == 'NODES':
nodes_list = json.loads(section['section-content']) # json-de-code section-content, as it is a string with an extra layer of json-encoding
for my_node in nodes_list:
# --- check & modify node here ---
section['section-content'] = json.dumps(nodes_list) # json-re-encode modified section-content
# example: json-re-encode branch content and upload to solman (http-put)
data = json.dumps(branch_content)
headers = {'X-CSRF-Token': token, 'Accept': 'application/json', 'Content-Type': 'application/json'}
url = 'http://'+ host + ':' + port + '/sap/opu/odata/sap/ProcessManagement/BranchContentImporterSet(BranchId=\'' + branch_id + '\',ChangeDocumentId=\'' + change_document_id + '\')/$value'
response = client.put(url, data=data, headers=headers)
# selection criteria (which nodes to update?)
select_obj_type = 'TESTDOCUMENT' # object type i.e. DOCUMENT, TESTDOCUMENT, PROC, ...
select_attr_description = re.compile('New.*') # description - Regular Expression or exact match (case sensitive!)
select_attr_smd_state = '0COPY_EDITING' # status i.e. 0IN_PROGRESS, 0COPY_EDITING, 0RELEASED, ...
# new attribute value(s) to be set (for nodes matching the selection criteria)
update_attr_smd_state = '0RELEASED' # status (see above)
The changes can be verified in Solution Documentation, List View:
import json
import requests
import re
import pprint
host = 'SOLMAN.HOSTNAME' #SolMan Hostname or IP
port = '50000' #Port (most likely 5000)
sap_client = '800' #Client
user = 'USERNAME' #Username
passwd = 'PASSWORD' #Password
solution_name = 'DELETE_ME' #Solution Name
branch_name = 'DEVELOPMENT' #Branch Name
change_document_id = '' # only needed if branch is under change control
SIMULATION_ONLY = False # simulate first (no import) then change this flag to False to perform actual import
# selection criteria (which nodes to update?)
select_obj_type = 'TESTDOCUMENT' # object type i.e. DOCUMENT, TESTDOCUMENT, PROC, ...
select_attr_description = re.compile('New.*') # description - Regular Expression or exact match (case sensitive!)
select_attr_smd_state = '0COPY_EDITING' # status i.e. 0IN_PROGRESS, 0COPY_EDITING, 0RELEASED, ...
# new attribute value(s) to be set (for nodes matching the selection criteria)
update_attr_smd_state = '0RELEASED' # status (see above)
def get_solution_by_name (solution_name):
# returns a list of solutions by name
response = requests.get('http://'+ host + ':' + port + '/sap/opu/odata/sap/ProcessManagement/SolutionSet?sap-client=' + sap_client, auth=(user, passwd), headers=headers)
return list(filter(lambda s: s['SolutionName'] == solution_name, response.json()['d']['results']))
def get_branch_by_name (solution_id, branch_name):
# returns a list of branches by name and solution_id
response = requests.get('http://'+ host + ':' + port + '/sap/opu/odata/sap/ProcessManagement/BranchSet?sap-client=' + sap_client, auth=(user, passwd), headers=headers)
return list(filter(lambda b: b['BranchName'] == branch_name and b['SolutionId'] == solution_id, response.json()['d']['results']))
def get_node_attribute_values(my_node, attribute_name):
# returns the value of an attribute of a node_id
res = []
for my_attribute in my_node['attributes']:
if my_attribute['attr_type'] == attribute_name: res = my_attribute['values']
return res
#Check connection and retrieve token
client = requests.session()
headers = {'X-CSRF-Token': 'Fetch', 'Accept': 'application/json', 'Content-Type': 'application/json'}
try:
response = client.get('http://' + host + ':' + port + '/sap/opu/odata/sap/ProcessManagement?sap-client=' + sap_client, auth=(user, passwd), headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print('Connection unsuccessful:\n', e)
exit(1)
token = response.headers['x-csrf-token']
print('Connection successful. Received token:', token)
# get solution id (by name), branch id (by name)
solution_id = get_solution_by_name(solution_name)[0]['SolutionId']
branch_id = get_branch_by_name(solution_id, branch_name)[0]['BranchId']
print('Fetching branch content...', end='')
response = client.get('http://'+ host + ':' + port + '/sap/opu/odata/sap/ProcessManagement/BranchContentSet(BranchId=\'' + branch_id + '\',ScopeId=\'SAP_DEFAULT_SCOPE\',SiteId=\'\',SystemRole=\'D\')/$value?sap-client=' + sap_client ,auth=(user, passwd),headers=headers)
branch_content = response.json()
print('done.')
match_counter = 0
# get the section-content of the NODES section...
for section in branch_content['sections']:
if section['section-id'] == 'NODES':
nodes_list = json.loads(section['section-content']) # json-de-code section-content, as it is a string with an extra layer of json-encoding
# ... and check for each node whether it matches the selection criteria
for my_node in nodes_list:
if my_node['obj_type'] == select_obj_type: # object tye matches selection criteria?
if get_node_attribute_values(my_node, '_SMD_STATE')[0] == select_attr_smd_state: # status matches selection criteria?
if re.search(select_attr_description, get_node_attribute_values(my_node, '_DESCRIPTION')[0]): # description matches selection criteria?
#pprint.pprint(my_node)
# ...if there is a match: update the node
match_counter-=-1 #;-)
print('Match ' + str(match_counter) + ' (before): \t' + str(my_node))
# each node's "attributes" is a list of dictionaries (each one having an attr_type and a value). To target the dictionary we want (having attr__type = _SMD_STATE) we need its index in the list.
attr_smd_state_index = next((index for (index, attribute) in enumerate(my_node['attributes']) if
attribute["attr_type"] == "_SMD_STATE"), None)
my_node['attributes'][attr_smd_state_index]['values'] = [update_attr_smd_state]
print('Match ' + str(match_counter) + ' (after 😞 \t' + str(my_node))
print('Total matching nodes found: \t' + str (match_counter))
section['section-content'] = json.dumps(nodes_list) # json-re-encode modified section-content
if SIMULATION_ONLY:
print('Simulation only --> skipping Import.')
exit(0)
# if this is not a simulation run: import changes to solution manager
url = 'http://'+ host + ':' + port + '/sap/opu/odata/sap/ProcessManagement/BranchContentImporterSet(BranchId=\'' + branch_id + '\',ChangeDocumentId=\'' + change_document_id + '\')/$value'
data = json.dumps(branch_content)
headers = {'X-CSRF-Token': token, 'Accept': 'application/json', 'Content-Type': 'application/json'}
print('Importing modified branch content (expecting http 204)...: ', end='')
response = client.put(url, data=data, headers=headers)
print(response)
You must be a registered user to add a comment. If you've already registered, sign in. Otherwise, register and sign in.
User | Count |
---|---|
16 | |
12 | |
11 | |
9 | |
8 | |
8 | |
8 | |
8 | |
7 | |
6 |