Hi, just started using the cookie-cutter sdk Has ...
# singer-tap-development
s
Hi, just started using the cookie-cutter sdk Has anyone ever had this error with next_token? My api call keeps on getting poluted by more and more useless characters until I get a 414:
Current stream:
Copy code
class OwnersStream(HubspotStream):
    """Define custom stream."""
    _LOG_REQUEST_METRIC_URLS=True
    name = "owners"
    path = "/crm/v3/owners"
    primary_keys = ["id"]
    replication_key = "updatedAt"
    records_jsonpath = "$.results[*]"
    next_page_token_jsonpath = "$.paging.next.link"
v
look at
HubspotStream
s
As in my mistake is obvious or I should be looking in hubspotstream?
Here is the source code:
Copy code
class HubspotStream(RESTStream):
    """Hubspot stream class."""

    # TODO: Set the API's base URL here:
    url_base = "<https://api.hubapi.com>"

    # OR use a dynamic url_base:
    # @property
    # def url_base(self) -> str:
    #     """Return the API URL root, configurable via tap settings."""
    #     return self.config["api_url"]

    records_jsonpath = "$[*]"  # Or override `parse_response`.
    next_page_token_jsonpath = "$.next_page"  # Or override `get_next_page_token`.


    @property
    def schema_filepath(self) -> Path:
        return SCHEMAS_DIR / f"{self.name}.json"

    @property
    def authenticator(self) -> BearerTokenAuthenticator:
        """Return a new authenticator object."""
        return BearerTokenAuthenticator.create_for_stream(
            self,
            token=self.config.get("access_token"),
        )

    @property
    def http_headers(self) -> dict:
        """Return the http headers needed."""
        headers = {}
        if "user_agent" in self.config:
            headers["User-Agent"] = self.config.get("user_agent")
        # If not using an authenticator, you may also provide inline auth headers:
        # headers["Private-Token"] = self.config.get("auth_token")
        return headers

    def get_next_page_token(
        self, response: requests.Response, previous_token: Optional[Any]
    ) -> Optional[Any]:
        """Return a token for identifying next page or None if no more pages."""
        # TODO: If pagination is required, return a token which can be used to get the
        #       next page. If this is the final page, return "None" to end the
        #       pagination loop.
        if self.next_page_token_jsonpath:
            all_matches = extract_jsonpath(
                self.next_page_token_jsonpath, response.json()
            )
            first_match = next(iter(all_matches), None)
            next_page_token = first_match
        else:
            next_page_token = response.headers.get("X-Next-Page", None)

        return next_page_token

    def get_url_params(
        self, context: Optional[dict], next_page_token: Optional[Any]
    ) -> Dict[str, Any]:
        """Return a dictionary of values to be used in URL parameterization."""
        params: dict = {}
        if next_page_token:
            params["page"] = next_page_token
        if self.replication_key:
            params["sort"] = "asc"
            params["order_by"] = self.replication_key
        return params

    def prepare_request_payload(
        self, context: Optional[dict], next_page_token: Optional[Any]
    ) -> Optional[dict]:
        """Prepare the data payload for the REST API request.

        By default, no payload will be sent (return None).
        """
        # TODO: Delete this method if no payload is required. (Most REST APIs.)
        return None

    def parse_response(self, response: requests.Response) -> Iterable[dict]:
        """Parse the response and return an iterator of result rows."""
        # TODO: Parse response body and return a set of records.
        yield from extract_jsonpath(self.records_jsonpath, input=response.json())

    def post_process(self, row: dict, context: Optional[dict]) -> dict:
        """As needed, append or transform raw data to match expected structure."""
        # TODO: Delete this method if not needed.
        return row
v
never mind that wasn't needed I get it now
next_page_token_jsonpath = "$.paging.next.link"
Have you looked at the response? It looks like $.paging.next.link is a full URL for the next page. You add that as a parameter of
page
(which hubspot's api probably ignores) , then hubspot's api gives you the same response back, and you add
page
again to the full url (including the old page)
Copy code
def get_url_params(
        self, context: Optional[dict], next_page_token: Optional[Any]
    ) -> Dict[str, Any]:
        """Return a dictionary of values to be used in URL parameterization."""
        params: dict = {}
        if next_page_token:
            params["page"] = next_page_token
Is where this gets set. Hope that helps!
s
The response looks good, that's what troubles me
I'll try it out thanks!
v
There's nothing to "try" but you'll get it eventually 🙂 keep plugging away, look closely at the response data and the sections I sent if you're stuck let me know what you've tried and maybe I can push you the right way :
😄
A good excercise would be to manually do a few pages in Postman or some HTTP tool so you get a feel for what the requests should be
s
Do you have any suggestions on how to debug (besides log, try, repeat? )
v
I tend to do that as it's faster in a lot of cases, you can use a debugger as well 🤷 preference I guess!
I think starting with Postman is going to help the most here as the issue is that
params["page"] = next_page_token
is wrong
That's my best guess, I've been wrong before hard to tell without being in it!
s
Got it, you're awesome, thanks!