source-dynamics-365-finance-and-operations: remove Pydanitic validation for documents

Pydantic valiation is arguably not necessary for this connector. It
primarily processes CSVs and performs very light transformation (convert
empty strings to `None`, convert boolean fields to actual boolean values,
add in the `_meta.op` field). Yielding validated Pydantic model instances
also causes the CDK to serialize documents with model_dump_json(),
performing some additional validation.

This connector has high CPU usage, and removing some Pydantic validation
should help speed up the connector. By performing the light transformation
itself and yielding raw dicts, the connector uses the CDK's faster `orjson`
serialization pathway when capturing documents, avoiding overhead from
Pydantic validation and serialization. This ends up reducing
steady state CPU usage from ~80% to ~35% when capturing a single
data-heavy binding.

Author: Alex-Bair

source-dynamics-365-finance-and-operations/source_dynamics_365_finance_and_operations/adls_gen2_client.py

@@ -1,6 +1,6 @@
 from datetime import datetime, timezone
 from logging import Logger
-from typing import AsyncGenerator, Any, Optional, TypeVar
+from typing import AsyncGenerator, Any, Optional
 
 from pydantic import BaseModel, Field, computed_field
 
@@ -11,8 +11,6 @@
 
 NEXT_PAGE_HEADER = "x-ms-continuation"
 
-_CSVRow = TypeVar('_CSVRow', bound=BaseModel)
-
 
 class ADLSFilesystemMetadata(BaseModel):
     """Represents metadata for a single filesystem in Azure Data Lake Storage Gen2."""
