fix(file-based): override primary_key in PermissionsFileBasedStream to avoid invalid parser-defined key (#903)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: rwask

airbyte_cdk/sources/file_based/stream/concurrent/adapters.py

@@ -155,10 +155,7 @@ def get_json_schema(self) -> Mapping[str, Any]:
 
     @property
     def primary_key(self) -> PrimaryKeyType:
-        return (
-            self._legacy_stream.config.primary_key
-            or self.get_parser().get_parser_defined_primary_key(self._legacy_stream.config)
-        )
+        return self._legacy_stream.primary_key
 
     def get_parser(self) -> FileTypeParser:
         return self._legacy_stream.get_parser()

airbyte_cdk/sources/file_based/stream/permissions_file_based_stream.py

@@ -7,6 +7,7 @@
 
 from airbyte_cdk.models import AirbyteLogMessage, AirbyteMessage, Level
 from airbyte_cdk.models import Type as MessageType
+from airbyte_cdk.sources.file_based.config.file_based_stream_config import PrimaryKeyType
 from airbyte_cdk.sources.file_based.file_based_stream_permissions_reader import (
     AbstractFileBasedStreamPermissionsReader,
 )
@@ -35,6 +36,10 @@ def __init__(
         super().__init__(**kwargs)
         self.stream_permissions_reader = stream_permissions_reader
 
+    @property
+    def primary_key(self) -> PrimaryKeyType:
+        return self.config.primary_key
+
     def _filter_schema_invalid_properties(
         self, configured_catalog_json_schema: Dict[str, Any]
     ) -> Dict[str, Any]: