基础知识

在 Python 中, 反射(Reflection)是一种编程语言特性, 允许程序在运行时检查、访问和修改对象的属性和方法。Python 提供了一组内置的函数和语法来实现反射, 这些函数和语法包括:

核心反射函数

• getattr(object, name, [default]): 获取对象的属性值。如果属性不存在, 则可选择返回默认值
• setattr(object, name, value): 设置对象的属性值
• hasattr(object, name): 检查对象是否具有指定的属性
• delattr(object, name): 删除对象的属性
• dir([object]): 返回对象的属性和方法列表。如果没有提供对象, 则返回当前作用域中的所有名称
• type(object): 返回对象的类型
• isinstance(object, classinfo): 检查对象是否是指定类的实例
• issubclass(class, classinfo): 检查一个类是否是另一个类的子类

反射机制使得 Python 中的对象具有动态性, 程序可以在运行时根据需要动态地访问和操作对象的属性和方法, 而不需要提前知道对象的具体类型或结构。这种特性在编写通用、灵活和可扩展的代码时非常有用。

基本用法

属性操作

class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age
    
    def greet(self):
        return f"Hello, I'm {self.name}"

# 创建对象
person = Person("Alice", 30)

# 检查属性是否存在
print(hasattr(person, "name"))      # True
print(hasattr(person, "height"))    # False

# 获取属性值
print(getattr(person, "name"))      # "Alice"
print(getattr(person, "age"))       # 30
print(getattr(person, "height", 170))  # 170 (默认值)

# 设置属性值
setattr(person, "age", 35)
setattr(person, "height", 165)
print(person.age)                   # 35
print(person.height)                # 165

# 删除属性
delattr(person, "height")
print(hasattr(person, "height"))    # False

方法调用

# 获取方法
greet_method = getattr(person, "greet")
print(greet_method())               # "Hello, I'm Alice"

# 动态调用方法
method_name = "greet"
if hasattr(person, method_name):
    method = getattr(person, method_name)
    result = method()
    print(result)                   # "Hello, I'm Alice"

类型检查

# 检查对象类型
print(type(person))                 # <class '__main__.Person'>
print(isinstance(person, Person))   # True
print(isinstance(person, object))   # True

# 检查类继承关系
print(issubclass(Person, object))   # True
print(issubclass(str, object))      # True

高级特性

动态属性访问

class DynamicObject:
    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)
    
    def __getattr__(self, name):
        """当属性不存在时调用"""
        return f"Attribute '{name}' not found"
    
    def __setattr__(self, name, value):
        """设置属性时的自定义行为"""
        if name.startswith('_'):
            raise AttributeError(f"Cannot set private attribute '{name}'")
        super().__setattr__(name, value)

# 使用示例
obj = DynamicObject(name="test", value=42)
print(obj.name)                     # "test"
print(obj.unknown)                  # "Attribute 'unknown' not found"

try:
    obj._private = "secret"
except AttributeError as e:
    print(e)                        # "Cannot set private attribute '_private'"

反射与装饰器结合

def reflect_methods(cls):
    """装饰器：为类添加反射方法"""
    def get_methods(self):
        """获取所有方法"""
        return [method for method in dir(self) 
                if callable(getattr(self, method)) and not method.startswith('_')]
    
    def call_method(self, method_name, *args, **kwargs):
        """动态调用方法"""
        if hasattr(self, method_name):
            method = getattr(self, method_name)
            if callable(method):
                return method(*args, **kwargs)
            else:
                raise TypeError(f"'{method_name}' is not callable")
        else:
            raise AttributeError(f"'{method_name}' not found")
    
    cls.get_methods = get_methods
    cls.call_method = call_method
    return cls

@reflect_methods
class Calculator:
    def add(self, a, b):
        return a + b
    
    def multiply(self, a, b):
        return a * b
    
    def divide(self, a, b):
        return a / b if b != 0 else None

# 使用示例
calc = Calculator()
print(calc.get_methods())           # ['add', 'multiply', 'divide']
print(calc.call_method('add', 5, 3))  # 8
print(calc.call_method('multiply', 4, 7))  # 28

