Standard または Advancedのライセンスで利用可能。
ポータル管理者およびバージョン管理者は、Python スクリプトを使用して、フィールド スタッフがブランチ バージョニングのデータに対して編集内容を反映した後に実行予定のレプリカ バージョンのリコンサイルを自動化することができます。
このワークフローは、ブランチ バージョン対応データを持つ同期対応フィーチャ サービスを使用して作成されたレプリカ バージョンの管理に適用できます。
ブランチ バージョン対応データを持つオフライン マップの作業についての詳細
注意:
このワークフローはブランチ バージョニングのためのものです。 従来のバージョニングでは、異なる管理タスクが必要とされます。 従来のバージョニングを使用している場合は、Python を使用してリコンサイル処理とポスト処理を自動化することもできます。
適用可能なシナリオ
次のセクションの Python の例では、フィーチャ サービス ArcGIS REST API および ArcPy ジオプロセシング ツールを使用して、整合性チェック属性ルールが設定されたレプリカ バージョンをリコンサイルおよびポスト処理します。 以下では、コード サンプルを使用する場面とサンプル スクリプトの処理内容について説明します。
注意:
このスクリプトは、特定のデータ設定を含むワークフローの例を表しています。 ワークフローがこの正確なデータ モデルに従わない場合もありますが、概念とスクリプト ロジックを使用して、組織向けに自動化されたスクリプトをカスタマイズすることができます。 たとえば、整合チェック ルールを使用しない場合もありますが、スクリプトを変更して整合チェック ルールの評価を削除することができます。 また、整合チェック ルールを使用する場合は、評価操作を実行する前に各バージョンをリコンサイルして整合チェックを実行し、各バージョンを個別にポストすることで、整合チェックに他のすべての編集者による編集が含まれていることを確認することができます。
前提条件:
この例では、データは次のように設定されています。
- フィーチャ サービスは、同期が有効で、[ダウンロードされたマップごとのバージョンの作成] オプションが選択された (ArcGIS Enterprise 10.8.1 以降が必要) ブランチ バージョン対応データを持っている必要があります。
- レプリカはサービスに、マップをオフラインにして作成されています。
- バージョン管理サーバーの機能がサービス上で有効になっています。
- スクリプトでポータル ユーザーに提供される認証情報は、バージョン管理権限が割り当てられた、デフォルトのポータル管理者ロールまたはカスタム ロールのメンバー用でなければなりません。
- また、この例では、データは整合性チェック属性ルールが適用され、整合性チェックサーバー機能が有効になっています。 これは、レプリカ バージョンの作業に必須ではありませんが、データの整合性を確保するための方法を提供します。
- ArcGIS REST API の replicas リソースを使用して、フィーチャ サービスのレプリカ バージョンのリストを取得します。
- ArcGIS REST API の versionsInfos リソースを使用して、バージョンのプロパティを取得します。 これらのプロパティを使用して、評価日が Null または修正日付が最終評価日より大きい (つまり、最終評価日以降、編集が行われている) バージョンのみを取得するためのフィルターを作成します。 フィルターの条件を満たすバージョンの名前を、評価対象の listOfVersionsToValidate リストに追加します。
- 評価リストの各バージョンについて、ArcGIS REST API の evaluate 操作を使用して、サービスの整合性チェック属性ルールを評価します。 評価に問題がなく、エラーが返されなかった場合、そのバージョンはリコンサイルおよびポストができます。 評価に問題はないが、エラーが返された場合、編集内容はポストされず、エラーを手動で調査および修正できるようメッセージが生成されます。
- エラーなしで問題なく評価が完了したバージョンに対しては、修正内容をポストするオプションが設定された [バージョンのリコンサイル (Reconcile Versions)] ツールが実行されます。
- すべてのレプリカ バージョンにリコンサイルのみ行う (ポストしない) オプションが設定された [バージョンのリコンサイル (Reconcile Versions)] ツールが、すべてのレプリカ バージョンに対して実行されます。
コードの例
以下の完結した Python コード例により、上記の操作が完了します。 このスクリプトには、完了した各操作の出力を保存するログ ファイルを生成し、スクリプトの完了後に表示できるオプションが含まれています。
# 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.')