source-salesforce-native: use ValidationContext class instead of dict

While the previous dict validation context approach worked and still
works with the IncrementalCSVProcessor typing changes in the previous
commit, I'm updating the connector to use a more structured
ValidationContext class so there's a clear example of the (currently)
preferred way to use Pydantic's validation contexts with the
IncrementalCSVProcessor.

Author: Alex-Bair

source-salesforce-native/source_salesforce_native/bulk_job_manager.py

@@ -15,6 +15,7 @@
     BulkJobCheckStatusResponse,
     SalesforceDataSource,
     SalesforceRecord,
+    ValidationContext,
 )
 
 INITIAL_SLEEP = 0.2
@@ -35,6 +36,8 @@
     encoding='utf-8'
 )
 
+BULK_VALIDATION_CONTEXT = ValidationContext(SalesforceDataSource.BULK_API)
+
 
 class BulkJobManager:
     def __init__(self, http: HTTPSession, log: Logger, instance_url: str):
@@ -103,12 +106,11 @@ async def _fetch_results(self, job_id: str, model_cls: type[SalesforceRecord]) -
             expected = int(count)
             received = 0
 
-            bulk_context = {'data_source': SalesforceDataSource.BULK_API}
             processor = IncrementalCSVProcessor(
                 body(),
                 model_cls,
                 CSV_CONFIG,
-                validation_context=bulk_context,
+                validation_context=BULK_VALIDATION_CONTEXT,
             )
             async for record in processor:
                 yield record

source-salesforce-native/source_salesforce_native/models.py

@@ -332,6 +332,11 @@ class SalesforceDataSource(StrEnum):
     BULK_API = "bulk_api"
 
 
+class ValidationContext:
+    def __init__(self, data_source: SalesforceDataSource):
+        self.data_source = data_source
+
+
 class SalesforceRecord(BaseDocument, extra="allow"):
     field_details: ClassVar[FieldDetailsDict]
 
@@ -353,10 +358,10 @@ def _transform_fields(cls, values: dict[str, Any], info: ValidationInfo) -> dict
                 "field_details must be set on the SalesforceRecord subclass before validation."
             )
 
-        context = info.context or {}
-        data_source: SalesforceDataSource | None = context.get('data_source', None)
-        if data_source is None:
-            raise ValueError("data_source must be provided in validation context")
+        if not info.context or not isinstance(info.context, ValidationContext):
+            raise RuntimeError(f"Validation context must be of type ValidationContext: {info.context}")
+
+        data_source = info.context.data_source
 
         transformed: dict[str, Any] = {}
 

source-salesforce-native/source_salesforce_native/rest_query_manager.py

@@ -10,6 +10,7 @@
     QueryResponse,
     SalesforceDataSource,
     SalesforceRecord,
+    ValidationContext,
 )
 
 
@@ -19,6 +20,8 @@
 MAX_URI_LENGTH = 16_384
 MAX_FIELDS_LENGTH = MAX_URI_LENGTH - 2_500
 
+REST_VALIDATION_CONTEXT = ValidationContext(SalesforceDataSource.REST_API)
+
 class Query:
     def __init__(self, http: HTTPSession, log: Logger, base_url: str, query: str):
         self.http = http
@@ -177,7 +180,7 @@ async def execute(
                 # current state of the record in Saleforce.
                 if str_to_dt(record[cursor_field]) <= end:
                     yield model_cls.model_validate(
-                        record, context={'data_source': SalesforceDataSource.REST_API}
+                        record, context=REST_VALIDATION_CONTEXT
                     )
 
                 # Delete completed records to avoid keeping them in memory.