Samples

Basic Concepts

A quality verification is a test run based on a specific quality specification. A quality specification holds one or more quality conditions. Each quality condition represents a test algorithm configured for one or more specific datasets.

In order to execute a verification the prosuite.verification.service class is used to create the communication channel to the server and to start the verification of a specific quality specification.

service = prosuite.verification.Service(host_name='localhost', port_nr=5151)
service.verify(specification=my_specification)

A quality verification can be based on an XML specification (exported from the ProSuite Data Dictionary) or on a specification created in code, containing a list of quality conditions. Quality conditions are created with the factory class prosuite.factories.quality_conditions.Conditions which contains all available test algorithms. Intellisense provides the method parameters and help/docstrings but for an overview over the available tests or a general introduction, refer to the ProSuite HTML help or the ‘Quick Reference’.

Before running the verification in Python, make sure the server is running, for example by starting prosuite-qa-microservice.exe. By default the communication channel is http://localhost:5151.

Verify a specification created in code

The Specification class holds a set of conditions that can be configured programmatically.

Define the Data Model (= workspace)
Create the Datasets (= feature classes, tables with optional filter) in the model
Create a Service instance containing the connection properties
Define the specification: create a prosuite.quality instance
Create Condition instances using the static Conditions class and add them to the specification
Optionally define the verification perimeter
Optionally define the verification output directory
Execute the verification

 import prosuite as ps

 model = ps.Model("TopoModel", "D:\Test Data\ExtractStGallen.gdb") # From "ProSuite Documentation / SampleData.zip"
 datasets = [ps.Dataset("TLM_FLIESSGEWAESSER", model),
             ps.Dataset("TLM_STRASSE", model)]

 service = ps.Service(host_name='localhost', port_nr=5151) # You might want to change this to the host and port of your ProSuite installation

 simpleSpecification = ps.Specification(
     name='MinimumLengthSpecification',
     description='A very simple quality specification checking feature and segment length of roads and rivers')

 for dataset in datasets:
     simpleSpecification.add_condition(ps.Conditions.qa_min_length_0(dataset, limit=10, is3_d=False))
     simpleSpecification.add_condition(ps.Conditions.qa_segment_length_0(dataset, 1.5, False))

 envelope = ps.EnvelopePerimeter(x_min=2750673, y_min=1215551, x_max=2765845, y_max=1206640)

 out_dir = 'C:/temp/verification_output' # You might want to change this to a directory that exists on your system, also make sure no Issue.gdb exists in this directory

 verification_responses = service.verify(specification=simpleSpecification, output_dir=out_dir, perimeter=envelope)

 for verification_response in verification_responses:
     print(verification_response.message)

Response Messages

Creating external issue file geodatabase
Starting quality verification using quality specification MinimumLengthSpecification with verification tile size 5000Extent: 15172 x 8911
X-Min: 2750673
Y-Min: 1206640
X-Max: 2765845
Y-Max: 1215551

Verifying quality conditions per cached tiles (container tests)
Processing tile 0 of 8: XMin: 2’750’673.00 YMin: 1’206’640.00 XMax: 2’755’673.00 YMax: 1’211’640.00
Processing tile 1 of 8: XMin: 2’755’673.00 YMin: 1’206’640.00 XMax: 2’760’673.00 YMax: 1’211’640.00
Processing tile 2 of 8: XMin: 2’760’673.00 YMin: 1’206’640.00 XMax: 2’765’673.00 YMax: 1’211’640.00
Processing tile 3 of 8: XMin: 2’765’673.00 YMin: 1’206’640.00 XMax: 2’765’845.00 YMax: 1’211’640.00
Processing tile 4 of 8: XMin: 2’750’673.00 YMin: 1’211’640.00 XMax: 2’755’673.00 YMax: 1’215’551.00
Processing tile 5 of 8: XMin: 2’755’673.00 YMin: 1’211’640.00 XMax: 2’760’673.00 YMax: 1’215’551.00
Processing tile 6 of 8: XMin: 2’760’673.00 YMin: 1’211’640.00 XMax: 2’765’673.00 YMax: 1’215’551.00
Processing tile 7 of 8: XMin: 2’765’673.00 YMin: 1’211’640.00 XMax: 2’765’845.00 YMax: 1’215’551.00
Quality verification finishedNumber of verified datasets: 2.
Number of verified conditions: 4
No category
QaMinLength(0) TLM_FLIESSGEWAESSER - errors: 290
QaMinLength(0) TLM_STRASSE - errors: 974
QaSegmentLength(0) TLM_FLIESSGEWAESSER - errors: 1733
QaSegmentLength(0) TLM_STRASSE - errors: 1939
Warning count: 0
Error count: 4’936
The quality specification is not fulfilled