元类与反射

class MetaReflect(type):
    """元类：自动为类添加反射功能"""
    
    def __new__(cls, name, bases, attrs):
        # 添加反射方法
        attrs['get_attributes'] = cls._get_attributes
        attrs['get_methods'] = cls._get_methods
        attrs['get_properties'] = cls._get_properties
        attrs['reflect_call'] = cls._reflect_call
        
        return super().__new__(cls, name, bases, attrs)
    
    @staticmethod
    def _get_attributes(self):
        """获取所有属性"""
        return {name: getattr(self, name) for name in dir(self) 
                if not name.startswith('_') and not callable(getattr(self, name))}
    
    @staticmethod
    def _get_methods(self):
        """获取所有方法"""
        return {name: getattr(self, name) for name in dir(self) 
                if not name.startswith('_') and callable(getattr(self, name))}
    
    @staticmethod
    def _get_properties(self):
        """获取所有属性（包括property）"""
        return {name: getattr(self.__class__, name) for name in dir(self.__class__) 
                if isinstance(getattr(self.__class__, name), property)}
    
    @staticmethod
    def _reflect_call(self, method_name, *args, **kwargs):
        """反射调用方法"""
        if hasattr(self, method_name):
            method = getattr(self, method_name)
            if callable(method):
                return method(*args, **kwargs)
        raise AttributeError(f"Method '{method_name}' not found")

class ReflectiveClass(metaclass=MetaReflect):
    def __init__(self, name):
        self.name = name
        self._private = "secret"
    
    def public_method(self):
        return f"Public method of {self.name}"
    
    def _private_method(self):
        return "Private method"
    
    @property
    def computed_property(self):
        return f"Computed property for {self.name}"

# 使用示例
obj = ReflectiveClass("test")
print(obj.get_attributes())         # {'name': 'test'}
print(obj.get_methods())            # {'public_method': <function ...>}
print(obj.get_properties())         # {'computed_property': <property ...>}
print(obj.reflect_call('public_method'))  # "Public method of test"

实际应用案例

1. 配置管理

class ConfigManager:
    def __init__(self, config_dict=None):
        self._config = config_dict or {}
    
    def get(self, key, default=None):
        """获取配置值"""
        return getattr(self, key, default)
    
    def set(self, key, value):
        """设置配置值"""
        setattr(self, key, value)
        self._config[key] = value
    
    def has(self, key):
        """检查配置是否存在"""
        return hasattr(self, key)
    
    def delete(self, key):
        """删除配置"""
        if hasattr(self, key):
            delattr(self, key)
            del self._config[key]
    
    def get_all(self):
        """获取所有配置"""
        return {key: getattr(self, key) for key in self._config}
    
    def load_from_dict(self, config_dict):
        """从字典加载配置"""
        for key, value in config_dict.items():
            self.set(key, value)
    
    def __getattr__(self, name):
        """动态获取配置值"""
        if name in self._config:
            return self._config[name]
        raise AttributeError(f"Configuration '{name}' not found")

# 使用示例
config = ConfigManager()
config.set('database_url', 'postgresql://localhost/mydb')
config.set('debug', True)
config.set('max_connections', 100)

print(config.get('database_url'))   # 'postgresql://localhost/mydb'
print(config.has('debug'))          # True
print(config.get_all())             # {'database_url': '...', 'debug': True, ...}

# 从字典加载
config_dict = {'api_key': 'secret123', 'timeout': 30}
config.load_from_dict(config_dict)
print(config.api_key)               # 'secret123'

2. 插件系统

class PluginManager:
    def __init__(self):
        self._plugins = {}
    
    def register_plugin(self, name, plugin_class):
        """注册插件"""
        if hasattr(plugin_class, 'execute'):
            self._plugins[name] = plugin_class()
        else:
            raise ValueError(f"Plugin {name} must have 'execute' method")
    
    def get_plugin(self, name):
        """获取插件"""
        return self._plugins.get(name)
    
    def list_plugins(self):
        """列出所有插件"""
        return list(self._plugins.keys())
    
    def execute_plugin(self, name, *args, **kwargs):
        """执行插件"""
        plugin = self.get_plugin(name)
        if plugin:
            return plugin.execute(*args, **kwargs)
        else:
            raise ValueError(f"Plugin '{name}' not found")
    
    def has_plugin(self, name):
        """检查插件是否存在"""
        return name in self._plugins

