쿼리 함수의 SSE 실행(Streaming Execute Query) (V2)

Query를 실행하고 text/event-stream 형식의 Server-Sent Events(SSE)로 결과를 받습니다. 일반 함수와 스트리밍 함수를 모두 지원합니다. 비스트리밍 함수는 전체 결과를 담은 데이터 이벤트 하나를, 스트리밍 함수는 결과 배치가 준비될 때마다 이벤트를 반환합니다.

본문에 ontology를 지정하면 해당 온톨로지 범위의 함수를 실행하고, 생략하면 전역 함수를 실행합니다. 일반 실행의 전역 함수 경로와 이 선택 입력을 구분합니다. 최신 버전의 기본 선택은 안정 버전의 가장 높은 의미 버전이며, PUBLISH_TIME은 최근 게시 기준과 기본 사전 릴리스 포함 동작을 사용합니다.

이 엔드포인트는 미리보기 기능이며 언제든 변경되거나 제거될 수 있습니다. 사용하려면 preview=true를 지정합니다.

HTTP 요청

POST /api/v2/functions/queries/{queryApiName}/streamingExecute

작업 식별자는 v2.streamingExecuteQuery입니다. OAuth2로 호출하는 타사 애플리케이션에는 api:functions-execute 권한 범위가 필요합니다. 각 요청에 Authorization: Bearer <접근 토큰>을 전달합니다.

경로 매개변수

쿼리 매개변수

요청 본문

StreamingExecuteQueryRequest 본문은 필수입니다. ontology는 RID 또는 API 이름을 받습니다. includePrerelease는 명시적 version과 함께 사용할 수 없고, branch와 version을 함께 지정하면 그 브랜치에 해당 버전이 있어야 합니다.

parameters의 구조

맵의 키는 ParameterId(string), 값은 DataValue(any)입니다.

키: 액션이나 쿼리 입력에 사용하는 매개변수의 고유 ID입니다. 매개변수는 Ontology Manager에서 확인하고 관리할 수 있습니다.

함수 입력이나 출력의 JSON 값입니다. 구조체 배열처럼 값이 중첩될 수 있습니다.

구체적인 JSON 표현과 중첩 값은 DataValue 인코딩을 따릅니다.

다음 본문의 식별자와 값은 가상 예시입니다.

{
  "latestVersionResolution": "PUBLISH_TIME",
  "parameters": {
    "price": 29.99
  },
  "version": "1.2.3",
  "branch": "ri.branch..branch.d827184f-ee0e-4351-8b70-efbe51e07252",
  "ontology": "example-ontology"
}

응답

SSE의 각 data 필드는 JSON으로 인코딩한 StreamingExecuteQueryResponse입니다. 메시지는 type: data 또는 type: error이며 빈 줄이 이벤트를 끝냅니다. 여러 data: 줄은 줄바꿈으로 연결하여 하나의 JSON 메시지로 처리합니다.

데이터를 일부 받은 뒤에도 실행 오류가 발생할 수 있습니다. 이때 오류 이벤트를 보낸 뒤 스트림이 종료되므로 HTTP 성공 상태만 확인해서는 안 됩니다. value가 배열이라고 가정하지 않고 함수의 출력 계약에 맞게 처리합니다.

분기 data

쿼리 결과의 배치입니다.

value의 구조

구체적인 JSON 표현과 중첩 값은 DataValue 인코딩을 따릅니다.

분기 error

쿼리 실행 중 발생한 오류입니다.

개별 오류

요청 예제

FOUNDRY_HOST에는 https://를 제외한 Foundry 호스트를, FOUNDRY_TOKEN에는 유효한 토큰을 제공합니다. 함수 API 이름, 온톨로지, 매개변수와 버전 및 브랜치는 실제 정의에 맞게 바꿉니다. 실험적 트랜잭션을 사용하지 않는 경우 transactionId를 쿼리에서 생략합니다.

예제는 빈 줄을 만날 때 모아 둔 data: 줄을 하나의 JSON 이벤트로 처리합니다. 데이터 이벤트의 JSON은 그대로 출력하고, 오류 이벤트나 HTTP 실패는 오류로 종료합니다. 이미 출력한 일부 배치를 실행 전체의 성공으로 처리하지 않습니다. 연결을 자동으로 재요청하지 않습니다.

cURL

Bash와 Python 3.7 이상을 사용합니다. pipefail로 HTTP 요청 또는 이벤트 처리 중 어느 쪽이 실패해도 파이프라인이 실패하도록 합니다.

