REST API 연결과 코드 기반 수집 및 내보내기

외부 시스템이 HTTP 기반 REST API를 제공하면 Foundry에서 웹훅이나 Python 외부 변환으로 호출할 수 있습니다. 선택 기준은 사용자가 누른 버튼에 즉시 반응하는 호출인지, 데이터를 수집하거나 내보내는 파이프라인인지입니다.

웹훅용 REST API 소스

REST API 소스에는 호출할 도메인, 도메인별 인증과 필요한 포트를 설정합니다. 이 소스 유형은 자체 동기화나 내보내기 기능을 제공하지 않습니다. REST 데이터 수집과 내보내기는 외부 변환으로 구현합니다. 이전 magritte-rest-v2 플러그인은 사용하지 않고, 기존 연결은 현재 REST API 소스의 웹훅이나 외부 변환으로 이전합니다.

https://my-domain.com은 다음 화면에서 사용하는 예시 도메인입니다. 실제 대상 도메인과 그 서비스가 요구하는 인증을 입력합니다.

외부 변환의 준비 조건

Python Code Repository에 transforms-external-systems 라이브러리를 추가하고 소스를 가져옵니다. 소스에서 HTTPS 연결과 사전에 구성한 HTTP 클라이언트를 얻고, 비밀 정보는 get_secret으로 읽습니다. 인터넷에서 직접 접근할 수 없는 API도 Foundry 워커와 에이전트 프록시 외부 통신 정책을 통해 접근할 수 있습니다.

Foundry 데이터셋, 미디어 집합이나 스트림을 같은 변환의 입력으로 사용하려면 소스의 Connection settings > Export configuration에서 Enable exports to this source를 켜야 합니다. 허용할 조직과 보안 표지를 지정하며, 해당 조직과 보안 표지를 제거할 권한이 필요합니다. 실제 요청 본문에 입력 데이터를 보내지 않더라도 같은 실행 작업에서 외부 연결을 사용할 수 있으므로 이 제어가 적용됩니다. 소스에 등록된 가상 테이블을 Python 변환에서 쓰는 경우에도 필요합니다. 이 보안 설정은 소스에 내장 내보내기 기능을 추가하는 설정과 다릅니다.

아래 예제는 서로 다른 변환 모듈입니다. 모듈 위쪽에 정의한 RID, 데이터셋 경로, URL과 식별자는 가상 예시값이므로 실제 소스와 API에 맞게 수정합니다. 클라이언트 비밀 정보와 개인 키는 코드에 넣지 않고 소스의 추가 비밀 정보로 보관합니다.

OAuth Client Credentials로 수집

Client Credentials Grant는 사용자 개입 없이 서비스 사이에서 인증할 때 사용합니다. 클라이언트는 토큰 엔드포인트에 자격 증명을 보내 짧은 수명의 액세스 토큰을 받고, 후속 리소스 요청의 Authorization 헤더에 Bearer 토큰을 넣습니다.

토큰 도메인과 리소스 API 도메인을 모두 소스에 구성합니다. 토큰 요청은 application/x-www-form-urlencoded 본문을 사용합니다.

인증 서버에 따라 audience나 resource 같은 추가 매개변수를 요구할 수 있습니다. 해당 서버가 요구하는 경우 토큰 요청의 data에 추가합니다.

페이지를 순회하는 수집 예제

이 예제의 리소스 API는 각 응답의 data 배열에 행을 반환하고, 다음 페이지가 있으면 nextPageToken을 반환하는 인터페이스입니다. 요청은 pageSize와 pageToken을 사용합니다. 다른 API에서는 실제 매개변수명과 응답 구조에 맞게 바꿉니다.

SOURCE_RID와 OUTPUT_PATH를 실제 소스 RID와 출력 경로로 수정하고, TOKEN_URL, RESOURCE_URL, SCOPES를 대상 API에 맞게 설정합니다. 소스에는 additionalSecretClientId와 additionalSecretClientSecret을 등록합니다. 각 리소스 객체는 출력의 문자열 열 record_json에 JSON으로 저장하므로 원천마다 다른 필드 집합을 유지하면서 빈 결과도 같은 스키마로 기록할 수 있습니다.