Issues written to C:\temp\verification_output\Issues.gdb

Verification report written to C:\temp\verification_output\verification.xml
Html report:
C:\temp\verification_output\verification.html
Quality specification report:
C:\temp\verification_output\qualityspecification.html

Control verification with the Issue class

The Issue class can be used to control the verification process. It can be used to stop the verification process when a certain issue condition is met. A sample Python script can be found in the ProSuite Python Samples repository.

issue_allowable = True

 for verification_response in verification_responses:
     if len(verification_response.issues) > 0:
         for issue in verification_response.issues:
             # Demo Prints

             # print(issue.description)
             # print(issue.involved_objects)
             # print(issue.geometry)
             # print(issue.issue_code)
             # print(issue.allowable)
             # print(issue.stop_condition)

             if issue.allowable is False:
                 print(f"Not allowed issue met: {issue.description} in {issue.involved_objects[0].table_name}")
                 print("Stopping verification")
                 issue_allowable = False
                 break

     if issue_allowable is False:
         break

Verification using XML Specification

Create a Service instance. In this example the service runs on a remote server machine.
Define the quality specification: create a XmlSpecification instance from a specification.qa.xml file.
Define the verification output directory
Optionally define the verification perimeter
Execute the verification

import prosuite

service = prosuite.verification.Service(host_name='arcgis_server', port_nr=5151)

xml_file = "\\share\QA\specifications\road_specification.qa.xml"
sde_file = "\\share\connection_files\production_QA_version.sde"

# Data source replacements: see notes below
xml_spec = prosuite.XmlSpecification(specification_file=xml_file,
                                     specification_name="Produktionsunterstuetzung",
                                     data_source_replacements=[["ProductionModel", sde_file]])

out_dir = '\\share\QA\results\verification_output'

for verification_response in service.verify(specification=xml_spec, output_dir = out_dir):
    print(verification_response.message_level)
    print(verification_response.service_call_status)
    print(verification_response.message)

Notes:

Directories: The specified paths must be accessible by the server, hence use UNC-paths.

Data Source Replacements: The datasets in the XML specifications use a Workspace ID as reference to the database. The Workspace ID is defined at the bottom of the XML file. Data source replacements are provided in the format of string pairs [<workspace_id>, <path_to_geodatabase>].

Examples:

[“TLM_Data”, “C:/full/path/to/connection.sde”]
[“ERM_Data”, “C:/full/path/to/data.gdb”]

For each Workspace ID defined at the very end of the XML file that should be replaced, provide a path to a file geodatabase or an sde file.

If no data source replacement is provided, a connectionString must be defined for the respective workspace in the XML. This is achieved by exporting the full workspace information when exporting the XML specification in the data dictionary editor by checking the Option ‘Export workspace connection information’.
If one or more datasource replacements are provided they will be used to open the referenced datasets in the XML by using the specified geodatabase for the respective workspace id.

For more details see the documentation of the verification service: prosuite.verification.Service.

Get specification names from XmlSpecification

import prosuite
xml_file = 'C:/temp/road_specification.qa.xml'
names_list = prosuite.XmlSpecification.get_specification_names(xml_file)
print(names_list)

Verification using DDX Specification

Using the DDX Specification, you can directly run quality verification defined in your Data Dictionary Editor through the Python API. This approach allows you to leverage predefined quality specifications and dataset configurations, making it simple to execute targeted data quality checks programmatically based on the settings in your data dictionary.

Create a Service instance. In this example, the service runs on a local server machine.
Define the quality specification by creating a DdxSpecification instance using the ddx_id and the project_short_name. These identifiers can be found in the Data Dictionary Editor.
Define the verification parameters, including any specific dataset ID and object IDs to verify. Both the dataset ID and object IDs can also be checked in the Data Dictionary Editor.
Optionally define advanced parameters such as update_issues_in_verified_model and save_verification_statistics.
Define the verification output directory.
Execute the verification.

