> ActQL_ v0.0-alpha
目次
ガイド

Python

actql パッケージは、ActQLをPythonのアプリの中で動かします。動かしておくサーバーはなく、同じAPIでリモートのHostやActQL Cloudにもつながります。TypeScriptとNodeをPythonの流儀で写したもので、メソッドは snake_case、状態は型付きのdataclass（dictが必要なら dataclasses.asdict()）、同期と async の両方のインターフェースがあります。

uv add actql
# or
pip install actql

Python 3.11以上が必要です。

短いクエリ

actql.open() が最短の道です。状態をディレクトリに置くローカルのエンジンと、1回限りのクエリのための sql() / aql() が手に入ります。

import actql

with actql.open(state_dir=".actql") as db:
    rows = db.sql(
        "SELECT id, name FROM crm.accounts WHERE owner = @owner",
        {"owner": "kim"},
    ).all()
    print(rows)

    with db.aql("FROM crm.accounts |> LIMIT 1000").rows() as reader:
        for row in reader:
            print(row["id"])

sql() は普通の SELECT を、aql() はPipe構文を受け取ります。第2引数はパラメーターのマップです。all() で全件を読み、rows() で順に流し読みします。どちらも with ブロックの中で実行します。ブロックを抜けるとハンドルは閉じられます。

同じ sql() と aql() は、connect() で作ったクライアントでも使えます。

短いクエリは読むだけです。Effectのあるプログラムは start_source を通します。

接続する

client = actql.connect(
    mode="local",
    installation=".actql",   # Storeのディレクトリ
    profile="mock",          # 省略すると設定した接続を使う
)

リモートのHostには：

client = actql.connect(
    mode="remote",
    endpoint=os.environ["ACTQL_ENDPOINT"],   # actql://host.acme.internal
    token=os.environ["ACTQL_TOKEN"],
)

token には文字列のほか、文字列を返す呼び出し可能オブジェクトも渡せます。endpoint には、actql:// の接続文字列か、Hostの https:// のURLを渡します。クライアントはコンテキストマネージャーとして使うか、自分で close() を呼んでください。閉じてもRunは取り消されません。

イベントループの中では、非同期のインターフェースを使います。同じメソッドを await するだけです。

async with await actql.AsyncClient.connect(mode="local", installation=".actql") as client:
    rows = await client.sql("SELECT 1 AS id").all()

client.sql(…) はクエリのハンドルを返し、.all() を await する（または .rows() を順に読む）ことでクエリが実行されます。読み取り1回につき await は1つです。

プログラムを実行する

source = open("churn-risk.aql").read()
execution = client.start_source(
    source,
    format="aql",
    parameters={"owner": "kim"},
    stop_before="commit",        # 任意：最初の書き込みの前で止める
)
state = execution.wait()
print(state.status)              # "succeeded", "waiting_checkpoint", "failed", …

start_source はプログラムを検査し、永続的なRunを始めます。wait() は次の安定した状態、つまり判断待ち、突き合わせ待ち、または最終状態で、status、checkpoint、effects、revision を持つ RunState のdataclassを返します。Runがすでに安定した状態なら、wait() はその状態をすぐに返します。その次の状態を待つには wait(since=state.revision) を使います。succeeded は、すべての計算とEffectに結果が出たという意味です。非同期のEffectは accepted のまま、外部システム側の処理が続いていることがあります。execution.ref はRunの参照です。別のプロセスがこのRunを必要とするなら保存しておきます。

成功したあとに、名前付きのResultを読みます。

rows = execution.result("updates").all(max_rows=10_000)

読み取りには上限があります。max_rows を渡すか、rows() で流し読みしてください。レコードを検証して型を付けるには、schema= にpydanticのモデル、またはPython SDKが受け付ける検証オブジェクトを渡します。合わないレコードがあれば例外になり、値が変換されることはありません。

Checkpoint

ポリシーが書き込みの判断を保留したとき、または stop_before="commit" を渡したとき：

state = execution.wait()
if state.status == "waiting_checkpoint":
    ckpt = state.checkpoint
    print(ckpt.id, ckpt.effect.target, len(ckpt.items))
    for change in ckpt.preview.changes:
        print(change.id, change.field, change.before, "→", change.after)

