Compute Module Python SDK

foundry-compute-modules는 함수 등록, 작업 수신과 결과 전송, 인증 보조 함수 및 SLS 로그를 제공합니다.

python -m pip install foundry-compute-modules

함수 서명과 시작

첫 인자는 호출 문맥인 context, 두 번째 인자는 입력 payload인 event입니다. 타입을 지정하지 않은 context는 딕셔너리이며 작업 ID와 해당 모드에서 제공하는 메타데이터를 담습니다. 반환값은 SDK가 JSON으로 직렬화할 수 있어야 합니다.

from compute_modules.annotations import function

@function
def stringify(context, event) -> str:
    return str(event["value"])

이 코드를 app.py로 저장하여 python app.py로 실행합니다. 데코레이터 방식은 SDK가 시작 루틴을 등록하므로 별도의 start 호출을 추가하지 않습니다. Foundry 전체에서 호출하려면 모듈이 Responsive가 된 뒤 Functions에서 발견된 명세를 Import합니다.

입력과 반환 타입을 지정하면 함수 스키마를 추론할 수 있습니다. TypedDict는 사전 접근, dataclass는 속성 접근을 사용합니다.

TypedDict 계산 함수

from typing import TypedDict
from compute_modules.annotations import function

class CalculateInput(TypedDict):
    x: float
    y: float
    operation: str

class CalculateOutput(TypedDict):
    result: float
    operation: str

@function
def calculate(context, event: CalculateInput) -> CalculateOutput:
    x, y, operation = event["x"], event["y"], event["operation"]
    if operation == "add":
        result = x + y
    elif operation == "subtract":
        result = x - y
    elif operation == "multiply":
        result = x * y
    elif operation == "divide":
        if y == 0:
            raise ValueError("The divisor must not be zero")
        result = x / y
    else:
        raise ValueError("Unsupported operation")
    return {"result": result, "operation": operation}

지원하지 않는 연산과 0으로 나누기는 예외로 전달합니다.

dataclass 입력

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

@dataclass
class EventInput:
    name: str
    timestamp: datetime.datetime
    value: float

@function
def process_event(context, event: EventInput) -> str:
    return f"Processed {event.name} at {event.timestamp} with value {event.value}"

입력의 최상위 타입은 TypedDict, dataclass 또는 필드와 생성자 타입 힌트가 일치하는 클래스여야 합니다. 단일 숫자도 그 숫자를 필드로 가진 입력 타입에 넣습니다.

스트리밍

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

class GenerateInput(TypedDict):
    count: int

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

결과 타입은 Array로 등록되지만 각 원소의 JSON을 순차 전송합니다. generator 전체는 JSON으로 직렬화할 수 없으므로 streaming=True가 필요합니다. 일반 list도 streaming을 켜면 원소별로 전송하며, 끄면 전체 결과를 한 JSON 값으로 보냅니다. dict는 스트리밍 대상 iterable에서 제외됩니다.

애플리케이션 권한과 토큰

Application permissions는 서비스 사용자 권한으로 동작합니다. 요청한 최종 사용자의 데이터 권한을 자동으로 대신 적용하는 모드가 아닙니다. FOUNDRY_HOST에는 프로토콜을 제외한 호스트를 지정하고 Foundry 호스트로 나가는 소스를 연결합니다.

import os
from compute_modules.auth import (
    RefreshingOauthToken,
    retrieve_third_party_id_and_creds,
)

client_id, client_secret = retrieve_third_party_id_and_creds()
refreshing_token = RefreshingOauthToken(
    hostname=os.environ["FOUNDRY_HOST"],
    scope=["api:datasets-read", "api:datasets-write"],
)

def dataset_token() -> str:
    token = refreshing_token.get_token()
    if not token:
        raise RuntimeError("Unable to obtain an application token")
    return token

보조 함수는 CLIENT_ID와 CLIENT_SECRET을 읽습니다. get_token()을 호출할 때 토큰이 없거나 기본 갱신 간격인 1,800초를 지났으면 다시 가져옵니다. 별도 타이머가 30분마다 요청하는 방식은 아닙니다. 인증 실패로 토큰을 얻지 못할 수 있으므로 위 함수는 그 상태를 실패로 처리합니다.

온톨로지 조회

Developer Console에서 Backend service와 Application permissions를 선택하여 Python OSDK를 생성하고 패키지를 설치합니다. 서비스 사용자에게 대상 객체를 읽을 권한을 부여하며 Compute Module에 그 앱의 client ID와 secret을 설정합니다.

가상 패키지명 demo_python_sdk와 객체 유형 Employee를 사용하는 예입니다. 실제 생성 패키지와 객체 유형 이름으로 바꿉니다.

import os
from demo_python_sdk import FoundryClient, ConfidentialClientAuth
from compute_modules.annotations import function

