Administradores del portal y administradores de versiones pueden utilizar scripts de Python para automatizar una conciliación programada de versiones de réplica después de que los trabajadores de campo hayan sincronizado ediciones con datos incluidos en versionado en rama.
Este flujo de trabajo es aplicable para administrar versiones de réplica que se crearon utilizando un servicio de entidades con sincronización habilitada con datos versionados en rama.
Más información sobre cómo trabajar con mapas sin conexión con datos versionados en rama
Escenario aplicable
El ejemplo de Python de la siguiente sección utiliza el servicio de entidades ArcGIS REST API y herramientas de geoprocesamiento ArcPy para conciliar y publicar versiones de réplica con la aplicación de reglas de atributos de validación. A continuación se describe cuándo se utilizaría la muestra de código y qué hace el script de muestra.
Nota:
Este script representa un ejemplo de flujo de trabajo con una configuración de datos específica. Es posible que sus flujos de trabajo no sigan este modelo de datos exacto, pero los conceptos y la lógica de script pueden seguir utilizándose para personalizar un script automatizado para su organización. Por ejemplo, es posible que no pueda utilizar reglas de validación, pero puede modificar el script para eliminar la evaluación de las reglas de validación. Además, en algunos casos, cuando se utilizan reglas de validación, puede ser conveniente conciliar, ejecutar la validación y publicar cada versión individualmente antes de ejecutar la operación de evaluación para asegurarse de que la validación incluya las ediciones realizadas por todos los demás editores.
Requisito previo:
En este ejemplo, la configuración de datos es la siguiente:
- El servicio de entidades debe tener habilitada la sincronización con datos versionados en rama con la opción seleccionada de Crear una versión para cada mapa descargado, que requiere ArcGIS Enterprise 10.8.1 o posterior.
- Se han creado réplicas en el servicio con un mapa sin conexión.
- La capacidad del servidor de administración de versiones se ha habilitado en el servicio.
- Las credenciales proporcionadas para el usuario del portal en el script deben ser para un miembro con el rol de administrador del portal predeterminado o un rol personalizado que tenga concedido el privilegio de administración de versiones.
- En este ejemplo, los datos también tienen reglas de atributos de validación aplicadas y la capacidad del servidor de validación habilitada. Este no es un requisito para trabajar con versiones de réplica, sino que proporciona una forma de asegurar la integridad de los datos.
- Obtenga una lista de versiones de réplica para el servicio de entidades utilizando el recurso replicas en ArcGIS REST API.
- Utilice el recurso versionsInfos en ArcGIS REST API para obtener las propiedades de la versión. Con estas propiedades, cree un filtro para obtener solo las versiones en las que la fecha de evaluación sea nula o la fecha modificada sea mayor que la fecha de la última evaluación (lo que significa que las ediciones se han realizado desde la última evaluación). Incorpore los nombres de las versiones que satisfacen el filtro para la lista listOfVersionsToValidate que se evaluará.
- Para cada versión de la lista de evaluación, utilice la operación evaluate en ArcGIS REST API para evaluar las reglas de atributos de validación en el servicio. Si la evaluación resulta satisfactoria sin que se devuelvan errores, la versión está preparada para conciliar y publicar. Si la evaluación resulta satisfactoria aunque se devuelven errores, no publique las ediciones; genere un mensaje para que los errores puedan inspeccionarse y solucionarse manualmente.
- Para versiones que se evaluaron satisfactoriamente sin errores, ejecute la herramienta Conciliar versiones con la opción de publicar ediciones.
- Ejecute la herramienta Conciliar versiones con la opción para conciliar únicamente (no publicar) todas las versiones de réplica.
Ejemplo de código
El ejemplo de código de Python completo siguiente completa las operaciones incluidas anteriormente. Este script incluye una opción para generar un archivo de registro que capture la salida de cada operación completada y pueda visualizarse después de que se complete el script.
# Import modules
import arcpy, traceback, urllib, json, urllib.request, urllib.parse, os, urllib.error, datetime
# Overwrite the reconcile log output each time the script is run
arcpy.env.overwriteOutput = True
# Script parameters
serviceName = "MyServiceName"
baseURL = "https://MyServer.MyDomain.com"
portalAdmin = "MyAdminUser"
portalAdminPwd = "MyAdmin.Password"
logFileScript = "C:/Logs/validateRecPostScriptLog.txt"
logfileOutputRecPost = 'C:/Logs/reconcile_log.txt'
# Choose to output a log file for the script
outputScriptReport = True
# Define functions
def openURL(url, params=None):
"""This function used to open a URL and returns the json response"""
try:
request_params = {'f':'pjson'}
if params:
request_params.update(params)
encodedParams = urllib.parse.urlencode(request_params)
request = urllib.request.urlopen(url, encodedParams.encode('UTF-8'))
response = request.read()
json_response = json.loads(response)
return json_response
except:
print (traceback.format_exc())
def versionInfo(versionOwner=""):
"""This function queries the versions owned by the versionOwner.
It returns a list of dictionaries."""
vmsUrlinfo = "{}/server/rest/services/{}/VersionManagementServer/versionInfos?&ownerFilter={}&includeHidden=&f=json&token={}".format(baseURL, serviceName, versionOwner, token)
response = openURL(vmsUrlinfo)
if response['success'] == True:
versionsDict = response['versions']
return versionsDict
else:
return("Unable to get version info")
def evaluateUrl(validationUrl, validate_params):
"""This function runs evaluate on the validation server
It returns the json response."""
evalJsonResp = openURL(validationUrl, validate_params)
if evalJsonResp['success'] == False:
return [False, evalJsonResp]
else:
return [True, evalJsonResp]
def generateMessage(msg, print_statement=True):
"""This function generates messages as the script runs. If print_statement
is set to True, print to the screen. If outputScriptReport is set to true,
write the message to the logfile"""
if outputScriptReport == True:
with open(logFileScript, 'a') as log_file:
log_file.write(msg + "\n")
if print_statement == True:
print(msg)
def recPostVersions(versionList, post):
"""This function runs the Reconcile Versions GP tool to reconcile
and optionally post to the feature service"""
if post == True:
postVersion = "POST"
elif post == False:
postVersion = "NO_POST"
# Reconcile and post the replica versions
# This tool is set to abort if there are conflicts and detects conflicts by object
arcpy.ReconcileVersions_management(featureService,
'ALL_VERSIONS',
'sde.DEFAULT',
versionList,
'NO_LOCK_ACQUIRED',
'ABORT_CONFLICTS',
'BY_OBJECT',
'FAVOR_EDIT_VERSION',
postVersion,
'KEEP_VERSION',
logfileOutputRecPost)
generateMessage(arcpy.GetMessages()+"\n")
# Start execution
generateMessage('Starting Validation/Reconcile/Post Automation Script... {:%Y-%b-%d %H:%M:%S}\n'.format(datetime.datetime.now()))
# Sign in to ArcGIS Enterprise
signIntoPortal = arcpy.SignInToPortal(baseURL+"/portal", portalAdmin, portalAdminPwd)
generateMessage("Signed into ArcGIS Enterprise {} as user {}".format(baseURL+"/portal", portalAdmin))
# Get the token returned by the SignIntoPortal arcpy function to use for making REST requests
token = signIntoPortal['token']
# Build the feature service URL
featureService = "{}/server/rest/services/{}/FeatureServer".format(baseURL, serviceName)
# Get a list of the replica versions from the REST endpoint
listOfRepVersions = []
replicaVersionsURL = featureService + "/replicas?returnVersion=true&f=pjson"
repVersionsJson = openURL(replicaVersionsURL, signIntoPortal)
for repVersion in repVersionsJson:
versionName = repVersion['replicaVersion']
listOfRepVersions.append(versionName)
# Create an empty list to append version names to validate
listOfVersionsToValidate = []
# Iterate through each version returned by the versionInfo() function to find
# the versions that need to be validated that are also in the listOfRepVersions list
for version in versionInfo():
print("")
# Parse the version info response, which is a python dictionary/json
# If the version name is sde.DEFAULT, pass since we do not want to evaluate the default version
if version['versionName'] == "sde.DEFAULT":
pass
# If the modifiedDate property is null, pass
elif version['modifiedDate'] == "None":
pass
# If the evaluation date is null, append the version name to the list to listOfVersions to be evaluated
elif version['evaluationDate'] == None:
if version['versionName'] in listOfRepVersions:
listOfVersionsToValidate.append(version['versionName'])
# If the evaluation date is not null, but it has been modifed since the last evaluation, add it to the list to be validated
elif version['evaluationDate'] != None and version['modifiedDate'] > version['evaluationDate']:
if version['versionName'] in listOfRepVersions:
listOfVersionsToValidate.append(version['versionName'])
# If none of these conditions are met
else:
generateMessage("Version {} will not be validated.".format(version['versionName']))
# Validate versions
generateMessage('The following versions will be validated: {}\n'.format(listOfVersionsToValidate))
# Create lists to contain versions where the validation passed or failed
failEval = []
passEval = []
# For each version in the list of versions, build the json request needed to validate
for i in listOfVersionsToValidate:
validate_params = { "gdbVersion": i,
"sessionId": "",
"evaluationArea": "",
"changesInVersion": "true",
"selection": "",
"evaluationType": '["validationRules"]',
"returnEdits": "true",
"async": "false",
"f": "pjson",
"token": token
}
# Build the REST URL used to validate the service
validationUrl = baseURL + "/server/rest/services/"+ serviceName +"/ValidationServer/evaluate"
# Call the evalVersion() function to validate the version
evalVersion = evaluateUrl(validationUrl, validate_params)
# If the evaluate failed, append to the failEval list
if evalVersion[0] == False:
generateMessage("Evalution of version {} failed".format(i))
generateMessage(str(evalVersion[1]))
failEval.append(i)
# If the evaluate passed, check to see if errors were returned
elif evalVersion[0] == True:
# If errors are returned, only reconcile this version
if evalVersion[1]['errorsIdentified'] != 0:
generateMessage("{} Errors were identified in version {}.\nThe version will be reconciled but will not be posted.\n".format((str(evalVersion[1]['errorsIdentified'])),i))
generateMessage(str(evalVersion[1]), False)
# If errors were not found this version can be posted
else:
generateMessage("Evaluation of version {} passed with no errors identified.\nThis version will be reconciled and posted.\n".format(i))
generateMessage(str(evalVersion[1]))
passEval.append(i)
# The versions that passed validation should be reconciled/posted
generateMessage('\nThe following versions passed validation and will be reconciled and posted: {}\n'.format(passEval))
# Run recPostVersions on the list of versions that passed evaluation with the option to post
recPostVersions(passEval, True)
# Open the reconcile log file and append the results to our report
with open(logfileOutputRecPost, 'r') as openRecLog:
generateMessage(openRecLog.read(), False)
# Run recPostVersions with the option to reconcile all replica versions, no post
recPostVersions(listOfRepVersions, False)
# Open the reconcile log file and append the results to our report
with open(logfileOutputRecPost, 'r') as openRecLog:
generateMessage(openRecLog.read(), False)
# Script execution complete
generateMessage('Validate, Reconcile, & Post Script Complete.')