Request params encryption

  • Is this your first time deploying Airbyte?: Yes
  • OS Version / Instance: MacOS
    Hey there,
    I am developing an http API source using the airbyte cdk.
    When I am using the request params method I am returning the following response:
                return {'startIndex': 0, 'resultsPerPage': self.max_results_per_page,
                        'pubStartDate': "2022-07-30T13:57:21:000 UTC%2B03:00", 'pubEndDate': "2022-07-31T13:57:21:000 UTC%2B03:00"}

But for some reason, those fields are being encrypted and the params that are being passed to the API are:

pubStartDate: "2022-07-30T13%3A57%3A21%3A000+UTC%252B03%3A00"
pubEndDate: "2022-07-31T13%3A57%3A21%3A000+UTC%252B03%3A00"

Is there anything I am doing wrong?
How can I send those date time strings correctly - without them being encrypted?

The API I am using: API Vulnerabilities

Thanks in advance, any help will be much appreciated!!

The full connector code:

class NvdVulnerabilitiesStream(HttpStream, ABC):
    url_base = "https://services.nvd.nist.gov/"
    max_results_per_page = 2000

    def __init__(self, days_to_fetch: int):
        auth = NoAuth()
        super().__init__(authenticator=auth)
        self.days_to_fetch = days_to_fetch

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:

        if response.json()['resultsPerPage'] < self.max_results_per_page:
            return None

        return {"startIndex": response.json()['startIndex'] + self.max_results_per_page, "resultsPerPage": self.max_results_per_page}

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:

        if not next_page_token and self.days_to_fetch != -1:
            end_time = datetime.now()
            start_time = end_time - timedelta(days=self.days_to_fetch)
            if not next_page_token:
                return {'startIndex': 0, 'resultsPerPage': self.max_results_per_page,
                        'pubStartDate': start_time.strftime(DATETIME_SCHEME), 'pubEndDate': end_time.strftime(DATETIME_SCHEME)}

        if not next_page_token:
            return {"startIndex": 0, "resultsPerPage": self.max_results_per_page}

        return {"startIndex": next_page_token['startIndex'], "resultsPerPage": next_page_token['resultsPerPage']}

    def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
        for vulnerability in response.json()["result"]["CVE_Items"]:
            yield vulnerability
        print(response.json()["startIndex"])
        time.sleep(6)


class Vulnerabilities(NvdVulnerabilitiesStream):
    primary_key = None

    def __init__(self, days_to_fetch: int):
        super().__init__(days_to_fetch)

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        return "rest/json/cves/1.0/"```

Hello and welcome to the community, @Niv! It’s great to hear that you’re developing a source. Could you please give me a link to the API documentation to that I can help you debug this?