def update_data_scan()

in datascan/python-api-sample-scripts/dq_update_scan.py [0:0]

• 24 lines of code
• 1 McCabe index (conditional complexity)

def update_data_scan():
    # Create a Dataplex client object
    print("Authenticating Dataplex Client...")
    dataplex_client = dataplex_v1.DataScanServiceClient()

    rules = [
        { # MODIFY BELOW
        "column": "COLUMN_NAME",
        "dimension": "COMPLETENESS",
        "non_null_expectation": {},
        "threshold": 0.98
        },
    ]

    data_scan_name = args.data_scan_name # format: "projects/{PROJECT-ID}/locations/{REGION}/dataScans/{DATA-SCAN-ID}"
    # Define a DataScan()
    data_scan = dataplex_v1.DataScan()
    data_scan.data_quality_spec.rules = rules
    data_scan.name = data_scan_name

    # Define an UpdateDataScanRequest()
    update_request = dataplex_v1.UpdateDataScanRequest()
    update_request.data_scan = data_scan
    update_request.update_mask = "dataQualitySpec"
    # one of: "data", "dataProfileSpec", "dataQualitySpec", "description", "displayName", "executionSpec", "labels"
    # See details at: https://cloud.google.com/dataplex/docs/reference/rest/v1/projects.locations.dataScans/patch

    print("Updating scan...")
    update_scan = dataplex_client.update_data_scan(request=update_request)
    update_scan_result = update_scan.result()
    updated_dq_spec = update_scan_result.data_quality_spec
    print("SUCCESS! Here is the updated DQ Spec --> ")
    print(updated_dq_spec)