REST

Foundry は、REST (representational state transfer) API を公開する外部システムと統合できます。REST API を同期、エクスポート、または対話的に呼び出す場合、それぞれ異なる方法が必要になることがあります。このページでは、REST API と安全かつ効率的に統合するための接続オプションを紹介します。

REST API ソース

REST API ソースは、Foundry アプリケーションからアクションを介して外部システムに直接、対話的な HTTP リクエストを送信するワークフローに使用できます。たとえば、クリックすると Webhook を使用して REST エンドポイントを呼び出すボタンを備えた Workshop アプリケーションを作成し、そのアプリケーションを既存のワークフローやソースシステムに接続できます。

HTTP エンドポイントへの Webhook には、Data Connection の REST API ソースタイプを使用してください。ベース URL、認証、および任意のポートを設定する必要があります。

以下の設定例は、Bearer トークン認証を使用して https://my-domain.com への接続を設定する方法を示しています。

Foundry の Webhook について詳しくはこちら。

Code Repositories の外部トランスフォーム

REST API の呼び出しが必要な同期やエクスポートを設定するには、外部トランスフォームを使用します。Python コードリポジトリにソースをインポートし、API にクエリを実行するカスタムロジックを記述するだけで設定できます。

コードリポジトリから API を呼び出す方法について詳しくはこちら。

例

以下の例は、複雑な外部トランスフォームの一般的なパターンを示しています。

OAuth クライアントクレデンシャルグラント

OAuth クライアントクレデンシャルグラント ↗は、サービス間で非対話的に REST API を呼び出す際の一般的な認証パターンです。クライアントは、OAuth2 サーバーのトークンエンドポイントで client_id と client_secret を有効期間の短いアクセストークンと交換し、その後のリソース API へのすべてのリクエストで、そのトークンを Bearer トークンとして Authorization ヘッダーに含めます。

トランスフォームを記述する前に、OAuth2 トークンエンドポイントのドメインとリソース API のドメインを REST API ソースに設定し、client_id と client_secret をソースの追加のシークレットとして保存してください。トランスフォームにシークレットをハードコードしないでください。

トークンエンドポイントへのリクエストでは、application/x-www-form-urlencoded コンテンツタイプを使用し、少なくとも以下のパラメーターを含める必要があります。

アクセストークンを取得したら、リソース API へのすべてのリクエストの Authorization ヘッダーに含めます。

Authorization: Bearer <access_token>

多くの API は、結果をページ単位で返します。一般的なパターンでは、レスポンスの nextPageToken フィールドで、さらに結果があることを示します。トランスフォームでは、nextPageToken が返されなくなるまでループする必要があります。

基本的な client_id/client_secret の例

以下の例では、アクセストークンをリクエストし、ページネーションを使用するリソース API を呼び出します。汎用的なプレースホルダーを使用しているため、OAuth2 で保護された任意の REST API に適用できます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import logging

import pandas as pd
from transforms.api import Output, transform_pandas
from transforms.external.systems import external_systems, Source, ResolvedSource

logger = logging.getLogger(__name__)


