file-mode-api: make filename_extractor optional and fallback to guid

Author: aldogonzalez8

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1450,13 +1450,14 @@ definitions:
               - "$ref": "#/definitions/CustomRecordExtractor"
               - "$ref": "#/definitions/DpathExtractor"
           filename_extractor:
-            description: Defines relative path and name to store the file
+            description: Defines the name to store the file. Stream name is automatically added to the file path. File unique ID can be used to avoid overwriting files. Random UUID will be used if the extractor is not provided.
             type: string
             interpolation_context:
               - config
               - record
             examples:
-              - "{{ record.relative_path }}/{{ record.file_name }}/"
+              - "{{ record.id }}/{{ record.file_name }}/"
+              - "{{ record.id }}_{{ record.file_name }}/"
       $parameters:
         type: object
         additional_properties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2003,11 +2003,12 @@ class FileUploader(BaseModel):
         None,
         description="Responsible for fetching the content of the file. If not defined, the assumption is that the whole response body is the file content",
     )
-    filename_extractor: str = Field(
-        ...,
-        description="File Name extractor.",
+    filename_extractor: Optional[str] = Field(
+        None,
+        description="Defines the name to store the file. Stream name is automatically added to the file path. File unique ID can be used to avoid overwriting files. Random UUID will be used if the extractor is not provided.",
         examples=[
-            "{{ record.relative_path }}/{{ record.file_name }}/",
+            "{{ record.id }}/{{ record.file_name }}/",
+            "{{ record.id }}_{{ record.file_name }}/",
         ],
     )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3353,7 +3353,7 @@ def create_file_uploader(
             download_target_extractor=download_target_extractor,
             config=config,
             parameters=model.parameters or {},
-            filename_extractor=model.filename_extractor,
+            filename_extractor=model.filename_extractor if model.filename_extractor else None,
         )
 
     def create_moving_window_call_rate_policy(

airbyte_cdk/sources/declarative/retrievers/file_uploader.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+import uuid
 from dataclasses import InitVar, dataclass, field
 from pathlib import Path
 from typing import Optional, Mapping, Union, Any
@@ -31,14 +32,15 @@ class FileUploader:
     config: Config
     parameters: InitVar[Mapping[str, Any]]
 
-    filename_extractor: Union[InterpolatedString, str]
+    filename_extractor: Optional[Union[InterpolatedString, str]] = None
     content_extractor: Optional[RecordExtractor] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
-        self._filename_extractor = InterpolatedString.create(
-            self.filename_extractor,
-            parameters=parameters,
-        )
+        if self.filename_extractor:
+            self.filename_extractor = InterpolatedString.create(
+                self.filename_extractor,
+                parameters=parameters,
+            )
 
     def upload(self, record: Record) -> None:
         mocked_response = SafeResponse()
@@ -60,9 +62,13 @@ def upload(self, record: Record) -> None:
         else:
             files_directory = Path(get_files_directory())
 
-            relative_path = self._filename_extractor.eval(self.config, record=record)
-            relative_path = relative_path.lstrip("/")
-            file_relative_path = Path(relative_path)
+            file_name = (
+                self.filename_extractor.eval(self.config, record=record)
+                if self.filename_extractor
+                else str(uuid.uuid4())
+            )
+            file_name = file_name.lstrip("/")
+            file_relative_path = Path(record.stream_name) / Path(file_name)
 
             full_path = files_directory / file_relative_path
             full_path.parent.mkdir(parents=True, exist_ok=True)
@@ -72,7 +78,7 @@ def upload(self, record: Record) -> None:
             file_size_bytes = full_path.stat().st_size
 
             logger.info("File uploaded successfully")
-            logger.info(f"File url: {str(full_path)} ")
+            logger.info(f"File url: {str(full_path)}")
             logger.info(f"File size: {file_size_bytes / 1024} KB")
             logger.info(f"File relative path: {str(file_relative_path)}")
 

unit_tests/sources/declarative/file/file_stream_manifest.yaml

@@ -161,7 +161,6 @@ definitions:
       download_target_extractor:
         type: DpathExtractor
         field_path: [ "content_url" ]
-      filename_extractor: "{{ record.relative_path }}/{{ record.file_name }}/"
 
 
 streams:

unit_tests/sources/declarative/file/test_file_stream.py

@@ -1,3 +1,5 @@
+import re
+
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 from unittest import TestCase
@@ -29,9 +31,12 @@ def _source(
     catalog: ConfiguredAirbyteCatalog,
     config: Dict[str, Any],
     state: Optional[List[AirbyteStateMessage]] = None,
+    yaml_file: Optional[str] = None,
 ) -> YamlDeclarativeSource:
+    if not yaml_file:
+        yaml_file = "file_stream_manifest.yaml"
     return YamlDeclarativeSource(
-        path_to_yaml=str(Path(__file__).parent / "file_stream_manifest.yaml"),
+        path_to_yaml=str(Path(__file__).parent / yaml_file),
         catalog=catalog,
         config=config,
         state=state,
@@ -43,11 +48,12 @@ def read(
     catalog: ConfiguredAirbyteCatalog,
     state_builder: Optional[StateBuilder] = None,
     expecting_exception: bool = False,
+    yaml_file: Optional[str] = None,
 ) -> EntrypointOutput:
     config = config_builder.build()
     state = state_builder.build() if state_builder else StateBuilder().build()
     return entrypoint_read(
-        _source(catalog, config, state), config, catalog, state, expecting_exception
+        _source(catalog, config, state, yaml_file), config, catalog, state, expecting_exception
     )
 
 
@@ -96,7 +102,32 @@ def test_get_article_attachments(self) -> None:
         file_reference = output.records[0].record.file_reference
         assert file_reference
         assert file_reference.file_url
+        assert re.match(r"^.*/article_attachments/[0-9a-fA-F-]{36}$", file_reference.file_url)
         assert file_reference.file_relative_path
+        assert re.match(
+            r"^article_attachments/[0-9a-fA-F-]{36}$", file_reference.file_relative_path
+        )
+        assert file_reference.file_size_bytes
+
+    def test_get_article_attachments_with_filename_extractor(self) -> None:
+        output = read(
+            self._config(),
+            CatalogBuilder()
+            .with_stream(ConfiguredAirbyteStreamBuilder().with_name("article_attachments"))
+            .build(),
+            yaml_file="test_file_stream_with_filename_extractor.yaml",
+        )
+
+        assert output.records
+        file_reference = output.records[0].record.file_reference
+        assert file_reference
+        assert file_reference.file_url
+        # todo: once we finally mock the response update to check file name
+        assert not re.match(r"^.*/article_attachments/[0-9a-fA-F-]{36}$", file_reference.file_url)
+        assert file_reference.file_relative_path
+        assert not re.match(
+            r"^article_attachments/[0-9a-fA-F-]{36}$", file_reference.file_relative_path
+        )
         assert file_reference.file_size_bytes
 
     def test_discover_article_attachments(self) -> None:

unit_tests/sources/declarative/file/test_file_stream_with_filename_extractor.yaml

@@ -0,0 +1,191 @@
+version: 2.0.0
+
+type: DeclarativeSource
+
+check:
+  type: CheckStream
+  stream_names:
+    - "articles"
+
+definitions:
+  bearer_authenticator:
+    type: BearerAuthenticator
+    api_token: "{{ config['credentials']['access_token'] }}"
+  basic_authenticator:
+    type: BasicHttpAuthenticator
+    username: "{{ config['credentials']['email'] + '/token' }}"
+    password: "{{ config['credentials']['api_token'] }}"
+
+  retriever:
+    type: SimpleRetriever
+    requester:
+      type: HttpRequester
+      url_base: https://{{ config['subdomain'] }}.zendesk.com/api/v2/
+      http_method: GET
+      authenticator:
+        type: SelectiveAuthenticator
+        authenticator_selection_path: ["credentials", "credentials"]
+        authenticators:
+          oauth2.0: "#/definitions/bearer_authenticator"
+          api_token: "#/definitions/basic_authenticator"
+    record_selector:
+      type: RecordSelector
+      extractor:
+        type: DpathExtractor
+        field_path:
+          ["{{ parameters.get('data_path') or parameters.get('name') }}"]
+      schema_normalization: Default
+    paginator:
+      type: DefaultPaginator
+      page_size_option:
+        type: RequestOption
+        field_name: "per_page"
+        inject_into: request_parameter
+      pagination_strategy:
+        type: CursorPagination
+        page_size: 100
+        cursor_value: '{{ response.get("next_page", {}) }}'
+        stop_condition: "{{ last_page_size == 0 }}"
+      page_token_option:
+        type: RequestPath
+
+  base_stream:
+    type: DeclarativeStream
+    schema_loader:
+      type: JsonFileSchemaLoader
+    retriever:
+      $ref: "#/definitions/retriever"
+
+  cursor_incremental_sync:
+    type: DatetimeBasedCursor
+    cursor_datetime_formats:
+      - "%s"
+      - "%Y-%m-%dT%H:%M:%SZ"
+      - "%Y-%m-%dT%H:%M:%S%z"
+    datetime_format: "%s"
+    cursor_field: "{{ parameters.get('cursor_field',  'updated_at') }}"
+    start_datetime:
+      datetime: "{{ timestamp(config.get('start_date')) | int if config.get('start_date') else day_delta(-730, '%s') }}"
+    start_time_option:
+      inject_into: request_parameter
+      field_name: "{{ parameters['cursor_filter'] }}"
+      type: RequestOption
+
+  base_incremental_stream:
+    $ref: "#/definitions/base_stream"
+    incremental_sync:
+      $ref: "#/definitions/cursor_incremental_sync"
+
+  # Incremental cursor-based streams
+  articles_stream:
+    $ref: "#/definitions/base_incremental_stream"
+    name: "articles"
+    primary_key: "id"
+    schema_loader:
+      type: InlineSchemaLoader
+      schema:
+        type: object
+        $schema: http://json-schema.org/schema#
+        properties:
+          id:
+            type: integer
+        additionalProperties: true
+    incremental_sync:
+      $ref: "#/definitions/cursor_incremental_sync"
+      start_time_option:
+        $ref: "#/definitions/cursor_incremental_sync/start_time_option"
+        field_name: "start_time"
+    retriever:
+      $ref: "#/definitions/retriever"
+      ignore_stream_slicer_parameters_on_paginated_requests: true
+      requester:
+        $ref: "#/definitions/retriever/requester"
+        path: "help_center/incremental/articles"
+      paginator:
+        type: DefaultPaginator
+        pagination_strategy:
+          type: CursorPagination
+          cursor_value: '{{ response.get("next_page", {}) }}'
+          stop_condition: "{{ config.get('ignore_pagination', False) or last_page_size == 0 }}"
+        page_token_option:
+          type: RequestPath
+      record_selector:
+        extractor:
+          type: DpathExtractor
+          field_path: ["articles"]
+
+  article_attachments_stream:
+    $ref: "#/definitions/base_incremental_stream"
+    name: "article_attachments"
+    primary_key: "id"
+    schema_loader:
+      type: InlineSchemaLoader
+      schema:
+        type: object
+        $schema: http://json-schema.org/schema#
+        properties:
+          id:
+            type: integer
+        additionalProperties: true
+    retriever:
+      $ref: "#/definitions/retriever"
+      ignore_stream_slicer_parameters_on_paginated_requests: true
+      requester:
+        $ref: "#/definitions/retriever/requester"
+        path: "help_center/articles/{{ stream_partition.article_id }}/attachments"
+      partition_router:
+        type: SubstreamPartitionRouter
+        parent_stream_configs:
+          - type: ParentStreamConfig
+            parent_key: "id"
+            partition_field: "article_id"
+            stream:
+              $ref: "#/definitions/articles_stream"
+            incremental_dependency: true
+      record_selector:
+        extractor:
+          type: DpathExtractor
+          field_path: ["article_attachments"]
+    file_uploader:
+      type: FileUploader
+      requester:
+        type: HttpRequester
+        url_base: "{{download_target}}"
+        http_method: GET
+        authenticator:
+          type: SelectiveAuthenticator
+          authenticator_selection_path: [ "credentials", "credentials" ]
+          authenticators:
+            oauth2.0: "#/definitions/bearer_authenticator"
+            api_token: "#/definitions/basic_authenticator"
+      download_target_extractor:
+        type: DpathExtractor
+        field_path: [ "content_url" ]
+      filename_extractor: "{{ record.id }}/{{ record.file_name }}/"
+
+
+streams:
+  - $ref: "#/definitions/articles_stream"
+  - $ref: "#/definitions/article_attachments_stream"
+
+spec:
+  type: Spec
+  connection_specification:
+    type: object
+    $schema: http://json-schema.org/draft-07/schema#
+    required:
+      - subdomain
+      - start_date
+    properties:
+      subdomain:
+        type: string
+        name: subdomain
+        order: 0
+        title: Subdomain
+      start_date:
+        type: string
+        order: 1
+        title: Start date
+        format: date-time
+        pattern: ^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$
+    additionalProperties: true