set -o pipefail
curl --fail --no-buffer -X POST \
	-H "Content-type: application/json" \
	-H "Authorization: Bearer ${FOUNDRY_TOKEN}" \
	"https://${FOUNDRY_HOST}/api/v2/functions/queries/myQueryFunction/streamingExecute?transactionId=transaction-1&preview=true" \
	-d '{"latestVersionResolution":"PUBLISH_TIME","parameters":{"price":29.99},"version":"1.2.3","branch":"ri.branch..branch.d827184f-ee0e-4351-8b70-efbe51e07252","ontology":"example-ontology"}' | python3 -c '
import json
import sys

def consume_sse(lines):
    data_lines = []
    first_line = True
    for raw_line in lines:
        line = raw_line.rstrip("\r\n")
        if first_line and line.startswith("\ufeff"):
            line = line[1:]
        first_line = False
        if line == "":
            if data_lines:
                event_data = "\n".join(data_lines)
                message = json.loads(event_data)
                if message["type"] == "error":
                    raise RuntimeError("{}: {} ({})".format(
                        message["errorCode"], message["errorName"],
                        message["errorInstanceId"]))
                if message["type"] != "data" or "value" not in message:
                    raise ValueError("Invalid streaming Query message")
                print(event_data, flush=True)
                data_lines.clear()
            continue
        field, _, value = line.partition(":")
        if field == "data":
            data_lines.append(value[1:] if value.startswith(" ") else value)
    if data_lines:
        raise RuntimeError("SSE stream ended before the event delimiter")

sys.stdin.reconfigure(encoding="utf-8-sig", newline=None)
consume_sse(sys.stdin)
'

Python

Requests를 설치한 Python 환경에서 사용합니다. UTF-8 문자와 CRLF 줄바꿈이 네트워크 조각 사이에서 나뉘어도 이어서 처리합니다. CR, LF와 CRLF만 줄바꿈으로 해석하며 종료 시 응답 연결을 닫습니다.

import codecs
import io
import json
import os
import requests

HOSTNAME = os.environ["FOUNDRY_HOST"]
TOKEN = os.environ["FOUNDRY_TOKEN"]
headers = {
    "content-type": "application/json",
    "authorization": "Bearer {}".format(TOKEN)
}
request_body = {
    "latestVersionResolution": "PUBLISH_TIME",
    "parameters": {
        "price": 29.99
    },
    "version": "1.2.3",
    "branch": "ri.branch..branch.d827184f-ee0e-4351-8b70-efbe51e07252",
    "ontology": "example-ontology"
}

def iter_sse_lines(chunks):
    decoder = io.IncrementalNewlineDecoder(
        codecs.getincrementaldecoder("utf-8-sig")(), True
    )
    pending = ""
    for chunk in chunks:
        lines = (pending + decoder.decode(chunk)).split("\n")
        pending = lines.pop()
        yield from lines
    lines = (pending + decoder.decode(b"", True)).split("\n")
    pending = lines.pop()
    yield from lines
    if pending:
        yield pending


def consume_sse(lines):
    data_lines = []
    first_line = True
    for raw_line in lines:
        line = raw_line.rstrip("\r\n")
        if first_line and line.startswith("\ufeff"):
            line = line[1:]
        first_line = False
        if line == "":
            if data_lines:
                event_data = "\n".join(data_lines)
                message = json.loads(event_data)
                if message["type"] == "error":
                    raise RuntimeError("{}: {} ({})".format(
                        message["errorCode"], message["errorName"],
                        message["errorInstanceId"]))
                if message["type"] != "data" or "value" not in message:
                    raise ValueError("Invalid streaming Query message")
                print(event_data, flush=True)
                data_lines.clear()
            continue
        field, _, value = line.partition(":")
        if field == "data":
            data_lines.append(value[1:] if value.startswith(" ") else value)
    if data_lines:
        raise RuntimeError("SSE stream ended before the event delimiter")

with requests.post('https://{}/api/v2/functions/queries/myQueryFunction/streamingExecute?transactionId=transaction-1&preview=true'.format(HOSTNAME), headers=headers, json=request_body, stream=True) as response:
    response.raise_for_status()
    consume_sse(iter_sse_lines(response.iter_content(chunk_size=8192)))

Java

Apache HttpClient 4.5와 Jackson Databind 2.x 및 그에 맞는 Core/Annotations 의존성이 필요합니다. StreamingExecuteQueryExample.java 파일에서 사용합니다. Java SE에는 JSON 파서가 포함되어 있지 않으므로 Jackson으로 이벤트 종류를 읽되 원래 JSON을 다시 직렬화하지 않습니다.

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;

