Batch multiple definition query

RiznaAbdulgafoor

import arcpy
import os

def main():
# --- Tool parameters ---
gdb = arcpy.GetParameterAsText(0) # Input Geodatabase
field_name = arcpy.GetParameterAsText(1) # Field to use in query (e.g., project_number)
where_clause = arcpy.GetParameterAsText(2) # SQL clause (e.g., project_number = 'ABC' OR project_number = 'XYZ')
clear_all = arcpy.GetParameter(3) # Boolean checkbox: True = clear all definition queries

# Get current project
aprx = arcpy.mp.ArcGISProject("CURRENT")

# Walk through all feature classes in the GDB
for dirpath, dirnames, fcs in arcpy.da.Walk(gdb, datatype="FeatureClass"):
for fc in fcs:
fc_path = os.path.join(dirpath, fc)

# If not clearing → check that the field exists before applying query
if not clear_all:
if not arcpy.ListFields(fc_path, field_name):
arcpy.AddWarning(f"{field_name} not found in {fc}, skipping")
continue

# Go through all maps and layers in the project
for m in aprx.listMaps():
for lyr in m.listLayers():
if lyr.isFeatureLayer:
try:
# Match the layer's data source with this feature class path
if os.path.normcase(os.path.normpath(lyr.dataSource)) == os.path.normcase(os.path.normpath(fc_path)):

# --- CLEAR ALL option ---
if clear_all:
lyr.definitionQuery = ""
arcpy.AddMessage(f"Cleared query on {lyr.name}")

# --- APPLY QUERY option ---
else:
if where_clause:
lyr.definitionQuery = where_clause
arcpy.AddMessage(f"Applied query on {lyr.name}: {where_clause}")
else:
arcpy.AddWarning(f"No where clause provided for {lyr.name}, skipped")

except Exception as e:
arcpy.AddWarning(f"Could not update {lyr.name}: {e}")

# Save changes to project
aprx.save()

if __name__ == "__main__":
main()

Hayden Welch

You're getting really heavy with the nesting and it's hard to follow the plot. You also haven't really asked a question, just posted code and a title explaining what your code seems to be attempting to do.

 

Either way I went ahead and tried to make your code a bit more readable:

import arcpy
import os
from pathlib import Path
from collections.abc import Iterator
from arcpy._mp import Layer, Map

def apply_query(lyr: Layer, query: str|None) -> None:
    if query is not None:
        lyr.updateDefinitionQueries([{'name': 'Query', 'sql': query, 'isActive': True}])
        return
    qs = lyr.listDefinitionQueries()
    for q in qs:
        q['isActive'] = False
    lyr.updateDefinitionQueries(qs)

def get_feature_layers(m: Map) -> Iterator[Layer]:
    yield from (l for l in m.listLayers() if l.isFeatureLayer)

def has_field(lay: Layer, field: str) -> bool:
    return [field in arcpy.da.SearchCursor(lay, '*').fields].pop()

def main():
    # --- Tool parameters ---
    gdb = arcpy.GetParameterAsText(0) # Input Geodatabase
    gdb_path = Path(gdb)
    field_name = arcpy.GetParameterAsText(1) # Field to use in query (e.g., project_number)
    where_clause = arcpy.GetParameterAsText(2) # SQL clause (e.g., project_number = 'ABC' OR project_number = 'XYZ')
    clear_all = arcpy.GetParameter(3) # Boolean checkbox: True = clear all definition queries
    
    aprx = arcpy.mp.ArcGISProject("CURRENT")
    
    # Iterate through tuples of map, layer using the get_feature_layers filter
    for map, layer in ((m, l) for m in aprx.listMaps() for l in get_feature_layers(m)):
        _ident = f'{map.name}->{layer.longName}' # For logging
        
        # This will make sure only layers that are sourced to the GDB are processed
        if not Path(layer.dataSource).is_relative_to(gdb_path):
            continue
        
        # This will make sure that the field exists in the layer if you aren't clearing
        if not clear_all and not has_field(layer, field_name):
            arcpy.AddWarning(f"{field_name} not found in {_ident}, skipping") 
            continue
        
        # This will apply the query or clear the query
        try: apply_query(layer, None if clear_all else where_clause)
        except Exception as e:
            arcpy.AddWarning(f"Could not update {_ident}: {e}")
            continue # Don't print the success message
        
        arcpy.AddMessage(f"Applied query on {_ident}: {where_clause}")
        
if __name__ == "__main__":
    main()