GraphQL Pagination in Custom Connector

[dprose]

The standard Shopify connector for Airbyte is great, but it doesn't yet include Company data from the B2B tools in Shopify. So I'm trying to find a way to extract that company data via a custom connector. Company data is available in the Shopify GraphQL api here. I've been able to authenticate and pull a single page of data into Airbyte. The problem that I'm having is with pagination. Here is Shopify's documentation on pagination.

The basic query that I'm using looks like this:

{ companies(first: 100, after: $cursor) {
    pageInfo {
      hasNextPage
      endCursor
    }
    nodes {
      id      
      externalId
      name
    }
  }
}

For pagination to work, I need a variable $cursor that is null on the first api call and then on subsequent api calls is set to the endCursor value that comes from the api response. I've tried every configuration I can think of in the UI and the yaml file to try to get this to work and haven't had any success. I tried looking into using the {{ next_page_token }} macro, but I don't understand what data that is pulling or how I would set it to null on the first api call.

Am I missing something obvious? Has anyone else been able to solve a similar problem? I'm attaching my yaml config here for reference as well.

version: 0.51.16
type: DeclarativeSource
check:
  type: CheckStream
  stream_names:
    - Companies
streams:
  - type: DeclarativeStream
    name: Companies
    primary_key:
      - id
    schema_loader:
      type: InlineSchemaLoader
      schema:
        $schema: http://json-schema.org/schema#
        properties:
          errors:
            items:
              properties:
                extensions:
                  properties:
                    argumentName:
                      type: string
                    code:
                      type: string
                    typeName:
                      type: string
                  type: object
                locations:
                  items:
                    properties:
                      column:
                        type: number
                      line:
                        type: number
                    type: object
                  type: array
                message:
                  type: string
                path:
                  items:
                    type: string
                  type: array
              type: object
            type: array
        type: object
    retriever:
      type: SimpleRetriever
      requester:
        type: HttpRequester
        url_base: >-
          https://classicalacademicpress.myshopify.com/admin/api/2023-07/graphql.json
        path: >-
          https://classicalacademicpress.myshopify.com/admin/api/2023-07/graphql.json
        http_method: POST
        request_parameters: {}
        request_headers:
          Content-Type: application/json
        authenticator:
          type: ApiKeyAuthenticator
          api_token: '{{ config[''api_key''] }}'
          inject_into:
            type: RequestOption
            field_name: X-Shopify-Access-Token
            inject_into: header
        request_body_json:
          query: |-
            { companies(first: 100, after: {{ next_page_token }}) {
                pageInfo {
                  hasNextPage
                  endCursor
                }
                nodes {
                  id      
                  externalId
                  name
                }
              }
            }
      record_selector:
        type: RecordSelector
        extractor:
          type: DpathExtractor
          field_path:
            - data
            - companies
            - nodes
      paginator:
        type: DefaultPaginator
        pagination_strategy:
          type: CursorPagination
          cursor_value: >-
            {{ response.get("data", {}).get("companies", {}).get("pageInfo",
            {}).get("endCursor", {}) }}
          stop_condition: >-
            {{ not response.get("data", {}).get("companies", {}).get("pageInfo",
            {}).get("hasNextPage", {}) }}
spec:
  connection_specification:
    $schema: http://json-schema.org/draft-07/schema#
    type: object
    required:
      - api_key
    properties:
      api_key:
        type: string
        title: API Key
        airbyte_secret: true
        order: 0
    additionalProperties: true
  documentation_url: https://example.org
  type: Spec
metadata:
  autoImportSchema:
    Companies: true

[timuric]

Also having the same issue with Saleor GraphQL API, it seems generally pagination with GraphQL is not possible?

[timuric]

found solution to this, you need to use variables and jinja if else predicate to add conditionally cursor

[dprose]

@timuric, thanks for posting! Can you give a quick code example of how you solved it?

[timuric]

Here you go, this is for Saleor GraphQL API which has similar GraphQL convention so should be easy to adopt to your query

spec:
  type: Spec
  connection_specification:
    type: object
    $schema: http://json-schema.org/draft-07/schema#
    required:
      - start_date
      - api_url
      - channel
      - page_size
    properties:
      api_url:
        type: string
        order: 1
        title: API_URL
        description: https://{saleor-instance}/graphql/
      channel:
        type: string
        order: 2
        title: channel
      page_size:
        type: string
        order: 3
        title: page_size
      start_date:
        type: string
        order: 0
        title: Start date
        format: date-time
        pattern: ^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$
    additionalProperties: true
type: DeclarativeSource
check:
  type: CheckStream
  stream_names:
    - Products
streams:
  - name: Products
    type: DeclarativeStream
    retriever:
      type: SimpleRetriever
      paginator:
        type: DefaultPaginator
        page_token_option:
          type: RequestOption
          field_name: variables
          inject_into: body_json
        pagination_strategy:
          type: CursorPagination
          cursor_value: >-
            {{ '{"after": "'  + response.data.products.pageInfo.endCursor + '"}'
            if response.data.products.pageInfo.endCursor else "" }}
          stop_condition: '{{ response.data.products.pageInfo.nextPage is false }}'
      requester:
        path: '{{ config[''api_url''] }}'
        type: HttpRequester
        url_base: https://docs.saleor.io/docs/3.x/
        http_method: POST
        authenticator:
          type: NoAuth
        request_headers:
          content-type: application/json
        request_body_json:
          query: >-
            query($after: String){   products(first: {{ config['page_size'] }},
            after: $after, channel: "{{ config['channel'] }}", sortBy: {    
            field: LAST_MODIFIED,     direction: DESC   }){     edges{      
            node{         id         name   updatedAt    }       cursor    
            }     pageInfo{       hasNextPage       endCursor     }   } }
        request_parameters: {}
      record_selector:
        type: RecordSelector
        extractor:
          type: DpathExtractor
          field_path:
            - data
            - products
            - edges
            - '*'
            - node
      partition_router: []
    primary_key: []
    schema_loader:
      type: InlineSchemaLoader
      schema:
        type: object
        $schema: http://json-schema.org/schema#
        properties:
          id:
            type: string
          name:
            type: string
          updatedAt:
            type: string
    incremental_sync:
      type: DatetimeBasedCursor
      cursor_field: updatedAt
      is_data_feed: true
      start_datetime:
        type: MinMaxDatetime
        datetime: '{{ config[''start_date''] }}'
        datetime_format: '%Y-%m-%dT%H:%M:%SZ'
      datetime_format: '%Y-%m-%dT%H:%M:%S.%f%z'
      lookback_window: PT1H
      cursor_datetime_formats:
        - '%Y-%m-%dT%H:%M:%S.%f%z'
version: 0.61.2
metadata:
  autoImportSchema:
    Products: true

[dprose]

Thanks so much!