import json

import polars as pl
from transforms.api import Output, transform
from transforms.external.systems import Source, external_systems

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
OUTPUT_PATH = "/Examples/REST/resources"
TOKEN_URL = "https://auth.example.com/oauth/token"
RESOURCE_URL = "https://api.example.com/api/v1/resources"
SCOPES = "resources.read"


@external_systems(api_source=Source(SOURCE_RID))
@transform.using(output=Output(OUTPUT_PATH))
def compute(api_source, output):
    client = api_source.get_https_connection().get_client()
    token_response = client.post(
        TOKEN_URL,
        data={
            "grant_type": "client_credentials",
            "client_id": api_source.get_secret("additionalSecretClientId"),
            "client_secret": api_source.get_secret("additionalSecretClientSecret"),
            "scope": SCOPES,
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    token_response.raise_for_status()
    token_body = token_response.json()
    headers = {"Authorization": f"Bearer {token_body['access_token']}"}
    rows = []
    page_token = None
    while True:
        params = {"pageSize": 100}
        if page_token is not None:
            params["pageToken"] = page_token
        response = client.get(
            RESOURCE_URL,
            headers=headers,
            params=params,
        )
        response.raise_for_status()
        body = response.json()
        rows.extend({"record_json": json.dumps(item, ensure_ascii=False)} for item in body["data"])
        page_token = body.get("nextPageToken")
        if not page_token:
            break
    output.write_table(pl.DataFrame(rows, schema={"record_json": pl.String}))

모든 페이지의 결과가 비어 있으면 record_json: String 스키마의 빈 테이블을 출력합니다. 후속 변환에서는 실제 리소스의 JSON 스키마에 따라 필요한 필드를 열로 추출합니다.

토큰은 만료됩니다. 이 예제는 한 토큰의 유효기간 안에 끝나는 수집을 전제로 합니다. 긴 수집에서는 토큰 응답의 expires_in으로 만료 시점을 계산하고 실행 중 새 토큰을 발급받아야 합니다. 토큰과 클라이언트 비밀 정보를 URL이나 로그에 기록하지 않습니다. Foundry API 자체를 호출하는 경우에는 Foundry 연결의 토큰 엔드포인트와 애플리케이션 구성을 사용합니다.

JWT 클라이언트 증명으로 NetSuite 계정 갱신

일부 OAuth 서버는 공유 비밀 정보 대신 개인 키로 서명한 JWT 클라이언트 증명을 요구합니다. NetSuite가 이 방식에 해당합니다. 클라이언트 ID, 인증서 ID와 개인 키를 소스의 additionalSecretClientId, additionalSecretCertificateId, additionalSecretPrivateCertificate에 저장합니다. 등록한 인증서와 그 개인 키가 서로 대응해야 합니다.

NetSuite 토큰 요청의 경로는 /services/rest/auth/oauth2/v1/token입니다. 다음 필드를 form 본문에 보냅니다.

JWT의 헤더에는 typ=JWT, 서명 알고리즘 alg와 등록한 인증서 ID인 kid가 들어갑니다. NetSuite는 PS256, PS384, PS512, ES256, ES384, ES512를 허용합니다. 아래 예제의 ES256에는 그 알고리즘에 맞는 EC 개인 키가 필요하며 Python 환경에는 PyJWT와 cryptography를 준비합니다.

입력 데이터셋의 스키마는 account_id: String, account_name: String입니다. 기존 계정 이름은 PATCH /services/rest/record/v1/account/{account_id}로 변경합니다. JSON 본문의 필드는 acctName이며, 문자열을 직접 이어 붙이지 않고 JSON 직렬화를 사용합니다. PATCH에서 생략한 필드는 그대로 유지하고 읽기 전용 필드는 변경할 수 없습니다.

소스 RID, 입력 계정 목록 경로와 출력 이력 경로를 모듈 상수에 지정합니다. 소스의 기본 URL은 해당 계정의 SuiteTalk API URL로 구성합니다. 입력 계정에 대한 수정 권한과 앞서 설명한 소스 내보내기 권한이 필요합니다.

from datetime import datetime, timedelta, timezone
from urllib.parse import quote, urljoin

import jwt
import polars as pl
from transforms.api import Input, Output, transform
from transforms.external.systems import Source, external_systems

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
ACCOUNT_UPDATES_PATH = "/Examples/NetSuite/account-updates"
RESULT_PATH = "/Examples/NetSuite/update-results"


@external_systems(netsuite=Source(SOURCE_RID))
@transform.using(
    account_updates=Input(ACCOUNT_UPDATES_PATH),
    output=Output(RESULT_PATH),
)
def update_account_names(netsuite, account_updates, output):
    connection = netsuite.get_https_connection()
    client = connection.get_client()
    token_url = urljoin(connection.url, "/services/rest/auth/oauth2/v1/token")
    now = datetime.now(timezone.utc)
    assertion = jwt.encode(
        {
            "iss": netsuite.get_secret("additionalSecretClientId"),
            "scope": "rest_webservices",
            "aud": token_url,
            "iat": now,
            "exp": now + timedelta(minutes=59),
        },
        netsuite.get_secret("additionalSecretPrivateCertificate"),
        algorithm="ES256",
        headers={"kid": netsuite.get_secret("additionalSecretCertificateId"), "typ": "JWT"},
    )
    token_response = client.post(
        token_url,
        data={
            "grant_type": "client_credentials",
            "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
            "client_assertion": assertion,
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    token_response.raise_for_status()
    token = token_response.json()["access_token"]
    responses = []
    for row in account_updates.polars().iter_rows(named=True):
        account_id = row["account_id"]
        url = urljoin(
            connection.url,
            "/services/rest/record/v1/account/" + quote(account_id, safe=""),
        )
        response = client.patch(
            url,
            json={"acctName": row["account_name"]},
            headers={"Authorization": f"Bearer {token}"},
        )
        responses.append({
            "account_id": account_id,
            "response_status": response.status_code,
            "response": response.text,
        })
    output.write_table(pl.DataFrame(responses, schema={
        "account_id": pl.String,
        "response_status": pl.Int32,
        "response": pl.String,
    }))

출력은 각 PATCH의 HTTP 상태와 실제 응답 본문을 담는 호출 이력입니다. 실패한 HTTP 응답도 이력에 포함되므로 빌드 성공만으로 모든 계정이 갱신되었다고 판단하면 안 됩니다. 성공한 갱신은 일반적으로 본문 없는 204 응답이며, 오류는 상태 코드와 응답 본문으로 확인합니다. 네트워크 예외는 변환을 실패시킬 수 있고 이미 완료한 외부 갱신은 Foundry 출력 실패로 되돌아가지 않습니다. 응답 본문을 저장하는 데이터셋도 필요한 접근 권한으로 제한합니다.

사용자 정의 서버 인증서를 사용하는 클라이언트

온프레미스 서버나 TLS 검사 환경의 자체 서명 인증서는 소스의 신뢰할 서버 인증서에 추가해야 합니다. 외부 변환의 기본 HTTPS 클라이언트에는 보통 이 인증서 구성이 자동 적용됩니다. 별도 Python 클라이언트가 REQUESTS_CA_BUNDLE을 사용하면 기본 CA 묶음과 소스의 인증서 묶음을 합친 파일을 지정해야 할 수 있습니다. 인증서 검증을 끄지 않습니다.

다음 예제는 Office365-REST-Python-Client로 SharePoint Online의 사이트 제목을 읽습니다. 인증은 Microsoft Entra 애플리케이션의 인증서 방식이고 서버 신뢰는 CA 묶음으로 처리합니다. 온프레미스 SharePoint Server에는 그 서버가 지원하는 인증 방식을 별도로 구성해야 하며 이 Entra 인증 구성을 그대로 적용한다고 가정하면 안 됩니다.

모듈 상수의 SOURCE_RID와 OUTPUT_PATH를 실제 소스와 출력 경로로 수정합니다. SITE_URL, TENANT, CLIENT_ID, CERT_THUMBPRINT는 사용할 사이트와 Entra 애플리케이션 및 인증서에 맞게 설정합니다. KEY_ENCRYPTED는 개인 키가 암호화되어 있을 때 True로 지정합니다.

기존 신뢰 묶음은 REQUESTS_CA_BUNDLE, CURL_CA_BUNDLE, Requests 기본 CA 묶음 순서로 선택합니다. 이 예제에서 CA 묶음 경로는 디렉터리가 아닌 파일을 가리켜야 합니다.

PEM 개인 키는 소스의 additionalSecretPrivateKey에 저장합니다. 암호화한 키는 additionalSecretPrivateKeyPassphrase도 준비합니다. 임시 CA 파일은 요청이 끝난 뒤 삭제하고 프로세스의 원래 환경 변수 값을 복원합니다.

import os
import tempfile
from pathlib import Path

import polars as pl
from office365.sharepoint.client_context import ClientContext
from requests.utils import DEFAULT_CA_BUNDLE_PATH
from transforms.api import Output, transform
from transforms.external.systems import Source, external_systems

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
OUTPUT_PATH = "/Examples/SharePoint/site-title"
SITE_URL = "https://contoso.sharepoint.com/sites/example"
TENANT = "contoso.onmicrosoft.com"
CLIENT_ID = "00000000-0000-0000-0000-000000000002"
CERT_THUMBPRINT = "0000000000000000000000000000000000000000"
KEY_ENCRYPTED = False


@external_systems(sharepoint=Source(SOURCE_RID))
@transform.using(output=Output(OUTPUT_PATH))
def compute(sharepoint, output):
    original_bundle = os.environ.get("REQUESTS_CA_BUNDLE")
    base_bundle = original_bundle or os.environ.get("CURL_CA_BUNDLE") or DEFAULT_CA_BUNDLE_PATH
    default_ca = Path(base_bundle).read_bytes()
    source_ca = Path(sharepoint.server_certificates_bundle_path).read_bytes()
    passphrase = None
    if KEY_ENCRYPTED:
        passphrase = sharepoint.get_secret("additionalSecretPrivateKeyPassphrase")
    with tempfile.TemporaryDirectory() as directory:
        combined = Path(directory) / "trusted-ca.pem"
        combined.write_bytes(default_ca + b"\n" + source_ca + b"\n")
        os.environ["REQUESTS_CA_BUNDLE"] = str(combined)
        try:
            client = ClientContext(SITE_URL).with_client_certificate(
                tenant=TENANT,
                client_id=CLIENT_ID,
                thumbprint=CERT_THUMBPRINT,
                private_key=sharepoint.get_secret("additionalSecretPrivateKey"),
                passphrase=passphrase,
            )
            web = client.web
            client.load(web)
            client.execute_query()
            title = web.properties["Title"]
        finally:
            if original_bundle is None:
                os.environ.pop("REQUESTS_CA_BUNDLE", None)
            else:
                os.environ["REQUESTS_CA_BUNDLE"] = original_bundle
    output.write_table(pl.DataFrame({"web_title": [title]}))

REQUESTS_CA_BUNDLE은 프로세스 전체 설정이므로 다른 스레드가 동시에 사용하는 요청에도 영향을 줄 수 있습니다. 이 예제는 해당 설정 변경과 요청을 순차 실행합니다.

Foundry 자체 API 호출

외부 시스템에서 Foundry 리소스를 사용하거나 Foundry 위에 별도 애플리케이션을 구축할 때에는 Foundry REST API를 사용합니다. 인증은 OAuth 2.0이며 요청과 응답에는 주로 JSON을 사용합니다. 온톨로지와 모델링 등의 리소스별 엔드포인트가 있습니다. 외부 REST API용 소스와 Foundry API 자체의 접근 권한 및 엔드포인트를 구분합니다.