Python (gql)

Step 1. Getting started

Step 1.1. Overview

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)

Step 1.2. Add dependencies

본 문서는 python-gql을 사용하여 Delivery Tracker API를 사용하는 방법을 안내 합니다. 이를 위해 우선 python-gql을 설치 합니다.

pip install gql[aiohttp]

Step 1.3. Add Transport class

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] = 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. Overview"에 기재되어 있는 main.py 코드를 작성하고 main.py를 실행하면 API를 동작 시켜보세요.

Step 2. Advanced

Schema validation 등의 추가적인 정보가 필요하신 경우 python-gql 문서를 참고해주세요.