コードから判断するか、人に渡してこのプロセスを終えるかを選べます。RunはStoreの中で待ちます。

if len(ckpt.items) <= 100:
    client.approve(ckpt.id)
    final = execution.wait(since=state.revision)
else:
    notify_reviewer(run_ref=execution.ref, checkpoint=ckpt.id)
    # ここで待つものはない。レビュー担当者がCLIか別の場所から承認し、
    # あとのプロセスが execution.ref にアタッチして結果を読む。

approve() と reject() は、クライアントに設定した承認者で判断に署名し、そのプレビューに結び付けます。承認すると同じRunが続きます。却下すると、Runは何も送らずに cancelled で終わります。

コードで判断する

クライアント自身を委譲先にすることもできます。connect() に decide 関数を渡すと、Hostは止まる前にプレビューを渡してそれを呼びます。関数は "permit"、"reject"、"defer" のどれかを返し、"defer" のときだけCheckpointが作られます。

def decide(decision):
    if decision.kind != "effect_set":        # 読み取り、LLMの呼び出し、Run自体
        return "permit"
    return "permit" if len(decision.preview.items) <= 100 else "defer"

client = actql.connect(mode="local", installation=".actql", decide=decide)

decision.kind は "run"、"read"、"infer"、"egress"、"effect_set" のどれかです。preview を持つのは effect_set の判断だけで、その items は変更されるレコード、fields は書き込まれる項目、target は対象のリソースです。関数ができるのは判断だけで、プレビューを変えることはできません。関数が例外を送出すれば、何も送られません。

レビュー用のツール向けには、client.checkpoints()、client.review_checkpoint(item)、client.resolve_checkpoints(items, signer=...) が、明示的な署名とページングを伴う同じ操作を提供します。

Runにアタッチする

execution = client.attach(saved_ref)
state = execution.status()       # 現在の状態。待たない

再起動のあとも、Runは止まった場所にそのままあります。応答が失われた書き込みは needs_reconciliation として現れます。

if state.status == "needs_reconciliation":
    after = execution.reconcile()   # 外部システムに何が起きたかを確認する。再送はしない
    print(after.status, after.effects)

execution.cancel() はRunに停止を求めます。同じプログラムを同じ request_id で始めると、もう一度実行する代わりに既存のRunが返ります。ただし、プログラム、パラメーター、入力、認可のスコープが同じときに限ります。同じ request_id に違う内容を渡すと RequestConflict が送出されます。

execution = client.start_source(source, format="aql", request_id="nightly-2026-10-15")

実行せずに検査する

outcome = client.check(source, kind="text", format="aql", scope="template_only")
if outcome.report.verdict != "valid":
    for reason in outcome.report.reasons:
        print(reason.code, reason.location)

client.plan(source, format="aql") は、actql explain が表示する内容をデータとして返します。

ファイルの内容を読むこと、SERIALIZE でファイルを作ること、Resultを受け取ることは、別々のものです。ファイルを扱うがその違いを説明しています。

エラー

失敗は、安定した code を持つ actql.ClientError のサブクラスとして送出されます。

例外いつ
CheckFailedプログラムが無効。報告が添付される
RequestHeldRunが始まる前にポリシーが判断を保留した（Run前のCheckpoint）。ほかのCheckpointと同じように解決してから、Runに attach する
RequestConflictその request_id が、別のプログラム、パラメーター、入力、またはスコープで使われていた
SubmissionUnknown開始への応答が失われた。リクエストを照会し、新しいIDで始め直さない
ResolutionUnknown承認への応答が失われた。同じ項目を再送すれば、Hostは繰り返しだと認識し、二重に適用しない
ClientClosedクライアントが閉じられている

応答が失われたことは、何も起きなかったことを意味しません。リクエストやRunはその識別情報で照会し、書き込みの再試行として再送はしないでください。

次に読む

Source: actql/actql/docs/guides/python.md at d904416674e4 · Edit on GitHub

TypeScriptとNode

Previous Page

ローカルとリモートのHost

Where a Program runs and who may do what: Local Hosts, remote Hosts, project and Host configuration, and the standard policy.