Read directory of SDE connection files and list database and all feature classes, tables, etc. to create an inventory

The way i'd approach it is using a FeatureReader with the directory being c:\\temp\\*.sde (read all files ending in SDE in that folder) then set the FeatureReader to only read the schema and process the output schema from there into the required format

@hkingsbury​ if I use the FeatureReader like you suggested, you have to set the Format to something.  I tried the Schema (Any Format) option but results in this:

Generic Reader: Unable to guess a valid reader name from dataset '<name of SDE connection file>.sde'. A NULL reader will be used instead.

If you set the Format to SDE, then you can only set it to 1 SDE connection file AND you have to set the Feature Types to Read.

Any other ideas?

Generic Reader: Unable to guess a valid reader name from dataset '<name of SDE connection file>.sde'. A NULL reader will be used instead.

You'll still need to set the FeatureReader as SDE, but just enable the read only schema option 

As I think more about this problem I realise that you can't then use a wild card as a feature type to read in the feature classes.

Attached is an example workspace that firstly uses ArcPy to get all the FCs in the SDE connection, then passes them through to a FeatureReader(SDE) to get the schema. You could also achieve the same with and SQLExecuter, but unsure if you have direct SQL access to the underlying DB

Thanks for the workbench. This works for 1 SDE connection but I'm trying to run an entire folder of them. The current Connection File user parameter only allows 1 selection - is there a way to use a wildcard like  c:\\temp\\*.sde instead?

Hello ,

thanks for ur workspace .

i have used it but I have found that the output of Python caller has not all feature classes .

could I ask if there is a reason for that ? Why did it read most of feature classes but not all of them ?

​@hkingsbury you wouldn’t happen to still have the workbench you attached back in 2023 would you?  I can’t find it on here anymore...

​@jonvelco I don’t sorry! It was likely using the ListFeatureClasses ArcPy function - https://pro.arcgis.com/en/pro-app/latest/arcpy/functions/listfeatureclasses.htm

And then outputting one feature per featureclass and using the featureclass name in a Reader to then read the schema feature.

​@hkingsbury thanks for the quick response and guidance.  I’ll look into that approach and see what I can come up with.

Is it me, or did all the attachments/files disappear after the forums were updated recently??

Here’s a screenshot of an example workflow using Directory and File Pathnames and PythonCaller.  Few highlights

• Directory and File Pathnames filtered to *.sde
• Python Compatibility set
• PythonCaller using arcpy.ListFeatureClasses()ArcPY

