Salesforce

from simple_salesforce import Salesforce

# Salesforce credentials (replace with environment variables in production)
username = ""
password = ""
security_token = ""

# 🔍 Specify object names to include (DANGER leaving the list empty will include ALL)
included_objects = ["Contact", "Account", "Opportunity"]  # Case-sensitive

# Connect to Salesforce
sf = Salesforce(username=username, password=password, security_token=security_token)

# Mapping Salesforce field types to SQL types
def map_salesforce_type(sf_type):
    mapping = {
        "string": "TEXT",
        "textarea": "TEXT",
        "picklist": "TEXT",
        "id": "VARCHAR(18)",
        "boolean": "BOOLEAN",
        "int": "INTEGER",
        "double": "FLOAT",
        "currency": "DECIMAL(18,2)",
        "percent": "DECIMAL(5,2)",
        "date": "DATE",
        "datetime": "TIMESTAMP",
        "email": "TEXT",
        "phone": "TEXT",
        "url": "TEXT",
        "reference": "VARCHAR(18)",
        "base64": "BYTEA",
        "location": "TEXT",
    }
    return mapping.get(sf_type, "TEXT")


# Get all object descriptions
objects = sf.describe()["sobjects"]

# Open file for writing schema output
with open("salesforce_schema.txt", "w") as file:
    for obj in objects:
        obj_name = obj["name"]
        
        # Apply filter if specified
        if included_objects and obj_name not in included_objects:
            continue
        
        try:
            obj_details = sf.__getattr__(obj_name).describe()
            field_defs = []

            for field in obj_details["fields"]:
                field_name = field["name"]
                field_type = map_salesforce_type(field["type"])
                field_defs.append(f"{field_name} {field_type}")

            field_list = ", ".join(field_defs)
            table_def = f"TABLE {obj_name} ({field_list});"
            print(table_def)
            file.write(table_def + "\n")

        except Exception as e:
            error_msg = f"-- Could not process {obj_name}: {e}"
            print(error_msg)
            file.write(error_msg + "\n")

print("\nFiltered schema has been saved to 'salesforce_schema.txt'")