import prosuite

service = prosuite.Service(host_name='localhost', port_nr=5151)

ddx_id = 9
project_short_name = "..."
specification = prosuite.DdxSpecification(ddx_id=ddx_id, project_short_name=project_short_name)

road_dataset_id = 138
road_object_ids_to_verify = [3606433]
rail_dataset_id = 141
rail_object_ids_to_verify = [2105461, 2105452, 2105593]

params = prosuite.VerificationParameters(user_name="UJR")

# Optionally provided object ID lists per dataset (use dataset ID from data dictionary)
params.add_objects_to_verify(road_dataset_id, road_object_ids_to_verify)
params.add_objects_to_verify(rail_dataset_id, rail_object_ids_to_verify)

# Optionally, write the issue to the error datasets in the data model
params.update_issues_in_verified_model = True

# Optionally, save the verification statistics to the respective data dictionary table (requires DDX schema 1.0.0.1)
params.save_verification_statistics = True

out_dir = 'C:/temp/verification_output'
extent = None  # Optionally define the verification extent

for verification_response in service.verify(specification=specification, output_dir=out_dir, perimeter=extent, parameters=params):
    print(verification_response.message_level)
    print(verification_response.service_call_status)
    print(verification_response.message)

Notes:

Directories: Ensure the specified paths are accessible by the server, especially for output directories.
update_issues_in_verified_model: If set to True, this updates issues within the error datasets in the verified model.
save_verification_statistics: If set to True, this option saves verification statistics into the Data Dictionary database.

Verification using DDX Specification with Automatic Object List Generation

In addition to specifying individual object IDs, you can also automatically retrieve all feature IDs within a dataset for comprehensive quality verification. This can be done by using a function (e.g., get_object_ids_from_feature_class) to fetch all object IDs, allowing you to apply checks across an entire dataset. This approach is especially useful for verifying large datasets.

Use this method to streamline quality verification for complete datasets or to apply selection queries for filtered checks. Simply generate the object list and add it to your verification parameters.

import arcpy

def get_object_ids_from_feature_class(feature_class, selection_query=None):
    object_ids = []
    with arcpy.da.SearchCursor(feature_class, ["OBJECTID"], selection_query) as cursor:
        for row in cursor:
            object_ids.append(row[0])
    return object_ids

ddx_id = 9
ddx_project_short_name = ""
road_dataset_id = 138


road_feature_class = r"C:\DIRA\... sde\feature_class"
road_selection_query = None  # Optional: SQL filter query

road_object_ids_to_verify = get_object_ids_from_feature_class(road_feature_class, road_selection_query)

Verification on Secure Channel

In this example, the grpc.ssl_channel_credentials object is created by a utility method, that gets the required root certificates automatically from the windows certificate store. For advanced scenarios or credentials on a non-windows platform, see the gRPC Python docs.

import prosuite
ssl_credentials = prosuite.utils.get_ssl_channel_credentials()

# if channel_credentials are passed to the Verification constructor, a secure channel will be established.
service = prosuite.verification.Service(host_name='localhost', port_nr=5151, channel_credentials=ssl_credentials)

Define a WKB perimeter

import prosuite
poly_as_hex_string = '01ee0300000100000001eb03000001000000050000004060e5e8cfd5434100c3640aa44f32410000000000000000f8065f282dd6434100c3640aa44f32410000000000000000f8065f282dd6434170d71262d64f324100000000000000004060e5e8cfd5434170d71262d64f324100000000000000004060e5e8cfd5434100c3640aa44f32410000000000000000'
wkb_perimeter = prosuite.WkbPerimeter(bytes.fromhex(poly_as_hex_string))

# the wkb_perimeter can be assigned to the perimeter parameter in verify()

Note

The variable ‘poly_as_hex_string’ is the hex string representation of a polygon or envelope. It can be produced for example from an arcpy.Geometry. Any arcpy.Geometry can be converted to WKB and encoded as hex based string:

poly_as_hex_string = arcpy_polygon_geometry.WKB.hex()

Acessing a verification response

service.verify() returns an iterable of ResponseVerification objects. It is iterable because the verification service returns a reponse stream. Hence the progress can be printed in real-time.

for verification_response in service.verify():
    print(verification_response.message_level)
    print(verification_response.service_call_status)
    print(verification_response.message)

