Python (gql)

Step 1. Getting started

Step 1.1. Overview

After the initial setup in Step 1, you'll have the following APIs at your fingertips.

main.py

from gql import gql, Client
from delivery_tracker_gql.transport import DeliveryTrackerAIOHTTPTransport

# Example query to get the time, code of the last event
# There are many other queries, which can be used to retrieve various information
# Please refer to the documentation and API schema on the official site.
TRACK_QUERY = gql(
    """
    query GetTrackLastEvent($carrierId: ID!, $trackingNumber: String!) {
        track(carrierId: $carrierId, trackingNumber: $trackingNumber) {
            lastEvent {
                time
                status {
                    code
                }
            }
        }
    } 
    """
)

# Create a Client that connects to the DeliveryTracker Server.
client = Client(
    transport=DeliveryTrackerAIOHTTPTransport(
        client_id="[YOUR_CLIENT_ID]",
        client_secret="[YOUR_CLIENT_SECRET]",
    ),
    fetch_schema_from_transport=False,
)

# Call the DeliveryTracker API.
# Use `await client.execute_async(TRACK_QUERY, ...)` inside the coroutine
result = client.execute(
    TRACK_QUERY,
    variable_values={
        "carrierId": "en.cjlogistics",
        "trackingNumber": "1234567890",
    },
)

print(result)

Step 1.2. Add dependencies

This document walks you through how to use the Delivery Tracker API using python-gql. To do this, first install python-gql.

pip install gql[aiohttp]

Step 1.3. Add Transport class

The AIOHTTPTransport built into python-gql does not provide Delivery Tracker Auth out of the box. For authentication, add a DeliveryTrackerAIOHTTPTransport that inherits from AIOHTTPTransport and implements Delivery Tracker Auth.

delivery_tracker_gql/transport.py

import json
import base64
import aiohttp
from typing import Any, Callable, Dict, Optional, Union
from aiohttp.typedefs import LooseHeaders
from graphql import DocumentNode, ExecutionResult
from gql.transport.aiohttp import AIOHTTPTransport
from gql.transport.exceptions import (
    TransportProtocolError,
    TransportServerError,
)


class DeliveryTrackerAIOHTTPTransport(AIOHTTPTransport):
    DELIVERY_TRACKER_API_GRAPHQL_ENDPOINT = "https://apis.tracker.delivery/graphql"
    DELIVERY_TRACKER_AUTH_TOKEN_ENDPOINT = "https://auth.tracker.delivery/oauth2/token"
    __credentials: str
    __auth_http_session: Optional[aiohttp.ClientSession] = None
    __access_token: Optional[str] = None

    def __init__(
        self,
        client_id: str,
        client_secret: str,
        headers: Optional[LooseHeaders] = None,
        timeout: Optional[int] = None,
        ssl_close_timeout: Optional[Union[int, float]] = 10,
        json_serialize: Callable = json.dumps,
        client_session_args: Optional[Dict[str, Any]] = None,
    ):
        self.__credentials = (
            "Basic "
            + base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
        )
        super().__init__(
            url=self.DELIVERY_TRACKER_API_GRAPHQL_ENDPOINT,
            headers=headers,
            timeout=timeout,
            ssl_close_timeout=ssl_close_timeout,
            json_serialize=json_serialize,
            client_session_args=client_session_args,
        )

    async def execute(
        self,
        document: DocumentNode,
        variable_values: Optional[Dict[str, Any]] = None,
        operation_name: Optional[str] = None,
        extra_args: Optional[Dict[str, Any]] = None,
        upload_files: bool = False,
    ) -> ExecutionResult:
        access_token = await self.__get_access_token()

        if extra_args is None:
            extra_args = {}
        if "headers" not in extra_args:
            extra_args["headers"] = {}

        if access_token is not None:
            extra_args["headers"]["Authorization"] = f"Bearer {access_token}"

        result = await super().execute(
            document=document,
            variable_values=variable_values,
            operation_name=operation_name,
            extra_args=extra_args,
            upload_files=upload_files,
        )

        if self.__has_unauthenticated_error(result):
            access_token = await self.__get_access_token(
                force_fetch_new_access_token=True
            )
            if access_token is not None:
                extra_args["headers"]["Authorization"] = f"Bearer {access_token}"

            result = await super().execute(
                document=document,
                variable_values=variable_values,
                operation_name=operation_name,
                extra_args=extra_args,
                upload_files=upload_files,
            )

        return result

    def __has_unauthenticated_error(self, execution_result: ExecutionResult) -> bool:
        if execution_result.errors is None:
            return False
        for error in execution_result.errors:
            try:
                if error["extensions"]["code"] == "UNAUTHENTICATED":
                    return True
            except:
                pass
        return False

    async def __get_access_token(
        self, force_fetch_new_access_token: bool = False
    ) -> Optional[str]:
        if self.__access_token is None or force_fetch_new_access_token:
            self.__access_token = await self.__fetch_new_access_token()
        return self.__access_token

    async def __fetch_new_access_token(self) -> Optional[str]:
        if self.__auth_http_session is None:
            self.__auth_http_session = aiohttp.ClientSession()

        auth_response = await self.__auth_http_session.post(
            url=self.DELIVERY_TRACKER_AUTH_TOKEN_ENDPOINT,
            headers={
                "Authorization": self.__credentials,
                "Content-Type": "application/x-www-form-urlencoded",
            },
            data=b"grant_type=client_credentials",
        )

        if auth_response.status >= 400:
            raise TransportServerError(
                f"Auth error: http response code={auth_response.status} body={await auth_response.text()}"
            )

        try:
            auth_response_body = await auth_response.json()
            access_token = auth_response_body["access_token"]
            assert isinstance(access_token, str)
            return access_token
        except Exception as e:
            raise TransportProtocolError("The access_token field was not found.") from e

Step 1.4. Execute your first query

"Step 1.1. Write the main.py code described in the "Overview" and run main.py to get the API working.

Step 2. Advanced

If you need additional information, such as schema validation, please see the python-gql documentation.