@function
def get_object(context, event) -> list[str]:
    host = os.environ["FOUNDRY_HOST"]
    auth = ConfidentialClientAuth(
        client_id=os.environ["CLIENT_ID"],
        client_secret=os.environ["CLIENT_SECRET"],
        hostname=host,
        should_refresh=True,
        scopes=["api:ontologies-read"],
    )
    client = FoundryClient(auth=auth, hostname=host)
    return [str(employee) for employee in client.ontology.objects.Employee.take(1)]

최대 한 객체를 읽어 문자열 목록으로 반환하며, 객체가 없으면 실제 빈 목록을 반환합니다. 조회 실패는 예외로 전달합니다. OSDK 객체 접근법은 Python OSDK에서 설명합니다.

데이터셋 파일 읽기와 쓰기

다음 코드는 데이터셋 읽기/쓰기 권한, Application permissions 및 Foundry 호스트의 소스 네트워크 정책이 필요합니다. requests를 의존성에 추가합니다. API는 상대 파일 경로를 받으며 선행 /는 허용하지 않습니다.

import os
from dataclasses import dataclass
from urllib.parse import quote
import requests
from compute_modules.annotations import function
from compute_modules.auth import RefreshingOauthToken

HOST = os.environ["FOUNDRY_HOST"]
CA_PATH = os.environ["DEFAULT_CA_PATH"]
TOKENS = RefreshingOauthToken(
    hostname=HOST,
    scope=["api:datasets-read", "api:datasets-write"],
)

def authorization() -> dict[str, str]:
    token = TOKENS.get_token()
    if not token:
        raise RuntimeError("Unable to obtain an application token")
    return {"Authorization": f"Bearer {token}"}

@dataclass
class UploadFileRequest:
    dataset_rid: str
    file_path: str
    file_content: str

@dataclass
class UploadFileResponse:
    status: int

@function
def upload_file(context, event: UploadFileRequest) -> UploadFileResponse:
    rid = quote(event.dataset_rid, safe="")
    path = quote(event.file_path, safe="")
    with requests.post(
        f"https://{HOST}/api/v2/datasets/{rid}/files/{path}/upload",
        params={"transactionType": "APPEND", "branchName": "master"},
        headers={**authorization(), "Content-Type": "application/octet-stream"},
        data=event.file_content.encode("utf-8"),
        verify=CA_PATH,
        timeout=60,
    ) as response:
        response.raise_for_status()
        return UploadFileResponse(status=response.status_code)

@dataclass
class GetFileRequest:
    dataset_rid: str
    file_path: str

@dataclass
class GetFileResponse:
    status: int
    file_content: str

@function
def get_file(context, event: GetFileRequest) -> GetFileResponse:
    rid = quote(event.dataset_rid, safe="")
    path = quote(event.file_path, safe="")
    with requests.get(
        f"https://{HOST}/api/v2/datasets/{rid}/files/{path}/content",
        params={"branchName": "master"},
        headers=authorization(),
        verify=CA_PATH,
        timeout=60,
    ) as response:
        response.raise_for_status()
        return GetFileResponse(
            status=response.status_code,
            file_content=response.content.decode("utf-8"),
        )

쓰기 요청은 master에 새 APPEND 트랜잭션을 만들고 커밋합니다. 읽기는 같은 브랜치의 최신 뷰에서 파일 내용을 가져옵니다. 위 함수는 UTF-8 텍스트용이며 바이너리 파일을 문자열로 디코딩하지 않습니다. 인증/권한 오류, 파일 부재, HTTP 실패와 UTF-8 디코딩 실패를 정상 결과로 감추지 않습니다.

구조화 로그

get_logger()는 세션 ID, 작업 ID와 프로세스 ID를 로그에 포함합니다. 입력 전체, 토큰이나 비밀을 로그에 남기지 않습니다.

import logging
from typing import TypedDict
from compute_modules.annotations import function
from compute_modules.logging import get_logger, set_internal_log_level

set_internal_log_level(logging.INFO)
logger = get_logger(__name__)
logger.setLevel(logging.INFO)

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

@function
def add(context, event: AddInput) -> int:
    try:
        result = event["x"] + event["y"]
        logger.info("Addition completed")
        return result
    except Exception:
        logger.exception("Addition failed")
        raise

SLS 심각도는 DEBUG 0, INFO 1, WARNING 2, ERROR 3, CRITICAL 4입니다. logger.setLevel(logging.INFO)이면 INFO 이상을 기록합니다. Python의 logging 상수 자체 숫자와 SLS 심각도 번호는 구분합니다. SDK 내부 로그 수준은 set_internal_log_level()로 별도 설정하며 기본은 ERROR입니다.

외부 소스

python -m pip install 'foundry-compute-modules[sources]'

import os
from compute_modules.sources_v2 import get_source

source = get_source(os.environ["SOURCE_API_NAME"])

SOURCE_API_NAME은 Data Connection에서 지정하고 Compute Module에 연결한 실제 소스 API 이름입니다. get_source()는 Source 객체를 반환하며 비밀 조회와 구성된 HTTPS 연결에 사용합니다. 소스가 없거나 필요한 extras를 설치하지 않았으면 오류가 발생합니다. 실행 문제는 복제본 진단에서 확인합니다.