はじめに
この記事は、Pythonで動的にコードを生成・実行する手法についての備忘録です。
目次
- Pythonで動的コードを実行する方法
-
eval()
とexec()
の基本 -
compile()
を使った高度な実行制御 - 動的コード生成のテクニック
- 実践的なユースケース
- セキュリティリスクと対策
- まとめと注意点
1. Pythonで動的コードを実行する方法
Pythonでは、主に以下の方法で動的にコードを実行できます:
-
eval()
: 式を評価して結果を返す -
exec()
: 文(複数行のコードブロックも可)を実行する -
compile()
: コードをバイトコードにコンパイルする - 動的モジュールインポート:
importlib
を使用 - AST(抽象構文木)による構文解析と実行
それぞれの特徴と用途を見ていきましょう。
2. eval()
とexec()
の基本
eval()
関数
eval()
は**式(expression)**を評価し、その結果を返します。式とは値を生成するコードのことで、例えば 1 + 1
、"hello" + " world"
、len([1, 2, 3])
などです。
# 基本的な使い方
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]
exec()
関数
exec()
は**文(statement)**を実行します。文とは、代入、条件分岐、ループなど、値を返さないコードのことです。複数行のコードブロックも実行できます。
# 単純な文の実行
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
名前空間の指定
eval()
とexec()
は、オプションで名前空間(グローバル変数とローカル変数)を指定できます。これにより、実行環境を制御できます。
# グローバル名前空間を指定
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はローカル名前空間に追加される)
3. compile()
を使った高度な実行制御
compile()
関数を使うと、コードをコンパイルしてコードオブジェクトを作成できます。このオブジェクトはeval()
やexec()
で実行できます。
# 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
compile()
のメリットは、一度コンパイルしたコードを何度も実行できること、またコンパイル時にエラーチェックができることです。
# コードを一度コンパイルして複数回実行
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. 動的コード生成のテクニック
実際にコードを動的に生成するテクニックを見ていきましょう。
テンプレートとしてのコード生成
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}")
動的クラス生成
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)
プログラム全体の動的生成と実行
複雑なプログラムを動的に生成し、実行する例です。
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'])
5. 実践的なユースケース
ユースケース1: 数式の動的評価
ユーザーが入力した数式を安全に評価する例です。
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
ユースケース2: 動的なデータ検証ルール
データ検証ルールを動的に構築して実行する例です。
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} は有効です")
ユースケース3: 単純なDSL(ドメイン特化言語)の実装
簡単なDSLを実装する例です。
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()
に渡すことは、リモートコード実行攻撃の危険があります。
主なリスク
- 任意のコード実行: 悪意のあるユーザーがシステムコマンドを実行する可能性
- データアクセス: 機密情報への不正アクセス
- リソース消費: 無限ループなどによるリソース枯渇攻撃
安全対策
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}")
追加のセキュリティ対策
-
タイムアウト設定: 実行時間に制限を設ける
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)
-
サンドボックス環境の使用: 別プロセスやコンテナでコードを実行
-
ホワイトリスト方式: 許可する操作だけを明示的に定義
⚠️ 注意事項: この記事に含まれるコードのうち、signal.alarm()
によるタイムアウト処理はUnix/Linux環境専用です。Windows環境では動作しないため、別のタイムアウト方法(例:threading.Timer
)を検討してください。
7. まとめと注意点
動的コードの生成・実行は非常に柔軟で強力な手法ですが、慎重な取り扱いが求められます。
メリット:動作の変更やコード生成が実行時に可能で、DSLやプラグイン構築にも応用できます。
注意点:セキュリティリスクやデバッグの難しさ、可読性の低下、パフォーマンスへの影響が懸念されます。
推奨対策:必要最小限の範囲で使用し、入力の検証やサンドボックス化、タイムアウト・例外処理を必ず導入することが重要です。
おわりに
Pythonの動的コード生成・実行機能は諸刃の剣です。適切な場面で慎重に使えば強力なツールになりますが、安全性を最優先に考える必要があります。