class BasePlugin:
    def execute(self, *args, **kwargs):
        raise NotImplementedError("Subclasses must implement execute method")

class HelloPlugin(BasePlugin):
    def execute(self, name="World"):
        return f"Hello, {name}!"

class MathPlugin(BasePlugin):
    def execute(self, operation, a, b):
        if operation == 'add':
            return a + b
        elif operation == 'multiply':
            return a * b
        else:
            raise ValueError(f"Unknown operation: {operation}")

# 使用示例
manager = PluginManager()
manager.register_plugin('hello', HelloPlugin)
manager.register_plugin('math', MathPlugin)

print(manager.list_plugins())       # ['hello', 'math']
print(manager.execute_plugin('hello', 'Alice'))  # "Hello, Alice!"
print(manager.execute_plugin('math', 'add', 5, 3))  # 8

3. API路由系统

class APIRouter:
    def __init__(self):
        self._routes = {}
    
    def route(self, path, methods=None):
        """路由装饰器"""
        def decorator(handler):
            if methods is None:
                methods = ['GET']
            
            for method in methods:
                key = f"{method}:{path}"
                self._routes[key] = handler
            
            return handler
        return decorator
    
    def handle_request(self, method, path, *args, **kwargs):
        """处理请求"""
        key = f"{method}:{path}"
        handler = self._routes.get(key)
        
        if handler:
            return handler(*args, **kwargs)
        else:
            raise ValueError(f"No handler for {method} {path}")
    
    def get_routes(self):
        """获取所有路由"""
        return self._routes

# 使用示例
router = APIRouter()

@router.route('/users', methods=['GET'])
def get_users():
    return {'users': ['Alice', 'Bob', 'Charlie']}

@router.route('/users', methods=['POST'])
def create_user(name):
    return {'message': f'Created user: {name}'}

@router.route('/users/<id>', methods=['GET'])
def get_user(id):
    return {'user': f'User {id}'}

# 处理请求
print(router.handle_request('GET', '/users'))  # {'users': ['Alice', 'Bob', 'Charlie']}
print(router.handle_request('POST', '/users', 'David'))  # {'message': 'Created user: David'}
print(router.get_routes())  # 显示所有注册的路由

4. 数据验证器

class Validator:
    def __init__(self):
        self._validators = {}
    
    def add_validator(self, field_name, validator_func):
        """添加验证器"""
        self._validators[field_name] = validator_func
    
    def validate(self, data):
        """验证数据"""
        errors = {}
        
        for field_name, validator in self._validators.items():
            if field_name in data:
                try:
                    validator(data[field_name])
                except ValueError as e:
                    errors[field_name] = str(e)
            else:
                errors[field_name] = "Field is required"
        
        if errors:
            raise ValueError(f"Validation errors: {errors}")
        
        return True

class UserValidator(Validator):
    def __init__(self):
        super().__init__()
        
        # 添加验证器
        self.add_validator('name', self._validate_name)
        self.add_validator('age', self._validate_age)
        self.add_validator('email', self._validate_email)
    
    def _validate_name(self, name):
        if not isinstance(name, str) or len(name) < 2:
            raise ValueError("Name must be a string with at least 2 characters")
    
    def _validate_age(self, age):
        if not isinstance(age, int) or age < 0 or age > 150:
            raise ValueError("Age must be an integer between 0 and 150")
    
    def _validate_email(self, email):
        if not isinstance(email, str) or '@' not in email:
            raise ValueError("Email must be a valid email address")

# 使用示例
validator = UserValidator()

# 有效数据
try:
    validator.validate({
        'name': 'Alice',
        'age': 25,
        'email': '[email protected]'
    })
    print("Validation passed")
except ValueError as e:
    print(f"Validation failed: {e}")

