Jinba FlowのサンドボックスでPythonを実行する

Pythonコードが必要なステップにはPYTHON_SANDBOX_RUNを使います。必須のcode入力がサンドボックス内で実行されます。最後の式の値が結果となり、ない場合は標準出力に出力したテキストが使われます。data_typeでSTRING（既定）、NUMBER、OBJECTのどれとして解析するか指定します。OBJECTでは最後の式をjson.dumps(...)にしてください（print(json.dumps(...))で出力しても構いません）。辞書やリストをそのまま最後の式にすると、JSONではなくPythonのrepr形式で返るため解析できません。

前のステップから数値を計算する

- id: first
  tool: INPUT_TEXT
  input:
    - name: value
      value: "1"
- id: second
  tool: INPUT_TEXT
  input:
    - name: value
      value: "2"
- id: sum
  tool: PYTHON_SANDBOX_RUN
  needs:
    - first
    - second
  input:
    - name: code
      value: 'int("{{steps.first.result}}") + int("{{steps.second.result}}")'
    - name: data_type
      value: NUMBER

結果はresult.dataに入り、この例では3になります。失敗した場合はlogsとerrorを確認します。NUMBERの結果は数値として解析できる必要があります。通常のテキストにはSTRING、JSONとして扱えるデータにはOBJECTを使います。

パッケージを追加してオブジェクトを返す

コードでライブラリが必要な場合は、ステップのconfigにPythonパッケージを追加します。この例では、前のステップの売上データをpandasで集計し、JSONオブジェクトとして返します。

- id: rows
  tool: INPUT_TEXT
  input:
    - name: value
      value: '[{"region": "east", "sales": 120}, {"region": "west", "sales": 80}, {"region": "east", "sales": 50}]'
- id: summarize
  tool: PYTHON_SANDBOX_RUN
  needs: [rows]
  config:
    - name: packages
      value: '["pandas"]'
  input:
    - name: code
      value: |
        import json
        import pandas as pd

        rows = json.loads(r'''{{steps.rows.result}}''')
        df = pd.DataFrame(rows)
        totals = {region: int(total) for region, total in df.groupby("region")["sales"].sum().items()}
        print(json.dumps({"totals": totals, "grand_total": int(df["sales"].sum())}))
    - name: data_type
      value: OBJECT

OBJECTでは標準出力に出力したJSONが解析されるため、後続ステップでは{{steps.summarize.result.data.grand_total}}（250）や{{steps.summarize.result.data.totals.east}}（170）で参照できます。前のステップの出力は、例のようにjson.loads(r'''...''')で読み込んでください。JSONをそのままPythonコードに埋め込むと、null、true、falseでエラーになります。

設定	既定値	説明
`packages`	`[]`	コード実行前にpipでインストールするPythonパッケージ。JSON配列またはカンマ区切りで指定
`dependencies`	`[]`	インストールするシステム（apt）パッケージ
`timeout`	`900000`	サンドボックスのタイムアウト（ミリ秒、15分）

egress ポリシーが設定された組織での動作

egress ポリシー（サンドボックスから接続できる外部ホストの制限）が設定された組織では、コードが動く前にサンドボックスがロックダウンされ、コードは管理者権限なしで実行されます。

sudo は使えず、/root と /usr/local には書き込めません。システムパッケージは dependencies 設定で指定してください。コードの実行前に、必要な権限でインストールされます。
packages はロックダウン前にインストールされるので、許可ホストの設定は不要です。コード内の pip install はロックダウン後に動くため、パッケージの配布元（pypi.org、files.pythonhosted.org）が許可ホストに含まれている必要があります。どちらもユーザー自身の領域（~/.local）にインストールされ、import はこれまでどおり使えます。自分で作った仮想環境（venv）の中で pip を使う場合は、PIP_USER=0 を設定してください。
サンドボックス内では、外部のホスト名はすべてダミーのアドレス 198.18.0.1 に解決されます。許可されたホストへの HTTP / HTTPS リクエストはホスト名で振り分けられるため、これまでどおり届きます。ホストの実際の IP アドレスを必要とするコードは動かず、ホスト名ではなく IP アドレスを直接指定したリクエストは拒否されます。
外に出られるのは、許可されたホストへの HTTP / HTTPS（ポート 80 / 443）だけです。それ以外のプロトコルや、DNS への直接の問い合わせは遮断されます。

ステップのデータとシークレットを参照する

前ステップのデータは steps.{step_name}.result.{key}、シークレットは secrets.{secret_name}.{key}（例: secrets.OAUTH.access_token）でアクセス可能。単一キーのシークレットは str(secrets.{secret_name}) で値を直接取得
サンドボックスへ渡されるのはコード中で名前を直接指定したシークレットのみ。secrets.NAME、secrets["NAME"]、secrets.get("NAME") のようにリテラル名で参照してください（f"Bearer {secrets.NAME}" のような f-string 内の参照も対象）。キーを変数で組み立てる参照（secrets[name]）や secrets.items()、json.dumps(secrets) のような全件参照は、サンドボックス起動前にエラーになります

大きな表の受け渡し: `tables`

