YAML DSLから関数を定義して使用する

YAML DSLシリーズ第四弾です。
前回までで、変数を含んだHello Worldと、文字列のフォーマット、条件分岐を実装しました。
今回はYAML DSLから関数を定義して、呼び出してみたいと思います。

#DSLをデザインする

前回までのDSLは以下です。ログイン処理を2回行っていますが、2回目はYAMLのアンカー/エイリアスを使っています。つまり実際には同じ命令群を二回記述していることになります。この部分を、関数として定義できると良さそうです。

login.yml

- set: [username, cuhey3]
- set: [password, rumiokubo]
# ログイン処理（成功する）
- when: &login
  - $: [check_auth, $: username, $: password]
  # 成功時処理
  -
    - print: 'Login Successful!'
    - print:
        $: [format, 'Welcome {}!', $: username]
  # 失敗時処理
  - print: 'Login Failed.'
- set: [password, yukaotsubo]
# ログイン処理（失敗する）
# YAMLのアンカー/エイリアスを使用して記述を再利用
- when: *login

先にDSLのデザインを行います。関数を定義する関数defineを新しく登場させてみました。

login.yml

# 関数を定義する関数define
- define:
    # 第一引数は関数名
    - login
    # 第二引数は引数リスト
    - [username, password]
    # 第三引数は具体的な処理
    -
      - when:
          - $: [check_auth, $: username, $: password]
          # 成功時処理
          -
            - print: 'Login Successful!'
            - print:
                $: [format, 'Welcome {}!', $: username]
          # 失敗時処理
          - print: 'Login Failed.'
# ログイン処理（成功）
- login: [cuhey3, rumiokubo]
# ログイン処理（失敗）
- login: [cuhey3, yukaotsubo]

#深く考えずに実装してみる

DSL解釈プログラムを、深く考えずに実装すると、以下のようになります。
（追加部分だけ抜粋します。全量は前回の記事をご覧ください。）

dsl_runner.py

def define(func_name, args_name, dsl):
    def func(*args_value):
        # 引数をグローバル変数にセット
        for i in range(len(args_name)):
            variables[args_name[i]] = args_value[i]
        execute_dsl(dsl)
    # 既存の関数一覧func_dictにマッピング
    func_dict[func_name] = func


# DSLから呼び出し可能な関数を列挙
func_dict = {
    'print': print,
    'set': _set,
    'get': _get,
    'format': _format,
    'check_auth': check_auth,
    'when': when,
    # 追加
    'define': define
}

前回までの実装では、関数スコープの変数を使用する仕組みがありません。
そのため、defineによって定義された関数が呼び出された時は、引数は実際にはグローバル変数にセットされ、DSL本体が実行されます。
これでも、今の用途の範囲では正しく動作します。

Login Successful!
Welcome cuhey3!
Login Failed.

しかし、やはり変数は正しいスコープで扱いたいです。また本来であれば、define関数を使って実行時に定義した関数は、環境のグローバル関数リストではなく、実行時のスコープ内で保持されるべきです。
どうやら、DSLから関数を正しく定義できるようにするには、関数のスコープやローカル変数を考慮した仕組みを整備する、ということが必要なようです。

#関数はローカル変数に相当するdictを受け取るようにする
結論から述べますと、Pythonが本来持っているスコープの概念や機能を、DSL上で再現するために外からうまく実装するのは、私には無理でした。（どう実装していいかすら考えがおよびませんでした。）
ですので、ローカル変数を擬似的かつ手っ取り早く再現する方法として、今回は

• 各関数を呼び出す際に、ローカル変数に相当するdictを一緒に渡す
• DSL上で提供される関数は、ローカル変数を引数として必ず受け取るようにする

という方針を取ることにしました。

今回のソースコードの修正は、既存の定義済み関数およびその呼び出しと広範に渡りますので、以下に全量のソースコードを記載します。また、合わせて変数名の見直しを行っていますが、意味合いとしては変わっていないはずです。
関数に渡すローカル変数に対して、下記のコードではdsl_localsという名前をつけています。
また、前回までに出てきていたグローバル変数であるuser_infoですとか、定義済み関数のリストは、dsl_globalsにまとめています。

dsl_runner.py

import yaml


#dsl_localsを使用しない関数のためのデコレータ
def ignore_locals(f):
    def wrapper(dsl_locals, *args, **kwargs):
        result = f(*args, **kwargs)
        if result:
            return result

    return wrapper


# 変数セット関数
def _set(dsl_locals, var_name, value):
    dsl_locals[var_name] = value


# 変数ゲット関数
def _get(dsl_locals, var_name):
    return dsl_locals.get(var_name) or dsl_globals.get(var_name)


# フォーマット関数
@ignore_locals
def _format(format_string, *args):
    return format_string.format(*args)


# ログインチェック関数
@ignore_locals
def check_auth(username, password):
    return dsl_globals['user_info'].get(username) == password


# 条件分岐関数
# when関数が呼ばれた時点で、predicateは真偽値に変換されている。
def when(dsl_locals, predicate, true_dsl, false_dsl):
    if predicate:
        execute_dsl(true_dsl, dsl_locals)
    else:
        execute_dsl(false_dsl, dsl_locals)


