file-based: add method in stream reader to obtain Identities schema

airbyte_cdk/sources/file_based/file_based_stream_reader.py

@@ -192,7 +192,7 @@ def get_file_acl_permissions(self, file: RemoteFile, logger: logging.Logger) ->
         ACL Permissions from files.
         """
         raise NotImplementedError(
-            f"{self.__class__.__name__} required to support ACL permissions, please update file_permissions_schema accordingly to obtain the required schema for each stream on the source implementation"
+            f"{self.__class__.__name__} required to support ACL permissions, please update file_permissions_schema accordingly."
         )
 
     @abstractmethod
@@ -202,7 +202,7 @@ def load_identity_groups(self, logger: logging.Logger) -> Iterable[Dict[str, Any
         identities.
         """
         raise NotImplementedError(
-            f"{self.__class__.__name__} required to support identities, please add schema for your identities stream in schemas folder"
+            f"{self.__class__.__name__} required to support identities, please update identities_schema."
         )
 
     @property
@@ -213,5 +213,16 @@ def file_permissions_schema(self) -> Dict[str, Any]:
         ACL Permissions from files.
         """
         raise NotImplementedError(
-            f"{self.__class__.__name__} required to support ACL Permissions, please return required json schema for your permissions streams"
+            f"{self.__class__.__name__} required to support ACL Permissions, please return required json schema for your permissions streams."
+        )
+
+    @property
+    @abstractmethod
+    def identities_schema(self) -> Dict[str, Any]:
+        """
+        This is required for connectors that will support syncing
+        identities.
+        """
+        raise NotImplementedError(
+            f"{self.__class__.__name__} required to support fetch Identities, please return required json schema for your Identities stream."
         )

airbyte_cdk/sources/file_based/stream/identities_stream.py

@@ -10,6 +10,7 @@
 from airbyte_cdk.sources.file_based.exceptions import FileBasedErrorsCollector
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader
 from airbyte_cdk.sources.streams.permissions.identities import Identities
+from airbyte_cdk.sources.streams.core import JsonSchema
 
 
 class FileIdentities(Identities):
@@ -40,3 +41,7 @@ def primary_key(self) -> PrimaryKeyType:
 
     def load_identity_groups(self) -> Iterable[Dict[str, Any]]:
         return self.stream_reader.load_identity_groups(logger=self.logger)
+
+    @cache
+    def get_json_schema(self) -> JsonSchema:
+        return self.stream_reader.identities_schema