[備忘録] Pythonで動的コード生成と実行を実装する 

## はじめに

この記事は、Pythonで動的にコードを生成・実行する手法についての備忘録です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/2186432b-bfc2-4adf-a29f-7a64f1dccaf7.png)


## 目次

1. Pythonで動的コードを実行する方法
2. `eval()`と`exec()`の基本
3. `compile()`を使った高度な実行制御
4. 動的コード生成のテクニック
5. 実践的なユースケース
6. セキュリティリスクと対策
7. まとめと注意点

## 1. Pythonで動的コードを実行する方法

Pythonでは、主に以下の方法で動的にコードを実行できます：

- `eval()`: 式を評価して結果を返す
- `exec()`: 文（複数行のコードブロックも可）を実行する
- `compile()`: コードをバイトコードにコンパイルする
- 動的モジュールインポート: `importlib`を使用
- AST(抽象構文木)による構文解析と実行

それぞれの特徴と用途を見ていきましょう。

## 2. `eval()`と`exec()`の基本

### `eval()`関数

`eval()`は**式(expression)**を評価し、その結果を返します。式とは値を生成するコードのことで、例えば `1 + 1`、`"hello" + " world"`、`len([1, 2, 3])`などです。

```python
# 基本的な使い方
result = eval('1 + 2 * 3')
print(result)  # 7

# 変数を使用
x = 10
result = eval('x * 2')
print(result)  # 20

# 関数呼び出し
result = eval('len("hello")')
print(result)  # 5

# リスト内包表記
result = eval('[i**2 for i in range(5)]')
print(result)  # [0, 1, 4, 9, 16]
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/f61f1335-18dd-48eb-98c5-01e6c9739265.png)


### `exec()`関数

`exec()`は**文(statement)**を実行します。文とは、代入、条件分岐、ループなど、値を返さないコードのことです。複数行のコードブロックも実行できます。

```python
# 単純な文の実行
exec('x = 100')
print(x)  # 100

# 複数行のコードブロック
code = '''
def greet(name):
    return f"Hello, {name}!"

message = greet("Python")
'''
exec(code)
print(message)  # "Hello, Python!"

# 条件分岐やループの実行
exec('''
for i in range(3):
    print(f"Count: {i}")
''')
# Count: 0
# Count: 1
# Count: 2
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/60d5ba09-6a97-43f4-85ee-10fc75e5c042.png)

### 名前空間の指定

`eval()`と`exec()`は、オプションで名前空間（グローバル変数とローカル変数）を指定できます。これにより、実行環境を制御できます。

```python
# グローバル名前空間を指定
global_ns = {'x': 10, 'y': 20}
result = eval('x + y', global_ns)
print(result)  # 30

# ローカル名前空間も指定
local_ns = {'y': 5}
result = eval('x + y', global_ns, local_ns)
print(result)  # 15（xはグローバルから、yはローカルから取得）

# exec()でも同様に指定可能
exec('z = x + y', global_ns, local_ns)
print(local_ns['z'])  # 15（zはローカル名前空間に追加される）
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/cc6a4bbd-c122-43ac-b10b-bafd184596ab.png)

## 3. `compile()`を使った高度な実行制御

`compile()`関数を使うと、コードをコンパイルしてコードオブジェクトを作成できます。このオブジェクトは`eval()`や`exec()`で実行できます。

```python
# eval用にコンパイル（式モード）
expr_code = compile('x + y', '<string>', 'eval')
result = eval(expr_code, {'x': 10, 'y': 20})
print(result)  # 30

# exec用にコンパイル（文モード）
stmt_code = compile('for i in range(3): print(i)', '<string>', 'exec')
exec(stmt_code)
# 0
# 1
# 2

# 対話モード（複数の文と式を含むコードをコンパイル）
# 注意: 'single'モードは対話式インタプリタ用で、各文を別々にコンパイルする必要があります
exec_code = compile('x = 10\ny = 20', '<string>', 'exec')
exec(exec_code)
eval_code = compile('x + y', '<string>', 'eval')
result = eval(eval_code)
print(result)  # 30
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/86215d05-52f1-4600-995a-a95029db6a0b.png)


`compile()`のメリットは、一度コンパイルしたコードを何度も実行できること、またコンパイル時にエラーチェックができることです。