public class StreamingExecuteQueryExample {
    private static final ObjectMapper JSON = new ObjectMapper();

    private static void consumeSse(BufferedReader reader) throws IOException {
        StringBuilder data = new StringBuilder();
        boolean hasData = false;
        boolean firstLine = true;
        String line;
        while ((line = reader.readLine()) != null) {
            if (firstLine && !line.isEmpty() && line.charAt(0) == '\uFEFF') {
                line = line.substring(1);
            }
            firstLine = false;
            if (line.isEmpty()) {
                if (hasData) {
                    String eventData = data.toString();
                    JsonNode message = JSON.readTree(eventData);
                    String type = message.path("type").asText();
                    if ("error".equals(type)) {
                        throw new IOException(message.path("errorCode").asText()
                                + ": " + message.path("errorName").asText()
                                + " (" + message.path("errorInstanceId").asText() + ")");
                    }
                    if (!"data".equals(type) || !message.has("value")) {
                        throw new IOException("Invalid streaming Query message");
                    }
                    System.out.println(eventData);
                    data.setLength(0);
                    hasData = false;
                }
            } else if (line.equals("data") || line.startsWith("data:")) {
                int start = line.equals("data") ? line.length() : 5;
                if (start < line.length() && line.charAt(start) == ' ') {
                    start++;
                }
                if (hasData) data.append('\n');
                data.append(line, start, line.length());
                hasData = true;
            }
        }
        if (hasData) throw new IOException("SSE stream ended before the event delimiter");
    }

    public static void main(String[] args) throws Exception {
        String HOSTNAME = System.getenv("FOUNDRY_HOST");
        String TOKEN = System.getenv("FOUNDRY_TOKEN");

        try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
            HttpPost request = new HttpPost("https://" + HOSTNAME + "/api/v2/functions/queries/myQueryFunction/streamingExecute?transactionId=transaction-1&preview=true");
            request.addHeader("content-type", "application/json");
            request.addHeader("authorization", String.format("Bearer %s", TOKEN));
            request.setEntity(new StringEntity("{\"latestVersionResolution\":\"PUBLISH_TIME\",\"parameters\":{\"price\":29.99},\"version\":\"1.2.3\",\"branch\":\"ri.branch..branch.d827184f-ee0e-4351-8b70-efbe51e07252\",\"ontology\":\"example-ontology\"}", ContentType.APPLICATION_JSON));
            try (CloseableHttpResponse response = httpClient.execute(request)) {
                int status = response.getStatusLine().getStatusCode();
                if (status < 200 || status >= 300) {
                    throw new IOException("HTTP " + status);
                }
                try (BufferedReader reader = new BufferedReader(new InputStreamReader(
                        response.getEntity().getContent(), StandardCharsets.UTF_8))) {
                    consumeSse(reader);
                }
            }
        }
    }
}

JavaScript

fetch를 제공하는 Node.js의 ES 모듈(.mjs)에서 사용합니다. 입력 스트림 오류도 처리하며 서비스 사용자 토큰을 브라우저로 전달하지 않습니다.

import { Readable } from "node:stream";
import { createInterface } from "node:readline";

const HOSTNAME = process.env.FOUNDRY_HOST;
const TOKEN = process.env.FOUNDRY_TOKEN;