# 无效数据
try:
    validator.validate({
        'name': 'A',
        'age': 200,
        'email': 'invalid-email'
    })
except ValueError as e:
    print(f"Validation failed: {e}")

性能优化技巧

1. 缓存反射结果

from functools import lru_cache

class OptimizedReflector:
    def __init__(self, obj):
        self.obj = obj
        self._cache = {}
    
    @lru_cache(maxsize=128)
    def get_attribute(self, name, default=None):
        """缓存属性获取结果"""
        return getattr(self.obj, name, default)
    
    def get_method(self, name):
        """获取方法（带缓存）"""
        if name not in self._cache:
            if hasattr(self.obj, name):
                method = getattr(self.obj, name)
                if callable(method):
                    self._cache[name] = method
                else:
                    raise TypeError(f"'{name}' is not callable")
            else:
                raise AttributeError(f"'{name}' not found")
        
        return self._cache[name]
    
    def call_method(self, name, *args, **kwargs):
        """调用方法（带缓存）"""
        method = self.get_method(name)
        return method(*args, **kwargs)

# 使用示例
class TestClass:
    def __init__(self):
        self.counter = 0
    
    def increment(self):
        self.counter += 1
        return self.counter
    
    def get_counter(self):
        return self.counter

obj = TestClass()
reflector = OptimizedReflector(obj)

# 多次调用相同方法（缓存会提高性能）
for _ in range(1000):
    reflector.call_method('increment')

2. 批量反射操作

class BatchReflector:
    def __init__(self, obj):
        self.obj = obj
    
    def get_multiple_attributes(self, names, defaults=None):
        """批量获取属性"""
        if defaults is None:
            defaults = {}
        
        result = {}
        for name in names:
            default = defaults.get(name, None)
            result[name] = getattr(self.obj, name, default)
        
        return result
    
    def set_multiple_attributes(self, attributes_dict):
        """批量设置属性"""
        for name, value in attributes_dict.items():
            setattr(self.obj, name, value)
    
    def call_multiple_methods(self, method_calls):
        """批量调用方法"""
        results = {}
        for method_name, args_kwargs in method_calls.items():
            if hasattr(self.obj, method_name):
                method = getattr(self.obj, method_name)
                if callable(method):
                    args = args_kwargs.get('args', ())
                    kwargs = args_kwargs.get('kwargs', {})
                    results[method_name] = method(*args, **kwargs)
                else:
                    results[method_name] = f"'{method_name}' is not callable"
            else:
                results[method_name] = f"'{method_name}' not found"
        
        return results

# 使用示例
class DataProcessor:
    def __init__(self):
        self.data = []
        self.processed = False
    
    def add_data(self, item):
        self.data.append(item)
    
    def process(self):
        self.data = [x * 2 for x in self.data]
        self.processed = True
    
    def get_sum(self):
        return sum(self.data)
    
    def get_count(self):
        return len(self.data)

processor = DataProcessor()
batch_reflector = BatchReflector(processor)

# 批量设置属性
batch_reflector.set_multiple_attributes({
    'data': [1, 2, 3, 4, 5],
    'processed': False
})

# 批量获取属性
attributes = batch_reflector.get_multiple_attributes(['data', 'processed'])
print(attributes)  # {'data': [1, 2, 3, 4, 5], 'processed': False}

# 批量调用方法
method_calls = {
    'process': {'args': (), 'kwargs': {}},
    'get_sum': {'args': (), 'kwargs': {}},
    'get_count': {'args': (), 'kwargs': {}}
}

results = batch_reflector.call_multiple_methods(method_calls)
print(results)  # {'process': None, 'get_sum': 30, 'get_count': 5}

调试和测试技巧

1. 反射调试工具