Advanced Parameters

Optionally, change advanced verification parameters, such as the Verification tile_size (the default is 5000m)

import prosuite

xml_file = 'C:/temp/road_specification.qa.xml'
service = prosuite.verification.Service(host_name='localhost', port_nr=5151)

xml_spec = prosuite.XmlSpecification(
    specification_file=xml_file, specification_name="Produktionsunterstuetzung",
                                     data_source_replacements=[["ProductionModel", sde_file]])

params = prosuite.verification.VerificationParameters(tile_size=10000)

out_dir = 'C:/temp/verification_output'

for verification_response in service.verify(specification=spec, output_dir=out_dir, parameters=params):
    print(verification_response)

for verification_response in service.verify(specification=spec, output_dir = out_dir):
    print(verification_response)

Start and stop the local service process

If no service is constantly running and the python script should run without interaction, e.g. as a batch job, the server process can be started directly from python on the local machine. In this example, an XML specification is used.

import time
import subprocess
import prosuite

# Start the service from a local server installation with the default port.
# It will fail and shut down immediately if another service is already serving on the same port.
server_process = subprocess.Popen(r"C:\ProSuite\Server\prosuite-qa-microservice.exe")

# Alternatively, provide a host name and custom port like this:
# server_process = subprocess.Popen(
#     [r"C:\ProSuite\Server\prosuite-qa-microservice.exe",
#     "--hostname", "LOCALHOST", "--port", "12345"])

# Wait for the process to start, initialize the ArcGIS license and the communication channl
time.sleep(10)

service = prosuite.verification.Service(host_name='LOCALHOST', port_nr=5151)

xml_file = "C:/Data/specifications/road_specification.qa.xml"
workspace = "C:/Data/TopographicData.gdb"

xml_spec = prosuite.XmlSpecification(specification_file=xml_file,
                                    specification_name="Produktionsunterstuetzung",
                                    data_source_replacements=[["ProductionModel", workspace]])

out_dir = 'C:/Temp/verification_output'

for verification_response in service.verify(specification=spec, output_dir = out_dir):
    print(verification_response)

# Stop the service
server_process.kill()

Issue Filters

Issue Filters allow filtering of issues found by specific conditions during the QA process. These filters are useful for focusing on particular types of issues and ignoring irrelevant ones.

Issue Filters can be configured for specific QA conditions and will be evaluated for each issue found. Typical use cases include spatial filtering or attribute-based conditions on the involved rows or features that caused the issue.

Common uses:

Filter by Area (IfWithin): Limit issues to a defined boundary or region.
Filter by Proximity (IfNear): Display issues that are near a specified feature.
Filter by Row Involvement (IfInvolvedRows): Focus on issues related to specific table rows.

For the full list of available Issue Filters, see the ProSuite QA Help for Tests Algorithm documentation.

Issue Filters are part of the module: prosuite.factories.issue_filters.

Basic Issue Filter Workflow with `IfWithin`

Use case: Focus on issues within a specific area, such as a district or a defined project boundary where data has already been prepared for a specific purpose.

import prosuite

# Create a service to verify the specification
service = prosuite.Service(host_name='localhost', port_nr=5151)

# Define the dataset (e.g., roads)
model = prosuite.Model("SimpleModel", r"C:\Data\SimpleCity.gdb")
roads = prosuite.Dataset("TLM_STRASSEN", model)

# Define the spatial filter: only include issues within a specific boundary
boundary = prosuite.EnvelopePerimeter(x_min=2750673, y_min=1215551, x_max=2765845, y_max=1206640)
issue_filter = prosuite.IssueFilters.IfWithin(boundary)

# Define the condition (e.g., minimum length of roads)
condition = prosuite.Conditions.qa_min_length_0(roads, limit=5.0, is3_d=False)

# Create a specification and add the condition
specification = prosuite.Specification(name="Check Road Length within Boundary")
specification.add_condition(condition)

# Apply the filter to the verification process
output_dir = r"C:\temp\filtered_road_verification"
for verification_response in service.verify(specification=specification, filters=[issue_filter], output_dir=output_dir):
    print(verification_response.message_level)
    print(verification_response.service_call_status)
    print(verification_response.message)

This workflow ensures that only the issues occurring within the specified area are considered during verification.

