Python 택배API 연동 가이드

1. 설치

필요한 패키지를 설치합니다.

pip install requests python-dotenv

웹훅 서버를 위해 Flask도 설치합니다.

pip install flask

2. 환경 변수 설정

API 키를 환경 변수로 관리합니다.

# .env 파일
DELIVERY_PUBLIC_KEY=pk_live_your_public_key
DELIVERY_SECRET_KEY=sk_live_your_secret_key
WEBHOOK_SECRET=your_webhook_secret

import os
from dotenv import load_dotenv

load_dotenv()
PUBLIC_KEY = os.getenv('DELIVERY_PUBLIC_KEY')
SECRET_KEY = os.getenv('DELIVERY_SECRET_KEY')
API_KEY = f'{PUBLIC_KEY}:{SECRET_KEY}'

3. 배송 조회

기본 예제 (단건 조회)

import requests
import os

PUBLIC_KEY = os.getenv('DELIVERY_PUBLIC_KEY')
SECRET_KEY = os.getenv('DELIVERY_SECRET_KEY')
BASE_URL = 'https://api.deliveryapi.co.kr/v1'

def track_delivery(courier_code: str, tracking_number: str) -> dict:
    """배송 조회 API 호출"""
    url = f'{BASE_URL}/tracking/trace'
    headers = {
        'Authorization': f'Bearer {PUBLIC_KEY}:{SECRET_KEY}',
        'Content-Type': 'application/json'
    }
    payload = {
        'items': [{
            'courierCode': courier_code,
            'trackingNumber': tracking_number
        }]
    }

    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()  # HTTP 에러 발생 시 예외

    # 첫 번째 결과 반환
    return response.json()['data']['results'][0]


# 사용 예시
if __name__ == '__main__':
    try:
        result = track_delivery('cj', '1234567890')

        if result['success']:
            data = result['data']
            print(f"배송 상태: {data['deliveryStatus']}")
            print(f"택배사: {data['courierName']}")

            # 배송 이력 출력
            for progress in data['progresses']:
                print(f"  {progress['dateTime']} - {progress['description']}")
        else:
            print(f"조회 실패: {result['message']}")

    except requests.exceptions.HTTPError as e:
        print(f"API 에러: {e.response.json()}")

다건 조회 (최대 100건)

def track_multiple(items: list[dict]) -> list[dict]:
    """여러 송장 동시 조회"""
    url = f'{BASE_URL}/tracking/trace'
    headers = {
        'Authorization': f'Bearer {PUBLIC_KEY}:{SECRET_KEY}',
        'Content-Type': 'application/json'
    }
    payload = {'items': items}

    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()['data']['results']


# 사용 예시
items = [
    {'courierCode': 'cj', 'trackingNumber': '123456789012'},
    {'courierCode': 'lotte', 'trackingNumber': '987654321098'},
    {'courierCode': 'hanjin', 'trackingNumber': '112233445566'}
]

results = track_multiple(items)
for result in results:
    if result['success']:
        data = result['data']
        print(f"{data['courierName']}: {data['deliveryStatus']}")
    else:
        print(f"조회 실패: {result['error']['message']}")

클래스로 래핑

import requests
from typing import Optional, List, Dict
from dataclasses import dataclass


@dataclass
class Progress:
    dateTime: str
    status: str
    location: str
    description: str
    statusCode: str = ""


@dataclass
class TrackingResult:
    courier_code: str
    tracking_number: str
    delivery_status: str
    delivery_status_text: str
    is_delivered: bool
    progresses: List[Progress]


class DeliveryAPI:
    def __init__(self, public_key: str, secret_key: str):
        self.public_key = public_key
        self.secret_key = secret_key
        self.base_url = 'https://api.deliveryapi.co.kr/v1'
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {public_key}:{secret_key}',
            'Content-Type': 'application/json'
        })

    def track(self, courier_code: str, tracking_number: str) -> TrackingResult:
        """배송 조회"""
        response = self.session.post(
            f'{self.base_url}/tracking/trace',
            json={'items': [{'courierCode': courier_code, 'trackingNumber': tracking_number}]}
        )
        response.raise_for_status()
        result = response.json()['data']['results'][0]

        if not result['success']:
            raise Exception(result['message'])

        data = result['data']
        return TrackingResult(
            courier_code=data['courierCode'],
            tracking_number=data['trackingNumber'],
            delivery_status=data['deliveryStatus'],
            delivery_status_text=data['deliveryStatusText'],
            is_delivered=data['isDelivered'],
            progresses=[
                Progress(**p) for p in data.get('progresses', [])
            ]
        )

    def get_couriers(self) -> List[Dict]:
        """지원 택배사 목록 조회"""
        response = self.session.get(f'{self.base_url}/tracking/couriers')
        response.raise_for_status()
        return response.json()['data']


# 사용 예시
api = DeliveryAPI(
    os.getenv('DELIVERY_PUBLIC_KEY'),
    os.getenv('DELIVERY_SECRET_KEY')
)
result = api.track('cj', '1234567890')
print(f"상태: {result.delivery_status_text}")