class ReflectionDebugger:
    def __init__(self, obj):
        self.obj = obj
    
    def inspect_object(self):
        """详细检查对象"""
        info = {
            'type': type(self.obj).__name__,
            'module': type(self.obj).__module__,
            'attributes': {},
            'methods': {},
            'properties': {},
            'private_attrs': {}
        }
        
        for name in dir(self.obj):
            try:
                attr = getattr(self.obj, name)
                
                if name.startswith('_'):
                    info['private_attrs'][name] = {
                        'type': type(attr).__name__,
                        'value': str(attr)[:100] + '...' if len(str(attr)) > 100 else str(attr)
                    }
                elif callable(attr):
                    if isinstance(attr, property):
                        info['properties'][name] = {
                            'type': 'property',
                            'fget': attr.fget.__name__ if attr.fget else None,
                            'fset': attr.fset.__name__ if attr.fset else None
                        }
                    else:
                        info['methods'][name] = {
                            'type': 'method',
                            'signature': str(attr)
                        }
                else:
                    info['attributes'][name] = {
                        'type': type(attr).__name__,
                        'value': str(attr)[:100] + '...' if len(str(attr)) > 100 else str(attr)
                    }
            except Exception as e:
                info['attributes'][name] = {'error': str(e)}
        
        return info
    
    def print_inspection(self):
        """打印检查结果"""
        info = self.inspect_object()
        
        print(f"=== Object Inspection: {info['type']} ===")
        print(f"Module: {info['module']}")
        
        print("\n--- Attributes ---")
        for name, details in info['attributes'].items():
            print(f"  {name}: {details}")
        
        print("\n--- Methods ---")
        for name, details in info['methods'].items():
            print(f"  {name}: {details}")
        
        print("\n--- Properties ---")
        for name, details in info['properties'].items():
            print(f"  {name}: {details}")
        
        print("\n--- Private Attributes ---")
        for name, details in info['private_attrs'].items():
            print(f"  {name}: {details}")

# 使用示例
class TestObject:
    def __init__(self):
        self.public_attr = "public"
        self._private_attr = "private"
    
    def public_method(self):
        return "public method"
    
    def _private_method(self):
        return "private method"
    
    @property
    def computed_property(self):
        return "computed"
    
    @computed_property.setter
    def computed_property(self, value):
        self._private_attr = value

obj = TestObject()
debugger = ReflectionDebugger(obj)
debugger.print_inspection()

2. 反射单元测试

import unittest
from unittest.mock import Mock, patch

class ReflectionTestCase(unittest.TestCase):
    def setUp(self):
        self.test_obj = Mock()
        self.test_obj.name = "test"
        self.test_obj.value = 42
        self.test_obj.method = Mock(return_value="result")
    
    def test_hasattr(self):
        """测试hasattr功能"""
        self.assertTrue(hasattr(self.test_obj, 'name'))
        self.assertTrue(hasattr(self.test_obj, 'method'))
        self.assertFalse(hasattr(self.test_obj, 'nonexistent'))
    
    def test_getattr(self):
        """测试getattr功能"""
        self.assertEqual(getattr(self.test_obj, 'name'), 'test')
        self.assertEqual(getattr(self.test_obj, 'nonexistent', 'default'), 'default')
        
        with self.assertRaises(AttributeError):
            getattr(self.test_obj, 'nonexistent')
    
    def test_setattr(self):
        """测试setattr功能"""
        setattr(self.test_obj, 'new_attr', 'new_value')
        self.assertEqual(self.test_obj.new_attr, 'new_value')
    
    def test_delattr(self):
        """测试delattr功能"""
        delattr(self.test_obj, 'name')
        self.assertFalse(hasattr(self.test_obj, 'name'))
        
        with self.assertRaises(AttributeError):
            delattr(self.test_obj, 'nonexistent')
    
    def test_method_call(self):
        """测试方法调用"""
        method = getattr(self.test_obj, 'method')
        result = method()
        self.assertEqual(result, 'result')
        self.test_obj.method.assert_called_once()

