chore: add PydanticValidator for EnterpriseCredentialInventory - BED-10126

src/openhound_github/models/__init__.py

@@ -7,6 +7,7 @@
 from .classic_personal_access_token import ClassicPersonalAccessToken
 from .enterprise import Enterprise
 from .enterprise_admin import EnterpriseAdmin
+from .enterprise_credential_inventory import EnterpriseCredentialInventory
 from .enterprise_helpers import enterprise_role_node_id, enterprise_team_node_id
 from .enterprise_managed_user import EnterpriseManagedUser
 from .enterprise_member import BaseUser, flatten_enterprise_member
@@ -140,6 +141,7 @@
     "DeployKey",
     "Enterprise",
     "EnterpriseAdmin",
+    "EnterpriseCredentialInventory",
     "EnterpriseManagedUser",
     "EnterpriseOrganization",
     "EnterpriseRole",

src/openhound_github/models/enterprise_credential_inventory.py

@@ -0,0 +1,19 @@
+from typing import Any
+
+from pydantic import BaseModel
+
+
+class EnterpriseCredentialInventory(BaseModel):
+    """A downloaded credential export, before credential-specific parsing.
+
+    Keep CSV fields as raw values so unknown credential types and new columns
+    survive collection. DLT returns dictionaries for the downstream parser.
+    """
+
+    export_id: str
+    as_of: str | None = None
+    created_with_fallback: bool
+    columns: list[str]
+    rows: list[dict[str, Any]]
+    enterprise_node_id: str
+    enterprise_name: str

src/openhound_github/resources/enterprise.py

@@ -37,6 +37,7 @@
     ClassicPersonalAccessToken,
     Enterprise,
     EnterpriseAdmin,
+    EnterpriseCredentialInventory,
     EnterpriseManagedUser,
     EnterpriseOrganization,
     EnterpriseRunner,
@@ -285,7 +286,7 @@ def _download_enterprise_credential_inventory(
 
 @app.transformer(
     name="enterprise_credential_inventory",
-    columns={"columns": {"data_type": "json"}, "rows": {"data_type": "json"}},
+    columns=EnterpriseCredentialInventory,
     parallelized=True,
 )
 def enterprise_credential_inventory(enterprise_data: Enterprise, ctx: SourceContext):

tests/test_classic_personal_access_tokens.py

@@ -6,18 +6,20 @@
 from unittest.mock import MagicMock
 
 import dlt
+import duckdb
 import pytest
 import requests
-import duckdb
+from dlt.common.schema.exceptions import DataValidationError
 from dlt.destinations import filesystem
+from dlt.extract.validation import PydanticValidator
 
 from openhound_github.kinds import edges as ek
 from openhound_github.lookup import GithubLookup
 from openhound_github.models.classic_personal_access_token import (
     ClassicPersonalAccessToken,
 )
-from openhound_github.models.enterprise_member import BaseUser
 from openhound_github.models.enterprise import Enterprise
+from openhound_github.models.enterprise_member import BaseUser
 from openhound_github.resources.enterprise import (
     SourceContext,
     _download_enterprise_credential_inventory,
@@ -53,6 +55,43 @@ def iter_content(self, chunk_size):
             yield self.content[offset : offset + chunk_size]
 
 
+def test_inventory_validator_preserves_raw_csv_for_token_parser(monkeypatch):
+    validator = enterprise_credential_inventory.validator
+    assert isinstance(validator, PydanticValidator)
+    monkeypatch.setattr(
+        validator, "table_name", enterprise_credential_inventory.name, raising=False
+    )
+    assert enterprise_credential_inventory.columns["columns"]["data_type"] == "json"
+    assert enterprise_credential_inventory.columns["rows"]["data_type"] == "json"
+
+    inventory = {
+        "export_id": "export-1",
+        "as_of": None,
+        "created_with_fallback": False,
+        "enterprise_node_id": "E_1",
+        "enterprise_name": "enterprise",
+        "columns": ["credential_type", "credential_id", "new_csv_column"],
+        "rows": [
+            {"credential_type": "classic_pat", "credential_id": "invalid"},
+            {"credential_type": "classic_pat", "credential_id": "42"},
+            {
+                "credential_type": "future_credential_type",
+                "credential_id": "007",
+                "new_csv_column": None,
+            },
+        ],
+    }
+    validated = validator(inventory)
+    assert isinstance(validated, dict)
+    assert validated == inventory
+    tokens = list(classic_personal_access_tokens.__wrapped__(validated))
+    assert [token["credential_id"] for token in tokens] == [42]
+
+    # Validate the envelope without applying credential-specific rules to CSV rows.
+    with pytest.raises(DataValidationError):
+        validator({**inventory, "rows": "not a list of rows"})
+
+
 def test_classic_pat_export_polls_one_job_and_deduplicates_org_rows(monkeypatch):
     client = MagicMock()
     client.post.return_value = Response(
@@ -180,6 +219,7 @@ def inventory_rows(client, slug, fallback_client=None, on_export_created=None):
         return {
             "export_id": "export-1",
             "as_of": "2026-10-05T20:00:00Z",
+            "created_with_fallback": False,
             "columns": ["credential_type", "credential_id", "hashed_token"],
             "rows": [
                 {