Updating Feature Service Coded Value Domain with ArcGIS API for Python

[James_Whitacre]

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.

[RandyBurton · Accepted 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
      }]
    }
  }]
}‍‍‍‍‍‍‍‍‍‍‍‍‍

[FPCWAGIS_Admin]

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. 

[RandyBurton]

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.

[CassKalinski]

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')

[CassKalinski]

'''
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

[LindsayRaabe_FPCWA]

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?)