Python (gql)

ステップ1. はじめに

ステップ1.1. 概要

Step 1の初期設定が完了すると、以下のようなAPIを簡単に利用することができます。

main.py

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

# 最近のEventのtime, codeを取得するクエリの例文
# その他様々なクエリが存在し、これにより様々な情報を照会することができます。
# 公式サイトのドキュメント及びAPIスキーマを参考にしてください。
TRACK_QUERY = gql(
    """
    query GetTrackLastEvent($carrierId: ID!、 $trackingNumber: String!) {
        track(carrierId: $carrierId, trackingNumber: $trackingNumber) {
            lastEvent {
                time
                status {
                    code
                }
            }
        }
    } 
    """
)

# DeliveryTracker Server と接続する Client を作成します。
client = Client(
    transport=DeliveryTrackerAIOHTTPTransport(
        client_id="[YOUR_CLIENT_ID]",
        client_secret="[YOUR_CLIENT_SECRET]",
    ),
    fetch_schema_from_transport=False,
)

# DeliveryTracker API を呼び出します。
# コルーチン内では `await client.execute_async(TRACK_QUERY, ...)` を使用してください。
result = client.execute(
    TRACK_QUERY,
    variable_values={
        "carrierId"："kr.cjlogistics",
        "trackingNumber"："1234567890",
    },
)

print(result)

ステップ1.2. 依存関係の追加

このドキュメントはpython-gqlを使ってDelivery Tracker APIを使う方法を説明します。 このため、まずpython-gqlをインストールします。

pip install gql[aiohttp] をインストールします。

ステップ1.3. トランスポートクラスの追加

python-gqlに組み込まれている AIOHTTPTransport はDelivery Tracker Authを内蔵していません。 認証のため AIOHTTPTransport を継承し、Delivery Tracker Authが実装された DeliveryTrackerAIOHTTPTransport を追加します。

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] = なし
    __access_token: Optional[str] = なし

    def __init__(
        self,
        client_id: str,
        client_secret: str,
        headers: Optional[LooseHeaders] = なし,
        timeout: Optional[int] = なし,
        ssl_close_timeout: Optional[Union[int, float]].= 10,
        json_serialize: Callable = json.dumps,
        client_session_args: Optional[Dict[str, Any]]: client_session_args.= 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：オプション[Dict[str, 任意]]]= なし、
        operation_name: Optional[str] = なし、
        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
        if 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がNoneまたは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.

ステップ1.4. 最初のクエリを実行する

"ステップ1.1. Overview"に記載されているmain.pyのコードを作成し、main.pyを実行するとAPIを動作させてみましょう。

ステップ2. 上級者向け

Schema validationなどの追加情報が必要な場合はpython-gqlドキュメントを参考してください。