I am trying to update an attribute coded value domain using the ArcGIS API for Python using the Update Servcie Definition method. I am an admin for my AGO org.
Here is my code so far (if you are testing it, you will need to used your own parameters):
import arcgis
username = 'username'
gis = arcgis.gis.GIS('https://www.arcgis.com', username)
url = 'https://domain.com/arcgis/rest/services/Service_name/FeatureServer/0'
field_name = 'Field_Name'
cv_code = 1
cv_name = 'Code Name'
fs = arcgis.features.FeatureLayer(url, gis)
update_dict = {"fields": [{"name": field_name, "domain": {"codedValues": [{"name": cv_name, "code": cv_code}]}}]}
fs.manager.update_definition(update_dict)
When I run in my Jupyter Notebook, I get the error below. What am I missing? Is my 'update_dict' not correct (I ma inclined to think it isn't)? Is there a better way to script this out? Thanks.
Error:
Unable to update feature service layer definition.
Invalid definition for ''.
Invalid definition for System.Collections.Generic.List`1[ESRI.ArcGIS.SDS.FieldInfo]
---------------------------------------------------------------------------
RuntimeError Traceback (most recent call last)
<ipython-input-7-4fbf02d770c9> in <module>()
----> 1 fs.manager.update_definition(update_dict)
C:\Program Files\ArcGIS\Pro\bin\Python\envs\arcgispro-py3\lib\site-packages\arcgis\features\managers.py in update_definition(self, json_dict)
981 u_url = self._url + "/updateDefinition"
982
--> 983 res = self._con.post(u_url, params)
984 self.refresh()
985 return res
C:\Program Files\ArcGIS\Pro\bin\Python\envs\arcgispro-py3\lib\site-packages\arcgis\_impl\connection.py in post(self, path, postdata, files, ssl, compress, is_retry, use_ordered_dict, add_token, verify_cert, token, try_json, out_folder, file_name, force_bytes, add_headers)
1151 verify_cert=verify_cert, is_retry=True)
1152
-> 1153 self._handle_json_error(resp_json['error'], errorcode)
1154 return None
1155 except AttributeError:
C:\Program Files\ArcGIS\Pro\bin\Python\envs\arcgispro-py3\lib\site-packages\arcgis\_impl\connection.py in _handle_json_error(self, error, errorcode)
1171
1172 errormessage = errormessage + "\n(Error Code: " + str(errorcode) +")"
-> 1173 raise RuntimeError(errormessage)
1174
1175 class _StrictURLopener(request.FancyURLopener):
RuntimeError: Unable to update feature service layer definition.
Invalid definition for ''.
Invalid definition for System.Collections.Generic.List`1[ESRI.ArcGIS.SDS.FieldInfo]
(Error Code: 400)
Follow up:
Ok, in investigating further, I found this idea: https://community.esri.com/ideas/12962.
Great idea, and a good solution for my immediate issue. However, I am foreseeing a need in my greater use case to automate other tasks in conjunction with this task or to have a set of tools (either Python or eventually shipped with ArcGIS Pro) that allows for editing/updating other service definition properties. I know this can be done with the ArcGIS API for Python, but it seems cumbersome when going back and forth between JSON and Python. I also don't have a lot of experience with working with JSON in Python, so that could be one of my hindrances here.
Any help is still greatly appreciated.
Solved! Go to Solution.
You may need to supply the domain name in your update_dict.
{
"fields": [{
"name": "Field_Name",
"domain": {
"type": "codedValue",
"name": "Domain_Name",
"codedValues": [{
"name": "Code Name",
"code": 1
}]
}
}]
}
You may need to supply the domain name in your update_dict.
{
"fields": [{
"name": "Field_Name",
"domain": {
"type": "codedValue",
"name": "Domain_Name",
"codedValues": [{
"name": "Code Name",
"code": 1
}]
}
}]
}
I know this thread is a little old now, but it's the closest I've found in my searching. Did you (or anyone) have any luck with getting Domain values in an ArcOnline feature service to update based on an external list via python (or another automated method)?
My requirement is to have a pick list (Domain) of Staff Names or Contractor Names updated on a regular basis (i.e. Weekly) from an external source so that the list is representative of our staff/contractors at all times.
I have used the Update Definition (Feature Layer) of the REST API to modify domains (see example 6). I think adding to the domain list works best, as deleting a domain value may leave some invalid values in the field. Instead of deleting a domain, I would recommend changing the name/description to indicate it is out of date. If the field needing the domain update is used for symbology, you will also need to edit and update the "types" section of the JSON file.
I have not used the ArcGIS API for Python for this purpose. For this, I would explore the update_definition section of the API documentation.
Came across this thread a month or so ago. Trying to answer the same question about updating domains from a list. It has been 19 months since the question, but thought the attached code might be helpful for future searchers.
The two modules are a mashup from a broader solution. See comments in the code for config and changes needed. This is set to update multiple domains across multiple feature layers. The input is an Excel file, with one sheet per field/domain.
'''
domain_update.py
Imports from a Excel file. Each sheet in the file needs to be named
to match the field name containing the domain. Each sheet has two
columns: Code and Description. This is the list of code/name pairs
you want the domain to reflect.
'''
import os
import json
import logging
import pandas as pd
from arcgis_helper import ArcGISManager, AGOAccountInfo
#<<<<<<<<<<<<<<< Configuration Required >>>>>>>>>>>>>>>
DOMAIN_LOAD_EXCEL_FILE = r'C:\path\to\your\excel.xlsx'
AGOL_URL = r'https://yoururl.arcgis.com/'
FS_ID = #'ITEM_ID_FOR_FEATURE_SERVICE'
#<<<<<<<<<<<<<<< End of config >>>>>>>>>>>>>>>
class BaseDataMapper:
def __init__(self, logger, arcgis_manager=None):
self.logger = logger
self.arcgis_manager = arcgis_manager
self._layer_configs = {}
def set_arcgis_manager(self, arcgis_manager):
self.arcgis_manager = arcgis_manager
class DataMapper(BaseDataMapper):
def __init(self, logger, arcgis_manager=None):
super().__init__(logger, arcgis_manager)
def update_domains(self, update_data):
'''
Update domains for the layers in the indicated feature service
JSON format for update_data:
{
"solutionID": "SOLUTION_FEATURE SERVICE_ID",
"domainValues": [
{
"field": "ATTRIBUTE_NAME",
"codedValues": [
{
"description": "TEXT DESCRIPTION",
"code": "CODE"
}
]
}
]
}
'''
self.logger.info('Decomposing incoming JSON')
updates = json.loads(update_data)
solution_id = updates['solutionID']
new_domain_values = updates['domainValues'][:]
self.logger.info('Getting layers from AGOL Comm Solution')
fs_layers = self.arcgis_manager.get_feature_service_layers(solution_id)
for layer in fs_layers:
lyr_indx = fs_layers.index(layer)
self.arcgis_manager.set_feature_layer(solution_id, lyr_indx)
lyr_attributes = self.arcgis_manager.get_layer_attribute_names()
for domain_update in new_domain_values:
field_name = domain_update['field']
if field_name in lyr_attributes:
msg = f'Attempting to update {field_name} domain for layer {layer.properties.name}'
self.logger.info(msg)
self.logger.info('Vetting incoming domain code type')
# new_coded_values = domain_update['codedValues']
new_coded_values = self._check_domain_code_types(field_name, domain_update['codedValues'])
self.logger.info('Checking for deprecated values that need to be retained')
current_values = self.arcgis_manager.get_attribute_unique_values(field_name)
if current_values:
new_coded_values = self.arcgis_manager.append_deprecated_domain_codes(field_name, new_coded_values)
self.logger.info('Verifying there is an "Unknown" value present in the domain codes')
new_coded_values = self._verify_unknown_code_present(field_name, new_coded_values)
self.logger.info('Updating the field definition with the new domain codes')
new_update_dict = self.arcgis_manager.create_domain_update_dictionary(field_name, new_coded_values)
try:
self.arcgis_manager.update_feature_layer_definition(new_update_dict)
except RuntimeError as update_err:
err_msg = f'Failed to update {field_name} domain for layer {layer.properties.name}. Error: {update_err}'
self.logger.error(err_msg)
else:
msg = f'{field_name} domain updated for layer {layer.properties.name}'
self.logger.info(msg)
else:
msg = f'Field domain {field_name} skipped for layer {layer.properties.name}. Field not in layer.'
self.logger.warning(msg)
return
def _check_domain_code_types(self, check_field, domain_values):
'''
Check the incoming code values and match the type to the related
field. Return an updated list as needed.
'''
field_type = self.arcgis_manager.get_attribute_type(check_field)
updated_domain_values = []
for coded_values in domain_values:
if field_type == 'esriFieldTypeString':
code_set = {
'name': coded_values['name'],
'code': str(coded_values['code'])
}
elif field_type in ['esriFieldTypeSmallInteger', 'esriFieldTypeInteger']:
code_set = {
'name': coded_values['name'],
'code': int(coded_values['code'])
}
else:
raise TypeError('Error in type for incoming domain data')
updated_domain_values.append(code_set)
return updated_domain_values
def _verify_unknown_code_present(self, check_field, domain_codes):
'''
Verify there is a 'unknown' present in the domain codes.
Insert if missing and return the verified code list.
'''
domain_descriptions = [domain['name'] for domain in domain_codes]
validated_domain_list = domain_codes
if 'Unknown' not in domain_descriptions:
field_type = self.arcgis_manager.get_attribute_type(check_field)
if field_type == 'esriFieldTypeString':
unknown_code = {"name": "Unknown", "code": "UNK"}
else:
unknown_code = {"name": "Unknown", "code": 0}
validated_domain_list.insert(0, unknown_code)
return validated_domain_list
def _create_domain_load(field_name, domain_data:pd.DataFrame):
'''
Take the data from a dataframe and put it into the json structure
needed for the domain processing
'''
domain_load = {
'field': field_name,
'codedValues': []
}
for row in domain_data.itertuples():
code_set = {
'name': row.Description,
'code': row.Code
}
domain_load['codedValues'].append(code_set)
return domain_load
def main():
'''Setup and load from Excel'''
# TODO: Roll your own on the logger
logger = setup_your_logger()
data_mapper = DataMapper(logger)
# TODO: Need to create a method to input your UID/PW
credentials = get_username_password()
ago_account_info = AGOAccountInfo()
ago_account_info.url = AGOL_URL
ago_account_info.username = credentials[0]
ago_account_info.password = credentials[1]
arcgis_manager = ArcGISManager(ago_account_info, False)
data_mapper.set_arcgis_manager(arcgis_manager)
file_to_upload = DOMAIN_LOAD_EXCEL_FILE
data_load = {
"solutionID": FS_ID,
"domainValues": []
}
logger.info('Getting the data from the manual upload template')
if os.path.exists(file_to_upload):
df_uploads = pd.read_excel(file_to_upload, sheet_name=None, usecols=['Code', 'Description'])
for df_name, df_data in df_uploads.items():
logger.info('Assembling JSON for %s field', df_name)
domain_update = _create_domain_load(df_name, df_data)
data_load["domainValues"].append(domain_update)
logger.info('Starting domain upload process')
data_mapper.update_domains(json.dumps(data_load))
else:
logger.critical('Excel template not found. Unable to process manual domain updates.')
if __name__ == '__main__':
main()
print('End of manual domain load script')
'''
arcgis_helper.py
'''
from arcgis.gis import GIS
class AGOAccountInfo:
url = None
username = None
password = None
class ArcGISManager:
# global variables
account_info = None
gis = None
feature_layer = None
test_run = False
def __init__(self, ago_account_info, test_run=False):
self.account_info = ago_account_info
self.gis = GIS(ago_account_info.url, ago_account_info.username, ago_account_info.password)
self.test_run = test_run
def set_feature_layer(self, feature_service_id, sublayer_index=None):
if sublayer_index == None:
sublayer_index = 0
self.feature_layer = self._get_feature_layer(feature_service_id, sublayer_index)
def _get_feature_layer(self, feature_service_id, sublayer_index):
feature_layer = None
feature_service = self.gis.content.get(feature_service_id)
if feature_service != None and len(feature_service.layers) > sublayer_index:
feature_layer = feature_service.layers[sublayer_index]
if feature_layer == None:
raise Exception('The layer with id {} could not be found.'.format(feature_service_id))
return feature_layer
def get_existing_features(self, where):
query_results = self.feature_layer.query(where=where)
features = query_results.features
return features
def get_feature_service_layers(self, feature_service_id):
'''
Gets the set of feature layers of a feature service. If
feature service does not exist or no layers are present,
raises exception.
'''
feature_layers = None
feature_service = self.gis.content.get(feature_service_id)
if feature_service and len(feature_service.layers) > 0:
feature_layers = feature_service.layers
elif feature_service and len(feature_service.layers) == 0:
raise ValueError(f'The feature service with id {feature_service_id} has no layers.')
elif feature_service is None:
raise ValueError(f'The feature service with id {feature_service_id} could not be found.')
return feature_layers
def update_feature_layer_definition(self, update_data):
'''
Updates the layer definition. See Esri doc for various
input data structures that can be used
https://developers.arcgis.com/rest/services-reference/online/update-definition-feature-layer-.htm
Results structure:
{
"error": {
"code": 400,
"message": "",
"details": [
"Unable to update feature service layer definition."
]
}
}
'''
return_value = False
results = self.feature_layer.manager.update_definition(update_data)
return_value = None
if 'success' in results:
return_value = True
else:
err_code = results['error']['code']
err_msg = results['error']['message']
err_details = results['error']['details']
err_string = f'Code: {err_code} Message: {err_msg} Details: {str(err_details)}'
raise RuntimeError(err_string)
return return_value
def get_attribute_type(self, check_field):
'''
Gets the data type associated with the given field.
Returns None if field not found.
'''
field_type = None
for field in self.feature_layer.properties.fields:
if field.name == check_field:
field_type = field.type
break
return field_type
def get_layer_attribute_names(self):
'''
Get the names of the current layers attributes
'''
layer_attributes = []
for field in self.feature_layer.properties.fields:
layer_attributes.append(field.name)
return layer_attributes
def get_domain_name(self, check_field):
'''
Gets the domain associated with the given field.
Returns None if field not found or if there is no
domain for the field.
'''
domain_name = None
for field in self.feature_layer.properties.fields:
if field.name == check_field and field.domain:
domain_name = field.domain.name
break
return domain_name
def get_domain_code_description(self, check_field, domain_code):
'''
Gets the description associated with the given field and domain code
'''
descrip = None
for field in self.feature_layer.properties.fields:
if field.name == check_field and field.domain:
for domain_pair in field.domain.codedValues:
if domain_pair.code == domain_code:
descrip = domain_pair.name
break
break
return descrip
def get_domain_type(self, check_field):
'''
Get the domain type for the given field
'''
domain_type = None
for field in self.feature_layer.properties.fields:
if field.name == check_field and field.domain:
domain_type = field.domain.type
break
return domain_type
def get_domain_codes(self, check_field):
'''
Gets a list of current domain codes for the field
'''
domain_codes = []
for field in self.feature_layer.properties.fields:
if field.name == check_field and field.domain:
for domain_pair in field.domain.codedValues:
domain_codes.append(domain_pair['code'])
break
return domain_codes
def get_domain_coded_values(self, check_field):
'''
Gets a list of current domain codedValues for the field
codedValues structure:
{
"name": "DESCRIPTION",
"code": "CODE"
}
'''
domain_coded_values = []
for field in self.feature_layer.properties.fields:
if field.name == check_field and field.domain:
domain_coded_values = field.domain.codedValues
break
return domain_coded_values
def append_deprecated_domain_codes(self, check_field, new_codes):
'''
Compares a codedValue list against the existing domain
codedValue list. Retains any codes that have dropped, but who
have been used in the field. Makes these as deprecated.
Returns an updated codedValues dictionary.
Structure for new_codes and updated_codes:
[
{
"name": "Description of code",
"code": "Domain code"
}
]
'''
updated_codes = new_codes[:]
update_code_list = [x['code'] for x in new_codes]
current_codes_used = self.get_attribute_unique_values(check_field)
for code in current_codes_used:
if code not in update_code_list:
current_name = self.get_domain_code_description(check_field, code)
if current_name:
new_name = '_'.join(['DEPRECATED', current_name])
updated_codes.append(
{
'name': new_name,
'code': code
}
)
else:
raise ValueError('Domain description missing')
return updated_codes
def create_domain_update_dictionary(self, check_field, new_coded_values):
'''
Create the dictionary to be used with the update-definition
method of ArcGIS layer.manager
'''
update_dictionary = {"fields": []}
domain_name = self.get_domain_name(check_field)
domain_type = self.get_domain_type(check_field)
update_dictionary['fields'].append(
{
'name': check_field,
'domain': {
'name': domain_name,
"type": domain_type,
'codedValues': new_coded_values
}
}
)
return update_dictionary
def get_attribute_unique_values(self, attribute):
'''
Get a list of unique values used in the field
of the current feature layer
'''
current_values = []
query = "1=1"
features = self.get_existing_features(query)
for feature in features:
# Checks if attribute is present and if it has a value other than null/None
if attribute in feature.attributes and feature.attributes[attribute]:
current_values.append(feature.attributes[attribute])
# Reduce to unique values
if current_values:
current_values = list(set(current_values))
return current_values
Hi @CassKalinski I'm hoping you can clarify a couple of things for me. It looks like you would execute the domain_update code, and that calls in the arcgis_helper code (presumably stored in the same folder?).
You also mention that this would be used to update domains in multiple feature services, though I only see an input parameter for 1 feature service ID. Do you have this code copied for each feature service and run as required? Or do you have another solution (i.e. such as a 3rd code that contains the feature service ID's which get passed to the domain_update.py and iterates through each ID?)
@LindsayRaabe_FPCWA It is one feature service ID but it will loop through all the layers in that service. He does say that in his original explanation.
Also want to thank @CassKalinski as I have found this a very useful script.