```python
# コードを一度コンパイルして複数回実行
calc_code = compile('a * b', '<string>', 'eval')

for a, b in [(2, 3), (5, 7), (10, 20)]:
    result = eval(calc_code, {'a': a, 'b': b})
    print(f"{a} * {b} = {result}")
# 2 * 3 = 6
# 5 * 7 = 35
# 10 * 20 = 200
```

## 4. 動的コード生成のテクニック

実際にコードを動的に生成するテクニックを見ていきましょう。

### テンプレートとしてのコード生成

```python
def generate_function(name, params):
    param_list = ', '.join(params)
    body = ', '.join([f'{p}={{{p}}}' for p in params])
    
    code = f'''
def {name}({param_list}):
    print(f"Function {name} called with {body}")
    return {' + '.join(params) if params else 0}
'''
    return code

# 関数を動的に生成
func_code = generate_function('add_numbers', ['a', 'b', 'c'])
print("生成されたコード:")
print(func_code)

# 生成したコードを実行
exec(func_code)

# 生成した関数を呼び出す
result = add_numbers(10, 20, 30)
print(f"結果: {result}")
```

出力: 
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/aacec750-74d1-4072-81c1-313c298cc3bd.png)


### 動的クラス生成

```python
def generate_class(class_name, fields):
    field_defs = '\n    '.join([f'{field} = None' for field in fields])
    init_params = ', '.join(['self'] + fields)
    init_body = '\n        '.join([f'self.{field} = {field}' for field in fields])
    
    # __repr__メソッドを動的に生成
    repr_fields = []
    for field in fields:
        repr_fields.append(f"{field}={{self.{field}}}")
    
    repr_body = ', '.join(repr_fields)
    
    code = f'''
class {class_name}:
    {field_defs}
    
    def __init__({init_params}):
        {init_body}
    
    def __repr__(self):
        return f"{class_name}({repr_body})"
'''
    return code

# クラスを動的に生成
person_class_code = generate_class('Person', ['name', 'age', 'email'])
print("生成されたクラスコード:")
print(person_class_code)

# 生成したコードを実行
exec(person_class_code)

# 生成したクラスを使用
person = Person("山田太郎", 30, "yamada@example.com")
print(person)  # Person(name=山田太郎, age=30, email=yamada@example.com)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/973f8ca4-9f6e-4fad-81f6-5b545cee1e92.png)


### プログラム全体の動的生成と実行

複雑なプログラムを動的に生成し、実行する例です。

```python
def generate_data_processor(data_type, operations):
    """データ処理プログラムを動的に生成する"""
    
    imports = []
    functions = []
    main_code = []
    
    # 必要なライブラリをインポート
    if data_type == 'csv':
        imports.append('import csv')
    elif data_type == 'json':
        imports.append('import json')
    
    if 'plot' in operations:
        imports.append('import matplotlib.pyplot as plt')
    
    # 関数を生成
    if 'read' in operations:
        if data_type == 'csv':
            functions.append('''
def read_data(file_path):
    data = []
    with open(file_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            data.append(row)
    return data
''')
        elif data_type == 'json':
            functions.append('''
def read_data(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        return json.load(f)
''')
    
    if 'process' in operations:
        functions.append('''
def process_data(data):
    # データを加工する処理
    results = []
    for item in data:
        # 仮の処理：数値フィールドを抽出して合計
        num_values = [float(val) for val in item.values() if str(val).replace('.', '', 1).isdigit()]
        if num_values:
            results.append(sum(num_values))
    return results
''')
    
    if 'plot' in operations:
        functions.append('''
def plot_data(data):
    plt.figure(figsize=(10, 6))
    plt.bar(range(len(data)), data)
    plt.title('Data Visualization')
    plt.xlabel('Index')
    plt.ylabel('Value')
    plt.savefig('output_plot.png')
    plt.close()
''')
    
    # メインコード
    main_code.append('''
def main():
    # ファイルパスは実際の環境に合わせて変更してください
    file_path = 'sample_data.csv' if data_type == 'csv' else 'sample_data.json'
    
    print(f"Processing {data_type} data...")
''')
    
    if 'read' in operations:
        main_code.append('    data = read_data(file_path)')
    
    if 'process' in operations:
        main_code.append('    results = process_data(data)')
        main_code.append('    print(f"Processed {len(results)} data points")')
    
    if 'plot' in operations:
        main_code.append('    plot_data(results)')
        main_code.append('    print("Plot saved as output_plot.png")')
    
    main_code.append('''
    print("Done!")

if __name__ == "__main__":
    main()
''')
    
    # 全てのコードを結合
    full_code = '\n'.join(imports) + '\n\n' + '\n'.join(functions) + '\n'.join(main_code)
    full_code = full_code.replace('data_type', f"'{data_type}'")
    
    return full_code