Basic Issue Filter Workflow with Multiple Filters

Use case: Focus on issues within a specific area that are also near a given feature (e.g., roads near rivers within a designated region).

import prosuite

# Create a service to verify the specification
service = prosuite.Service(host_name='localhost', port_nr=5151)

# Define the dataset (e.g., roads and rivers)
model = prosuite.Model("SimpleModel", r"C:\Data\SimpleCity.gdb")
roads = prosuite.Dataset("TLM_STRASSEN", model)
rivers = prosuite.Dataset("TLM_FLUSS", model)

# Define multiple issue filters:
# 1. Only include issues within a specific boundary.
boundary = prosuite.EnvelopePerimeter(x_min=2750673, y_min=1215551, x_max=2765845, y_max=1206640)
area_filter = prosuite.IssueFilters.IfWithin(boundary)

# 2. Additionally, include only issues that are near the river features.
proximity_filter = prosuite.IssueFilters.IfNear(rivers, distance=50)

# Combine filters into a list
combined_filters = [area_filter, proximity_filter]

# Define the condition (e.g., minimum length of roads)
condition = prosuite.Conditions.qa_min_length_0(roads, limit=5.0, is3_d=False)

# Create a specification and add the condition
specification = prosuite.Specification(name="Check Road Length within Boundary and Near Rivers")
specification.add_condition(condition)

# Apply the combined filters to the verification process
output_dir = r"C:\temp\filtered_combined_verification"
for verification_response in service.verify(specification=specification, filters=combined_filters, output_dir=output_dir):
    print(verification_response.message_level)
    print(verification_response.service_call_status)
    print(verification_response.message)

This workflow demonstrates how to use multiple filters to refine the QA check, focusing on both spatial boundaries and proximity to other features.

Transformers

Transformers allow dynamic preprocessing of datasets, often before they are passed into a QA condition. The process happens in-memory without altering the original dataset or physically writing new tables.

Common Uses of Transformers

Transformers can be categorized into two main groups based on the complexity of the transformation:

1. Spatial Filtering:: These transformers filter the input datasets using a spatial constraint (TrOnlyContainedFeatures, TrOnlyDisjointFeatures, TrOnlyIntersectingFeatures)
2. Geometry Transformation:: These transformers perform a transformation of a single geometry (TrGeometryToPoints, TrMultipolygonToPolygon, TrLineToPolygon, TrProject etc.)
3. Topological Operations:: These transformers perform a topological operation involving several features (TrDissolve, TrIntersect, TrSpatialJoin)
4. Creation of Tables:: If a complex SQL query (e.g. containing sub-queries) is required to create a new table, the TrMakeTable transformer can be used. This allows for dynamic SQL execution to filter or transform data before it is used in QA checks. For Joins between multiple datasets, the TrJoinInMemory transformer can be used to perform in-memory joins across databases while TrJoin uses standard ArcGIS functionality to create a database join.

For the full list of available transformers, see the ProSuite QA Help for Tests Algorithm documentation:

Transformers are part of the module: prosuite.factories.transformers.

Basic Transformer Workflow with `TrMakeTable`

In this example, we demonstrate how to use the TrMakeTable transformer to create a filtered subset of data based on an SQL query containing a sub-query. This is helpful when you want to extract specific records or perform conditional filtering before using the data in a QA check.

Use case: Filter a dataset to select only specific road types (e.g., roads that belong to a particular street type) and then perform a quality check.

import prosuite

service = prosuite.Service(host_name='localhost', port_nr=5151)

# Load the geodatabase model
model = prosuite.Model("SimpleModel", r"C:\Data\SimpleCity.gdb")

# Define the dataset (e.g., roads)
roads = prosuite.Dataset("Roads", model)

# Define the SQL query to filter the roads based on a specific type
sql_query = "SELECT * FROM TLM_STRASSEN WHERE OBJEKTART IN (SELECT OBJEKTART FROM TLM_STRASSE_AUS_EINFAHRT)"
oid_field = "OBJECTID"

# Apply the TrMakeTable transformer to filter the dataset
filtered_roads = prosuite.Transformers.tr_make_table_1(roads, sql_query, oid_field)

# Perform a quality check on the filtered dataset (e.g., check that roads have a minimum length)
condition = prosuite.Conditions.qa_min_length_0(filtered_roads, limit=5.0, is3_d=False)

