New version with benchmark lookup by custom name and value (#12)

* new version with benchmark lookup by custom name/val

* precommit

* fixes

Author: kflemin

.gitignore

@@ -99,3 +99,4 @@ salesforce-config*.json
 !salesforce-config-example.json
 seed-config*.json
 !seed-config-example.json
+example_usage.py

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Change Log
 
+## Version 0.2.0
+
+Added a method to look up a benchmark with a custom ID name / value pair
+Updated testing to first retrieve a benchmark by custom ID and then update it
+Dependency updates
+
 ## Version 0.1.1
 
 Updated Python compatibility for v3.9-v3.12, and reformatted with Ruff

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "seed-salesforce"
-version = "0.1.1"
+version = "0.2.0"
 description = "Package for connecting SEED data to Salesforce"
 authors = ["Nicholas Long <nicholas.long@nrel.gov>", "Katherine Fleming <katherine.fleming@nrel.gov>"]
 license = "BSD-3-Clause"
@@ -28,7 +28,7 @@ python = ">=3.9, <3.13"
 python-dateutil = "*"
 simple-salesforce = "^1.12.6"
 
-[tool.poetry.dev-dependencies]
+[tool.poetry.group.dev.dependencies]
 mypy = "^1.11.2"
 pre-commit = "^3.8.0"
 pytest = "^7.1.2"

pyproject.toml

@@ -2,6 +2,7 @@
 
 import json
 import logging
+import re
 from pathlib import Path
 from typing import Optional
 
@@ -129,13 +130,15 @@ def get_first_benchmark(self) -> dict:
         else:
             raise Exception("Failed to return a Benchmark")
 
-    def get_benchmark_by_custom_id(self, salesforce_benchmark_id: str) -> dict:
-        """Return the benchmark by the Salesforce Benchmark ID.
+    def get_benchmark_by_custom_id(self, custom_id_name: str, custom_id_value: str) -> dict:
+        """Return the benchmark by the custom Salesforce Benchmark ID.
 
         Args:
-            salesforce_benchmark_id (str): Salesforce Benchmark ID of the property to return
-            Note: this is not necessarily the Benchmark ID (it's a separate field)
-            # TODO: make this configurable?
+            custom_id_name (str): Name of the custom ID field
+            custom_id_value (str): Value of the custom ID field
+            Note: this is not necessarily the Benchmark ID (it's a separate field that should be unique
+            and serves as the "key" for retrieving the benchmark, but it is not necessarily the actual
+            record ID in Salesforce)
 
         Returns:
             dict:  OrderedDict([('attributes',
@@ -145,18 +148,23 @@ def get_benchmark_by_custom_id(self, salesforce_benchmark_id: str) -> dict:
             ...
         """
 
+        # Validate field name to prevent SQL injection
+        # Field names in Salesforce should only contain alphanumeric characters, underscores, and end with __c for custom fields
+        if not re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*(__c)?$", custom_id_name):
+            raise ValueError(f"Invalid field name: {custom_id_name}")
+
         benchmark_exist = self.connection.query(
-            format_soql("Select Id from Benchmark__c where Salesforce_Benchmark_ID__c = {}", salesforce_benchmark_id),
+            format_soql(f"Select Id from Benchmark__c where {custom_id_name} = {{}}", custom_id_value),  # noqa: S608
         )
         if len(benchmark_exist["records"]) == 1:
             # if there is a single record, then it exist, but
             # we need to get the entire record for the request
             rec = self.get_benchmark_by_id(benchmark_exist["records"][0]["Id"])
             return rec
         elif len(benchmark_exist["records"]) > 1:
-            # there are multiple properties with the same name, raise error
+            # there are multiple properties with the same custom ID, raise error
             raise Exception(
-                f"Failed to return Benchmark {salesforce_benchmark_id}...multiple benchmarks with that name found",
+                f"Failed to return Benchmark {custom_id_value}...multiple benchmarks with that custom ID found",
             )
         else:
             # there is no property, return empty dict
@@ -307,6 +315,26 @@ def get_property_by_id(self, property_id: str) -> dict:
         except BaseException:
             raise Exception("Error retrieving property by ID")
 
+    def get_accounts(self) -> list:
+        """Get all accounts in salesforce
+
+        Returns:
+            list: list of accounts in salesforce
+        """
+        soql_query = "SELECT Id, Name FROM Account"
+        results = self.connection.query_all(soql_query)
+        return results["records"]
+
+    def get_contacts(self) -> list:
+        """Get all contacts in salesforce
+
+        Returns:
+            list: list of contacts in salesforce
+        """
+        soql_query = "SELECT Id, Name, Email FROM Contact"
+        results = self.connection.query_all(soql_query)
+        return results["records"]
+
     def get_account_by_account_id(self, account_id: str) -> dict:
         """Return the account by the account ID.
 

seed_salesforce/salesforce_client.py

@@ -165,28 +165,31 @@ def test_benchmark(self):
         benchmark = self.sf.get_first_benchmark()
         print(f" Benchmark: {benchmark}")
         assert benchmark is not None
-        benchmark_id = benchmark["Id"]
-        salesforce_benchmark_id = benchmark["Salesforce_Benchmark_ID__c"]
 
         # can you retrieve by "Salesforce Benchmark ID" custom field?
+        salesforce_benchmark_id = benchmark["Salesforce_Benchmark_ID__c"]
         print(" ...retrieving benchmark by Salesforce Benchmark ID...")
-        bench_by_custom_id = self.sf.get_benchmark_by_custom_id(salesforce_benchmark_id)
+        bench_by_custom_id = self.sf.get_benchmark_by_custom_id("Salesforce_Benchmark_ID__c", salesforce_benchmark_id)
         print(f" benchmark by custom id: {bench_by_custom_id}")
-        assert bench_by_custom_id["Id"] == benchmark_id
+        # We are not retrieving by the actual record ID,
+        # but rather a custom field that should be unique but is not necessarily the record ID
+        actual_id = bench_by_custom_id["Id"]
+        # assert that bench_by_custom_id has the same Salesforce Benchmark Id as the original custom ID
+        assert bench_by_custom_id["Salesforce_Benchmark_ID__c"] == salesforce_benchmark_id
 
         # can you update a benchmark field?
         print(" ...updating benchmark...")
         # TODO: again assuming this field exists...it might not?
         energy_star_score = benchmark["ENERGY_STAR_Score__c"]
         new_energy_star_score = 20
         args = {"ENERGY_STAR_Score__c": new_energy_star_score}
-        bench_updated = self.sf.update_benchmark(salesforce_benchmark_id, **args)
+        bench_updated = self.sf.update_benchmark(actual_id, **args)
         print(f"benchmark updated: {bench_updated}")
 
         # retrieve again to see if it was updated
-        bench2 = self.sf.get_benchmark_by_custom_id(salesforce_benchmark_id)
+        bench2 = self.sf.get_benchmark_by_custom_id("Salesforce_Benchmark_ID__c", salesforce_benchmark_id)
         assert bench2["ENERGY_STAR_Score__c"] == new_energy_star_score
 
         # restore value
         args["ENERGY_STAR_Score__c"] = energy_star_score
-        self.sf.update_benchmark(salesforce_benchmark_id, **args)
+        self.sf.update_benchmark(actual_id, **args)