# CSVデータを読み込み、処理し、プロットするプログラムを生成
program = generate_data_processor('csv', ['read', 'process', 'plot'])
print("生成されたプログラム:")
print(program)

# 実際の環境で実行する場合は以下のようにする
# with open('generated_program.py', 'w') as f:
#     f.write(program)
# import subprocess
# subprocess.run(['python', 'generated_program.py'])
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/381be302-ae5e-4c16-9dd6-eef541715cad.png)


## 5. 実践的なユースケース

### ユースケース1: 数式の動的評価

ユーザーが入力した数式を安全に評価する例です。

```python
def safe_eval_expression(expr, variables=None):
    """安全に数式を評価する関数"""
    if variables is None:
        variables = {}
    
    # 許可する演算子と関数のリスト
    allowed_names = {
        'abs': abs, 'round': round, 'min': min, 'max': max,
        'sum': sum, 'len': len, 'pow': pow, 'int': int, 'float': float
    }
    
    # 変数をallowed_namesに追加
    for name, value in variables.items():
        if isinstance(value, (int, float, list, tuple)):
            allowed_names[name] = value
    
    # evalを制限された名前空間で実行
    try:
        return eval(expr, {"__builtins__": {}}, allowed_names)
    except Exception as e:
        return f"エラー: {str(e)}"

# 使用例
expressions = [
    "2 + 2 * 3",
    "abs(-15) + max(4, 7, 2)",
    "pow(2, 8)",
    "round(3.14159, 2)"
]

for expr in expressions:
    result = safe_eval_expression(expr)
    print(f"{expr} = {result}")

# 変数を使った例
variables = {'x': 10, 'y': 5, 'values': [1, 2, 3, 4, 5]}
expr = "x * y + sum(values)"
result = safe_eval_expression(expr, variables)
print(f"{expr} = {result}")  # x * y + sum(values) = 65
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/98d8af91-b9d0-4cbe-8288-ebcfca6b3dba.png)

### ユースケース2: 動的なデータ検証ルール

データ検証ルールを動的に構築して実行する例です。

```python
def create_validator(rules):
    """検証ルールから検証関数を動的に生成"""
    function_code = '''
def validate_data(data):
    errors = []
'''
    
    for field, rule in rules.items():
        condition = rule['condition']
        error_msg = rule['error']
        
        check_code = f'''
    # {field}のチェック
    if "{field}" in data:
        if not ({condition}):
            errors.append(f"{error_msg}")
    else:
        errors.append(f"{field}が存在しません")
'''
        function_code += check_code
    
    function_code += '''
    return errors
'''
    
    # 関数を生成して返す
    local_vars = {}
    exec(function_code, globals(), local_vars)
    return local_vars['validate_data']

# 検証ルールの定義
validation_rules = {
    'age': {
        'condition': 'isinstance(data["age"], int) and 0 <= data["age"] <= 120',
        'error': '年齢は0〜120の整数である必要があります'
    },
    'email': {
        'condition': 'isinstance(data["email"], str) and "@" in data["email"]',
        'error': 'メールアドレスの形式が不正です'
    },
    'username': {
        'condition': 'isinstance(data["username"], str) and 3 <= len(data["username"]) <= 20',
        'error': 'ユーザー名は3〜20文字である必要があります'
    }
}

# 検証関数を動的に生成
validator = create_validator(validation_rules)

# テストデータ
test_data = [
    {'username': 'john', 'age': 25, 'email': 'john@example.com'},  # 有効
    {'username': 'a', 'age': 150, 'email': 'invalid-email'},  # 無効
    {'username': 'alice', 'age': '30', 'email': 'alice@example.com'}  # 無効（ageが文字列）
]