I ended up using a Directory and File Pathnames reader pointed at a folder of SDE connections as ​@vn1 suggested.  However, instead of passing the name of the feature classes to a Feature Reader (Format = SDE, Connection = @value(path_windows), Feature types to Read = @value(FEATURECLASS) I just did the work in Python then processed the results. 

Open to constructive feedback or tweaks to the Python (see below)

Here’s what the workbench looks like:

Here’s what an example report looks like:

Code for the 3 Python callers is below - let me know if you see any issues!

Feature Classes:

import json  # For dictionary serialization
import fme
import fmeobjects
import arcpy


class FeatureProcessor(object):
    def __init__(self):
        """Base constructor for class members."""
        pass

    def has_support_for(self, support_type: int):
        """This method is called by FME to determine if the PythonCaller supports Bulk mode."""
        return support_type == fmeobjects.FME_SUPPORT_FEATURE_TABLE_SHIM

    def input(self, feature: fmeobjects.FMEFeature):
        # Retrieve the value of the 'path_windows' attribute
        path_windows = feature.getAttribute('path_windows')
        path_filename = feature.getAttribute('path_filename')

        # Retrieve the workspace path from the feature attribute
        connFile = feature.getAttribute('path_unix')
        arcpy.env.workspace = connFile

        # Extract database name, instance, and authentication username from the SDE connection
        try:
            desc = arcpy.Describe(connFile)
            connection_properties = desc.connectionProperties

            # Extract the database name, instance, and authentication username from connection properties
            database_name = connection_properties.database
            instance = connection_properties.instance
            auth_username = connection_properties.user

            if not database_name:
                print(f"Database name not found in connection: {connFile}")
                return
            if not instance:
                print(f"Instance not found in connection: {connFile}")
                return
            if not auth_username:
                print(f"Authentication username not found in connection: {connFile}")
                return

        except Exception as e:
            print(f"Error retrieving connection details from SDE connection: {str(e)}")
            return

        # Retrieve feature classes from the workspace
        featureClasses = arcpy.ListFeatureClasses()

        if featureClasses is None:
            # If no feature classes are found, print warning or skip processing
            print(f"No feature classes found in the workspace: {connFile}")
            return  # Exit the method to avoid further processing

        for fc in featureClasses:
            try:
                # Describe the feature class
                desc = arcpy.Describe(fc)

                # Retrieve metadata
                geometry_type = desc.shapeType
                spatial_ref = desc.spatialReference.name
                has_z = desc.hasZ
                has_m = desc.hasM
                alias = desc.aliasName

                # Retrieve field information
                fields = arcpy.ListFields(fc)
                if fields is None:
                    print(f"No fields found in feature class: {fc}")
                    continue

                field_info = []
                field_info_json = {"fields": []}

                for field in fields:
                    field_details = {
                        "field_name": field.name,
                        "field_type": field.type,
                        "length": field.length,
                        "precision": None,
                        "scale": None,
                        "domain_name": field.domain if field.domain else None,
                    }

                    if field.type in ["Double", "Single", "Integer", "SmallInteger"]:
                        field_details["precision"] = field.precision
                        field_details["scale"] = field.scale

                    field_info.append(
                        f"{field.name} ({field.type}, Length: {field.length}, "
                        f"Precision: {field_details['precision']}, Scale: {field_details['scale']})"
                    )
                    field_info_json["fields"].append(field_details)

                field_info_str = ";".join(field_info)

                # Serialize the field information to JSON
                field_info_json_str = json.dumps(field_info_json)

                # Get feature count
                feature_count = arcpy.GetCount_management(fc)[0]

                # Create a new FME feature to output
                fcFeature = fmeobjects.FMEFeature()

                # Set attributes
                fcFeature.setAttribute('FEATURECLASS', fc)
                fcFeature.setAttribute('GEOMETRY_TYPE', geometry_type)
                fcFeature.setAttribute('SPATIAL_REFERENCE', spatial_ref)
                fcFeature.setAttribute('HAS_Z', has_z)
                fcFeature.setAttribute('HAS_M', has_m)
                fcFeature.setAttribute('ALIAS', alias)
                fcFeature.setAttribute('FIELDS', field_info_str)
                fcFeature.setAttribute('FEATURE_COUNT', feature_count)
                fcFeature.setAttribute('FIELDS_JSON', field_info_json_str)
                fcFeature.setAttribute('DATABASE_NAME', database_name)
                fcFeature.setAttribute('INSTANCE', instance)
                fcFeature.setAttribute('AUTH_USERNAME', auth_username)
                fcFeature.setAttribute('path_windows', path_windows)
                fcFeature.setAttribute('path_filename', path_filename)

                # Output the feature to FME
                self.pyoutput(fcFeature)

            except Exception as e:
                print(f"Error processing feature class '{fc}': {str(e)}")
                continue

    def close(self):
        """This method is called once all the FME Features have been processed from input()."""
        pass

    def process_group(self):
        """This method is called by FME for each group when group processing mode is enabled."""
        pass

    def reject_feature(self, feature: fmeobjects.FMEFeature, code: str, message: str):
        """This method can be used to output a feature to the <Rejected> port."""
        feature.setAttribute("fme_rejection_code", code)
        feature.setAttribute("fme_rejection_message", message)
        self.pyoutput(feature, output_tag="<Rejected>")

Tables:

import json  # For dictionary serialization
import fme
import fmeobjects
import arcpy


class FeatureProcessor(object):
    def __init__(self):
        """Base constructor for class members."""
        pass

    def has_support_for(self, support_type: int):
        """This method is called by FME to determine if the PythonCaller supports Bulk mode."""
        return support_type == fmeobjects.FME_SUPPORT_FEATURE_TABLE_SHIM

    def input(self, feature: fmeobjects.FMEFeature):
        # Retrieve the value of the 'path_windows' attribute
        path_windows = feature.getAttribute('path_windows')
        path_filename = feature.getAttribute('path_filename')

        # Retrieve the workspace path from the feature attribute
        connFile = feature.getAttribute('path_unix')
        arcpy.env.workspace = connFile

        # Extract database name, instance, and authentication username from the SDE connection
        try:
            desc = arcpy.Describe(connFile)
            connection_properties = desc.connectionProperties

            # Extract the database name, instance, and authentication username from connection properties
            database_name = connection_properties.database
            instance = connection_properties.instance
            auth_username = connection_properties.user

            if not database_name:
                print(f"Warning: Database name not found in connection: {connFile}")
                return  # Exit if the database name is not found
            if not instance:
                print(f"Warning: Instance not found in connection: {connFile}")
                return  # Exit if the instance is not found
            if not auth_username:
                print(f"Warning: Authentication username not found in connection: {connFile}")
                return  # Exit if the username is not found

        except Exception as e:
            print(f"Error: Failed to retrieve connection details from SDE connection: {str(e)}")
            return  # Exit if an error occurs while describing the connection file

        # Retrieve tables from the workspace
        tables = arcpy.ListTables()

        if not tables:
            # If no tables are found, print a warning and skip processing
            print(f"Warning: No tables found in the workspace: {connFile}")
            return  # Exit the method to avoid further processing

        for table in tables:
            try:
                # Describe the table
                desc = arcpy.Describe(table)
                alias = desc.aliasName

                # Retrieve field information
                fields = arcpy.ListFields(table)
                if not fields:
                    print(f"Warning: No fields found in table: {table}")
                    continue

                field_info = []
                field_info_json = {"fields": []}

                for field in fields:
                    field_details = {
                        "field_name": field.name,
                        "field_type": field.type,
                        "length": field.length,
                        "precision": None,
                        "scale": None,
                        "domain_name": field.domain if field.domain else None,
                    }

                    if field.type in ["Double", "Single", "Integer", "SmallInteger"]:
                        field_details["precision"] = field.precision
                        field_details["scale"] = field.scale

                    field_info.append(
                        f"{field.name} ({field.type}, Length: {field.length}, "
                        f"Precision: {field_details['precision']}, Scale: {field_details['scale']})"
                    )
                    field_info_json["fields"].append(field_details)

                field_info_str = ";".join(field_info)

                # Serialize the field information to JSON
                field_info_json_str = json.dumps(field_info_json)

                # Get table row count
                row_count = arcpy.GetCount_management(table)[0]

                # Create a new FME feature to output
                tableFeature = fmeobjects.FMEFeature()

                # Set attributes
                tableFeature.setAttribute('TABLE_NAME', table)
                tableFeature.setAttribute('ALIAS', alias)
                tableFeature.setAttribute('FIELDS', field_info_str)
                tableFeature.setAttribute('ROW_COUNT', row_count)
                tableFeature.setAttribute('FIELDS_JSON', field_info_json_str)
                tableFeature.setAttribute('DATABASE_NAME', database_name)
                tableFeature.setAttribute('INSTANCE', instance)
                tableFeature.setAttribute('AUTH_USERNAME', auth_username)
                tableFeature.setAttribute('path_windows', path_windows)
                tableFeature.setAttribute('path_filename', path_filename)

                # Output the feature to FME
                self.pyoutput(tableFeature)

            except Exception as e:
                print(f"Error: Failed to process table '{table}': {str(e)}")
                continue

    def close(self):
        """This method is called once all the FME Features have been processed from input()."""
        pass

    def process_group(self):
        """This method is called by FME for each group when group processing mode is enabled."""
        pass

    def reject_feature(self, feature: fmeobjects.FMEFeature, code: str, message: str):
        """This method can be used to output a feature to the <Rejected> port."""
        feature.setAttribute("fme_rejection_code", code)
        feature.setAttribute("fme_rejection_message", message)
        self.pyoutput(feature, output_tag="<Rejected>")

Domains:

import fme
import fmeobjects
import arcpy


class FeatureProcessor(object):
    def __init__(self):
        """Base constructor for class members."""
        pass

    def has_support_for(self, support_type: int):
        """This method is called by FME to determine if the PythonCaller supports Bulk mode."""
        return support_type == fmeobjects.FME_SUPPORT_FEATURE_TABLE_SHIM

    def input(self, feature: fmeobjects.FMEFeature):
        # Retrieve the workspace path and 'path_windows' attribute from the input feature
        connFile = feature.getAttribute('path_unix')
        path_windows = feature.getAttribute('path_windows')
        path_filename = feature.getAttribute('path_filename')
        arcpy.env.workspace = connFile

        # Extract database name, instance, and authentication username from the SDE connection
        try:
            desc = arcpy.Describe(connFile)
            connection_properties = desc.connectionProperties

            # Extract the database name, instance, and authentication username
            database_name = connection_properties.database
            instance = connection_properties.instance
            auth_username = connection_properties.user

            if not database_name:
                print(f"Database name not found in connection: {connFile}")
                return
            if not instance:
                print(f"Instance not found in connection: {connFile}")
                return
            if not auth_username:
                print(f"Authentication username not found in connection: {connFile}")
                return

        except Exception as e:
            print(f"Error retrieving connection details from SDE connection: {str(e)}")
            return

        # Retrieve all domains in the workspace
        domains = arcpy.da.ListDomains(arcpy.env.workspace)

        # Check if any domains were found
        if not domains:
            print(f"No domains were found in the workspace: {connFile}")
            return

        for domain in domains:
            domain_name = domain.name
            domain_type = domain.domainType
            description = domain.description

            if domain_type == "CodedValue":
                # Process coded value domains
                for code, value in domain.codedValues.items():
                    # Output a feature for each code-value pair
                    outputFeature = fmeobjects.FMEFeature()
                    outputFeature.setAttribute("DOMAIN_NAME", domain_name)
                    outputFeature.setAttribute("DOMAIN_TYPE", domain_type)
                    outputFeature.setAttribute("CODE", code)
                    outputFeature.setAttribute("VALUE", value)
                    outputFeature.setAttribute("DESCRIPTION", description)
                    outputFeature.setAttribute("DATABASE_NAME", database_name)
                    outputFeature.setAttribute("INSTANCE", instance)
                    outputFeature.setAttribute("AUTH_USERNAME", auth_username)
                    outputFeature.setAttribute("path_windows", path_windows)
                    outputFeature.setAttribute("path_filename", path_filename)
                    self.pyoutput(outputFeature)

            elif domain_type == "Range":
                # Process range domains
                min_val, max_val = domain.range
                # Output a single feature for the range domain
                outputFeature = fmeobjects.FMEFeature()
                outputFeature.setAttribute("DOMAIN_NAME", domain_name)
                outputFeature.setAttribute("DOMAIN_TYPE", domain_type)
                outputFeature.setAttribute("MIN_VALUE", min_val)
                outputFeature.setAttribute("MAX_VALUE", max_val)
                outputFeature.setAttribute("DESCRIPTION", description)
                outputFeature.setAttribute("DATABASE_NAME", database_name)
                outputFeature.setAttribute("INSTANCE", instance)
                outputFeature.setAttribute("AUTH_USERNAME", auth_username)
                outputFeature.setAttribute("path_windows", path_windows)
                outputFeature.setAttribute("path_filename", path_filename)
                self.pyoutput(outputFeature)

            else:
                # Process unknown domain types
                outputFeature = fmeobjects.FMEFeature()
                outputFeature.setAttribute("DOMAIN_NAME", domain_name)
                outputFeature.setAttribute("DOMAIN_TYPE", domain_type)
                outputFeature.setAttribute("DESCRIPTION", "Unknown domain type")
                outputFeature.setAttribute("DATABASE_NAME", database_name)
                outputFeature.setAttribute("INSTANCE", instance)
                outputFeature.setAttribute("AUTH_USERNAME", auth_username)
                outputFeature.setAttribute("path_windows", path_windows)
                self.pyoutput(outputFeature)

    def close(self):
        """This method is called once all the FME Features have been processed from input()."""
        pass

    def process_group(self):
        """This method is called by FME for each group when group processing mode is enabled."""
        pass

    def reject_feature(self, feature: fmeobjects.FMEFeature, code: str, message: str):
        """This method can be used to output a feature to the <Rejected> port."""
        feature.setAttribute("fme_rejection_code", code)
        feature.setAttribute("fme_rejection_message", message)
        self.pyoutput(feature, output_tag="<Rejected>")