file-based: add identities stream and rename acl toggle

Author: aldogonzalez8

airbyte_cdk/sources/file_based/config/abstract_file_based_spec.py

@@ -11,6 +11,9 @@
 
 from airbyte_cdk import OneOfOptionConfig
 from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig
+from airbyte_cdk.sources.file_based.config.identities_based_stream_config import (
+    IdentitiesStreamConfig,
+)
 from airbyte_cdk.sources.utils import schema_helpers
 
 
@@ -22,12 +25,17 @@ class Config(OneOfOptionConfig):
 
     delivery_type: Literal["use_records_transfer"] = Field("use_records_transfer", const=True)
 
-    sync_metadata: bool = Field(
-        title="Make stream sync files metadata",
-        description="If enabled, streams will sync files metadata instead of files data.",
+    sync_acl_permissions: bool = Field(
+        title="Include ACL Permissions",
+        description="Joins Document allowlists to each stream.",
         default=False,
         airbyte_hidden=True,
     )
+    identities: Optional[IdentitiesStreamConfig] = Field(
+        title="Identities configuration",
+        description="Configuration for identities",
+        airbyte_hidden=True,
+    )
 
 
 class DeliverRawFiles(BaseModel):

airbyte_cdk/sources/file_based/config/identities_based_stream_config.py

@@ -0,0 +1,7 @@
+from pydantic.v1 import BaseModel, Field
+from typing import Literal
+
+
+class IdentitiesStreamConfig(BaseModel):
+    name: Literal["identities"] = Field("identities", const=True, airbyte_hidden=True)
+    domain: str = Field(title="Domain", description="The domain of the identities.")

airbyte_cdk/sources/file_based/config/permissions.py

@@ -0,0 +1,33 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+
+import uuid
+from datetime import datetime
+from enum import Enum
+from pydantic.v1 import BaseModel
+
+
+class RemoteFileIdentityType(Enum):
+    USER = "user"
+    GROUP = "group"
+
+
+class RemoteFileIdentity(BaseModel):
+    id: uuid.UUID
+    remote_id: str
+    parent_id: str | None = None
+    name: str | None = None
+    description: str | None = None
+    email_address: str | None = None
+    member_email_addresses: list[str] | None = None
+    type: RemoteFileIdentityType
+    modified_at: datetime
+
+
+class RemoteFilePermissions(BaseModel):
+    id: str
+    file_path: str
+    allowed_identity_remote_ids: list[str] | None = None
+    denied_identity_remote_ids: list[str] | None = None
+    publicly_accessible: bool = False

airbyte_cdk/sources/file_based/file_based_source.py

@@ -33,6 +33,9 @@
     FileBasedStreamConfig,
     ValidationPolicy,
 )
+from airbyte_cdk.sources.file_based.config.identities_based_stream_config import (
+    IdentitiesStreamConfig,
+)
 from airbyte_cdk.sources.file_based.discovery_policy import (
     AbstractDiscoveryPolicy,
     DefaultDiscoveryPolicy,
@@ -49,7 +52,11 @@
     DEFAULT_SCHEMA_VALIDATION_POLICIES,
     AbstractSchemaValidationPolicy,
 )
