Python SDK

Compute Module 向けの Python SDK を使用すると、Foundry と統合するデプロイ済み関数を作成できます。この SDK は、関数登録用のデコレーター、認証ヘルパー、構造化ログ、および Foundry リソースを操作するためのユーティリティを提供します。

Python SDK は、次の主要なモジュールを提供します。

関数を定義する

基本的な関数

@function で注釈を付けたすべての関数は、Compute Module 内のエンドポイントとして自動的に登録されます。

Copied!
1
2
3
4
5
6
7
8
9
from compute_modules.annotations import function

@function
def my_function(context, event):
    """
    context： メタデータ（トークン、資格情報、ジョブIDなど）を含む Dict
    event： 関数パラメーターを含む Dict
    """
    return str(event['value'])

関数シグネチャの要件：

1. context: 呼び出しに関するメタデータ（ジョブID、資格情報など）
2. event: 入力ペイロード（TypedDict または dataclass で型を指定できます）

戻り値は JSON にシリアライズ可能である必要があります。

型付き関数（推奨）

型付きの入力と出力を使用すると、スキーマの自動推論が可能になり、IDE のサポートも向上します。

TypedDict の使用：

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
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:
    ops = {
        'add': lambda a, b: a + b,
        'subtract': lambda a, b: a - b,
        'multiply': lambda a, b: a * b,
        'divide': lambda a, b: a / b if b != 0 else float('inf')
    }
    result = ops.get(event['operation'], lambda a, b: 0)(event['x'], event['y'])
    return {'result': result, 'operation': event['operation']}

dataclass の使用：

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
from dataclasses import dataclass
from compute_modules.annotations import function
import datetime

@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}"

型システムとスキーマの自動推論について詳しくはこちら。

ストリーミング出力

結果セットが大きい場合は、ストリーミングを使用してメモリーの問題を回避し、結果を段階的に返します。

Copied!
1
2
3
4
5
6
7
from compute_modules.annotations import function
from typing import Iterable

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

認証

アプリケーション権限（3PA）

アプリケーション権限（サードパーティアプリケーション認可）を使用して、ユーザーの代理で Foundry リソースにアクセスします。RefreshingOauthToken クラスは、30分ごとにトークンを自動的に更新します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from compute_modules.auth import RefreshingOauthToken, retrieve_third_party_id_and_creds

HOSTNAME = "yourenrollment.palantirfoundry.com"

# クライアントIDとシークレットを取得します
client_id, client_secret = retrieve_third_party_id_and_creds()

# 自動更新トークンを作成します（デフォルトでは30分ごとに更新されます）
refreshing_token = RefreshingOauthToken(
    hostname=HOSTNAME,
    scope=["api:datasets-read", "api:datasets-write"]
)

# トークンを取得します（常に有効なトークンを返します）
access_token = refreshing_token.get_token()

認証モードについて詳しくはこちら。

Foundry リソースを操作する

オントロジー SDK を使用する

Python SDK は、オントロジーのオブジェクトを操作するためにオントロジー SDK と統合されています。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import os
from compute_modules.annotations import function
from compute_modules.auth import retrieve_third_party_id_and_creds
from your_osdk_package import FoundryClient, ConfidentialClientAuth

foundry_url = os.environ["FOUNDRY_URL"]
CLIENT_ID, CLIENT_CREDS = retrieve_third_party_id_and_creds()

@function
def get_object(context, event):
    auth = ConfidentialClientAuth(
        client_id=CLIENT_ID,
        client_secret=CLIENT_CREDS,
        hostname=foundry_url,
        should_refresh=True,
        scopes=[
            "api:ontologies-read",
            "api:ontologies-write",
        ],
    )
    client = FoundryClient(auth=auth, hostname=foundry_url)
    employee = client.ontology.objects.Employee
    return str(employee.take(1))

オントロジー SDK との統合について詳しくはこちら。

データセットを読み書きする

Foundry 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
62
63
64
65
66
67
68
69
70
import logging
import os
import requests
from dataclasses import dataclass
from compute_modules.annotations import function
from compute_modules.auth import RefreshingOauthToken
from compute_modules.logging import get_logger

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

FOUNDRY_URL = os.environ["FOUNDRY_URL"]
BASE_URL = f"https://{FOUNDRY_URL}"
refreshing_token = RefreshingOauthToken(
    hostname=FOUNDRY_URL,
    scope=["api:datasets-read", "api:datasets-write"]
)
CA_PATH = os.environ["DEFAULT_CA_PATH"]

@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:
    """データセットにファイルを書き込みます。"""
    logger.info(f"Uploading file to path: {event.file_path}")
    url = f"{BASE_URL}/api/v2/datasets/{event.dataset_rid}/files/{event.file_path}/upload"
    response = requests.post(
        url,
        params={"transactionType": "APPEND", "branchName": "master"},
        headers={
            "Authorization": f"Bearer {refreshing_token.get_token()}",
            "Content-Type": "application/octet-stream",
        },
        data=event.file_content,
        verify=CA_PATH,
    )
    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:
    """データセットからファイルの内容を読み取ります。"""
    logger.info(f"Getting file from {event.file_path}")
    url = f"{BASE_URL}/api/v2/datasets/{event.dataset_rid}/files/{event.file_path}/content"
    response = requests.get(
        url,
        headers={"Authorization": f"Bearer {refreshing_token.get_token()}"},
        verify=CA_PATH,
    )
    return GetFileResponse(
        status=response.status_code,
        file_content=response.content.decode('utf-8'),
    )

ログ

この SDK は、セッションID、ジョブID、プロセスIDを自動的に含む構造化ログを提供します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import logging
from compute_modules.logging import get_logger

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

@function
def my_function(context, event) -> str:
    logger.debug(f"Input: {event}")
    try:
        result = process(event)
        logger.info(f"Success: {result}")
        return result
    except Exception as e:
        logger.error(f"Failed: {e}", exc_info=True)
        raise

ログレベル：

Copied!
1
2
3
4
5
logger.debug("Debug logs")     # severity = 0
logger.info("Info logs")       # severity = 1
logger.warning("Warning logs") # severity = 2
logger.error("Error logs")     # severity = 3
logger.critical("Critical logs") # severity = 4

logger.setLevel(logging.INFO) を使用して、最小ログレベルを設定します。重大度が設定したレベル以上のログのみが表示されます。

デバッグとログの表示について詳しくはこちら。

ソースを使用する

Compute Module は、設定済みのソースを通じて外部システムにアクセスできます。この SDK は、ソースを使用するためのユーティリティを提供します。

Copied!
1
2
3
4
from compute_modules.sources_v2 import get_source

# ソースの資格情報にアクセスします
source = get_source("my-source-identifier")

ソースの設定と使用について詳しくはこちら。

GitHub リポジトリ

Python SDK はオープンソースで、GitHub で利用可能です。

• palantir/python-compute-module ↗