Move log_formatter as argument

Author: lazebnyi

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -528,6 +528,7 @@
 from airbyte_cdk.sources.declarative.transformations.keys_to_snake_transformation import (
     KeysToSnakeCaseTransformation,
 )
+from airbyte_cdk.sources.http_logger import format_http_message
 from airbyte_cdk.sources.message import (
     InMemoryMessageRepository,
     LogAppenderMessageRepositoryDecorator,
@@ -2382,6 +2383,7 @@ def create_dynamic_schema_loader(
             primary_key=None,
             stream_slicer=combined_slicers,
             transformations=[],
+            enable_logs=False,
         )
         schema_type_identifier = self._create_component_from_model(
             model.schema_type_identifier, config=config, parameters=model.parameters or {}
@@ -2968,6 +2970,7 @@ def create_simple_retriever(
             ]
         ] = None,
         use_cache: Optional[bool] = None,
+        enable_logs: bool = True,
         **kwargs: Any,
     ) -> SimpleRetriever:
         def _get_url() -> str:
@@ -3132,6 +3135,13 @@ def _get_url() -> str:
             )
 
         if self._limit_slices_fetched or self._emit_connector_builder_messages:
+            log_formatter = lambda response: format_http_message(
+                response,
+                f"Stream '{self.name}' request",
+                f"Request performed in order to extract records for stream '{self.name}'",
+                self.name,
+            ) if enable_logs else None
+
             return SimpleRetrieverTestReadDecorator(
                 name=name,
                 paginator=paginator,
@@ -3144,6 +3154,12 @@ def _get_url() -> str:
                 config=config,
                 maximum_number_of_slices=self._limit_slices_fetched or 5,
                 ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
+                log_formatter=lambda response: format_http_message(
+                    response,
+                    f"Stream '{name}' request",
+                    f"Request performed in order to extract records for stream '{name}'",
+                    name,
+                ) if enable_logs else None,
                 parameters=model.parameters or {},
             )
         return SimpleRetriever(

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -10,7 +10,6 @@
 from typing import (
     Any,
     Callable,
-    Dict,
     Iterable,
     List,
     Mapping,
@@ -42,7 +41,6 @@
 from airbyte_cdk.sources.declarative.requesters.requester import Requester
 from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
-from airbyte_cdk.sources.http_logger import format_http_message
 from airbyte_cdk.sources.source import ExperimentalClassWarning
 from airbyte_cdk.sources.streams.core import StreamData
 from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState
@@ -93,6 +91,7 @@ class SimpleRetriever(Retriever):
     cursor: Optional[DeclarativeCursor] = None
     ignore_stream_slicer_parameters_on_paginated_requests: bool = False
     additional_query_properties: Optional[QueryProperties] = None
+    log_formatter: Optional[Callable[[requests.Response], Any]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._paginator = self.paginator or NoPagination(parameters=parameters)
@@ -353,6 +352,7 @@ def _fetch_next_page(
                 stream_slice=stream_slice,
                 next_page_token=next_page_token,
             ),
+            log_formatter=self.log_formatter,
         )
 
     # This logic is similar to _read_pages in the HttpStream class. When making changes here, consider making changes there as well.
@@ -664,51 +664,6 @@ def __post_init__(self, options: Mapping[str, Any]) -> None:
     def stream_slices(self) -> Iterable[Optional[StreamSlice]]:  # type: ignore
         return islice(super().stream_slices(), self.maximum_number_of_slices)
 
-    def _fetch_next_page(
-        self,
-        stream_state: Mapping[str, Any],
-        stream_slice: StreamSlice,
-        next_page_token: Optional[Mapping[str, Any]] = None,
-    ) -> Optional[requests.Response]:
-        is_dynamic_schema_call = self.name == "dynamic_properties"
-        return self.requester.send_request(
-            path=self._paginator_path(
-                next_page_token=next_page_token,
-                stream_state=stream_state,
-                stream_slice=stream_slice,
-            ),
-            stream_state=stream_state,
-            stream_slice=stream_slice,
-            next_page_token=next_page_token,
-            request_headers=self._request_headers(
-                stream_state=stream_state,
-                stream_slice=stream_slice,
-                next_page_token=next_page_token,
-            ),
-            request_params=self._request_params(
-                stream_state=stream_state,
-                stream_slice=stream_slice,
-                next_page_token=next_page_token,
-            ),
-            request_body_data=self._request_body_data(
-                stream_state=stream_state,
-                stream_slice=stream_slice,
-                next_page_token=next_page_token,
-            ),
-            request_body_json=self._request_body_json(
-                stream_state=stream_state,
-                stream_slice=stream_slice,
-                next_page_token=next_page_token,
-            ),
-            log_formatter=lambda response: format_http_message(
-                response,
-                f"Stream '{self.name}' request",
-                f"Request performed in order to extract records for stream '{self.name}'",
-                self.name,
-                is_auxiliary=is_dynamic_schema_call,  # Mark schema discovery calls as auxiliary for cleaner logs
-            ),
-        )
-
 
 @deprecated(
     "This class is experimental. Use at your own risk.",