-from airbyte_cdk.sources.file_based.stream import AbstractFileBasedStream, DefaultFileBasedStream
+from airbyte_cdk.sources.file_based.stream import (
+    AbstractFileBasedStream,
+    DefaultFileBasedStream,
+    IdentitiesStream,
+)
 from airbyte_cdk.sources.file_based.stream.concurrent.adapters import FileBasedStreamFacade
 from airbyte_cdk.sources.file_based.stream.concurrent.cursor import (
     AbstractConcurrentFileBasedCursor,
@@ -157,13 +164,17 @@ def check_connection(
         errors = []
         tracebacks = []
         for stream in streams:
+            if isinstance(stream, IdentitiesStream):
+                # Probably need to check identities endpoint/api access but will skip for now.
+                continue
             if not isinstance(stream, AbstractFileBasedStream):
                 raise ValueError(f"Stream {stream} is not a file-based stream.")
             try:
                 parsed_config = self._get_parsed_config(config)
                 availability_method = (
                     stream.availability_strategy.check_availability
-                    if self._use_file_transfer(parsed_config) or self._sync_metadata(parsed_config)
+                    if self._use_file_transfer(parsed_config)
+                    or self._sync_acl_permissions(parsed_config)
                     else stream.availability_strategy.check_availability_and_parsability
                 )
                 (
@@ -289,6 +300,12 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
                     )
 
                 streams.append(stream)
+
+            if self._add_identities_stream(parsed_config):
+                identities_stream = self._make_identities_stream(
+                    stream_config=parsed_config.delivery_method.identities
+                )
+                streams.append(identities_stream)
             return streams
 
         except ValidationError as exc:
@@ -312,7 +329,19 @@ def _make_default_stream(
             cursor=cursor,
             use_file_transfer=self._use_file_transfer(parsed_config),
             preserve_directory_structure=self._preserve_directory_structure(parsed_config),
-            sync_metadata=self._sync_metadata(parsed_config),
+            sync_acl_permissions=self._sync_acl_permissions(parsed_config),
+        )
+
+    def _make_identities_stream(
+        self,
+        stream_config: IdentitiesStreamConfig,
+    ) -> Stream:
+        return IdentitiesStream(
+            config=stream_config,
+            catalog_schema=self.stream_schemas.get(stream_config.name),
+            stream_reader=self.stream_reader,
+            discovery_policy=self.discovery_policy,
+            errors_collector=self.errors_collector,
         )
 
     def _get_stream_from_catalog(
@@ -419,11 +448,19 @@ def _preserve_directory_structure(parsed_config: AbstractFileBasedSpec) -> bool:
         return True
 
     @staticmethod
-    def _sync_metadata(parsed_config: AbstractFileBasedSpec) -> bool:
+    def _sync_acl_permissions(parsed_config: AbstractFileBasedSpec) -> bool:
         if (
             FileBasedSource._use_records_transfer(parsed_config)
-            and hasattr(parsed_config.delivery_method, "sync_metadata")
-            and parsed_config.delivery_method.sync_metadata is not None
+            and hasattr(parsed_config.delivery_method, "sync_acl_permissions")
+            and parsed_config.delivery_method.sync_acl_permissions is not None
         ):
-            return parsed_config.delivery_method.sync_metadata
+            return parsed_config.delivery_method.sync_acl_permissions
         return False
+
+    @staticmethod
+    def _add_identities_stream(parsed_config: AbstractFileBasedSpec) -> bool:
+        return (
+            FileBasedSource._sync_acl_permissions(parsed_config)
+            and parsed_config.delivery_method.identities is not None
+            and parsed_config.delivery_method.identities.domain
+        )

airbyte_cdk/sources/file_based/file_based_stream_reader.py

@@ -155,14 +155,14 @@ def preserve_directory_structure(self) -> bool:
             return self.config.delivery_method.preserve_directory_structure
         return True
 
-    def sync_metadata(self) -> bool:
+    def sync_acl_permissions(self) -> bool:
         if (
             self.config
             and self.use_records_transfer()
-            and hasattr(self.config.delivery_method, "sync_metadata")
-            and self.config.delivery_method.sync_metadata is not None
+            and hasattr(self.config.delivery_method, "sync_acl_permissions")
+            and self.config.delivery_method.sync_acl_permissions is not None
         ):
-            return self.config.delivery_method.sync_metadata
+            return self.config.delivery_method.sync_acl_permissions
         return False
 
     @abstractmethod
@@ -203,25 +203,16 @@ def _get_file_transfer_paths(self, file: RemoteFile, local_directory: str) -> Li
         absolute_file_path = path.abspath(local_file_path)
         return [file_relative_path, local_file_path, absolute_file_path]
 
-    def get_file_metadata(self, file: RemoteFile, logger: logging.Logger) -> Dict[str, Any]:
+    def get_file_acl_permissions(self, file: RemoteFile, logger: logging.Logger) -> Dict[str, Any]:
         """
         This is required for connectors that will support syncing
-        metadata from files.
+        ACL Permissions from files.
         """
         return {}
 
-    def get_metadata_schema(self) -> Dict[str, Any]:
-        """ "
-        Base schema to emit metadata records for a file,
-        override in stream reader implementation if the requirements
-        are different.
-        """
-        return {
-            "type": "object",
-            "properties": {
-                "id": {"type": "string"},
-                "file_path": {"type": "string"},
-                "allowed_identity_remote_ids": {"type": "array", "items": "string"},
-                "is_public": {"type": "boolean"},
-            },
-        }
+    def load_identity_groups(self) -> Iterable[Dict[str, Any]]:
+        """
+        This is required for connectors that will support syncing
+        identities.
+        """
+        yield {}

airbyte_cdk/sources/file_based/schema_helpers.py

@@ -23,6 +23,31 @@
     "properties": {"data": {"type": "object"}, "file": {"type": "object"}},
 }
 
+remote_file_permissions_schema = {
+    "type": "object",
+    "properties": {
+        "id": {"type": "string"},
+        "file_path": {"type": "string"},
+        "allowed_identity_remote_ids": {"type": "array", "items": "string"},
+        "publicly_accessible": {"type": "boolean"},
+    },
+}
+
+remote_file_identity_schema = {
+    "type": "object",
+    "properties": {
+        "id": {"type": "string"},
+        "remote_id": {"type": "string"},
+        "parent_id": {"type": ["null", "string"]},
+        "name": {"type": ["null", "string"]},
+        "description": {"type": ["null", "string"]},
+        "email_address": {"type": ["null", "string"]},
+        "member_email_addresses": {"type": ["null", "array"]},
+        "type": {"type": "string"},
+        "modified_at": {"type": "string"},
+    },
+}
+
 
 @total_ordering
 class ComparableType(Enum):

airbyte_cdk/sources/file_based/stream/__init__.py

@@ -1,4 +1,5 @@
 from airbyte_cdk.sources.file_based.stream.abstract_file_based_stream import AbstractFileBasedStream
 from airbyte_cdk.sources.file_based.stream.default_file_based_stream import DefaultFileBasedStream
+from airbyte_cdk.sources.file_based.stream.identities_stream import IdentitiesStream
 
-__all__ = ["AbstractFileBasedStream", "DefaultFileBasedStream"]
+__all__ = ["AbstractFileBasedStream", "DefaultFileBasedStream", "IdentitiesStream"]

airbyte_cdk/sources/file_based/stream/default_file_based_stream.py

@@ -30,6 +30,7 @@
     file_transfer_schema,
     merge_schemas,
     schemaless_schema,
+    remote_file_permissions_schema,
 )
 from airbyte_cdk.sources.file_based.stream import AbstractFileBasedStream
 from airbyte_cdk.sources.file_based.stream.cursor import AbstractFileBasedCursor
@@ -47,7 +48,7 @@ class DefaultFileBasedStream(AbstractFileBasedStream, IncrementalMixin):
 
     FILE_TRANSFER_KW = "use_file_transfer"
     PRESERVE_DIRECTORY_STRUCTURE_KW = "preserve_directory_structure"
-    SYNC_METADATA_KW = "sync_metadata"
+    SYNC_ACL_PERMISSIONS_KW = "sync_acl_permissions"
     FILES_KEY = "files"
     DATE_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
     ab_last_mod_col = "_ab_source_file_last_modified"
@@ -57,7 +58,7 @@ class DefaultFileBasedStream(AbstractFileBasedStream, IncrementalMixin):
     airbyte_columns = [ab_last_mod_col, ab_file_name_col]
     use_file_transfer = False
     preserve_directory_structure = True
-    sync_metadata = False
+    sync_acl_permissions = False
 
     def __init__(self, **kwargs: Any):
         if self.FILE_TRANSFER_KW in kwargs:
@@ -66,8 +67,8 @@ def __init__(self, **kwargs: Any):
             self.preserve_directory_structure = kwargs.pop(
                 self.PRESERVE_DIRECTORY_STRUCTURE_KW, True
             )
-        if self.SYNC_METADATA_KW in kwargs:
-            self.sync_metadata = kwargs.pop(self.SYNC_METADATA_KW, False)
+        if self.SYNC_ACL_PERMISSIONS_KW in kwargs:
+            self.sync_acl_permissions = kwargs.pop(self.SYNC_ACL_PERMISSIONS_KW, False)
         super().__init__(**kwargs)
 
     @property
@@ -109,8 +110,8 @@ def _filter_schema_invalid_properties(
                     self.ab_file_name_col: {"type": "string"},
                 },
             }