4. 택배사 목록 조회

def get_couriers() -> list:
    """지원 택배사 목록 조회"""
    url = f'{BASE_URL}/tracking/couriers'
    headers = {'Authorization': f'Bearer {PUBLIC_KEY}:{SECRET_KEY}'}

    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()['data']


# 결과 예시
couriers = get_couriers()
for c in couriers:
    print(f"{c['code']}: {c['name']}")

# lotte: 롯데택배
# cj: CJ대한통운
# hanjin: 한진택배
# ...

5. 웹훅 수신 서버 (Flask)

배송 상태 변경 알림을 받는 웹훅 서버를 구현합니다.

from flask import Flask, request, jsonify
import hmac
import hashlib
import os

app = Flask(__name__)
WEBHOOK_SECRET = os.getenv('WEBHOOK_SECRET')


def verify_signature(payload: bytes, timestamp: str, signature: str) -> bool:
    """웹훅 서명 검증"""
    message = f"{timestamp}.".encode() + payload
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        message,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, f"sha256={expected}")


@app.route('/webhook/delivery', methods=['POST'])
def handle_webhook():
    # 서명 검증
    signature = request.headers.get('X-Webhook-Signature', '')
    timestamp = request.headers.get('X-Webhook-Timestamp', '')
    if not verify_signature(request.data, timestamp, signature):
        return jsonify({'error': 'Invalid signature'}), 401

    data = request.json
    event = data.get('event')
    items = data.get('items', [])

    for item in items:
        if event == 'tracking.polled':
            print(f"배송 상태 변경: {item['trackingNumber']}")
            print(f"새 상태: {item['currentStatus']}")
            # 비즈니스 로직 처리

        elif event == 'tracking.completed':
            print(f"전체 배송 완료: {item['trackingNumber']}")
            # 배송 완료 처리

    # 200 응답 (필수)
    return jsonify({'received': True}), 200


if __name__ == '__main__':
    app.run(port=3000, debug=True)

6. 에러 처리 및 재시도

import time
from requests.exceptions import HTTPError, ConnectionError, Timeout


def track_with_retry(courier: str, tracking_number: str, max_retries: int = 3) -> dict:
    """재시도 로직이 포함된 배송 조회"""
    for attempt in range(1, max_retries + 1):
        try:
            return track_delivery(courier, tracking_number)

        except (ConnectionError, Timeout) as e:
            # 네트워크 에러 - 재시도
            if attempt < max_retries:
                delay = 2 ** attempt  # 지수 백오프
                print(f"재시도 {attempt}/{max_retries}, {delay}초 후...")
                time.sleep(delay)
                continue
            raise

        except HTTPError as e:
            status_code = e.response.status_code

            # 5xx 서버 에러 - 재시도
            if status_code >= 500 and attempt < max_retries:
                delay = 2 ** attempt
                print(f"서버 에러, 재시도 {attempt}/{max_retries}")
                time.sleep(delay)
                continue

            # 4xx 클라이언트 에러 - 재시도 불가
            raise


# 에러 코드별 처리
def handle_api_error(error: HTTPError) -> str:
    """API 에러 메시지 반환"""
    try:
        error_data = error.response.json()
        code = error_data.get('code')
    except:
        return '알 수 없는 오류가 발생했습니다'

    error_messages = {
        'INVALID_TRACKING_NUMBER': '잘못된 송장번호입니다',
        'COURIER_NOT_SUPPORTED': '지원하지 않는 택배사입니다',
        'RATE_LIMIT_EXCEEDED': '요청 한도를 초과했습니다',
        'UNAUTHORIZED': 'API 키가 유효하지 않습니다',
    }

    return error_messages.get(code, '일시적인 오류가 발생했습니다')

7. 비동기 처리 (asyncio + aiohttp)

import aiohttp
import asyncio


async def track_delivery_async(courier_code: str, tracking_number: str) -> dict:
    """비동기 배송 조회"""
    url = f'{BASE_URL}/tracking/trace'
    headers = {
        'Authorization': f'Bearer {PUBLIC_KEY}:{SECRET_KEY}',
        'Content-Type': 'application/json'
    }
    payload = {
        'items': [{
            'courierCode': courier_code,
            'trackingNumber': tracking_number
        }]
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, json=payload) as response:
            response.raise_for_status()
            data = await response.json()
            return data['data']['results'][0]


# 여러 송장 동시 조회
async def track_multiple(tracking_list: list) -> list:
    """여러 송장번호 동시 조회"""
    tasks = [
        track_delivery_async(item['courier'], item['tracking_number'])
        for item in tracking_list
    ]
    return await asyncio.gather(*tasks, return_exceptions=True)


# 사용 예시
# asyncio.run(track_delivery_async('lotte', '1234567890'))

다른 언어 가이드

• JavaScript 연동 가이드
• Java 연동 가이드
• PHP 연동 가이드
• C# 연동 가이드