Compute Module 함수 등록과 자료형

Functions 모드의 모듈을 다른 Foundry 앱에서 호출하려면 함수를 등록해야 합니다. Functions 화면에서 명세를 직접 입력하거나 SDK가 게시한 명세를 가져올 수 있습니다. HTTP 서버를 연결하는 경우에는 OpenAPI 서버 연동을, 소스에서 명세와 이미지를 만드는 경우에는 Functions CLI를 사용합니다.

화면에서 수동 등록

Functions > Add function을 선택합니다.

입력 필드들은 하나의 JSON 객체로 묶입니다. 예를 들어 arg1, arg2와 arg3를 정의하면 다음과 같이 전달합니다.

{
  "arg1": "hello",
  "arg2": 2,
  "arg3": "1969-07-20"
}

API 이름은 com.<namespace>.computemodules.<MyApiName> 형식입니다. namespace는 소문자이며 특수문자를 포함하지 않고, MyApiName은 camel case이며 특수문자를 포함하지 않습니다. API 이름을 변경하면 기존 호출 코드도 바꿔야 합니다.

Test 탭에서 함수를 호출해 볼 수 있고 Save하면 플랫폼에서 호출할 수 있습니다. Compute Module 함수 버전은 항상 0.0.0이며 수정하면 같은 버전을 덮어씁니다. 이전 버전을 지정하여 호출하는 경로가 아니라 최신 게시 정의를 사용합니다.

명세를 JSON으로 게시

SDK를 사용하지 않는 클라이언트는 POST_SCHEMA_URI에 함수 명세 배열을 게시할 수 있습니다. 다음 명세는 Integer 입력 x와 y를 더한 뒤 String으로 반환하는 add에 대응합니다.

from typing import TypedDict
from compute_modules.annotations import function

class AddInput(TypedDict):
    x: int
    y: int

@function
def add(context, event: AddInput) -> str:
    return str(event["x"] + event["y"])

schemas.json:

[
  {
    "functionName": "add",
    "inputs": [
      {
        "name": "x",
        "dataType": {
          "integer": {},
          "type": "integer"
        },
        "required": true,
        "constraints": []
      },
      {
        "name": "y",
        "dataType": {
          "integer": {},
          "type": "integer"
        },
        "required": true,
        "constraints": []
      }
    ],
    "output": {
      "single": {
        "dataType": {
          "string": {},
          "type": "string"
        }
      },
      "type": "single"
    }
  }
]

사용자 클라이언트의 시작 단계에서 다음처럼 게시할 수 있습니다. SDK 자동 명세 게시와 수동 경로 중 사용하는 방식을 정합니다.

import json
import os
from pathlib import Path
import requests


def publish_schemas() -> None:
    with open("schemas.json", encoding="utf-8") as file:
        schemas = json.load(file)
    token = Path(os.environ["MODULE_AUTH_TOKEN"]).read_text().strip()
    with requests.post(
        os.environ["POST_SCHEMA_URI"],
        json=schemas,
        headers={"Module-Auth-Token": token, "Content-Type": "application/json"},
        verify=os.environ["DEFAULT_CA_PATH"],
        timeout=30,
    ) as response:
        response.raise_for_status()
        if response.status_code != 204:
            raise RuntimeError(f"Unexpected schema status: {response.status_code}")

functionName은 실제 함수 이름과 일치해야 하며 모든 입력과 출력을 정의합니다. 수신, 재시도와 결과 전송은 사용자 클라이언트 프로토콜에서 설명합니다.

Python의 자동 스키마 추론

아래 Python 타입 및 직렬화 조건은 foundry-compute-modules 0.34.0에 적용됩니다. 이 배포는 Python 3.10 이상, 4.0 미만을 요구합니다.

타입 추론은 등록한 함수의 입력과 반환 annotation을 FunctionSpecs로 바꿉니다. 입력 최상위 타입은 필드가 있는 복합 타입이어야 합니다. 원시값 하나만 입력받더라도 TypedDict나 dataclass의 필드로 감쌉니다.

from typing import TypedDict
from compute_modules.annotations import function

class HelloInput(TypedDict):
    planet: str

@function
def hello(context, event: HelloInput) -> str:
    return "Hello " + event["planet"] + "!"

여러 자료형을 받는 dataclass 예입니다. Date는 날짜, Timestamp는 시점으로 구분합니다.

from dataclasses import dataclass
import datetime
import decimal
from compute_modules.annotations import function

@dataclass
class TypedInput:
    bytes_value: bytes
    bool_value: bool
    date_value: datetime.date
    decimal_value: decimal.Decimal
    float_value: float
    int_value: int
    str_value: str
    datetime_value: datetime.datetime
    other_date_value: datetime.datetime

@function
def date_difference(context, event: TypedInput) -> str:
    difference = event.other_date_value - event.datetime_value
    return f"The difference between the provided dates is {difference}"

일반 클래스도 필드와 생성자의 이름 및 타입 힌트가 일치하면 입력으로 사용할 수 있습니다.

from compute_modules.annotations import function

class GoodInput:
    some_flag: bool
    some_value: int

    def __init__(self, some_flag: bool, some_value: int) -> None:
        self.some_flag = some_flag
        self.some_value = some_value

@function
def get_value(context, event: GoodInput) -> int:
    return event.some_value

다음 조건에서는 추론 또는 입력 변환이 실패합니다.

• 클래스 필드 타입 힌트가 없거나 생성자 인자의 이름과 일치하지 않습니다.
• 생성자 타입 힌트가 없거나 필드 타입과 다릅니다.
• 생성자에 *args 또는 **kwargs를 사용합니다.
• dict에 키/값 타입 매개변수를 지정하지 않습니다.
• 허용되지 않는 Map 키 타입 또는 일반 다중 타입 union을 사용합니다.