@@ -119,11 +117,16 @@ async def read_file(
         return await self.http.request(self.log, url)
 
     async def stream_csv(
-        self, 
+        self,
         path: str,
-        model: type[_CSVRow],
         field_names: list[str],
-    ) -> AsyncGenerator[_CSVRow, None]:
+    ) -> AsyncGenerator[dict[str, Any], None]:
+        """
+        Stream and parse a CSV file, yielding dict rows.
+
+        Transformations applied:
+        - Empty strings converted to None
+        """
         url_without_sas = f"{self.base_url}/{self.filesystem}/{path}"
 
         self.log.debug(f"Streaming CSV contents from /{path}.", {
@@ -134,11 +137,10 @@ async def stream_csv(
 
         _, body = await self.http.request_stream(self.log, url)
 
-        processor = IncrementalCSVProcessor(
-            body(),
-            model,
-            fieldnames=field_names
-        )
+        async for row in IncrementalCSVProcessor(body(), fieldnames=field_names):
+            # Convert empty strings to None
+            for key, value in row.items():
+                if value == "":
+                    row[key] = None
 
-        async for item in processor:
-            yield item
+            yield row

source-dynamics-365-finance-and-operations/source_dynamics_365_finance_and_operations/api.py

@@ -1,7 +1,7 @@
 import asyncio
 from datetime import datetime
 from logging import Logger
-from typing import AsyncGenerator
+from typing import AsyncGenerator, cast
 
 # Some functions that send a request to Azure for listing files or reading
 # metadata files are wrapped with the alru_cache decorator. alru_cache
@@ -141,7 +141,7 @@ async def read_csvs_in_folder(
     folder: str,
     table_name: str,
     client: ADLSGen2Client,
-) -> AsyncGenerator[BaseTable, None]:
+) -> AsyncGenerator[dict, None]:
     folder_contents = await get_folder_contents_for_table(folder, table_name, client)
 
     csvs: list[ADLSPathMetadata] = []
@@ -164,16 +164,36 @@ async def read_csvs_in_folder(
         csvs.sort(key=lambda c: c.last_modified_datetime)
 
         for csv in csvs:
-            async for row in client.stream_csv(csv.name, table_model, table_model.field_names):
-                yield row
+            async for row in client.stream_csv(csv.name, table_model.field_names):
+                yield transform_row(row, table_model.boolean_fields)
+
+
+def transform_row(row: dict[str, str], boolean_fields: frozenset[str]) -> dict[str, str | bool | dict[str, str]]:
+    """
+    Apply Dynamics 365-specific transformations to a CSV row.
+
+    Transformations:
+    - Convert boolean fields from "True"/"False"/empty strings to actual booleans
+    - Add _meta field with operation type based on IsDelete field
+      (IsDelete is "True" for deletions, "" otherwise)
+    """
+    result = cast(dict[str, str | bool | dict[str, str]], row)
+
+    for field_name in boolean_fields:
+        value = row.get(field_name)
+        result[field_name] = value.lower() == "true" if value else False
+
+    result["_meta"] = {"op": "d" if result.get("IsDelete") else "u"}
+
+    return result
 
 
 async def fetch_changes(
     client: ADLSGen2Client,
     table_name: str,
     log: Logger,
     log_cursor: LogCursor,
-) -> AsyncGenerator[BaseTable | LogCursor, None]:
+) -> AsyncGenerator[dict | LogCursor, None]:
     assert isinstance(log_cursor, datetime)
 
     finalized_folders = await get_finalized_timestamp_folders(client)

source-dynamics-365-finance-and-operations/source_dynamics_365_finance_and_operations/models.py

@@ -1,4 +1,3 @@
-from enum import StrEnum
 from typing import ClassVar, Literal, Annotated
 
 from estuary_cdk.capture.common import (
@@ -8,7 +7,7 @@
 )
 from estuary_cdk.incremental_csv_processor import BaseCSVRow
 
-from pydantic import BaseModel, Field, model_validator, BeforeValidator
+from pydantic import BaseModel, Field, BeforeValidator
 
 
 class AzureSASToken(BaseModel):
@@ -60,6 +59,11 @@ class Attribute(BaseModel, extra="allow"):
 
 
 class BaseTable(BaseCSVRow, extra="allow"):
+    """
+    Used for schema generation and table metadata. Not used to validate actual
+    documents - we yield raw dicts to avoid Pydantic's serialization/validation
+    overhead that's not necessary in this connector.
+    """
     name: ClassVar[str]
     field_names: ClassVar[list[str]]
     field_types: ClassVar[dict[str, str]]
@@ -70,24 +74,6 @@ class BaseTable(BaseCSVRow, extra="allow"):
     Id: str
     IsDelete: bool | None
 
-    @model_validator(mode='before')
-    @classmethod
-    def convert_boolean_fields(cls, values: dict) -> dict:
-        for field_name in cls.boolean_fields:
-            value = values.get(field_name)
-            if value is not None:
-                assert isinstance(value, str)
-                values[field_name] = value.lower() == 'true'
-        return values
-
-    @model_validator(mode='after')
-    def set_meta_op(self) -> 'BaseTable':
-        if self.IsDelete:
-            self.meta_ = BaseTable.Meta(op='d')
-        else:
-            self.meta_ = BaseTable.Meta(op='u')
-        return self
-
 
 def model_from_entity(entity: ModelDotJson.Entity) -> type[BaseTable]:
     field_names = [attr.name for attr in entity.attributes]

source-dynamics-365-finance-and-operations/tests/test_api.py

@@ -0,0 +1,94 @@
+from source_dynamics_365_finance_and_operations.api import transform_row
+
+
+class TestTransformRow:
+    """Tests for the transform_row helper function."""
+
+    def test_converts_boolean_field_true(self):
+        """Boolean field with 'true' string should become True."""
+        row = {"IsActive": "true", "Name": "Test"}
+        boolean_fields = frozenset({"IsActive"})
+
+        result = transform_row(row, boolean_fields)
+
+        assert result["IsActive"] is True
+        assert result["Name"] == "Test"
+
+    def test_converts_boolean_field_false(self):
+        """Boolean field with 'false' string should become False."""
+        row = {"IsActive": "false", "Name": "Test"}
+        boolean_fields = frozenset({"IsActive"})
+
+        result = transform_row(row, boolean_fields)
+
+        assert result["IsActive"] is False
+
+    def test_converts_boolean_field_case_insensitive(self):
+        """Boolean conversion should be case-insensitive."""
+        row = {"IsActive": "TRUE", "IsEnabled": "False", "IsValid": "TrUe"}
+        boolean_fields = frozenset({"IsActive", "IsEnabled", "IsValid"})
+
+        result = transform_row(row, boolean_fields)
+
+        assert result["IsActive"] is True
+        assert result["IsEnabled"] is False
+        assert result["IsValid"] is True
+
+    def test_boolean_field_none_becomes_false(self):
+        """Boolean field with None value should become False."""
+        row = {"IsActive": None, "Name": "Test"}
+        boolean_fields = frozenset({"IsActive"})
+
+        result = transform_row(row, boolean_fields)
+
+        assert result["IsActive"] is False
+
+    def test_multiple_boolean_fields(self):
+        """Multiple boolean fields should all be converted."""
+        row = {"IsActive": "true", "IsDeleted": "false", "IsEnabled": "true"}
+        boolean_fields = frozenset({"IsActive", "IsDeleted", "IsEnabled"})
+
+        result = transform_row(row, boolean_fields)
+
+        assert result["IsActive"] is True
+        assert result["IsDeleted"] is False
+        assert result["IsEnabled"] is True
+
+    def test_meta_op_delete_when_isdelete_true(self):
+        """_meta.op should be 'd' when IsDelete is True."""
+        row = {"IsDelete": "True", "Name": "Test"}
+        boolean_fields = frozenset()
+
+        result = transform_row(row, boolean_fields)
+
+        assert result["_meta"] == {"op": "d"}
+
+    def test_meta_op_update_when_isdelete_empty(self):
+        """_meta.op should be 'u' when IsDelete is empty string."""
+        row = {"IsDelete": "", "Name": "Test"}
+        boolean_fields = frozenset()
+
+        result = transform_row(row, boolean_fields)
+
+        assert result["_meta"] == {"op": "u"}
+
+    def test_meta_op_update_when_isdelete_false(self):
+        """_meta.op should be 'u' when IsDelete is 'False' and converted to bool."""
+        row = {"IsDelete": "False", "Name": "Test"}
+        boolean_fields = frozenset({"IsDelete"})
+
+        result = transform_row(row, boolean_fields)
+
+        assert result["IsDelete"] is False
+        assert result["_meta"] == {"op": "u"}
+
+    def test_mutates_row_in_place(self):
+        """transform_row should mutate the row in place and return it."""
+        row = {"IsActive": "true"}
+        boolean_fields = frozenset({"IsActive"})
+
+        result = transform_row(row, boolean_fields)
+
+        assert result is row
+        assert row["IsActive"] is True
+        assert "_meta" in row