@external_systems(
    api_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<output_dataset_rid>"),
)
def compute(api_source: ResolvedSource) -> pd.DataFrame:
    base_url = api_source.get_https_connection().url
    client = api_source.get_https_connection().get_client()

    client_id = api_source.get_secret("additionalSecretClientId")
    client_secret = api_source.get_secret("additionalSecretClientSecret")

    token_response = client.post(
        base_url + "/oauth/token",
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
            "scope": "<space-separated-scopes>",
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    token_response.raise_for_status()
    access_token = token_response.json()["access_token"]

    auth_headers = {"Authorization": f"Bearer {access_token}"}

    results = []
    page_token = None
    while True:
        params = {"pageSize": 100}
        if page_token:
            params["pageToken"] = page_token

        response = client.get(
            base_url + "/api/v1/<resource>",
            headers=auth_headers,
            params=params,
        )
        response.raise_for_status()
        body = response.json()

        results.extend(body.get("data", []))

        page_token = body.get("nextPageToken")
        if not page_token:
            break

        logger.info(f"Fetched {len(results)} items so far, continuing to next page.")

    return pd.DataFrame(results)

このパターンを適用するには、トークンエンドポイントのパス、リソース URL、および scope の値を、対象の API が必要とするものに置き換えます。一部の OAuth2 サーバーでは、audience や resource などの追加パラメーターが必要です。これらをトークンリクエストの data 辞書に追加します。

対象の API が Foundry API 自体である場合（トランスフォームからの Foundry 間の呼び出し）、Foundry 固有のトークンエンドポイント、スコープ、およびセットアップ手順については、コードから Foundry API を呼び出すを参照してください。

JWT クライアントアサーションを使用するバリエーション (NetSuite)

NetSuite を含む一部の OAuth2 サーバーでは、通常の client_secret の代わりに、署名付き JWT クライアントアサーションを使用してクライアントを認証する必要があります。以下の例では、POST /account ↗ エンドポイントを使用して、勘定科目の入力データセットから NetSuite の勘定科目名を更新します。このグラントタイプを有効にするには、client_id、certificate_id、および certificate_private_key を追加のシークレットとしてソースに追加します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
from transforms.api import (
    transform,
    Output,
    Input,
    TransformInput,
    TransformOutput,
    TransformContext,
)
from transforms.external.systems import external_systems, Source, ResolvedSource
import datetime
import jwt
from urllib.parse import urljoin
import logging

logger = logging.getLogger(__name__)

@external_systems(
    netsuite_rest_source=Source("<source_rid>")
)
@transform(
    output=Output("<dataset_rid>"),
    account_updates=Input("<dataset_rid>"),  # スキーマ [account_id: String, account_name: String] を持つデータセット
)
def update_account_names(
    netsuite_rest_source: ResolvedSource,
    account_updates: TransformInput,
    output: TransformOutput,
    ctx: TransformContext,
):
    # --- 接続とシークレットを設定する ---
    base_url = netsuite_rest_source.get_https_connection().url
    client = netsuite_rest_source.get_https_connection().get_client()
    client_id = netsuite_rest_source.get_secret("additionalSecretClientId")
    certificate_id = netsuite_rest_source.get_secret("additionalSecretCertificateId")
    certificate_private_key = netsuite_rest_source.get_secret("additionalSecretPrivateCertificate")

    # --- ヘルパー：JWT トークンを作成する ---
    def make_jwt_token(
        url, client_id, certificate_id, certificate_private_key, lifetime_in_minutes=59
    ):
        current_timestamp = datetime.datetime.now()
        expiration = current_timestamp + datetime.timedelta(minutes=lifetime_in_minutes)

        payload = {
            "iss": client_id,
            "scope": "rest_webservices",
            "aud": url,
            "iat": current_timestamp,
            "exp": expiration,
        }

        additional_headers = {
            "kid": certificate_id,
        }

        return jwt.encode(
            payload,
            certificate_private_key,
            algorithm="ES256",
            headers=additional_headers,
        )

    # --- ヘルパー：OAuth2 アクセストークンを取得する ---
    def get_oauth2_access_token():
        url = urljoin(base_url, "/services/rest/auth/oauth2/v1/token")
        payload = {
            "grant_type": "client_credentials",
            "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
            "client_assertion": make_jwt_token(
                url,
                client_id,
                certificate_id,
                certificate_private_key,
            ),
        }
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        response = client.post(url, data=payload, headers=headers)
        return response.json()["access_token"]

    # --- 更新用のデータを準備する ---
    account_update_data = [
        {
            "account_id": row.account_id,
            "payload": f'{{"acctName": "{row.account_name}"}}',
        }
        for row in account_updates.dataframe().collect()
    ]

    # --- アカウントを更新する ---
    token = get_oauth2_access_token()
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}",
    }

    responses = []
    for account in account_update_data:
        account_id = account["account_id"]
        payload = account["payload"]
        logger.info(f"Updating account: {account_id} with payload {payload}")
        url = urljoin(base_url, f"/services/rest/record/v1/account/{account_id}")
        response = client.patch(url, data=payload, headers=headers)
        responses.append(
            {
                "account_id": account_id,
                "response_status": response.status_code,
                "response": response.text,
            }
        )

    output.write_dataframe(ctx.spark_session.createDataFrame(responses))

自己署名サーバー証明書を使用する

オンプレミスシステムでは、接続が信頼されるようにソースに追加する必要がある自己署名サーバー証明書を使用していることがあります。通常、これらの証明書は、外部トランスフォームで提供される組み込み HTTPS クライアントに自動的に追加されます。ただし、一部の Python クライアントは REQUESTS_CA_BUNDLE 環境変数を使用することがあります。この場合は、この変数を上書きする必要があります。

以下の例は、SharePoint 用 Python クライアント ↗ Office365-REST-Python-Client を使用してオンプレミスの SharePoint ソースからデータを読み取るために、REQUESTS_CA_BUNDLE を上書きする方法を示しています。この上書きは、このクライアントを使用するために必要な手順です。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from pyspark.sql import DataFrame
from transforms.api import Output, transform, lightweight
from transforms.external.systems import external_systems, Source
import pandas as pd
import polars as pl
import tempfile
import os
from office365.sharepoint.client_context import ClientContext

@lightweight
@external_systems(
    sharepoint_rest=Source("<source_rid>")
)
@transform(
    output=Output("<dataset_rid>"),
)
def compute(ctx,  output, sharepoint_rest) -> DataFrame:

    # 1. デフォルトの証明書の環境変数にカスタム証明書を追加します
    cert_file = tempfile.NamedTemporaryFile(delete=False)
    with open(cert_file.name, 'w') as tmp_f:
        with open(os.environ.get("REQUESTS_CA_BUNDLE"), 'r') as ca_f:
            with open(sharepoint_rest.server_certificates_bundle_path, 'r') as source_ca_f:
                tmp_f.write(ca_f.read())
                tmp_f.write(source_ca_f.read())
    cert_file.close()

    os.environ["REQUESTS_CA_BUNDLE"] = cert_file.name # REQUESTS_CA_BUNDLE にソースの自己署名証明書が含まれるようになりました


    # 2. クライアント証明書認証を使用して Sharepoint に接続します。
    client = ClientContext("<sharepoint_url>").with_client_certificate(
        tenant="<tenant_id>",
        client_id="<client_id>",
        thumbprint="<thumbprint>",
        private_key=sharepoint_rest.get_secret("additionalSecretPrivateKey"),
        passphrase=sharepoint_rest.get_secret("additionalSecretPrivateKeyPassphrase"), # 任意、秘密鍵がパスワードで暗号化されている場合
    )

    # 3. Web タイトルを取得し、DataFrame として返します
    current_web = client.web
    client.load(current_web)
    client.execute_query()

    data = [{"web_title": current_web.properties['Title']}]
    output.write_table(pl.from_pandas(pd.DataFrame.from_records(data)))

Foundry REST API

Foundry プラットフォーム上でアプリケーションを構築する場合は、Foundry REST API を使用します。Foundry API は認証に OAuth 2.0 プロトコルを使用し、主に JSON のリクエストとレスポンスを使用します。また、オントロジーおよび Modeling のリソースをサポートしています。

Foundry API について詳しくはこちら。