# Add the condition to a specification and verify
specification = prosuite.Specification("Check Road Length")
specification.add_condition(condition)

server = prosuite.Service(host_name="localhost", port_nr=5151)
output_dir = r"C:\temp\filtered_road_verification"

for verification_response in service.verify(specification=specification, output_dir=output_dir):
     print(verification_response.message_level)
     print(verification_response.service_call_status)
     print(verification_response.message)

Intersection of Features Using `TrIntersect`

The TrIntersect transformer is used to calculate intersections between two spatial datasets (e.g. roads and rivers). This can be useful to identify where features cross or overlap, such as finding all points where roads intersect rivers.

Use case: Find intersections between roads and rivers to identify potential bridge locations.

import prosuite

service = prosuite.Service(host_name='localhost', port_nr=5151)

# Load the geodatabase model
model = prosuite.Model("SimpleModel", r"C:\Data\SimpleCity.gdb")

# Define two input feature classes (roads and rivers)
roads = prosuite.Dataset("Roads", model)
rivers = prosuite.Dataset("Rivers", model)
lakes = prosuite.Dataset("Lakes", model)

# Apply the TrIntersect transformer
intersect_dataset = prosuite.Transformers.tr_intersect(roads, rivers)

# Make sure only point intersections (and no lines, in case of roads running along rivers)
intersect_dataset.result_dimension = 1

# Use the result in a QA condition (e.g. check that intersections exist)

#Test that the bridges of rails are not within a lake:
condition = prosuite.Conditions.qa_interior_intersects_other_0(intersect_dataset, lakes)

# Add to a specification and verify
specification = prosuite.Specification("Check Road-River Intersections")
specification.add_condition(condition)

server = prosuite.Service(host_name="localhost", port_nr=5151)
output_dir = r"C:\temp\intersection_verification"

for verification_response in service.verify(specification=specification, output_dir=output_dir):
     print(verification_response.message_level)
     print(verification_response.service_call_status)
     print(verification_response.message)

DDX Verification with Transformer (TrMakeTable)

This example demonstrates how a quality specification can be loaded from the ProSuite Data Dictionary (DDX) and adapted with an SQL-based transformer (TrMakeTable) replacing an existing dataset. This shows how a configured input dataset can be adapted to using a dynamic query before the QA test runs.