# 関数定義関数
def define(dsl_locals, func_name, args_name, dsl):
    def func(*args_value):
        # 引数名、引数値をローカル変数にセット
        # DSL上の変数宣言からは添字が一つずれる（dsl_localsが存在するため）
        for i in range(0, len(args_name)):
            dsl_locals[args_name[i]] = args_value[i + 1]
        execute_dsl(dsl, dsl_locals)

    _functions = dsl_locals.get('_functions') or {}
    _functions[func_name] = func
    dsl_locals['_functions'] = _functions


# DSLから呼び出し可能な関数を列挙
dsl_globals = {
    # ログインチェックで使用するusernameとpasswordのペア
    'user_info': {
        'cuhey3': 'rumiokubo'
    },
    # DSLから呼び出し可能な定義済み関数を列挙
    '_functions': {
        'print': ignore_locals(print),
        'set': _set,
        'get': _get,
        'format': _format,
        'check_auth': check_auth,
        'when': when,
        'define': define
    }
}


# 引数を評価する関数
def evaluate_args(dsl_locals, args):
    result = []
    for arg in args:
        if type(arg) is dict and '$' in arg:
            dollar_val = arg.get('$')
            if type(dollar_val) is list:
                func_name = dollar_val[0]
                func = find_func_in_scope(dsl_locals, func_name)
                result.append(
                    func(dsl_locals, *evaluate_args(dsl_locals,
                                                    dollar_val[1:])))
            else:
                result.append(_get(dsl_locals, dollar_val))
        else:
            result.append(arg)
    return result


# dsl_localsまたはdsl_globalsから関数を見つける
def find_func_in_scope(dsl_locals, func_name):
    func = dsl_globals['_functions'].get(func_name) or (
        dsl_locals.get('_functions') or {}).get(func_name)
    assert func, 'function not found: ' + func_name
    return func


# DSL実行関数
def execute_dsl(dsl, dsl_locals=None):
    # dsl_localsが渡されなかった時は初期化する
    if dsl_locals is None:
        dsl_locals = {}
    # 命令が一つだけ渡される場合も許容する
    if type(dsl) is not list:
        dsl = [dsl]
    for inst in dsl:
        for func_name, raw_args in inst.items():
            func = find_func_in_scope(dsl_locals, func_name)
            # 引数が一つだけ渡される場合も許容する
            if type(raw_args) is not list:
                raw_args = [raw_args]
            args = evaluate_args(dsl_locals, raw_args)
            func(dsl_locals, *args)


with open('login.yml') as yaml_file:
    # YAMLのロード
    dsl = yaml.load(yaml_file)
    # スコープが効いているか確認するためトップレベルをリストにする
    for d in dsl:
        print('--- DSL start ---')
        execute_dsl(d)
        print('--- DSL end ---')

実装のポイントとしては、以下のようになります。

• DSLから呼び出し可能なすべての定義済み関数の第一引数はdsl_locals（関数定義を共通化する上での制約…）
• ただしその全てが関数の中でdsl_localsを参照するわけではないので、＠ignore_localsデコレータで引数を省略可能
• ローカル変数dsl_localsをいつ初期化するかはよく考える必要がある。現実装はexecute_dslがdsl_locals引数なしで呼び出された時。

検証用に、少しDSLの構造を変えて、トップレベルをサブDSLのリストにします。

login.yml

# DSL 1
- # 関数を定義する関数define
    - define:
        # 第一引数は関数名
        - login
        # 第二引数は引数リスト
        - [username, password]
        # 第三引数は処理
        -
          - when:
              - $: [check_auth, $: username, $: password]
              # 成功時処理
              -
                - print: 'Login Successful!'
                - print:
                    $: [format, 'Welcome {}!', $: username]
              # 失敗時処理
              - print: 'Login Failed.'
    # ログイン処理（成功）
    - login: [cuhey3, rumiokubo]
    # ログイン処理（失敗）
    - login: [cuhey3, yukaotsubo]
# DSL 2
-
    - login: [cuhey3, rumiokubo]
    # ログイン処理（失敗）
    - login: [cuhey3, yukaotsubo]

実行すると以下のようになります。
一つ目のDSLは問題なく実行できていますが、二つ目のDSLを実行した際、関数loginが見つからないため、実行に失敗します。これは想定どおりの動きです。

--- DSL start ---
Login Successful!
Welcome cuhey3!
Login Failed.
--- DSL end ---
--- DSL start ---
Traceback (most recent call last):
  File "/home/ubuntu/workspace/dsl_runner.py", line 129, in <module>
    execute_dsl(d)
  File "/home/ubuntu/workspace/dsl_runner.py", line 115, in execute_dsl
    func = find_func_in_scope(dsl_locals, func_name)
  File "/home/ubuntu/workspace/dsl_runner.py", line 101, in find_func_in_scope
    assert func, 'function not found: ' + func_name
AssertionError: function not found: login

以上で、DSL上から関数を定義し、また関数内でローカル変数を扱う仕組みを実装することができました。