airbyte_cdk.sources.declarative.requesters.request_options.request_options_provider

 1#
 2# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 3#
 4
 5from abc import abstractmethod
 6from dataclasses import dataclass
 7from typing import Any, Mapping, Optional, Union
 8
 9from airbyte_cdk.sources.types import StreamSlice, StreamState
10
11
12@dataclass
13class RequestOptionsProvider:
14    """
15    Defines the request options to set on an outgoing HTTP request
16
17    Options can be passed by
18    - request parameter
19    - request headers
20    - body data
21    - json content
22    """
23
24    @abstractmethod
25    def get_request_params(
26        self,
27        *,
28        stream_state: Optional[StreamState] = None,
29        stream_slice: Optional[StreamSlice] = None,
30        next_page_token: Optional[Mapping[str, Any]] = None,
31    ) -> Mapping[str, Any]:
32        """
33        Specifies the query parameters that should be set on an outgoing HTTP request given the inputs.
34
35        E.g: you might want to define query parameters for paging if next_page_token is not None.
36        """
37        pass
38
39    @abstractmethod
40    def get_request_headers(
41        self,
42        *,
43        stream_state: Optional[StreamState] = None,
44        stream_slice: Optional[StreamSlice] = None,
45        next_page_token: Optional[Mapping[str, Any]] = None,
46    ) -> Mapping[str, Any]:
47        """Return any non-auth headers. Authentication headers will overwrite any overlapping headers returned from this method."""
48
49    @abstractmethod
50    def get_request_body_data(
51        self,
52        *,
53        stream_state: Optional[StreamState] = None,
54        stream_slice: Optional[StreamSlice] = None,
55        next_page_token: Optional[Mapping[str, Any]] = None,
56    ) -> Union[Mapping[str, Any], str]:
57        """
58        Specifies how to populate the body of the request with a non-JSON payload.
59
60        If returns a ready text that it will be sent as is.
61        If returns a dict that it will be converted to a urlencoded form.
62        E.g. {"key1": "value1", "key2": "value2"} => "key1=value1&key2=value2"
63
64        At the same time only one of the 'request_body_data' and 'request_body_json' functions can be overridden.
65        """
66
67    @abstractmethod
68    def get_request_body_json(
69        self,
70        *,
71        stream_state: Optional[StreamState] = None,
72        stream_slice: Optional[StreamSlice] = None,
73        next_page_token: Optional[Mapping[str, Any]] = None,
74    ) -> Mapping[str, Any]:
75        """
76        Specifies how to populate the body of the request with a JSON payload.
77
78        At the same time only one of the 'request_body_data' and 'request_body_json' functions can be overridden.
79        """