Switch ArcGIS Feature Layer View to new source Feature Layer

switch_view.py

def switch_view_to_new_source(view_item: Union[str, Item], new_source_item: Union[str, Item], gis: Optional[GIS] = None) -> bool:
    """
    Switch an ArcGIS Online or ArcGIS Enterprise Feature Layer View from one source to another with _significant_ error catching.

    Before switching to the Feature Layer View to the new source, the following checks are performed.
    
    - Layer ID: Ensure layer ids match.
    - Geometry Type: Ensure the correct geometry type such as point, line or polygon.
    - Spatial Reference: Ensure geometry spatial references match.
    - Needed Fields: Ensure fields needed by the view are in the new source.
    - Field Data Types: Ensure fields in the new source are the correct data types, match the data types needed in the view.

    Args:
        view_item: Item or item id for the view being switched.
        new_source_item: Item or item id for the new feature layer source.
        gis: GIS object instance where the view and new source reside logged in as a user with permissions to modify the view.
    """
    # ensure there is a GIS to work with
    if gis is None and arcgis.env.active_gis is None:
        raise EnvironmentError('Cannot access an active GIS in the environment. Please populate the gis parameter with an instantiated GIS object.')
    elif gis is None:
        gis = arcgis.env.active_gis

    # if the provided items are strings, try to retrieve the Items from the GIS
    if isinstance(view_item, str):
        view_itm = gis.content.get(view_item)

        if view_itm is None:
            raise ValueError(f'Cannot locate an item in {gis.properties.url} using the provided view item id ({view_item}).')
        else:
            logging.debug(f'Created Item for view item id: {view_item}'
            view_item = view_itm

    if isinstance(new_source_item, str):
        src_itm = gis.content.get(new_source_item)

        if src_itm is None:
            raise ValueError(f'Cannot locate an item in {gis.properties.url} using the provided new source item id ({new_source_item}).')
        else:
            logging.debug(f'Created Item for new source item id: {new_source_item}')
            new_source_item = src_itm
            
    # ensure provided view item is truly a view
    if not 'View Service' in view_item.typeKeywords:
        raise ValueError('The provided view item does not appear to be a Feature Layer View.')

    # ensure provided new source is a feature service
    if new_source_item.type != 'Feature Service':
        raise ValueError('The provided new source does not appear to be a Feature Service.')
    
    # ensure both the view and source only have one layer
    if len(view_item.layers) > 0 or len(new_source_item.layers) > 0:
        raise ValueError('This function can only switch sources for a layer view and new source with one feature layer in each.')
    
    # get the layer from the view item
    view_layer = view_item.layers[0]
    
    # get the layer from the new source item
    new_source_layer = new_source_item.layers[0]
    
    # ensure the layer indexes match
    if view_layer.properties.id == new_source_layer.properties.id:
        logger.debug('View and new source layer ids (layer indices) match.')
    else:
        raise ValueError(f'The layer ids (layer indices) do not match. The view layer index is {view_layer.properties.id} and the new source layer id is {new_source_layer.properties.id}')
    
    # ensure geometry types match
    if view_layer.properties.geometryType == new_source_layer.properties.geometryType:
        logger.debug(f'Geometry types match, are both {view_layer.properties.geometryType}')
    else:
        raise ValueError(f'The view layer geometry type ({view_layer.properties.geometryType}) and new source layer geometry type ({new_source_layer.properties.geometryType}) do not match.')
    
    # get both the wkid and latest wkid values in sets
    view_sr_set = set((view_layer.properties.spatialReference.get('wkid'), view_layer.properties.spatialReference.get('latestWkid')))
    new_sr_set = set((new_source_layer.properties.spatialReference.get('wkid'), new_source_layer.properties.spatialReference.get('latestWkid')))
    
    # get rid of any null values - common if latest wkid is not present
    for sr_set in (view_sr_set, new_sr_set):
        sr_set.discard(None)
    
    # determine if spatial references are matching by determining if any wkids are common between the view and new source
    if len(view_sr_set.intersection(new_sr_set)) == 0:
        raise ValueError('The view and new source spatial references do not match.')
    else:
        logger.debug('The view and new source spatial references match.')
    
    # get dictionaries of the field name and field type in both the view and new source
    get_field_def = lambda lyr: {fld.get('name'): fld.get('type') for fld in view_layer.properties.fields}
    
    view_flds = get_field_def(view_lyr)
    new_source_flds = get_field_def(new_source_layer)
    
    # get any fields in the view, but missing from the source
    missing_fld_set = set(view_flds.keys()).difference(new_source_flds.keys())
    
    if len(missing_fld_set) > 0:
        raise ValueError(f'Fields necessary for the view ({", ".join(missing_fld_set)}) cannot be found in the new source fields.')
    else:
        logger.debug('All fields necessary for the view are present in the new source layer.')
    
    # check the data types to ensure the field data types are compatible
    bad_fld_typ_nm_lst = [(v_nm, v_typ, new_source_flds.get(v_nm)) for v_nm, v_typ in view_flds.items() if v_typ != new_source_flds.get(v_nm)]
    
    # if any incongruent field types
    if len(bad_fld_typ_nm_lst) > 0:
    
        # create a list of field types with incongruent field types
        bad_fld_typ_lst = [f'{nm}: {v_typ} / {n_typ}' for nm, v_typ, n_typ in bad_fld_typ_nm_lst]
        bad_fld_typ_str = ', '.join(bad_fld_typ_lst)
    
        # notify in error where issues exist
        raise ValueError(f'Incompatible field types detected between the view and new source layers - {bad_fld_typ_str}')
    
    else:
        logging.debug('All field types needed by the view are the correct data type in the new source layer.')
    
    # create the dictionary for removing the current feature layer source from the definition
    del_dict = { "layers": [ {"id": view_layer.properties.id} ] }
    
    # use the dictionary to remove the current feature layer source from the layer definition
    res = view_lyr.manager.delete_from_definition(del_dict)
    
    if res.get('success'):
        logging.debug('Successfully removed old source from view.')
    else:
        raise Exception(f'Unable to remove the old source from the view.')
    
    # create the dictionary for adding the new source to the view
    add_dict = {
      "layers": [
        {
          "adminLayerInfo": {
            "viewLayerDefinition": {
              "sourceServiceName": new_source_item.name,
              "sourceLayerId": new_source_layer.properties.id,
              "sourceLayerFields": "*"
            },
            "geometryField": {
              "name": "Shape"
            },
            "xssTrustedFields": ""
          }, 
          "id": new_source_layer.properties.id,
          "name": new_source_item.name
        }
      ]
    }
    
    # use the dictionary to add the new source feature layer to the view definition
    for attempt_idx in range(3):
        res = view_lyr.manager.add_to_definition(add_dict)
    
        # save result to variable
        add_success = res.get('success')
    
        # if successful, break out of attempt loop
        if add_success:
            logging.debug('Successfully replaced source in view.')
            break        
    
    if not add_success:
        raise Exception(f'Unable to add new source to the view.')
    
    # if everything successful this far, return boolean true
    return True