ステップの出力は JSON としてステップ間を移動します。数十万行の DataFrame はこの経路に載りません。シリアライズの時点でサンドボックスのメモリを使い切り、ステップが結果を返す前に停止します。このために、Python ステップには tables ヘルパーが用意されています。 tables.write(df) はフレームをストレージへ書き出し、保存先を指す数百バイトのマーカーを返します。このマーカーがステップの結果になります。次のステップは tables.read(...) でマーカーから DataFrame を復元します。行が JSON になることはありません。

- id: csv_file
  tool: INPUT_FILE
  input:
    - name: description
      value: "ピボットする売上 CSV"
    - name: file_extension
      value: csv
- id: load
  tool: PYTHON_SANDBOX_RUN
  needs: [csv_file]
  config:
    - name: packages
      value: '["pandas"]'
  input:
    - name: code
      value: |
        import json
        import pandas as pd
        df = pd.read_csv(steps.csv_file.result)
        json.dumps(tables.write(df))
    - name: data_type
      value: OBJECT
- id: pivot
  tool: PYTHON_SANDBOX_RUN
  needs: [load]
  config:
    - name: packages
      value: '["pandas"]'
  input:
    - name: code
      value: |
        import json
        df = tables.read(steps.load.result.data)
        pv = df.pivot_table(index=["rep_month", "branch_cd"],
                            columns="product_cd", values="amt_00",
                            aggfunc="sum")
        json.dumps(tables.write(pv, name="pivot"))
    - name: data_type
      value: OBJECT

INPUT_FILE はファイル URL を文字列で返すため、steps.csv_file.result が URL そのものになり、pd.read_csv でそのまま読み込めます。マーカーは dict です。そのため、他の構造化された結果と同じように返す必要があります。最後の式を json.dumps(...) にし、data_type に OBJECT を指定してください。指定しない場合 data_type は STRING になり、次のステップには dict の Python 表記が渡されるため、tables.read が読み取れません。 tables.write(df, name="result", fmt="csv", index=None) はフレームを公開し、マーカーを返します。

name — 保存するオブジェクト名。1 つのステップから複数の表を公開できます。使えるのは英数字と .、_、- です。.. と、先頭・末尾のドットは使えません。
fmt — "csv"（既定）または "parquet"。Parquet には pyarrow が必要で、すべてのサンドボックスイメージに入っているわけではありません。無い場合は tables.write がその旨を伝えます。
index — フレームのインデックスを列として書き出すかどうか。既定の None は「意味を持つときだけ書き出す」動作です。groupby(...).agg(...) や pivot_table の集計キー、resample の DatetimeIndex は保存され、単なる連番インデックスは書き出されません（絞り込み後のフレームに余分な列が増えません）。明示したい場合は True / False を渡します。

列は単一レベルである必要があります。values を複数指定したピボットは先に平坦化してください: df.columns = ["_".join(str(part) for part in column) for column in df.columns.to_flat_index()] df は Series ではなく DataFrame である必要があります。 df.groupby([...])["amt"].sum() は Series を返すため、先に .to_frame() を呼んでください。集計キーは名前付きインデックスなので、列として書き出されます。 tables.read(marker) は DataFrame を返します。戻り値は平坦な形です。 tables.write が書き出した内容はすべて列として復元され、書き出し時にインデックスだった階層も列になります。marker["__jinbatable"]["columns"] にその一覧が、rows と sizeBytes に行数とサイズが入っているため、後続ステップは表を取得しなくても形を確認できます。読み出し時には rows と実際に読めた行数を照合します。送信が途中で切れて短くなった表は、そのまま返さずエラーになります。 tables.is_table(value) は値がマーカーかどうかを返します。表と行データのどちらも受け取るステップで使えます。 サイズの上限. 1 回の tables.write で送れるサイズは、デプロイ先が 1 リクエストで受け取れる大きさに制限されます。これはツール側の制限ではなくデプロイ先の性質です。マネージド環境ではプラットフォームの入口が 32 MiB を超えるリクエストボディを拒否しますが、自己ホスト環境では上限が異なることも、上限が無いこともあり、代わりにリクエストタイムアウトで途中の送信が切断されることもあります。tables.write は「サイズ超過で拒否された」のか「ステータスが返る前に接続が切られた」のかを区別して報告し、デプロイ先が上限を宣言している場合はその値を示します。いずれの場合も、書き出す前に集計・絞り込みを行うか、name を変えて複数の表に分けてください。読み出し側にこの制限はありません。保存済みの表は、サンドボックスのメモリが許す限りまとめて読み込めます — 逆に言えば非常に大きい表ではメモリが制約になります。tables.read は表全体を一度に展開するためです。補足:

表は同じフローの後続ステップから読めます。以前の実行結果を再利用する reused ステップからも読めます。別のフローからは読めません。
forEach の各繰り返しには専用の保存先が割り当てられます。すべての繰り返しが既定の名前のまま tables.write(df) を呼んでも、互いに上書きしません。
forEach の item として渡されたマーカーも tables.read で読めます。
小さい結果にこの仕組みは不要です。行をそのまま返すステップは今までどおりです。

​前のステップから数値を計算する

​パッケージを追加してオブジェクトを返す

​egress ポリシーが設定された組織での動作

​ステップのデータとシークレットを参照する

​大きな表の受け渡し: tables

前のステップから数値を計算する

パッケージを追加してオブジェクトを返す

egress ポリシーが設定された組織での動作

ステップのデータとシークレットを参照する

大きな表の受け渡し: `tables`