# 検証実行
for i, data in enumerate(test_data):
    errors = validator(data)
    if errors:
        print(f"データ {i+1} は無効: {', '.join(errors)}")
    else:
        print(f"データ {i+1} は有効です")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/567669be-49fb-4fd5-9b93-7c8b7b132cb2.png)

### ユースケース3: 単純なDSL（ドメイン特化言語）の実装

簡単なDSLを実装する例です。

```python
class SimpleDSL:
    """シンプルなドメイン特化言語インタプリタ"""
    
    def __init__(self):
        self.variables = {}
    
    def execute(self, script):
        """DSLスクリプトを実行"""
        lines = script.strip().split('\n')
        result = None
        
        for i, line in enumerate(lines):
            line = line.strip()
            if not line or line.startswith('#'):
                continue
            
            try:
                if line.startswith('SET '):
                    # 変数設定: SET x = 10
                    parts = line[4:].split('=', 1)
                    var_name = parts[0].strip()
                    expr = parts[1].strip()
                    self.variables[var_name] = self._eval_expr(expr)
                
                elif line.startswith('PRINT '):
                    # 出力: PRINT x + 5
                    expr = line[6:].strip()
                    value = self._eval_expr(expr)
                    print(value)
                    result = value
                
                elif line.startswith('IF '):
                    # 条件: IF x > 10 THEN PRINT "Large"
                    parts = line[3:].split('THEN', 1)
                    condition = parts[0].strip()
                    action = parts[1].strip() if len(parts) > 1 else ""
                    
                    if self._eval_expr(condition):
                        result = self.execute(action)
                
                elif line.startswith('REPEAT '):
                    # ループ: REPEAT 5 TIMES PRINT i
                    parts = line.split('TIMES', 1)
                    count_expr = parts[0][7:].strip()
                    count = int(self._eval_expr(count_expr))
                    action = parts[1].strip()
                    
                    for i in range(count):
                        self.variables['i'] = i
                        result = self.execute(action)
            
            except Exception as e:
                print(f"エラー(行 {i+1}): {str(e)}")
        
        return result
    
    def _eval_expr(self, expr):
        """式を評価"""
        # DSL内の特殊構文を処理
        expr = expr.replace('"', "'")  # ダブルクォートをシングルクォートに変換
        
        # 安全な評価のために制限された名前空間を使用
        safe_globals = {"__builtins__": {}}
        safe_functions = {
            'len': len, 'str': str, 'int': int, 'float': float,
            'min': min, 'max': max, 'abs': abs, 'sum': sum
        }
        
        return eval(expr, safe_globals, {**safe_functions, **self.variables})

# DSLの使用例
dsl_script = '''
# 変数の設定
SET x = 10
SET y = 20
SET name = "Python"

# 計算と出力
PRINT x + y
PRINT "Hello, " + name

# 条件分岐
IF x > 5 THEN PRINT "x is greater than 5"

# ループ
REPEAT 3 TIMES PRINT "Count: " + str(i)
'''

dsl = SimpleDSL()
dsl.execute(dsl_script)
```

## 6. セキュリティリスクと対策

動的コード実行には重大なセキュリティリスクがあります。特に、ユーザー入力を直接`eval()`や`exec()`に渡すことは、リモートコード実行攻撃の危険があります。

### 主なリスク

1. **任意のコード実行**: 悪意のあるユーザーがシステムコマンドを実行する可能性
2. **データアクセス**: 機密情報への不正アクセス
3. **リソース消費**: 無限ループなどによるリソース枯渇攻撃

### 安全対策

