Merge branch 'master' into leti/experiment-add-source-in-signup

Author: letiescanciano

airbyte-api/src/main/java/io/airbyte/api/client/AirbyteApiClient.java

@@ -16,6 +16,7 @@
 import io.airbyte.api.client.generated.SourceApi;
 import io.airbyte.api.client.generated.SourceDefinitionApi;
 import io.airbyte.api.client.generated.SourceDefinitionSpecificationApi;
+import io.airbyte.api.client.generated.StateApi;
 import io.airbyte.api.client.generated.WorkspaceApi;
 import io.airbyte.api.client.invoker.generated.ApiClient;
 
@@ -46,8 +47,8 @@ public class AirbyteApiClient {
   private final WorkspaceApi workspaceApi;
   private final HealthApi healthApi;
   private final DbMigrationApi dbMigrationApi;
-
   private final AttemptApi attemptApi;
+  private final StateApi stateApi;
 
   public AirbyteApiClient(final ApiClient apiClient) {
     connectionApi = new ConnectionApi(apiClient);
@@ -64,6 +65,7 @@ public AirbyteApiClient(final ApiClient apiClient) {
     healthApi = new HealthApi(apiClient);
     dbMigrationApi = new DbMigrationApi(apiClient);
     attemptApi = new AttemptApi(apiClient);
+    stateApi = new StateApi(apiClient);
   }
 
   public ConnectionApi getConnectionApi() {
@@ -122,4 +124,8 @@ public AttemptApi getAttemptApi() {
     return attemptApi;
   }
 
+  public StateApi getStateApi() {
+    return stateApi;
+  }
+
 }

airbyte-api/src/main/openapi/config.yaml

@@ -63,6 +63,8 @@ tags:
     description: Export/Import Airbyte Configuration and Database resources.
   - name: attempt
     description: Interactions with attempt related resources.
+  - name: state
+    description: Interactions with state related resources.
 
 paths:
   /v1/workspaces/create:
@@ -1389,7 +1391,7 @@ paths:
   /v1/state/get:
     post:
       tags:
-        - connection
+        - state
       summary: Fetch the current state for a connection.
       operationId: getState
       requestBody:
@@ -1412,7 +1414,7 @@ paths:
   /v1/state/create_or_update:
     post:
       tags:
-        - connection
+        - state
         - internal
       summary: Create or update the state for a connection.
       operationId: createOrUpdateState
@@ -1994,7 +1996,7 @@ paths:
   /v1/web_backend/state/get_type:
     post:
       tags:
-        - connection
+        - web_backend
       summary: Fetch the current state type for a connection.
       operationId: getStateType
       requestBody:
@@ -2169,6 +2171,26 @@ paths:
           $ref: "#/components/responses/NotFoundResponse"
         "422":
           $ref: "#/components/responses/InvalidInputResponse"
+  /v1/jobs/get_normalization_status:
+    post:
+      tags:
+        - jobs
+        - internal
+      summary: Get normalization status to determine if we can bypass normalization phase
+      operationId: getAttemptNormalizationStatusesForJob
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/JobIdRequestBody"
+      responses:
+        "200":
+          description: Successful operation
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AttemptNormalizationStatusReadList"
+
   /v1/health:
     get:
       tags:
@@ -2242,6 +2264,7 @@ paths:
             application/json:
               schema:
                 $ref: "#/components/schemas/InternalOperationResult"
+
 components:
   securitySchemes:
     bearerAuth:
@@ -4838,6 +4861,26 @@ components:
       properties:
         succeeded:
           type: boolean
+    AttemptNormalizationStatusReadList:
+      type: object
+      properties:
+        attemptNormalizationStatuses:
+          type: array
+          items:
+            $ref: "#/components/schemas/AttemptNormalizationStatusRead"
+    AttemptNormalizationStatusRead:
+      type: object
+      properties:
+        attemptNumber:
+          $ref: "#/components/schemas/AttemptNumber"
+        hasRecordsCommitted:
+          type: boolean
+        recordsCommitted:
+          type: integer
+          format: int64
+        hasNormalizationFailed:
+          type: boolean
+
     InvalidInputProperty:
       type: object
       required:

airbyte-bootloader/build.gradle

@@ -11,7 +11,7 @@ dependencies {
     implementation project(':airbyte-protocol:protocol-models')
     implementation project(':airbyte-persistence:job-persistence')
 
-    implementation 'io.temporal:temporal-sdk:1.8.1'
+    implementation libs.temporal.sdk
     implementation libs.flyway.core
 
     testImplementation libs.platform.testcontainers.postgresql

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.4.0
+Low-code: Add support for custom error messages on error response filters
+
+## 0.3.0
+Publish python typehints via `py.typed` file. 
+
 ## 0.2.3
 - Propagate options to InterpolatedRequestInputProvider
 

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py

@@ -49,10 +49,10 @@ def __post_init__(self, options: Mapping[str, Any]):
     def max_retries(self) -> Union[int, None]:
         return self.error_handlers[0].max_retries
 
-    def should_retry(self, response: requests.Response) -> ResponseStatus:
+    def interpret_response(self, response: requests.Response) -> ResponseStatus:
         should_retry = None
         for retrier in self.error_handlers:
-            should_retry = retrier.should_retry(response)
+            should_retry = retrier.interpret_response(response)
             if should_retry.action == ResponseAction.SUCCESS:
                 return response_status.SUCCESS
             if should_retry == response_status.IGNORE or should_retry.action == ResponseAction.RETRY:

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/error_handlers/default_error_handler.py

@@ -15,6 +15,7 @@
 from airbyte_cdk.sources.declarative.requesters.error_handlers.http_response_filter import HttpResponseFilter
 from airbyte_cdk.sources.declarative.requesters.error_handlers.response_action import ResponseAction
 from airbyte_cdk.sources.declarative.requesters.error_handlers.response_status import ResponseStatus
+from airbyte_cdk.sources.declarative.types import Config
 from dataclasses_jsonschema import JsonSchemaMixin
 
 
@@ -91,6 +92,7 @@ class DefaultErrorHandler(ErrorHandler, JsonSchemaMixin):
 
     DEFAULT_BACKOFF_STRATEGY = ExponentialBackoffStrategy
 
+    config: Config
     options: InitVar[Mapping[str, Any]]
     response_filters: Optional[List[HttpResponseFilter]] = None
     max_retries: Optional[int] = 5
@@ -102,9 +104,11 @@ def __post_init__(self, options: Mapping[str, Any]):
 
         if not self.response_filters:
             self.response_filters.append(
-                HttpResponseFilter(ResponseAction.RETRY, http_codes=HttpResponseFilter.DEFAULT_RETRIABLE_ERRORS, options={})
+                HttpResponseFilter(
+                    ResponseAction.RETRY, http_codes=HttpResponseFilter.DEFAULT_RETRIABLE_ERRORS, config=self.config, options={}
+                )
             )
-            self.response_filters.append(HttpResponseFilter(ResponseAction.IGNORE, options={}))
+            self.response_filters.append(HttpResponseFilter(ResponseAction.IGNORE, config={}, options={}))
 
         if not self.backoff_strategies:
             self.backoff_strategies = [DefaultErrorHandler.DEFAULT_BACKOFF_STRATEGY()]
@@ -122,20 +126,20 @@ def max_retries(self, value: Union[int, None]):
         if not isinstance(value, property):
             self._max_retries = value
 
-    def should_retry(self, response: requests.Response) -> ResponseStatus:
+    def interpret_response(self, response: requests.Response) -> ResponseStatus:
         request = response.request
 
         if request not in self._last_request_to_attempt_count:
             self._last_request_to_attempt_count = {request: 1}
         else:
             self._last_request_to_attempt_count[request] += 1
         for response_filter in self.response_filters:
-            filter_action = response_filter.matches(response)
-            if filter_action is not None:
-                if filter_action == ResponseAction.RETRY:
-                    return ResponseStatus(ResponseAction.RETRY, self._backoff_time(response, self._last_request_to_attempt_count[request]))
-                else:
-                    return ResponseStatus(filter_action)
+            matched_status = response_filter.matches(
+                response=response, backoff_time=self._backoff_time(response, self._last_request_to_attempt_count[request])
+            )
+            if matched_status is not None:
+                return matched_status
+
         if response.ok:
             return response_status.SUCCESS
         # Fail if the response matches no filters

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/error_handlers/error_handler.py

@@ -26,7 +26,7 @@ def max_retries(self) -> Union[int, None]:
         pass
 
     @abstractmethod
-    def should_retry(self, response: requests.Response) -> ResponseStatus:
+    def interpret_response(self, response: requests.Response) -> ResponseStatus:
         """
         Evaluate response status describing whether a failing request should be retried or ignored.
 

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py

@@ -6,8 +6,11 @@
 from typing import Any, Mapping, Optional, Set, Union
 
 import requests
+from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
 from airbyte_cdk.sources.declarative.requesters.error_handlers.response_action import ResponseAction
+from airbyte_cdk.sources.declarative.requesters.error_handlers.response_status import ResponseStatus
+from airbyte_cdk.sources.declarative.types import Config
 from airbyte_cdk.sources.streams.http.http import HttpStream
 from dataclasses_jsonschema import JsonSchemaMixin
 
@@ -22,25 +25,43 @@ class HttpResponseFilter(JsonSchemaMixin):
         http_codes (Set[int]): http code of matching requests
         error_message_contains (str): error substring of matching requests
         predicate (str): predicate to apply to determine if a request is matching
+        error_message (Union[InterpolatedString, str): error message to display if the response matches the filter
     """
 
     TOO_MANY_REQUESTS_ERRORS = {429}
     DEFAULT_RETRIABLE_ERRORS = set([x for x in range(500, 600)]).union(TOO_MANY_REQUESTS_ERRORS)
 
     action: Union[ResponseAction, str]
+    config: Config
     options: InitVar[Mapping[str, Any]]
     http_codes: Set[int] = None
     error_message_contains: str = None
     predicate: Union[InterpolatedBoolean, str] = ""
+    error_message: Union[InterpolatedString, str] = ""
 
     def __post_init__(self, options: Mapping[str, Any]):
         if isinstance(self.action, str):
             self.action = ResponseAction[self.action]
         self.http_codes = self.http_codes or set()
         if isinstance(self.predicate, str):
             self.predicate = InterpolatedBoolean(condition=self.predicate, options=options)
+        self.error_message = InterpolatedString.create(string_or_interpolated=self.error_message, options=options)
 
-    def matches(self, response: requests.Response) -> Optional[ResponseAction]:
+    def matches(self, response: requests.Response, backoff_time: Optional[float] = None) -> Optional[ResponseStatus]:
+        filter_action = self._matches_filter(response)
+        if filter_action is not None:
+            error_message = self._create_error_message(response)
+            if filter_action == ResponseAction.RETRY:
+                return ResponseStatus(
+                    response_action=ResponseAction.RETRY,
+                    retry_in=backoff_time,
+                    error_message=error_message,
+                )
+            else:
+                return ResponseStatus(filter_action, error_message=error_message)
+        return None
+
+    def _matches_filter(self, response: requests.Response) -> Optional[ResponseAction]:
         """
         Apply the filter on the response and return the action to execute if it matches
         :param response: The HTTP response to evaluate
@@ -55,6 +76,14 @@ def matches(self, response: requests.Response) -> Optional[ResponseAction]:
         else:
             return None
 
+    def _create_error_message(self, response: requests.Response) -> str:
+        """
+        Construct an error message based on the specified message template of the filter.
+        :param response: The HTTP response which can be used during interpolation
+        :return: The evaluated error message string to be emitted
+        """
+        return self.error_message.eval(self.config, response=response.json(), headers=response.headers)
+
     def _response_matches_predicate(self, response: requests.Response) -> bool:
         return self.predicate and self.predicate.eval(None, response=response.json(), headers=response.headers)
 

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/error_handlers/response_status.py

@@ -9,20 +9,22 @@
 
 class ResponseStatus:
     """
-    ResponseAction amended with backoff time if a action is RETRY
+    ResponseAction amended with backoff time if an action is RETRY
     """
 
-    def __init__(self, response_action: Union[ResponseAction, str], retry_in: Optional[float] = None):
+    def __init__(self, response_action: Union[ResponseAction, str], retry_in: Optional[float] = None, error_message: str = ""):
         """
         :param response_action: response action to execute
         :param retry_in: backoff time (if action is RETRY)
+        :param error_message: the error to be displayed back to the customer
         """
         if isinstance(response_action, str):
             response_action = ResponseAction[response_action]
         if retry_in and response_action != ResponseAction.RETRY:
             raise ValueError(f"Unexpected backoff time ({retry_in} for non-retryable response action {response_action}")
         self._retry_in = retry_in
         self._action = response_action
+        self._error_message = error_message
 
     @property
     def action(self):
@@ -34,6 +36,11 @@ def retry_in(self) -> Optional[float]:
         """How long to backoff before retrying a response. None if no wait required."""
         return self._retry_in
 
+    @property
+    def error_message(self) -> str:
+        """The message to be displayed when an error response is received"""
+        return self._error_message
+
     @classmethod
     def retry(cls, retry_in: Optional[float]) -> "ResponseStatus":
         """

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/http_requester.py

@@ -60,7 +60,7 @@ def __post_init__(self, options: Mapping[str, Any]):
         if type(self.http_method) == str:
             self.http_method = HttpMethod[self.http_method]
         self._method = self.http_method
-        self.error_handler = self.error_handler or DefaultErrorHandler(options=options)
+        self.error_handler = self.error_handler or DefaultErrorHandler(options=options, config=self.config)
         self._options = options
 
     # We are using an LRU cache in should_retry() method which requires all incoming arguments (including self) to be hashable.
@@ -88,9 +88,9 @@ def get_method(self):
     # use a tiny cache to limit the memory footprint. It doesn't have to be large because we mostly
     # only care about the status of the last response received
     @lru_cache(maxsize=10)
-    def should_retry(self, response: requests.Response) -> ResponseStatus:
+    def interpret_response_status(self, response: requests.Response) -> ResponseStatus:
         # Cache the result because the HttpStream first checks if we should retry before looking at the backoff time
-        return self.error_handler.should_retry(response)
+        return self.error_handler.interpret_response(response)
 
     def get_request_params(
         self,

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/requester.py

@@ -70,9 +70,9 @@ def get_request_params(
         """
 
     @abstractmethod
-    def should_retry(self, response: requests.Response) -> ResponseStatus:
+    def interpret_response_status(self, response: requests.Response) -> ResponseStatus:
         """
-        Specifies conditions for backoff based on the response from the server.
+        Specifies conditions for backoff, error handling and reporting based on the response from the server.
 
         By default, back off on the following HTTP response statuses:
          - 429 (Too Many Requests) indicating rate limiting