Use case: Run a DDX-based quality verification, but only on features that match certain criteria (e.g. only features with status = ‘ACTIVE`), without modifying the original view or dataset.

import logging
import prosuite

from prosuite.quality import Parameter
from prosuite.data_model.transformed_dataset import TransformedDataset

# === Configuration ===
HOSTNAME = "localhost"
PORT = 5151
SPECIFICATION_ID = 5
PROJECT_NAME = "MY_PROJECT"
OUTPUT_DIR = r"C:\temp\verification_output"

TARGET_DATASET_NAME = "my_schema.v_feature_edges"
OID_FIELD = "object_id"
 SQL_FILTER = """
 SELECT object_id, change_column, type_column, geom
 FROM schema_name.source_view
 WHERE object_id NOT IN (
     SELECT other.object_id
     FROM schema_name.reference_table other
     WHERE other.object_id = schema_name.source_view.object_id
     AND other.status_column = 'some_terminated_status'
 )
 AND (
     change_column = 'some_terminated_status' OR change_column = 'active_status'
 )
 """


 # === Logging Setup ===
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")


 def redirect_datasets(params: list[Parameter]):
     for param in params:
         if not param.is_dataset_parameter():
             continue

         if param.is_dataset_value():
             dataset = param.dataset
             if dataset.model.default_database_schema_owner and '.' not in dataset.name:
                 dataset.name = f"{dataset.model.default_database_schema_owner}.{dataset.name}"

             if dataset.name == TARGET_NAME:
                 transformed = prosuite.Transformers.tr_make_table_1(dataset, SQL_FILTER, OID_FIELD)
                 param.dataset = transformed
                 logging.info(f"➔ Applied SQL filter on {dataset.name}")

         elif param.is_transformed_dataset_value():
             transformed = param.dataset
             if not transformed.transformer_descriptor.startswith("TrMakeTable"):
                 redirect_datasets(transformed.parameters)


 def main():
     service = prosuite.Service(host_name=HOST, port_nr=PORT)
     spec = prosuite.DdxService(host_name=HOST, port_nr=PORT).get_specification(SPEC_ID)
     spec.project_short_name = "shortname"

     logging.info(f"Loaded Specification: {spec.name} with {len(spec.get_conditions())} conditions")

     for cond in spec.get_conditions():
         logging.info(f"Processing condition: {cond.name}")
         redirect_datasets(cond.parameters)

     perimeter = prosuite.EnvelopePerimeter(x_min=2599839, y_min=1200036, x_max=2599900, y_max=1200050)
     params = prosuite.VerificationParameters(user_name="user01")

     logging.info("=== Starting QA Verification ===")

     result = None
     for response in service.verify(specification=spec, output_dir=OUTPUT_DIR, perimeter=perimeter, parameters=params):
         result = response
         logging.info(f"[{response.message_level}] {response.message}")

     if result and result.service_call_status == "Finished":
         logging.info("=== Verification completed successfully ===")
     else:
         logging.error("Verification failed or was incomplete.")


 if __name__ == "__main__":
     main()

Explanation:

The TrMakeTable transformer filters records using a custom SQL query before the QA condition executes.
The DDX specification is reused as-is, no modification in the Data Dictionary is required.
Works with PostgreSQL/PostGIS views or other SQL-based sources supported by the QA microservice.

Export transformed datasets using `QaExportTables`

Use case: After applying SQL-based transformers like TrMakeTable, you may want to inspect the intermediate results in a file geodatabase. This is helpful for understanding which features are being passed into QA checks.

import prosuite
import os

service = prosuite.Service(host_name='localhost', port_nr=5151)

# Load the model and base dataset
model = prosuite.Model("MyModel", r"C:\\Data\\MyDatabase.gdb")
roads = prosuite.Dataset("v_roads", model)

# Apply an SQL filter via TrMakeTable
sql = """
 SELECT object_id, change_column, type_column, geom
 FROM schema_name.source_view
 WHERE object_id NOT IN (
     SELECT other.object_id
     FROM schema_name.reference_table other
     WHERE other.object_id = schema_name.source_view.object_id
     AND other.status_column = 'some_terminated_status'
 )
 AND (
     change_column = 'some_terminated_status' OR change_column = 'active_status'
 )
 """
filtered_roads = prosuite.Transformers.tr_make_table_1(roads, sql, object_id_field="object_id")

# Define a basic QA condition
condition = prosuite.Conditions.qa_min_length_0(filtered_roads, limit=10.0, is3_d=False)

# Create specification and add the condition
spec = prosuite.Specification(name="Check Active Roads")
spec.add_condition(condition)

# Export transformed dataset to GDB for inspection
export_gdb = r"C:\\temp\\qa_export\\transformed_data.gdb"
spec.add_condition(
    prosuite.Conditions.qa_export_tables_0([filtered_roads], export_gdb)
)

# Run verification
for response in service.verify(specification=spec, output_dir=r"C:\\temp\\qa_export"):
    print(response.message_level)
    print(response.message)

Custom Names for Conditions and Transformers

ProSuite automatically generates names for QA conditions and transformers by combining the descriptor (e.g., TrMakeTable, QaMinLength) with the name of the first involved dataset. Dots in dataset names are replaced by underscores.

Example of an automatically generated name:

TrMakeTable1_my_schema_v_roads

This naming logic applies to both transformers and QA conditions.

Assigning custom names:

You can override the default name:s by explicitly setting the name attribute of the transformer or condition. This is particularly useful for clarity in output directories and verification reports.

transformer = prosuite.Transformers.tr_make_table_1(
 base_table="my_schema.v_roads",
 sql="""
     SELECT object_id, status, category, geom
     FROM my_schema.v_roads AS r
     WHERE object_id NOT IN (
         SELECT h.object_id
         FROM my_schema.roads_history h
         WHERE h.object_id = r.object_id
         AND h.version_end = '9999-12-31'
     )
     AND (
         status = 'active' OR status = 'open'
     )
 """,
 object_id_field="object_id"
 )
 transformer.name = "FilteredRoads_ActiveOrOpen"


condition = prosuite.Conditions.qa_min_length_0(transformer, limit=5.0, is3_d=False)
condition.name = "CheckMinLengthOfActiveRoads"