これまで作ってきた検査や変換のCLIは、終了コードを確認して初めて次の処理へ進めます。複数のコマンドを手作業で実行すると、順番の取り違えや、失敗を見落として後続処理を動かす事故が起こり得ます。
今回は、JSONに記載したPython CLIを上から順に実行し、失敗、出力形式の違反、タイムアウトが起きた時点で停止するワークフロー実行器を作ります。Python 3.10以上の標準ライブラリだけを使い、シェルは介しません。
この実装は、各処理を一つのトランザクションにするものではありません。途中で止まっても、完了済みの処理が作成したファイルや外部システムへの変更は戻りません。各CLIは元データを直接更新せず、新しい出力を作る設計にしておく必要があります。
今回固定する実行契約
| 項目 | 仕様 |
|---|---|
| 動作環境 | Python 3.10以上、ローカルのPOSIX環境 |
| 設定 | BOMなしUTF-8のJSON。最大262,144バイト、最大20ステップ |
| 実行順 | stepsの記載順。並列実行はしない |
| 作業ディレクトリ | シンボリックリンク解決後の設定JSONが置かれたディレクトリ |
| 成功条件 | 子CLIの終了コードが0、標準エラー出力が空、標準出力が1行のJSONオブジェクト |
| 停止条件 | 子CLIの非ゼロ終了、出力契約違反、起動失敗、タイムアウト、割り込み |
| レポート | 実行前には存在しないディレクトリへreport.jsonを作る |
| 対象外 | 完了済み処理の取り消し、定期実行、並列処理、別端末への分散実行 |
この実装ではshell=Falseを指定するため、argvの要素はシェル展開されません。ただし、argvからシェル自体を明示的に起動すれば、そのシェルは構文を解釈します。設定には任意の実行ファイルを指定できるため、信頼できない相手から受け取った設定は実行しないでください。
1．子CLIの出力契約を確認する
最初に、実行器の動作確認専用CLIを用意します。次のコードをsample_gate.pyとして保存します。
from __future__ import annotations
import argparse
import json
import sys
def compact(value: object) -> str:
return json.dumps(value, ensure_ascii=False, separators=(',', ':'))
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument('step')
parser.add_argument('--fail', action='store_true')
args = parser.parse_args()
if args.fail:
print(
compact({
'status': 'failed',
'step': args.step,
'error': 'requested_failure',
}),
file=sys.stderr,
)
return 7
print(compact({'status': 'passed', 'step': args.step}))
return 0
if __name__ == '__main__':
raise SystemExit(main())
通常時は標準出力へ1行のJSONを書き、終了コード0を返します。--failを付けた場合は標準エラー出力へJSONを書き、終了コード7を返します。これは業務判定を模したものではなく、停止動作を確認するためのテスト用CLIです。
2．実行順をJSONで定義する
次の内容をworkflow.jsonとして、sample_gate.pyと同じディレクトリへ保存します。
{
"version": 1,
"steps": [
{
"id": "input_check",
"argv": ["{python}", "sample_gate.py", "input_check"],
"timeout_seconds": 10
},
{
"id": "candidate_check",
"argv": ["{python}", "sample_gate.py", "candidate_check"],
"timeout_seconds": 10
},
{
"id": "publish_check",
"argv": ["{python}", "sample_gate.py", "publish_check"],
"timeout_seconds": 10
}
]
}{python}は、実行器自身を動かしているPythonのパスへ置き換えます。項目全体が{python}と一致した場合だけ置き換えます。
argvはコマンドを一つの文字列にせず、実行ファイルと引数を別々の要素にします。パイプ、リダイレクト、ワイルドカードなどのシェル構文は使えません。
3．ワークフロー実行器を作る
次のコードをworkflow_runner.pyとして保存します。
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
import re
import signal
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import BinaryIO
MAX_CONFIG_BYTES = 262_144
MAX_CAPTURE_BYTES = 65_536
MAX_STEPS = 20
ID_PATTERN = re.compile(r'[a-z][a-z0-9_-]{0,31}')
@dataclass(frozen=True)
class Step:
id: str
argv: tuple[str, ...]
timeout_seconds: int
class ConfigError(Exception):
pass
class StepFailure(Exception):
def __init__(
self,
exit_code: int,
kind: str,
message: str,
returncode: int | None,
duration_ms: int,
stdout_meta: dict[str, object],
stderr_meta: dict[str, object],
) -> None:
super().__init__(message)
self.exit_code = exit_code
self.kind = kind
self.message = message
self.returncode = returncode
self.duration_ms = duration_ms
self.stdout_meta = stdout_meta
self.stderr_meta = stderr_meta
def to_record(self, step_id: str) -> dict[str, object]:
return {
'id': step_id,
'status': 'interrupted' if self.exit_code == 130 else 'failed',
'kind': self.kind,
'message': self.message,
'returncode': self.returncode,
'duration_ms': self.duration_ms,
'stdout': self.stdout_meta,
'stderr': self.stderr_meta,
}
class CliParser(argparse.ArgumentParser):
def error(self, message: str) -> None:
raise ConfigError(message)
def compact(value: object) -> str:
return json.dumps(
value,
ensure_ascii=True,
allow_nan=False,
separators=(',', ':'),
)
def emit_error(kind: str, message: str) -> None:
print(
compact({'status': 'error', 'kind': kind, 'message': message}),
file=sys.stderr,
)
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat(
timespec='seconds'
).replace('+00:00', 'Z')
def elapsed_ms(started_ns: int) -> int:
return (time.monotonic_ns() - started_ns) // 1_000_000
def object_from_pairs(
pairs: list[tuple[str, object]],
) -> dict[str, object]:
result: dict[str, object] = {}
for key, value in pairs:
if key in result:
raise ValueError(f'duplicate JSON key: {key}')
result[key] = value
return result
def reject_constant(value: str) -> None:
raise ValueError(f'non-standard JSON constant: {value}')
def parse_json_float(value: str) -> float:
result = float(value)
if not math.isfinite(result):
raise ValueError(f'JSON number is outside the finite range: {value}')
return result
def parse_json_text(text: str) -> object:
return json.loads(
text,
object_pairs_hook=object_from_pairs,
parse_constant=reject_constant,
parse_float=parse_json_float,
)
def require_keys(
value: object,
expected: set[str],
label: str,
) -> dict[str, object]:
if type(value) is not dict:
raise ConfigError(f'{label} must be an object')
actual = set(value)
missing = sorted(expected - actual)
unknown = sorted(actual - expected)
if missing or unknown:
parts: list[str] = []
if missing:
parts.append(f'missing keys: {missing}')
if unknown:
parts.append(f'unknown keys: {unknown}')
raise ConfigError(f'{label}: ' + ', '.join(parts))
return value
def load_workflow(path: Path) -> tuple[tuple[Step, ...], str]:
raw = path.read_bytes()
if len(raw) > MAX_CONFIG_BYTES:
raise ConfigError('workflow file is too large')
if raw.startswith(bytes((0xEF, 0xBB, 0xBF))):
raise ConfigError('UTF-8 BOM is not allowed')
try:
root_value = parse_json_text(raw.decode('utf-8'))
except (UnicodeDecodeError, ValueError, RecursionError) as exc:
raise ConfigError(f'invalid workflow JSON: {exc}') from exc
root = require_keys(root_value, {'version', 'steps'}, 'root')
if type(root['version']) is not int or root['version'] != 1:
raise ConfigError('version must be integer 1')
raw_steps = root['steps']
if type(raw_steps) is not list:
raise ConfigError('steps must be an array')
if not 0 < len(raw_steps) <= MAX_STEPS:
raise ConfigError(f'steps must contain 1 to {MAX_STEPS} items')
steps: list[Step] = []
used_ids: set[str] = set()
for index, raw_step in enumerate(raw_steps):
label = f'steps[{index}]'
step = require_keys(
raw_step,
{'id', 'argv', 'timeout_seconds'},
label,
)
step_id = step['id']
if type(step_id) is not str or ID_PATTERN.fullmatch(step_id) is None:
raise ConfigError(f'{label}.id has an invalid format')
if step_id in used_ids:
raise ConfigError(f'duplicate step id: {step_id}')
used_ids.add(step_id)
raw_argv = step['argv']
if type(raw_argv) is not list or not 1 <= len(raw_argv) <= 32:
raise ConfigError(f'{label}.argv must contain 1 to 32 items')
argv: list[str] = []
for arg in raw_argv:
if type(arg) is not str or not arg or len(arg) > 4096:
raise ConfigError(f'{label}.argv contains an invalid item')
if any(
ord(char) < 32
or ord(char) == 127
or 0xD800 <= ord(char) <= 0xDFFF
for char in arg
):
raise ConfigError(
f'{label}.argv contains a control character or surrogate'
)
argv.append(arg)
timeout = step['timeout_seconds']
if type(timeout) is not int or not 1 <= timeout <= 3600:
raise ConfigError(
f'{label}.timeout_seconds must be an integer from 1 to 3600'
)
steps.append(Step(step_id, tuple(argv), timeout))
return tuple(steps), hashlib.sha256(raw).hexdigest()
def empty_stream_meta() -> dict[str, object]:
return {
'bytes': 0,
'sha256': hashlib.sha256(b'').hexdigest(),
}
def read_capture(handle: BinaryIO) -> tuple[bytes | None, dict[str, object]]:
handle.flush()
size = os.fstat(handle.fileno()).st_size
if size > MAX_CAPTURE_BYTES:
return None, {'bytes': size, 'sha256': None}
handle.seek(0)
data = handle.read()
return data, {
'bytes': size,
'sha256': hashlib.sha256(data).hexdigest(),
}
def stop_process_group(process: subprocess.Popen[bytes]) -> None:
try:
os.killpg(process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
except OSError:
if process.poll() is None:
process.kill()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
def parse_child_result(data: bytes) -> dict[str, object]:
if not data.endswith(b'\n'):
raise ValueError('stdout must end with one LF')
body = data[:-1]
if not body or b'\n' in body or b'\r' in body:
raise ValueError('stdout must contain exactly one line')
try:
value = parse_json_text(body.decode('utf-8'))
except (UnicodeDecodeError, ValueError, RecursionError) as exc:
raise ValueError(f'stdout is not strict JSON: {exc}') from exc
if type(value) is not dict:
raise ValueError('stdout JSON must be an object')
return value
def run_step(step: Step, cwd: Path) -> dict[str, object]:
started_ns = time.monotonic_ns()
empty_meta = empty_stream_meta()
process: subprocess.Popen[bytes] | None = None
try:
with tempfile.TemporaryFile() as stdout_handle, tempfile.TemporaryFile() as stderr_handle:
argv = [
sys.executable if item == '{python}' else item
for item in step.argv
]
process = subprocess.Popen(
argv,
cwd=str(cwd),
stdin=subprocess.DEVNULL,
stdout=stdout_handle,
stderr=stderr_handle,
shell=False,
start_new_session=True,
)
try:
process.wait(timeout=step.timeout_seconds)
except subprocess.TimeoutExpired:
stop_process_group(process)
_, stdout_meta = read_capture(stdout_handle)
_, stderr_meta = read_capture(stderr_handle)
raise StepFailure(
5,
'timeout',
f'timeout after {step.timeout_seconds} seconds',
process.returncode,
elapsed_ms(started_ns),
stdout_meta,
stderr_meta,
)
except KeyboardInterrupt:
stop_process_group(process)
_, stdout_meta = read_capture(stdout_handle)
_, stderr_meta = read_capture(stderr_handle)
raise StepFailure(
130,
'interrupted',
'execution was interrupted',
process.returncode,
elapsed_ms(started_ns),
stdout_meta,
stderr_meta,
)
stdout_data, stdout_meta = read_capture(stdout_handle)
stderr_data, stderr_meta = read_capture(stderr_handle)
duration_ms = elapsed_ms(started_ns)
if process.returncode != 0:
raise StepFailure(
3,
'child_failed',
f'child exited with status {process.returncode}',
process.returncode,
duration_ms,
stdout_meta,
stderr_meta,
)
if stdout_data is None or stderr_data is None:
raise StepFailure(
4,
'output_too_large',
f'each output stream must be at most {MAX_CAPTURE_BYTES} bytes',
process.returncode,
duration_ms,
stdout_meta,
stderr_meta,
)
if stderr_meta['bytes'] != 0:
raise StepFailure(
4,
'unexpected_stderr',
'successful child wrote to stderr',
process.returncode,
duration_ms,
stdout_meta,
stderr_meta,
)
try:
result = parse_child_result(stdout_data)
except ValueError as exc:
raise StepFailure(
4,
'invalid_stdout',
str(exc),
process.returncode,
duration_ms,
stdout_meta,
stderr_meta,
) from exc
return {
'id': step.id,
'status': 'passed',
'returncode': process.returncode,
'duration_ms': duration_ms,
'stdout': stdout_meta,
'stderr': stderr_meta,
'result': result,
}
except StepFailure:
raise
except OSError as exc:
if process is not None and process.poll() is None:
stop_process_group(process)
raise StepFailure(
6,
'launch_or_io_error',
f'{type(exc).__name__}: {exc}',
None if process is None else process.returncode,
elapsed_ms(started_ns),
empty_meta,
empty_meta,
) from exc
def write_report(run_dir: Path, report: dict[str, object]) -> Path:
temporary = run_dir / '.report.json.tmp'
final = run_dir / 'report.json'
payload = (
json.dumps(
report,
ensure_ascii=True,
allow_nan=False,
indent=2,
sort_keys=True,
).encode('utf-8')
+ b'\n'
)
with temporary.open('xb') as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, final)
return final
def main(argv: list[str] | None = None) -> int:
parser = CliParser()
parser.add_argument('workflow')
parser.add_argument('run_dir')
try:
args = parser.parse_args(argv)
except ConfigError as exc:
emit_error('argument_error', str(exc))
return 2
try:
workflow_path = Path(args.workflow).resolve(strict=True)
if not workflow_path.is_file():
raise ConfigError('workflow path is not a regular file')
steps, config_sha256 = load_workflow(workflow_path)
except (OSError, RuntimeError, ConfigError) as exc:
emit_error('config_error', str(exc))
return 2
try:
run_dir_input = Path(args.run_dir)
os.mkdir(run_dir_input, 0o700)
run_dir = run_dir_input.resolve(strict=True)
except (OSError, RuntimeError) as exc:
emit_error('run_directory_error', str(exc))
return 6
started_at = utc_now()
results: list[dict[str, object]] = []
status = 'passed'
failed_step: str | None = None
uncompleted: list[str] = []
exit_code = 0
next_index = 0
try:
while next_index < len(steps):
step = steps[next_index]
try:
result = run_step(step, workflow_path.parent)
except StepFailure as exc:
results.append(exc.to_record(step.id))
status = 'interrupted' if exc.exit_code == 130 else 'failed'
failed_step = step.id
uncompleted = [item.id for item in steps[next_index + 1:]]
exit_code = exc.exit_code
break
results.append(result)
next_index += 1
except KeyboardInterrupt:
status = 'interrupted'
uncompleted = [item.id for item in steps[next_index:]]
exit_code = 130
report: dict[str, object] = {
'schema_version': 1,
'status': status,
'config_sha256': config_sha256,
'started_at': started_at,
'finished_at': utc_now(),
'failed_step': failed_step,
'uncompleted_step_ids': uncompleted,
'steps': results,
}
try:
report_path = write_report(run_dir, report)
except (OSError, TypeError, ValueError, RecursionError) as exc:
emit_error('report_write_error', str(exc))
return 6
print(compact({
'status': status,
'report': str(report_path),
'failed_step': failed_step,
}))
return exit_code
def entry() -> int:
try:
return main()
except KeyboardInterrupt:
emit_error('interrupted', 'execution was interrupted')
return 130
if __name__ == '__main__':
raise SystemExit(entry())
4．正常系を実行する
run-okがまだ存在しないことを確認してから、3ファイルを置いたディレクトリで実行します。
python3 workflow_runner.py workflow.json run-ok
runner_status=$?
printf 'exit=%s\n' "$runner_status"
python3 -m json.tool run-ok/report.jsonすべての子CLIが契約を満たすため、実行器の終了コードは0になります。レポートのstatusはpassed、3件のステップもすべてpassedとなり、uncompleted_step_idsは空です。
同じrun-okを指定して再実行すると、既存ディレクトリを上書きせず終了コード6で停止します。既存のファイルやシンボリックリンクと同じ名前も使えません。実行ごとに新しいディレクトリ名を割り当ててください。
5．途中で失敗した場合を確認する
次の内容をworkflow_fail.jsonとして保存します。2番目のステップだけに--failを追加しています。
{
"version": 1,
"steps": [
{
"id": "input_check",
"argv": ["{python}", "sample_gate.py", "input_check"],
"timeout_seconds": 10
},
{
"id": "candidate_check",
"argv": ["{python}", "sample_gate.py", "candidate_check", "--fail"],
"timeout_seconds": 10
},
{
"id": "publish_check",
"argv": ["{python}", "sample_gate.py", "publish_check"],
"timeout_seconds": 10
}
]
}run-failが存在しない状態で実行します。
python3 workflow_runner.py workflow_fail.json run-fail
runner_status=$?
printf 'exit=%s\n' "$runner_status"
python3 -m json.tool run-fail/report.json実行器の終了コードは3になります。レポートではinput_checkが成功し、candidate_checkのreturncodeが7になります。3番目のpublish_checkは実行されず、uncompleted_step_idsへ記録されます。
実行器は、失敗した子CLIの標準出力や標準エラー出力をレポートへ複製しません。上限内ならバイト数とSHA-256を残し、65,536バイトを超えたストリームはバイト数とnullを記録します。入力値や個人情報をエラーへ出すCLIを誤って接続した場合に、レポートへ同じ情報を増やさないためです。
終了コードを後続処理の条件にする
| 終了コード | 意味 | 対応 |
|---|---|---|
| 0 | 全ステップが成功 | レポートを確認し、次の工程へ進む |
| 2 | 引数または設定JSONが不正 | 設定の項目、型、重複キーを直す |
| 3 | 子CLIが非ゼロで終了 | 該当CLIの入力と個別レポートを確認する |
| 4 | 子CLIが0で終了したが、成功時の出力契約に違反 | 標準出力、標準エラー出力、JSON形式を直す |
| 5 | タイムアウト | 処理量、外部待ち、制限時間を確認する |
| 6 | 起動、ディレクトリ、レポート書き込みの失敗 | パス、権限、空き容量を確認する |
| 130 | Ctrl+Cによる中断 | 途中生成物を確認してから新しい実行を開始する |
子CLI固有の終了コードは、そのまま実行器の終了コードにはしません。たとえば子CLIの終了コード7はレポートへ保存し、実行器は一律に3を返します。これにより、実行器を呼び出す側は共通の分類だけを扱えます。
実務へ置き換えるときの注意点
- 成功時の標準出力には、件数、判定、レポートパスなどの業務メタデータだけを出す
- 大量ログを出す処理は接続しない。終了コード0でも、各出力が65,536バイトを超えれば契約違反として停止する
- 出力は一時ファイルまたは新しいパスへ作り、検査完了前のファイルを公開名へ切り替えない
- タイムアウト時はプロセスグループへ
SIGKILLを送るため、子CLIの終了処理は実行されない - バックグラウンド化や別セッション化を行うCLIは対象にしない
- 設定の
argvへAPIキーやパスワードを書かない。引数は同じ端末の別プロセスから見える場合がある - 子CLIは親プロセスの環境変数を継承する。実行用アカウントへ必要以上の秘密情報を持たせない
標準出力と標準エラー出力は、一時ファイルへ受けてからサイズを確認します。このためメモリへ無制限に読み込むことはありませんが、終了またはタイムアウトまでは一時領域を消費します。大量のログを生成するコマンドを制御する用途には、ログ量を実行中に制限する別の仕組みが必要です。
config_sha256は、どの設定バイト列を読み込んだ実行かを照合するための値です。署名ではないため、設定とレポートの両方を書き換えられる利用者に対する改ざん証明にはなりません。
最初は二つの処理から接続する
実務へ導入するときは、検査と変換など、依存関係が明確な二つのCLIから始めます。それぞれを単独で実行し、終了コード、標準出力、標準エラー出力を確認してから設定へ追加します。
この実行器が保証するのは、登録した順序で一つずつ起動し、失敗後のステップを実行しないことです。各処理の正しさ、完了済み変更の取り消し、外部サービス側の結果までは保証しません。その境界を分けることで、小さなPython処理を一つの業務手順へまとめられます。