Python元编程实战:动态代码生成与AST操作高级技巧
一、Python元编程核心概念
传统编程
- 编写静态代码
- 直接操作数据
- 编译时确定行为
- 有限的运行时灵活性
元编程
- 代码生成代码
- 操作程序结构
- 运行时改变行为
- 极高的灵活性
动态类生成示例
def create_class(class_name, attributes):
"""动态创建类工厂"""
def __init__(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
cls_dict = {
'__init__': __init__,
'__slots__': tuple(attributes),
'__repr__': lambda self: f"{class_name}({', '.join(
f'{k}={getattr(self, k)!r}' for k in attributes
)})"
}
return type(class_name, (), cls_dict)
# 使用示例
Person = create_class('Person', ['name', 'age'])
p = Person(name='Alice', age=30)
print(p) # 输出: Person(name='Alice', age=30)
二、AST操作实战
1. 代码安全分析器
import ast
class SecurityVisitor(ast.NodeVisitor):
UNSAFE_FUNCS = {'eval', 'exec', 'open', '__import__'}
def visit_Call(self, node):
if isinstance(node.func, ast.Name) and node.func.id in self.UNSAFE_FUNCS:
print(f"⚠️ 安全警告: 检测到危险函数调用 - {node.func.id}")
self.generic_visit(node)
def analyze_code_security(code):
"""分析代码中的安全隐患"""
tree = ast.parse(code)
SecurityVisitor().visit(tree)
# 示例检测
code = """
import os
eval('os.system("rm -rf /")')
"""
analyze_code_security(code) # 输出: ⚠️ 安全警告: 检测到危险函数调用 - eval
2. 动态SQL生成器
class QueryBuilder:
def __init__(self, table):
self.table = table
self._select = []
self._where = []
def select(self, *columns):
self._select.extend(columns)
return self
def where(self, condition):
self._where.append(condition)
return self
def build(self):
if not self._select:
self._select = ['*']
query = f"SELECT {', '.join(self._select)} FROM {self.table}"
if self._where:
query += " WHERE " + " AND ".join(self._where)
return query
# 链式调用示例
query = (QueryBuilder("users")
.select("name", "email")
.where("age > 18")
.where("status = 'active'")
.build())
print(query) # 输出: SELECT name, email FROM users WHERE age > 18 AND status = 'active'
三、ORM框架实战案例
迷你ORM实现
class ModelMeta(type):
"""元类:将类属性映射为数据库字段"""
def __new__(cls, name, bases, attrs):
fields = {}
for k, v in attrs.items():
if isinstance(v, Field):
fields[k] = v
v.name = k
attrs['_fields'] = fields
attrs['_tablename'] = attrs.get('__tablename__', name.lower())
return super().__new__(cls, name, bases, attrs)
class Field:
"""字段描述符"""
def __init__(self, type_, default=None):
self.type = type_
self.default = default
self.name = None
def __get__(self, instance, owner):
return instance.__dict__.get(self.name, self.default)
def __set__(self, instance, value):
if not isinstance(value, self.type):
raise TypeError(f"Expected {self.type}, got {type(value)}")
instance.__dict__[self.name] = value
class Model(metaclass=ModelMeta):
"""模型基类"""
def __init__(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
@classmethod
def create_table_sql(cls):
columns = []
for name, field in cls._fields.items():
columns.append(f"{name} {field.type.__name__.upper()}")
return f"CREATE TABLE {cls._tablename} ({', '.join(columns)})"
# 使用示例
class User(Model):
__tablename__ = 'app_users'
id = Field(int)
name = Field(str)
email = Field(str, default='')
print(User.create_table_sql())
# 输出: CREATE TABLE app_users (id INTEGER, name STRING, email STRING)
四、生产环境最佳实践
- 性能优化:缓存生成的代码对象
- 安全防护:限制动态代码的执行环境
- 调试技巧:使用
inspect
模块追踪生成代码 - 类型提示:为动态生成的类添加类型注解
- 文档生成:自动为动态API生成文档字符串