API 튜토리얼

이 튜토리얼에서는 D.Hub API로 데이터 수집부터 파이프라인 실행까지 전 과정을 단계별로 실습합니다. cURL을 중심으로 진행하며, 각 단계의 Python 코드도 함께 제공합니다.

사전 준비

D.Hub 인스턴스 접속 URL ({host})
사용자 계정 (email, password)
cURL 또는 Python 3.x 환경 (cURL 트랙은 JSON 파싱용 jq 권장)

Placeholder 표기 안내

cURL 예시에는 {host}, you@example.com, col-abc123 같은 자리표시자가 나옵니다. 컬렉션·데이터셋·파이프라인 ID는 각 단계의 응답에서 정해지므로 아래 패턴처럼 환경 변수로 이어 받습니다. 자리표시자를 그대로 사용하면 요청이 동작하지 않습니다.

RESP=$(curl -s -X POST https://{host}/api/v1/collections ...)
COLLECTION_ID=$(echo "$RESP" | jq -r .id)
echo "COLLECTION_ID=$COLLECTION_ID"

{host}와 계정 정보만 환경에 맞게 한 번 치환하면 나머지 ID는 jq로 자동 추출됩니다.

1단계. 로그인하고 토큰 발급받기

가장 먼저 API 인증에 필요한 JWT 토큰을 발급받습니다. 로그인은 이메일과 비밀번호로 합니다.

cURL

RESP=$(curl -s -X POST https://{host}/api/v1/auth/login \
  -H "Content-Type: application/json" \
  -d '{"email": "you@example.com", "password": "your-password"}')
TOKEN=$(echo "$RESP" | jq -r .access_token)
echo "TOKEN=${TOKEN:0:24}..."

응답

{
  "access_token": "eyJhbGciOiJIUzI1NiIs...",
  "token_type": "bearer"
}

refresh_token은 응답 본문이 아니라 HttpOnly 쿠키로 설정됩니다. 이 튜토리얼에서는 이후 요청에 access_token만 사용합니다.

Python

import requests

HOST = "https://{host}"

login_response = requests.post(
    f"{HOST}/api/v1/auth/login",
    json={"email": "you@example.com", "password": "your-password"},
)
tokens = login_response.json()
headers = {"Authorization": f"Bearer {tokens['access_token']}"}

2단계. 컬렉션 만들기

데이터셋과 파이프라인을 관리할 컬렉션을 생성합니다.

cURL

RESP=$(curl -s -X POST https://{host}/api/v1/collections \
  -H "Authorization: Bearer ${TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"name": "seoul_traffic_analysis", "alias": "서울시 교통 분석", "description": "서울시 교통 데이터 수집 및 분석 프로젝트"}')
COLLECTION_ID=$(echo "$RESP" | jq -r .id)
echo "COLLECTION_ID=$COLLECTION_ID"

응답

{
  "id": "col-abc123",
  "name": "seoul_traffic_analysis",
  "alias": "서울시 교통 분석",
  "description": "서울시 교통 데이터 수집 및 분석 프로젝트",
  "created_at": "2026-03-12T09:00:00Z"
}

Python

collection = requests.post(
    f"{HOST}/api/v1/collections",
    headers=headers,
    json={
        "name": "seoul_traffic_analysis",
        "alias": "서울시 교통 분석",
        "description": "서울시 교통 데이터 수집 및 분석 프로젝트",
    },
).json()
collection_id = collection["id"]

3단계. 데이터셋 만들고 CSV 파일 업로드하기

컬렉션 내에 데이터셋을 생성하고 CSV 파일을 업로드합니다.

3-1. 데이터셋 생성

RESP=$(curl -s -X POST https://{host}/api/v1/datasets \
  -H "Authorization: Bearer ${TOKEN}" \
  -H "Content-Type: application/json" \
  -d "{\"name\": \"traffic_data\", \"alias\": \"교통량 데이터\", \"type\": \"delta\", \"description\": \"서울시 주요 도로 교통량\", \"collection_id\": \"${COLLECTION_ID}\"}")
DATASET_ID=$(echo "$RESP" | jq -r .id)
echo "DATASET_ID=$DATASET_ID"

3-2. CSV 파일 업로드

업로드는 데이터셋 하위 엔드포인트에 multipart/form-data로 보내며, 파일 필드 이름은 files 입니다.

curl -X POST https://{host}/api/v1/datasets/${DATASET_ID}/upload \
  -H "Authorization: Bearer ${TOKEN}" \
  -F "files=@traffic_data.csv"

Python

dataset = requests.post(
    f"{HOST}/api/v1/datasets",
    headers=headers,
    json={
        "name": "traffic_data",
        "alias": "교통량 데이터",
        "type": "delta",
        "description": "서울시 주요 도로 교통량",
        "collection_id": collection_id,
    },
).json()
dataset_id = dataset["id"]

with open("traffic_data.csv", "rb") as f:
    requests.post(
        f"{HOST}/api/v1/datasets/{dataset_id}/upload",
        headers=headers,
        files={"files": ("traffic_data.csv", f, "text/csv")},
    )

4단계. 데이터 조회하기

업로드된 데이터셋에서 최대 10개 행을 JSON으로 조회합니다. limit은 반환할 행 수를 제한하며 커서를 만들지 않습니다.

cURL

curl -X GET "https://{host}/api/v1/datasets/${DATASET_ID}/table?format=json&limit=10" \
  -H "Authorization: Bearer ${TOKEN}"

Python

table_data = requests.get(
    f"{HOST}/api/v1/datasets/{dataset_id}/table",
    headers=headers,
    params={"format": "json", "limit": 10},
).json()

SQL로 조회

컬럼 선택·필터·집계가 필요하면 POST /api/v1/datasets/{id}/table/query에 {"query": "SELECT ...", "limit": N} 형태로 SQL을 보냅니다. 지원 문법은 SQL 참조에서 확인합니다.

5단계. 파이프라인 만들고 실행하기

데이터를 가공하는 파이프라인을 생성하고 실행합니다.

5-1. 출력 데이터셋과 파이프라인 생성

결과를 저장할 데이터셋을 만든 뒤, 입력 행을 결과 데이터셋으로 복사하는 배치 파이프라인을 만듭니다. 파이프라인 생성 요청에는 type과 하나 이상의 steps를 포함합니다.

OUTPUT_RESP=$(curl -s -X POST https://{host}/api/v1/datasets \
  -H "Authorization: Bearer ${TOKEN}" \
  -H "Content-Type: application/json" \
  -d "{\"name\": \"traffic_data_copy\", \"alias\": \"교통량 데이터 복사본\", \"type\": \"delta\", \"collection_id\": \"${COLLECTION_ID}\"}")
OUTPUT_DATASET_ID=$(echo "$OUTPUT_RESP" | jq -r .id)

PIPELINE_PAYLOAD=$(jq -n \
  --arg collection_id "$COLLECTION_ID" \
  --arg input_id "$DATASET_ID" \
  --arg output_id "$OUTPUT_DATASET_ID" \
  '{
    name: "traffic_copy_pipeline",
    alias: "교통량 데이터 복사",
    type: "batch",
    collection_id: $collection_id,
    steps: [{
      name: "copy_rows",
      alias: "행 복사",
      transform: {op: "select_filter", config: {input: "source"}},
      inputs: {source: {dataset: $input_id, options: {read_mode: "batch"}}},
      outputs: {result: {dataset: $output_id, options: {write_mode: "overwrite"}}}
    }]
  }')

RESP=$(curl -s -X POST https://{host}/api/v1/pipelines \
  -H "Authorization: Bearer ${TOKEN}" \
  -H "Content-Type: application/json" \
  -d "$PIPELINE_PAYLOAD")
PIPELINE_ID=$(echo "$RESP" | jq -r .id)
echo "PIPELINE_ID=$PIPELINE_ID"

5-2. 파이프라인 실행 (배치)

파이프라인 실행은 배치(batch) 엔드포인트로 시작합니다.

RESP=$(curl -s -X POST https://{host}/api/v1/pipelines/${PIPELINE_ID}/batch \
  -H "Authorization: Bearer ${TOKEN}")
BATCH_ID=$(echo "$RESP" | jq -r .id)
echo "BATCH_ID=$BATCH_ID"

Python

output_dataset = requests.post(
    f"{HOST}/api/v1/datasets",
    headers=headers,
    json={
        "name": "traffic_data_copy",
        "alias": "교통량 데이터 복사본",
        "type": "delta",
        "collection_id": collection_id,
    },
).json()
output_dataset_id = output_dataset["id"]

pipeline = requests.post(
    f"{HOST}/api/v1/pipelines",
    headers=headers,
    json={
        "name": "traffic_copy_pipeline",
        "alias": "교통량 데이터 복사",
        "type": "batch",
        "collection_id": collection_id,
        "steps": [
            {
                "name": "copy_rows",
                "alias": "행 복사",
                "transform": {
                    "op": "select_filter",
                    "config": {"input": "source"},
                },
                "inputs": {
                    "source": {
                        "dataset": dataset_id,
                        "options": {"read_mode": "batch"},
                    }
                },
                "outputs": {
                    "result": {
                        "dataset": output_dataset_id,
                        "options": {"write_mode": "overwrite"},
                    }
                },
            }
        ],
    },
).json()
pipeline_id = pipeline["id"]

run = requests.post(
    f"{HOST}/api/v1/pipelines/{pipeline_id}/batch",
    headers=headers,
).json()
batch_id = run["id"]

파이프라인 스텝 구성

이 예제는 모든 행을 복사하는 select_filter 변환 스텝 하나를 사용합니다. 컬럼 선택·필터·조인 같은 세부 스텝은 웹 UI의 파이프라인 에디터에서 구성한 뒤 API로 실행할 수도 있습니다.

6단계. 실행 결과 확인하기

6-1. 배치 상태 조회

curl -X GET https://{host}/api/v1/pipelines/${PIPELINE_ID}/batch \
  -H "Authorization: Bearer ${TOKEN}"

6-2. 실행 트레이스 조회

특정 배치의 단계별 처리 결과(트레이스)를 확인합니다.

curl -X GET https://{host}/api/v1/trace/pipelines/${PIPELINE_ID}/batches/${BATCH_ID} \
  -H "Authorization: Bearer ${TOKEN}"

Python

import time

while True:
    batch = requests.get(
        f"{HOST}/api/v1/pipelines/{pipeline_id}/batch",
        headers=headers,
    ).json()

    state = (batch.get("state") or {}).get("state")
    print(f"상태: {state}")
    if state in ("ready", "failed"):
        break
    time.sleep(3)

trace = requests.get(
    f"{HOST}/api/v1/trace/pipelines/{pipeline_id}/batches/{batch_id}",
    headers=headers,
).json()

전체 흐름 요약

단계	API 엔드포인트	HTTP 메서드
로그인	`/api/v1/auth/login`	POST
컬렉션 생성	`/api/v1/collections`	POST
데이터셋 생성	`/api/v1/datasets`	POST
CSV 업로드	`/api/v1/datasets/{id}/upload`	POST
데이터 조회	`/api/v1/datasets/{id}/table`	GET
파이프라인 생성	`/api/v1/pipelines`	POST
파이프라인 실행	`/api/v1/pipelines/{id}/batch`	POST
배치 상태 조회	`/api/v1/pipelines/{id}/batch`	GET
트레이스 조회	`/api/v1/trace/pipelines/{id}/batches/{batch_id}`	GET

다음 단계

API 인증 — 서비스 토큰과 토큰 갱신 방식을 적용합니다.
오류 처리 — 상태 코드별 오류 처리와 재시도 기준을 확인합니다.
API 클라이언트 도구 — 사용하는 언어와 도구에 맞는 호출 방식을 선택합니다.

1단계. 로그인하고 토큰 발급받기​

cURL​

응답​

Python​

2단계. 컬렉션 만들기​

cURL​

응답​

Python​

3단계. 데이터셋 만들고 CSV 파일 업로드하기​

3-1. 데이터셋 생성​

3-2. CSV 파일 업로드​

Python​

4단계. 데이터 조회하기​

cURL​

Python​

5단계. 파이프라인 만들고 실행하기​

5-1. 출력 데이터셋과 파이프라인 생성​

5-2. 파이프라인 실행 (배치)​

Python​

6단계. 실행 결과 확인하기​

6-1. 배치 상태 조회​

6-2. 실행 트레이스 조회​

Python​

전체 흐름 요약​

다음 단계​

1단계. 로그인하고 토큰 발급받기

cURL

응답

Python

2단계. 컬렉션 만들기

cURL

응답

Python

3단계. 데이터셋 만들고 CSV 파일 업로드하기

3-1. 데이터셋 생성

3-2. CSV 파일 업로드

Python

4단계. 데이터 조회하기

cURL

Python

5단계. 파이프라인 만들고 실행하기

5-1. 출력 데이터셋과 파이프라인 생성

5-2. 파이프라인 실행 (배치)

Python

6단계. 실행 결과 확인하기

6-1. 배치 상태 조회

6-2. 실행 트레이스 조회

Python

전체 흐름 요약

다음 단계