-        elif self.sync_metadata:
-            return self.stream_reader.get_metadata_schema()
+        elif self.sync_acl_permissions:
+            return remote_file_permissions_schema
         else:
             return super()._filter_schema_invalid_properties(configured_catalog_json_schema)
 
@@ -193,9 +194,9 @@ def read_records_from_slice(self, stream_slice: StreamSlice) -> Iterable[Airbyte
                         yield stream_data_to_airbyte_message(
                             self.name, record, is_file_transfer_message=True
                         )
-                elif self.sync_metadata:
+                elif self.sync_acl_permissions:
                     try:
-                        metadata_record = self.stream_reader.get_file_metadata(
+                        metadata_record = self.stream_reader.get_file_acl_permissions(
                             file, logger=self.logger
                         )
                         yield stream_data_to_airbyte_message(
@@ -310,8 +311,8 @@ def get_json_schema(self) -> JsonSchema:
     def _get_raw_json_schema(self) -> JsonSchema:
         if self.use_file_transfer:
             return file_transfer_schema
-        elif self.sync_metadata:
-            return self.stream_reader.get_metadata_schema()
+        elif self.sync_acl_permissions:
+            return remote_file_permissions_schema
         elif self.config.input_schema:
             return self.config.get_input_schema()  # type: ignore
         elif self.config.schemaless: