Pipeline Run with File — 파일 업로드와 함께 파이프라인 실행

Authorization: Bearer sk_live_xxxxxxxxxxxxxxxx

{
  "logicalDate": "2026-05-22T00:00:00+09:00",
  "note": "daily import from analytics",
  "conf": {
    "any_user_key": "any_value"
  }
}

DataRetrievalOperator(
    connection_id="tenant_{companyId}_uploaded_file_default",
    query="{{ dag_run.conf['uploaded_file_path'] }}",
)

{
  "code": "SUCCESS",
  "message": "OK",
  "data": {
    "dagRunId": "manual__<uuid>",
    "dagId": "tenant__abc123",
    "state": "QUEUED",
    "logicalDate": "2026-05-22T00:00:00+09:00",
    "startDate": null,
    "note": "Triggered with uploaded file sales.parquet"
  },
  "traceId": "00-xxxxx-xxxxx-01"
}

curl -X POST "${HOST}/open/v1/pipelines/${PIPELINE_ID}/runs/with-file" \
  -H "Authorization: Bearer ${API_KEY}" \
  -F "file=@./sales.parquet"

curl -X POST "${HOST}/open/v1/pipelines/${PIPELINE_ID}/runs/with-file" \
  -H "Authorization: Bearer ${API_KEY}" \
  -F "file=@./sales.parquet" \
  -F 'metadata={"note":"daily","logicalDate":"2026-05-22T00:00:00+09:00","conf":{"source":"ga4"}};type=application/json'

const fd = new FormData();
fd.append("file", fileBlob, "sales.parquet");
 
// metadata는 별도 Blob으로 감싸야 Content-Type을 application/json으로 지정 가능
fd.append(
  "metadata",
  new Blob(
    [JSON.stringify({ note: "daily", conf: { source: "ga4" } })],
    { type: "application/json" }
  )
);
 
const res = await fetch(
  `${HOST}/open/v1/pipelines/${pipelineId}/runs/with-file`,
  {
    method: "POST",
    headers: {
      Authorization: `Bearer ${API_KEY}`,
      // Content-Type 절대 명시 금지 — fetch가 multipart boundary 자동 생성
    },
    body: fd,
  }
);
 
const json = await res.json();
console.log(json.data.dagRunId); // "manual__<uuid>"

import requests, json
 
with open("sales.parquet", "rb") as f:
    res = requests.post(
        f"{HOST}/open/v1/pipelines/{pipeline_id}/runs/with-file",
        headers={"Authorization": f"Bearer {API_KEY}"},
        files={
            "file": ("sales.parquet", f, "application/octet-stream"),
            "metadata": (
                None,
                json.dumps({"note": "daily"}),
                "application/json",
            ),
        },
    )
 
print(res.json()["data"]["dagRunId"])

[Client]
  │  multipart: file + metadata
  ▼
[app-api]
  │  · 검증 (size, 확장자, 비어있음)
  │  · MultipartFile → temp file (메모리 적재 회피)
  │  · StorageService.streamUpload() → ingest/{companyId}/{uuid}.{ext}
  │  · DataConnectorService.triggerCompanyDagRun(pipelineId, conf={uploaded_file_*})
  ▼ (Airflow REST API v2, 회사별 Bearer 토큰)
[Airflow DAG (Pipeline Builder 생성)]
  │  · DataRetrievalOperator(conn_id=..., query="{{ dag_run.conf['uploaded_file_path'] }}")
  │  · 런타임에 UploadedFileConnector가 fsspec으로 스토리지에서 다운로드 + 파싱
  ▼
[downstream transform/load operators — 기존 그대로]