const response = await fetch(`https://${HOSTNAME}/api/v2/functions/queries/myQueryFunction/streamingExecute?transactionId=transaction-1&preview=true`, {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${TOKEN}`,
    },
    body: '{"latestVersionResolution":"PUBLISH_TIME","parameters":{"price":29.99},"version":"1.2.3","branch":"ri.branch..branch.d827184f-ee0e-4351-8b70-efbe51e07252","ontology":"example-ontology"}'
 });
if (!response.ok) {
    await response.body?.cancel();
    throw new Error(`HTTP ${response.status}`);
}
const input = Readable.fromWeb(response.body);
input.setEncoding("utf8");
const lines = createInterface({ input, crlfDelay: Infinity });
let inputError;
input.on("error", (error) => { inputError = error; lines.close(); });
let firstLine = true;
let dataLines = [];
try {
    for await (let line of lines) {
        if (firstLine && line.startsWith("\uFEFF")) line = line.slice(1);
        firstLine = false;
        if (line === "") {
            if (dataLines.length > 0) {
                const eventData = dataLines.join("\n");
                const message = JSON.parse(eventData);
                if (message.type === "error") {
                    throw new Error(`${message.errorCode}: ${message.errorName} (${message.errorInstanceId})`);
                }
                if (message.type !== "data" || !Object.hasOwn(message, "value")) {
                    throw new Error("Invalid streaming Query message");
                }
                console.log(eventData);
                dataLines = [];
            }
        } else if (line === "data" || line.startsWith("data:")) {
            let value = line === "data" ? "" : line.slice(5);
            if (value.startsWith(" ")) value = value.slice(1);
            dataLines.push(value);
        }
    }
    if (inputError) throw inputError;
    if (dataLines.length > 0) throw new Error("SSE stream ended before the event delimiter");
} finally {
    lines.close();
    input.destroy();
}

PowerShell

PowerShell 7에서 실행합니다. 스크립트 범위의 ErrorActionPreference를 Stop으로 설정합니다. ResponseHeadersRead에서는 HttpClient 제한 시간이 헤더 수신까지 적용되므로 본문 읽기 제한 시간과 사용자 취소는 애플리케이션에서 별도로 관리합니다.

$ErrorActionPreference = 'Stop'
$TOKEN = $env:FOUNDRY_TOKEN
$HOSTNAME = $env:FOUNDRY_HOST
$body = @'
{"latestVersionResolution":"PUBLISH_TIME","parameters":{"price":29.99},"version":"1.2.3","branch":"ri.branch..branch.d827184f-ee0e-4351-8b70-efbe51e07252","ontology":"example-ontology"}
'@
$client = [System.Net.Http.HttpClient]::new()
$request = $null
$response = $null
$stream = $null
$reader = $null
try {
    $request = [System.Net.Http.HttpRequestMessage]::new(
        [System.Net.Http.HttpMethod]::Post, "https://$HOSTNAME/api/v2/functions/queries/myQueryFunction/streamingExecute?transactionId=transaction-1&preview=true"
    )
    $request.Headers.Authorization = [System.Net.Http.Headers.AuthenticationHeaderValue]::new("Bearer", $TOKEN)
    $request.Content = [System.Net.Http.StringContent]::new($body, [System.Text.Encoding]::UTF8, "application/json")
    $response = $client.SendAsync($request, [System.Net.Http.HttpCompletionOption]::ResponseHeadersRead).GetAwaiter().GetResult()
    [void] $response.EnsureSuccessStatusCode()
    $stream = $response.Content.ReadAsStreamAsync().GetAwaiter().GetResult()
    $reader = [System.IO.StreamReader]::new($stream, [System.Text.Encoding]::UTF8, $true)
    $dataLines = [System.Collections.Generic.List[string]]::new()
    while ($null -ne ($line = $reader.ReadLine())) {
        if ($line -ceq "") {
            if ($dataLines.Count -gt 0) {
                $eventData = [string]::Join("`n", $dataLines)
                $message = ConvertFrom-Json -InputObject $eventData -AsHashtable
                if ($message["type"] -ceq "error") {
                    throw "$($message["errorCode"]): $($message["errorName"]) ($($message["errorInstanceId"]))"
                }
                if ($message["type"] -cne "data" -or -not $message.ContainsKey("value")) {
                    throw "Invalid streaming Query message"
                }
                [Console]::WriteLine($eventData)
                $dataLines.Clear()
            }
        }
        elseif ($line -ceq "data" -or $line.StartsWith("data:", [System.StringComparison]::Ordinal)) {
            $value = if ($line -ceq "data") { "" } else { $line.Substring(5) }
            if ($value.StartsWith(" ")) { $value = $value.Substring(1) }
            $dataLines.Add($value)
        }
    }
    if ($dataLines.Count -gt 0) { throw "SSE stream ended before the event delimiter" }
}
finally {
    foreach ($resource in @($reader, $stream, $response, $request, $client)) {
        if ($null -ne $resource) { $resource.Dispose() }
    }
}

데이터 뒤에 오류가 오는 스트림

상품 배치 일부를 반환한 뒤 실패하는 가상 스트림입니다. 빈 줄이 각 이벤트의 끝입니다. 이 예시는 전송 형식을 보여 주며, 위 요청의 가상 가격 함수와는 별도의 출력 시나리오입니다.

data: {"type":"data","value":[{"productId":"SKU-001","price":29.99}]}

data: {"type":"error","errorCode":"INVALID_ARGUMENT","errorName":"QueryRuntimeError","errorInstanceId":"3f8a9c7b-2e4d-4a1f-9b8c-7d6e5f4a3b2c","errorDescription":"Division by zero","parameters":{}}

오류 이벤트의 errorCode, errorName, errorInstanceId는 필수이고 추가 설명과 매개변수는 선택입니다. 이미 받은 배치를 사용할 수 있는지는 애플리케이션의 처리 정책에 따라 판단하되 전체 실행을 성공으로 표시하지 않습니다.