class CustomReflectionTest(unittest.TestCase):
    def test_dynamic_attribute_access(self):
        """测试动态属性访问"""
        class DynamicClass:
            def __getattr__(self, name):
                return f"Dynamic: {name}"
        
        obj = DynamicClass()
        self.assertEqual(obj.any_attribute, "Dynamic: any_attribute")
    
    def test_property_reflection(self):
        """测试属性反射"""
        class PropertyClass:
            def __init__(self):
                self._value = 0
            
            @property
            def value(self):
                return self._value
            
            @value.setter
            def value(self, val):
                self._value = val
        
        obj = PropertyClass()
        
        # 测试属性获取
        self.assertEqual(getattr(obj, 'value'), 0)
        
        # 测试属性设置
        setattr(obj, 'value', 42)
        self.assertEqual(obj.value, 42)

if __name__ == '__main__':
    unittest.main()

常见陷阱和解决方案

1. 属性名冲突

class SafeReflector:
    def __init__(self, obj):
        self.obj = obj
    
    def safe_getattr(self, name, default=None):
        """安全获取属性，避免与内置方法冲突"""
        # 检查是否是内置方法
        if hasattr(self.obj, '__getattr__'):
            try:
                return getattr(self.obj, name, default)
            except AttributeError:
                return default
        
        # 检查属性是否存在
        if hasattr(self.obj, name):
            attr = getattr(self.obj, name)
            # 避免调用描述符
            if hasattr(attr, '__get__'):
                return attr.__get__(self.obj, type(self.obj))
            return attr
        
        return default
    
    def safe_setattr(self, name, value):
        """安全设置属性"""
        # 检查是否是只读属性
        if hasattr(self.obj, name):
            attr = getattr(self.obj, name)
            if hasattr(attr, '__set__'):
                try:
                    attr.__set__(self.obj, value)
                    return
                except AttributeError:
                    raise AttributeError(f"Cannot set read-only attribute '{name}'")
        
        setattr(self.obj, name, value)

# 使用示例
class ReadOnlyClass:
    def __init__(self):
        self._value = 42
    
    @property
    def value(self):
        return self._value

obj = ReadOnlyClass()
reflector = SafeReflector(obj)

print(reflector.safe_getattr('value'))  # 42

try:
    reflector.safe_setattr('value', 100)
except AttributeError as e:
    print(f"Error: {e}")  # "Cannot set read-only attribute 'value'"

2. 循环引用问题

import weakref

class WeakReflector:
    def __init__(self, obj):
        self._obj_ref = weakref.ref(obj)
    
    def get_object(self):
        """获取对象（如果还存在）"""
        obj = self._obj_ref()
        if obj is None:
            raise ReferenceError("Object has been garbage collected")
        return obj
    
    def getattr(self, name, default=None):
        """安全获取属性"""
        try:
            obj = self.get_object()
            return getattr(obj, name, default)
        except ReferenceError:
            return default
    
    def setattr(self, name, value):
        """安全设置属性"""
        obj = self.get_object()
        setattr(obj, name, value)

# 使用示例
class TestClass:
    def __init__(self):
        self.data = "test"

obj = TestClass()
weak_reflector = WeakReflector(obj)

print(weak_reflector.getattr('data'))  # "test"

# 删除对象
del obj

try:
    weak_reflector.getattr('data')
except ReferenceError as e:
    print(f"Object collected: {e}")

3. 性能优化建议

import time
from functools import wraps