```python
def very_safe_eval(expr, allowed_names=None):
    """
    非常に制限された環境で式を評価する
    allowed_names: 許可する変数や関数の辞書
    """
    if allowed_names is None:
        allowed_names = {}
    
    # 組み込み関数へのアクセスを完全に遮断
    forbidden = {'__import__', 'eval', 'exec', 'compile', 'globals', 'locals', 'open', 
                'getattr', 'setattr', 'delattr', '__class__', '__base__', '__subclasses__'}
    
    # 禁止された単語がないか確認
    for word in forbidden:
        if word in expr:
            raise ValueError(f"式に禁止された単語が含まれています: {word}")
    
    # AST（抽象構文木）を使って式を解析し、安全かチェック
    import ast
    
    try:
        # 式を解析
        parsed = ast.parse(expr, mode='eval')
        
        # 式の内容を検証する関数
        class SafetyChecker(ast.NodeVisitor):
            def visit_Name(self, node):
                if node.id not in allowed_names:
                    raise NameError(f"変数 '{node.id}' は許可されていません")
                return self.generic_visit(node)
            
            def visit_Attribute(self, node):
                raise AttributeError("属性アクセスは許可されていません")
            
            def visit_Call(self, node):
                if isinstance(node.func, ast.Name):
                    if node.func.id not in allowed_names:
                        raise NameError(f"関数 '{node.func.id}' は許可されていません")
                else:
                    raise TypeError("関数呼び出しの形式が不正です")
                return self.generic_visit(node)
        
        # 安全性チェック実行
        checker = SafetyChecker()
        checker.visit(parsed)
        
        # 式をコンパイルして実行
        compiled_expr = compile(parsed, '<string>', 'eval')
        return eval(compiled_expr, {"__builtins__": {}}, allowed_names)
    
    except Exception as e:
        return f"エラー: {str(e)}"

# 使用例
allowed_vars = {
    'x': 10,
    'y': 20,
    'add': lambda a, b: a + b,
    'multiply': lambda a, b: a * b
}

expressions = [
    "x + y",                          # 安全: 許可された変数を使用
    "add(x, y)",                      # 安全: 許可された関数を使用
    "multiply(x, 5)",                 # 安全: 許可された関数と定数を使用
    "__import__('os').system('ls')",  # 危険: 禁止された関数
    "open('/etc/passwd').read()",     # 危険: ファイルアクセス
    "globals()",                      # 危険: グローバル変数へのアクセス
    "x.__class__",                    # 危険: 属性アクセス
]

print("安全な式の評価:")
for expr in expressions[:3]:  # 安全な式だけを評価
    result = very_safe_eval(expr, allowed_vars)
    print(f"{expr} => {result}")

print("\n危険な式のチェック:")
for expr in expressions[3:]:  # 危険な式
    try:
        result = very_safe_eval(expr, allowed_vars)
        print(f"{expr} => {result}")
    except ValueError as e:
        print(f"{expr} => エラー: {e}")

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/e622efa9-6406-4bb2-8c63-65dfe403dd21.png)


### 追加のセキュリティ対策

1. **タイムアウト設定**: 実行時間に制限を設ける
   ```python
   import signal
   
   def timeout_handler(signum, frame):
       raise TimeoutError("コードの実行がタイムアウトしました")
   
   def run_with_timeout(code, timeout=1):
       """タイムアウト付きでコードを実行"""
       signal.signal(signal.SIGALRM, timeout_handler)
       signal.alarm(timeout)
       
       try:
           result = eval(code)
           signal.alarm(0)  # タイマーをリセット
           return result
       except TimeoutError as e:
           return str(e)
   ```

2. **サンドボックス環境の使用**: 別プロセスやコンテナでコードを実行
3. **ホワイトリスト方式**: 許可する操作だけを明示的に定義

⚠️ **注意事項**: この記事に含まれるコードのうち、`signal.alarm()`によるタイムアウト処理はUnix/Linux環境専用です。Windows環境では動作しないため、別のタイムアウト方法（例：`threading.Timer`）を検討してください。


## 7. まとめと注意点

動的コードの生成・実行は非常に柔軟で強力な手法ですが、慎重な取り扱いが求められます。

メリット：動作の変更やコード生成が実行時に可能で、DSLやプラグイン構築にも応用できます。

注意点：セキュリティリスクやデバッグの難しさ、可読性の低下、パフォーマンスへの影響が懸念されます。

推奨対策：必要最小限の範囲で使用し、入力の検証やサンドボックス化、タイムアウト・例外処理を必ず導入することが重要です。

## おわりに

Pythonの動的コード生成・実行機能は諸刃の剣です。適切な場面で慎重に使えば強力なツールになりますが、安全性を最優先に考える必要があります。

## 参考リンク

- [Python公式ドキュメント: 組み込み関数](https://docs.python.org/ja/3/library/functions.html)
- [PEP 551 -- セキュリティの透明性](https://www.python.org/dev/peps/pep-0551/)
- [AST モジュール](https://docs.python.org/ja/3/library/ast.html)