context에 타입을 지정한다면 compute_modules.context.QueryContext를 사용합니다. 지정하지 않으면 딕셔너리입니다.

Python 스키마 타입 표

이 표는 스키마 추론 결과입니다. 표에 있는 타입을 Python 값으로 그대로 반환할 때 JSON 전송까지 가능한지는 다음 직렬화 조건을 함께 적용합니다.

Byte, Short, Long과 Double 별칭은 compute_modules.function_registry.types에 있습니다. NewType 표시 자체가 값의 범위나 JSON 전송을 검증하는 것은 아닙니다.

Map 키로는 bytes, bool, Byte, date, Decimal, Double, float, int, Long, Short, str와 datetime 같은 스칼라 타입을 추론할 수 있습니다. JSON 결과의 객체 키로 자동 변환할 수 있는 범위와는 다르므로 키의 전송 표현을 따로 맞춰야 합니다.

필드의 required와 값의 None 허용도 별개입니다. 생성자 기본값 또는 TypedDict의 선택 키는 필드 제공 의무를 정하고, Optional 타입은 값의 선택성을 표현합니다. 반환 annotation이 없으면 SDK는 String 명세를 기본으로 삼으므로 실제 반환값에 맞는 타입을 명시합니다.

일반 결과의 JSON 직렬화

일반 결과는 한 JSON 값으로 직렬화합니다.

스키마가 binary, decimal, set 또는 Struct로 추론되었다고 해서 해당 Python 객체를 그대로 일반 결과로 반환할 수 있는 것은 아닙니다. 반환할 표현을 JSON 값으로 구성하고 등록한 입출력 계약과 맞춥니다. 날짜나 Decimal 같은 Map 키도 일반 결과 encoder가 문자열 키로 자동 변경하지 않습니다.

Timestamp의 전송 형식은 UTC ISO 문자열입니다. 0.34.0의 입력 변환은 Z로 끝나는 문자열을 시간대 정보가 없는 datetime으로 만듭니다. 반면 출력 encoder는 시간대 정보가 없는 datetime을 실행 환경의 로컬 시간으로 해석한 뒤 UTC로 변환합니다. 따라서 UTC 입력값을 그대로 반환하려면 value.replace(tzinfo=datetime.timezone.utc)로 그 값의 시간대를 명시합니다. 이미 시간대가 지정된 datetime은 해당 시간대를 기준으로 UTC 문자열로 변환됩니다.

스트리밍 결과

@function(streaming=True)는 dict를 제외한 iterable의 각 원소를 별도 JSON 값으로 직렬화합니다. 이 경로는 일반 결과용 CustomJSONEncoder 대신 원소별 json.dumps를 사용하므로 각 원소가 그 자체로 JSON에 들어갈 수 있어야 합니다. dataclass, datetime, bytes나 Decimal 원소를 자동 변환하는 것으로 간주하지 않습니다.

from typing import Iterable, TypedDict
from compute_modules.annotations import function

class CountInput(TypedDict):
    count: int

class ValuesInput(TypedDict):
    values: list[str]

@function(streaming=True)
def string_generator(context, event: CountInput) -> Iterable[str]:
    for i in range(event["count"]):
        yield f"string {i}"

@function
def string_list(context, event: CountInput) -> list[str]:
    return [f"string {i}" for i in range(event["count"])]

@function(streaming=True)
def unique_strings(context, event: ValuesInput) -> set[str]:
    return set(event["values"])

첫 함수는 원소를 순차 전송하고 두 번째 함수는 전체 목록을 한 번에 보냅니다. 세 번째는 중복을 없앤 문자열을 전송하되 순서는 보장하지 않습니다. generator를 반환하면서 streaming을 끄면 등록 스키마가 만들어져도 실행 시 전체 직렬화가 실패합니다.

발견한 함수 가져오기

자동 추론한 명세는 모듈을 배포하고 실행하여 Responsive가 된 뒤 화면에 나타납니다. Functions 탭에서 발견한 함수를 확인합니다.

함수를 선택하고 Import를 선택합니다.

함수가 보이지 않으면 복제본 상태와 로그를 확인합니다. 모듈이 아직 실행되지 않거나 명세 게시가 실패한 경우를 구분합니다.

TypeScript v1 함수에서 호출

먼저 Compute Module 함수를 API 이름으로 등록하고 모듈을 실행합니다. TypeScript 함수 저장소의 functions.json에서 enableResourceGeneration을 true로 설정합니다.

Resource imports > Add > Query Functions를 선택합니다.

Compute Module 자체는 특정 온톨로지에 묶이지 않지만 이 가져오기 흐름에서는 온톨로지 하나를 선택해야 합니다. 함수 API 이름을 검색해 선택하고 Confirm selection을 누릅니다.

작업공간을 다시 빌드하여 생성 코드를 갱신합니다.

다음 가상 API 이름은 com.mycustomnamespace.computemodules.Add이며 위에서 정의한 String 반환 add를 호출합니다.

import { Function } from "@foundry/functions-api";
import { add } from "@mycustomnamespace/computemodules";

export class MyFunctions {
  @Function()
  public async myFunction(): Promise<string> {
    return await add({ x: 50, y: 50 });
  }
}

실시간 미리보기에는 Compute Module과 TypeScript 코드가 같은 프로젝트에 있어야 하고 모듈이 실행 중이어야 합니다. 반환 타입은 등록한 함수의 실제 타입과 일치해야 합니다.

이 경로는 function-executor를 거치므로 Compute Module 함수가 5분 미만에 끝나야 합니다. 5분을 넘기는 호출은 시간 초과됩니다. 이 제약을 쿼리를 받지 않는 Pipeline 모드의 전체 실행 수명에 적용하지 않습니다.