Merge branch 'main' into christo/manifest-server-errors

Author: ChristoGrab

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2208,6 +2208,14 @@ def _build_concurrent_cursor(
             and stream_slicer
             and not isinstance(stream_slicer, SinglePartitionRouter)
         ):
+            if isinstance(model.incremental_sync, IncrementingCountCursorModel):
+                # We don't currently support usage of partition routing and IncrementingCountCursor at the
+                # same time because we didn't solve for design questions like what the lookback window would
+                # be as well as global cursor fall backs. We have not seen customers that have needed both
+                # at the same time yet and are currently punting on this until we need to solve it.
+                raise ValueError(
+                    f"The low-code framework does not currently support usage of a PartitionRouter and an IncrementingCountCursor at the same time. Please specify only one of these options for stream {stream_name}."
+                )
             return self.create_concurrent_cursor_from_perpartition_cursor(  # type: ignore # This is a known issue that we are creating and returning a ConcurrentCursor which does not technically implement the (low-code) StreamSlicer. However, (low-code) StreamSlicer and ConcurrentCursor both implement StreamSlicer.stream_slices() which is the primary method needed for checkpointing
                 state_manager=self._connector_state_manager,
                 model_type=DatetimeBasedCursorModel,
@@ -2395,21 +2403,12 @@ def create_http_requester(
 
         api_budget = self._api_budget
 
-        # Removes QueryProperties components from the interpolated mappings because it has been designed
-        # to be used by the SimpleRetriever and will be resolved from the provider from the slice directly
-        # instead of through jinja interpolation
-        request_parameters: Optional[Union[str, Mapping[str, str]]]
-        if isinstance(model.request_parameters, Mapping):
-            request_parameters = self._remove_query_properties(model.request_parameters)
-        else:
-            request_parameters = model.request_parameters
-
         request_options_provider = InterpolatedRequestOptionsProvider(
             request_body=model.request_body,
             request_body_data=model.request_body_data,
             request_body_json=model.request_body_json,
             request_headers=model.request_headers,
-            request_parameters=request_parameters,
+            request_parameters=model.request_parameters,  # type: ignore  # QueryProperties have been removed in `create_simple_retriever`
             query_properties_key=query_properties_key,
             config=config,
             parameters=model.parameters or {},
@@ -3199,7 +3198,8 @@ def _get_url(req: Requester) -> str:
 
         query_properties: Optional[QueryProperties] = None
         query_properties_key: Optional[str] = None
-        if self._query_properties_in_request_parameters(model.requester):
+        self._ensure_query_properties_to_model(model.requester)
+        if self._has_query_properties_in_request_parameters(model.requester):
             # It is better to be explicit about an error if PropertiesFromEndpoint is defined in multiple
             # places instead of default to request_parameters which isn't clearly documented
             if (
@@ -3211,7 +3211,7 @@ def _get_url(req: Requester) -> str:
                 )
 
             query_properties_definitions = []
-            for key, request_parameter in model.requester.request_parameters.items():  # type: ignore # request_parameters is already validated to be a Mapping using _query_properties_in_request_parameters()
+            for key, request_parameter in model.requester.request_parameters.items():  # type: ignore # request_parameters is already validated to be a Mapping using _has_query_properties_in_request_parameters()
                 if isinstance(request_parameter, QueryPropertiesModel):
                     query_properties_key = key
                     query_properties_definitions.append(request_parameter)
@@ -3225,6 +3225,16 @@ def _get_url(req: Requester) -> str:
                 query_properties = self._create_component_from_model(
                     model=query_properties_definitions[0], config=config
                 )
+
+            # Removes QueryProperties components from the interpolated mappings because it has been designed
+            # to be used by the SimpleRetriever and will be resolved from the provider from the slice directly
+            # instead of through jinja interpolation
+            if hasattr(model.requester, "request_parameters") and isinstance(
+                model.requester.request_parameters, Mapping
+            ):
+                model.requester.request_parameters = self._remove_query_properties(
+                    model.requester.request_parameters
+                )
         elif (
             hasattr(model.requester, "fetch_properties_from_endpoint")
             and model.requester.fetch_properties_from_endpoint
@@ -3361,7 +3371,7 @@ def _should_limit_slices_fetched(self) -> bool:
         return bool(self._limit_slices_fetched or self._emit_connector_builder_messages)
 
     @staticmethod
-    def _query_properties_in_request_parameters(
+    def _has_query_properties_in_request_parameters(
         requester: Union[HttpRequesterModel, CustomRequesterModel],
     ) -> bool:
         if not hasattr(requester, "request_parameters"):
@@ -4175,3 +4185,26 @@ def create_grouping_partition_router(
             deduplicate=model.deduplicate if model.deduplicate is not None else True,
             config=config,
         )
+
+    def _ensure_query_properties_to_model(
+        self, requester: Union[HttpRequesterModel, CustomRequesterModel]
+    ) -> None:
+        """
+        For some reason, it seems like CustomRequesterModel request_parameters stays as dictionaries which means that
+        the other conditions relying on it being QueryPropertiesModel instead of a dict fail. Here, we migrate them to
+        proper model.
+        """
+        if not hasattr(requester, "request_parameters"):
+            return
+
+        request_parameters = requester.request_parameters
+        if request_parameters and isinstance(request_parameters, Dict):
+            for request_parameter_key in request_parameters.keys():
+                request_parameter = request_parameters[request_parameter_key]
+                if (
+                    isinstance(request_parameter, Dict)
+                    and request_parameter.get("type") == "QueryProperties"
+                ):
+                    request_parameters[request_parameter_key] = QueryPropertiesModel.parse_obj(
+                        request_parameter
+                    )

airbyte_cdk/sources/file_based/file_based_stream_reader.py

@@ -3,13 +3,15 @@
 #
 
 import logging
+import time
 from abc import ABC, abstractmethod
 from datetime import datetime
 from enum import Enum
 from io import IOBase
 from os import makedirs, path
-from typing import Any, Callable, Iterable, List, MutableMapping, Optional, Set, Tuple
+from typing import Any, Iterable, List, MutableMapping, Optional, Set, Tuple
 
+from airbyte_protocol_dataclasses.models import FailureType
 from wcmatch.glob import GLOBSTAR, globmatch
 
 from airbyte_cdk.models import AirbyteRecordMessageFileReference
@@ -19,8 +21,9 @@
     preserve_directory_structure,
     use_file_transfer,
 )
+from airbyte_cdk.sources.file_based.exceptions import FileSizeLimitError
 from airbyte_cdk.sources.file_based.file_record_data import FileRecordData
-from airbyte_cdk.sources.file_based.remote_file import RemoteFile
+from airbyte_cdk.sources.file_based.remote_file import RemoteFile, UploadableRemoteFile
 
 
 class FileReadMode(Enum):
@@ -34,6 +37,7 @@ class AbstractFileBasedStreamReader(ABC):
     FILE_NAME = "file_name"
     LOCAL_FILE_PATH = "local_file_path"
     FILE_FOLDER = "file_folder"
+    FILE_SIZE_LIMIT = 1_500_000_000
 
     def __init__(self) -> None:
         self._config = None
@@ -113,16 +117,6 @@ def filter_files_by_globs_and_start_date(
                     seen.add(file.uri)
                     yield file
 
-    @abstractmethod
-    def file_size(self, file: RemoteFile) -> int:
-        """Utility method to get size of the remote file.
-
-        This is required for connectors that will support writing to
-        files. If the connector does not support writing files, then the
-        subclass can simply `return 0`.
-        """
-        ...
-
     @staticmethod
     def file_matches_globs(file: RemoteFile, globs: List[str]) -> bool:
         # Use the GLOBSTAR flag to enable recursive ** matching
@@ -153,9 +147,8 @@ def include_identities_stream(self) -> bool:
             return include_identities_stream(self.config)
         return False
 
-    @abstractmethod
     def upload(
-        self, file: RemoteFile, local_directory: str, logger: logging.Logger
+        self, file: UploadableRemoteFile, local_directory: str, logger: logging.Logger
     ) -> Tuple[FileRecordData, AirbyteRecordMessageFileReference]:
         """
         This is required for connectors that will support writing to
@@ -173,7 +166,53 @@ def upload(
                    - file_size_bytes (int): The size of the referenced file in bytes.
                    - source_file_relative_path (str): The relative path to the referenced file in source.
         """
-        ...
+        if not isinstance(file, UploadableRemoteFile):
+            raise TypeError(f"Expected UploadableRemoteFile, got {type(file)}")
+
+        file_size = file.size
+
+        if file_size > self.FILE_SIZE_LIMIT:
+            message = f"File size exceeds the {self.FILE_SIZE_LIMIT / 1e9} GB limit."
+            raise FileSizeLimitError(
+                message=message, internal_message=message, failure_type=FailureType.config_error
+            )
+
+        file_paths = self._get_file_transfer_paths(
+            source_file_relative_path=file.source_file_relative_path,
+            staging_directory=local_directory,
+        )
+        local_file_path = file_paths[self.LOCAL_FILE_PATH]
+        file_relative_path = file_paths[self.FILE_RELATIVE_PATH]
+        file_name = file_paths[self.FILE_NAME]
+
+        logger.info(
+            f"Starting to download the file {file.file_uri_for_logging} with size: {file_size / (1024 * 1024):,.2f} MB ({file_size / (1024 * 1024 * 1024):.2f} GB)"
+        )
+        start_download_time = time.time()
+
+        file.download_to_local_directory(local_file_path)
+
+        write_duration = time.time() - start_download_time
+        logger.info(
+            f"Finished downloading the file {file.file_uri_for_logging} and saved to {local_file_path} in {write_duration:,.2f} seconds."
+        )
+
+        file_record_data = FileRecordData(
+            folder=file_paths[self.FILE_FOLDER],
+            file_name=file_name,
+            bytes=file_size,
+            id=file.id,
+            mime_type=file.mime_type,
+            created_at=file.created_at,
+            updated_at=file.updated_at,
+            source_uri=file.uri,
+        )
+        file_reference = AirbyteRecordMessageFileReference(
+            staging_file_url=local_file_path,
+            source_file_relative_path=file_relative_path,
+            file_size_bytes=file_size,
+        )
+        return file_record_data, file_reference
 
     def _get_file_transfer_paths(
         self, source_file_relative_path: str, staging_directory: str

airbyte_cdk/sources/file_based/file_types/file_transfer.py

@@ -7,7 +7,7 @@
 from airbyte_cdk.models import AirbyteRecordMessageFileReference
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader
 from airbyte_cdk.sources.file_based.file_record_data import FileRecordData
-from airbyte_cdk.sources.file_based.remote_file import RemoteFile
+from airbyte_cdk.sources.file_based.remote_file import UploadableRemoteFile
 from airbyte_cdk.sources.utils.files_directory import get_files_directory
 
 
@@ -17,7 +17,7 @@ def __init__(self) -> None:
 
     def upload(
         self,
-        file: RemoteFile,
+        file: UploadableRemoteFile,
         stream_reader: AbstractFileBasedStreamReader,
         logger: logging.Logger,
     ) -> Iterable[Tuple[FileRecordData, AirbyteRecordMessageFileReference]]:

airbyte_cdk/sources/file_based/remote_file.py

@@ -1,7 +1,7 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-
+from abc import ABC, abstractmethod
 from datetime import datetime
 from typing import Optional
 
@@ -16,3 +16,42 @@ class RemoteFile(BaseModel):
     uri: str
     last_modified: datetime
     mime_type: Optional[str] = None
+
+
+class UploadableRemoteFile(RemoteFile, ABC):
+    """
+    A file in a file-based stream that supports uploading(file transferring).
+    """
+
+    id: Optional[str] = None
+    created_at: Optional[str] = None
+    updated_at: Optional[str] = None
+
+    @property
+    @abstractmethod
+    def size(self) -> int:
+        """
+        Returns the file size in bytes.
+        """
+        ...
+
+    @abstractmethod
+    def download_to_local_directory(self, local_file_path: str) -> None:
+        """
+        Download the file from remote source to local storage.
+        """
+        ...
+
+    @property
+    def source_file_relative_path(self) -> str:
+        """
+        Returns the relative path of the source file.
+        """
+        return self.uri
+
+    @property
+    def file_uri_for_logging(self) -> str:
+        """
+        Returns the URI for the file being logged.
+        """
+        return self.uri