def measure_performance(func):
    """性能测量装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"{func.__name__} took {end_time - start_time:.6f} seconds")
        return result
    return wrapper

class OptimizedReflection:
    def __init__(self, obj):
        self.obj = obj
        self._attr_cache = {}
        self._method_cache = {}
    
    @measure_performance
    def get_attribute_cached(self, name, default=None):
        """带缓存的属性获取"""
        if name not in self._attr_cache:
            self._attr_cache[name] = getattr(self.obj, name, default)
        return self._attr_cache[name]
    
    @measure_performance
    def get_method_cached(self, name):
        """带缓存的方法获取"""
        if name not in self._method_cache:
            if hasattr(self.obj, name):
                method = getattr(self.obj, name)
                if callable(method):
                    self._method_cache[name] = method
                else:
                    raise TypeError(f"'{name}' is not callable")
            else:
                raise AttributeError(f"'{name}' not found")
        return self._method_cache[name]
    
    def clear_cache(self):
        """清除缓存"""
        self._attr_cache.clear()
        self._method_cache.clear()

# 性能测试
class PerformanceTest:
    def __init__(self):
        self.data = list(range(1000))
    
    def process_data(self):
        return sum(self.data)
    
    def get_data(self):
        return self.data

# 测试性能
test_obj = PerformanceTest()
reflector = OptimizedReflection(test_obj)

# 第一次调用（较慢）
reflector.get_attribute_cached('data')
reflector.get_method_cached('process_data')

# 第二次调用（较快，使用缓存）
reflector.get_attribute_cached('data')
reflector.get_method_cached('process_data')

最佳实践

1. 错误处理

class RobustReflector:
    def __init__(self, obj):
        self.obj = obj
    
    def safe_reflect(self, operation, *args, **kwargs):
        """安全的反射操作"""
        try:
            if operation == 'getattr':
                return getattr(self.obj, *args, **kwargs)
            elif operation == 'setattr':
                return setattr(self.obj, *args, **kwargs)
            elif operation == 'hasattr':
                return hasattr(self.obj, *args, **kwargs)
            elif operation == 'delattr':
                return delattr(self.obj, *args, **kwargs)
            else:
                raise ValueError(f"Unknown operation: {operation}")
        except AttributeError as e:
            print(f"Attribute error: {e}")
            return None
        except TypeError as e:
            print(f"Type error: {e}")
            return None
        except Exception as e:
            print(f"Unexpected error: {e}")
            return None
    
    def reflect_with_fallback(self, attr_name, fallback_func):
        """带回退的反射操作"""
        try:
            return getattr(self.obj, attr_name)
        except AttributeError:
            return fallback_func()

# 使用示例
class FallbackClass:
    def fallback_method(self):
        return "fallback result"

obj = FallbackClass()
reflector = RobustReflector(obj)

# 安全反射
result = reflector.safe_reflect('getattr', 'nonexistent', 'default')
print(result)  # "default"

# 带回退的反射
def fallback():
    return "computed value"

result = reflector.reflect_with_fallback('nonexistent', fallback)
print(result)  # "computed value"

2. 类型安全

from typing import Any, Optional, Callable, Dict, List

class TypedReflector:
    def __init__(self, obj: Any):
        self.obj = obj
        self._type_cache: Dict[str, type] = {}
    
    def get_typed_attribute(self, name: str, expected_type: type, default: Any = None) -> Any:
        """获取类型安全的属性"""
        try:
            value = getattr(self.obj, name, default)
            if isinstance(value, expected_type):
                return value
            else:
                print(f"Warning: {name} is {type(value)}, expected {expected_type}")
                return default
        except Exception as e:
            print(f"Error getting {name}: {e}")
            return default
    
    def call_method_with_types(self, method_name: str, 
                              args: List[Any] = None, 
                              kwargs: Dict[str, Any] = None,
                              expected_return_type: Optional[type] = None) -> Any:
        """类型安全的方法调用"""
        if args is None:
            args = []
        if kwargs is None:
            kwargs = {}
        
        try:
            method = getattr(self.obj, method_name)
            if not callable(method):
                raise TypeError(f"'{method_name}' is not callable")
            
            result = method(*args, **kwargs)
            
            if expected_return_type and not isinstance(result, expected_return_type):
                print(f"Warning: {method_name} returned {type(result)}, expected {expected_return_type}")
            
            return result
        except Exception as e:
            print(f"Error calling {method_name}: {e}")
            return None

# 使用示例
class TypedClass:
    def __init__(self):
        self.name: str = "test"
        self.count: int = 42
        self.data: List[int] = [1, 2, 3]
    
    def get_name(self) -> str:
        return self.name
    
    def get_count(self) -> int:
        return self.count
    
    def process_data(self, multiplier: int) -> List[int]:
        return [x * multiplier for x in self.data]

obj = TypedClass()
typed_reflector = TypedReflector(obj)

# 类型安全的属性获取
name = typed_reflector.get_typed_attribute('name', str)
count = typed_reflector.get_typed_attribute('count', int)
print(f"Name: {name}, Count: {count}")

# 类型安全的方法调用
result = typed_reflector.call_method_with_types('get_name', expected_return_type=str)
print(f"Result: {result}")

processed = typed_reflector.call_method_with_types('process_data', args=[2], expected_return_type=list)
print(f"Processed: {processed}")

3. 文档和注释

class DocumentedReflector:
    """
    文档化的反射器类
    
    这个类提供了安全的反射操作，包括属性访问、方法调用等。
    所有操作都有适当的错误处理和类型检查。
    
    Attributes:
        obj: 要反射的对象
        _cache: 内部缓存，用于提高性能
    """
    
    def __init__(self, obj: Any):
        """
        初始化反射器
        
        Args:
            obj: 要反射的对象
        """
        self.obj = obj
        self._cache: Dict[str, Any] = {}
    
    def get_attribute(self, name: str, default: Any = None) -> Any:
        """
        获取对象属性
        
        Args:
            name: 属性名
            default: 默认值，当属性不存在时返回
            
        Returns:
            属性值或默认值
            
        Raises:
            AttributeError: 当属性不存在且没有提供默认值时
        """
        if name in self._cache:
            return self._cache[name]
        
        value = getattr(self.obj, name, default)
        self._cache[name] = value
        return value
    
    def set_attribute(self, name: str, value: Any) -> None:
        """
        设置对象属性
        
        Args:
            name: 属性名
            value: 要设置的值
            
        Raises:
            AttributeError: 当属性是只读时
        """
        setattr(self.obj, name, value)
        # 更新缓存
        self._cache[name] = value
    
    def call_method(self, method_name: str, *args, **kwargs) -> Any:
        """
        调用对象方法
        
        Args:
            method_name: 方法名
            *args: 位置参数
            **kwargs: 关键字参数
            
        Returns:
            方法调用的结果
            
        Raises:
            AttributeError: 当方法不存在时
            TypeError: 当属性不是可调用对象时
        """
        method = getattr(self.obj, method_name)
        if not callable(method):
            raise TypeError(f"'{method_name}' is not callable")
        
        return method(*args, **kwargs)
    
    def get_method_signature(self, method_name: str) -> Optional[str]:
        """
        获取方法签名
        
        Args:
            method_name: 方法名
            
        Returns:
            方法签名字符串，如果方法不存在则返回None
        """
        try:
            method = getattr(self.obj, method_name)
            if callable(method):
                import inspect
                return str(inspect.signature(method))
        except (AttributeError, TypeError):
            pass
        return None

# 使用示例
class ExampleClass:
    """示例类，用于演示反射功能"""
    
    def __init__(self, name: str, value: int):
        self.name = name
        self.value = value
    
    def process(self, multiplier: int = 1) -> int:
        """处理数据"""
        return self.value * multiplier
    
    def get_info(self) -> str:
        """获取信息"""
        return f"{self.name}: {self.value}"

# 创建文档化的反射器
obj = ExampleClass("test", 42)
doc_reflector = DocumentedReflector(obj)

# 使用反射器
print(doc_reflector.get_attribute('name'))  # "test"
print(doc_reflector.get_method_signature('process'))  # "(multiplier: int = 1) -> int"
print(doc_reflector.call_method('get_info'))  # "test: 42"

扩展资源

1. 相关库

• inspect: Python标准库，提供更强大的内省功能
• types: Python标准库，用于类型检查和操作
• abc: Python标准库，抽象基类，支持反射
• dataclasses: Python标准库，自动生成特殊方法
• attrs: 第三方库，简化类定义和反射
• pydantic: 第三方库，数据验证和反射

2. 学习资源

• Python官方文档: https://docs.python.org/3/library/functions.html#getattr
• Python内省教程: https://docs.python.org/3/library/inspect.html
• Python元编程指南: https://docs.python.org/3/reference/datamodel.html

3. 社区资源

• Stack Overflow: Python反射相关问题
• GitHub: Python反射相关项目
• Reddit: r/Python社区讨论

link

• Python反射机制详解 (opens new window)
• Python内省和反射 (opens new window)
• Python元编程 (opens new window) </rewritten_file>