3.12.反射

1.概述

反射（Reflection）是指在程序运行时检查、访问和修改其自身状态和行为的能力。Python 作为一种高度动态的语言，天然支持强大的反射机制。

2.核心价值

动态性：在运行时操作对象
灵活性：根据条件动态调用方法
通用性：编写可重用的通用代码
内省：检查对象的结构和能力

3.基本反射函数

3.1.核心函数

函数	描述	返回值
`hasattr(obj, name)`	检查对象是否有指定属性	布尔值
`getattr(obj, name[, default])`	获取对象属性值	属性值或默认值
`setattr(obj, name, value)`	设置对象属性值	None
`delattr(obj, name)`	删除对象属性	None
`dir(obj)`	获取对象所有属性和方法	列表

3.2.基本用法

class Example:
    def __init__(self):
        self.value = 42

example = Example()

# 检查属性是否存在
print(hasattr(example, 'value'))  # True

# 获取属性值
print(getattr(example, 'value'))  # 42

# 设置属性值
setattr(example, 'new_attr', 'hello')
print(example.new_attr)  # hello

# 删除属性
delattr(example, 'new_attr')
print('new_attr' in dir(example))  # False

3.3.安全的属性访问

class Config:
    def __init__(self):
        self.host = "localhost"
        self.port = 8080

config = Config()

# 安全的属性访问，提供默认值
timeout = getattr(config, 'timeout', 30)  # 如果不存在，返回默认值
print(f"Timeout: {timeout}")  # Timeout: 30

debug = getattr(config, 'debug', False)
print(f"Debug: {debug}")  # Debug: False

4.内省函数

4.1.类型检查函数

函数	描述	示例
`type(obj)`	返回对象类型	`type(123)` → `<class 'int'>`
`isinstance(obj, class)`	检查对象是否为指定类型	`isinstance(123, int)` → `True`
`issubclass(cls, class)`	检查类继承关系	`issubclass(Bar, Foo)` → `True`

4.2.对象信息函数

函数	描述	示例
`dir(obj)`	获取对象所有属性和方法	`dir(obj)` → `['attr1', 'method1', ...]`
`vars(obj)`	获取对象的属性字典	`vars(obj)` → `{'attr1': 'value1'}`
`obj.__dict__`	对象的属性字典	`obj.__dict__` → `{'attr1': 'value1'}`

4.3.示例代码

class Foo:
    def __init__(self):
        self.attr1 = "hello"

class Bar(Foo):
    def method1(self):
        return "world"

foo = Foo()
bar = Bar()

# 类型检查
print(type(foo))                    # <class '__main__.Foo'>
print(isinstance(bar, Foo))         # True
print(issubclass(Bar, Foo))         # True

# 对象信息
print(hasattr(foo, "attr1"))        # True
print(getattr(foo, "attr1"))        # hello
print(vars(foo))                    # {'attr1': 'hello'}
print(dir(foo))                     # ['__class__', ..., 'attr1']

4.4.自动发现对象成员

def show_object_members(obj):
    print(f"对象类型: {type(obj)}")
    for attr in dir(obj):
        if attr.startswith('_'):
            continue
        value = getattr(obj, attr)
        if callable(value):
            print(f"[方法] {attr}()")
        else:
            print(f"[属性] {attr} = {value}")

# 使用示例
show_object_members(123)

5.动态方法调用

5.1.基本模式

动态调用对象方法的步骤：

用字符串确定方法名
用 getattr(obj, method_name) 获取方法对象
检查该方法对象是否可调用（callable() 可选）
用参数调用该方法

5.2.示例代码

class ExampleObj:
    def calculate(self, x, y):
        return x + y

obj = ExampleObj()
method_name = "calculate"
args = (3, 5)

if hasattr(obj, method_name):
    method = getattr(obj, method_name)
    if callable(method):
        try:
            result = method(*args)
            print("Result:", result)  # Result: 8
        except Exception as e:
            print(f"调用方法 {method_name} 时发生异常:", e)
    else:
        print(f"{method_name} 不是可调用对象")
else:
    print(f"对象没有方法 {method_name}")

5.3.带参数处理的方法调用

class APIHandler:
    def get_user(self, user_id, include_profile=False):
        user = {"id": user_id, "name": f"User{user_id}"}
        if include_profile:
            user["profile"] = {"email": f"user{user_id}@example.com"}
        return user
    
    def create_post(self, title, content, tags=None, published=True):
        post = {
            "title": title,
            "content": content,
            "tags": tags or [],
            "published": published
        }
        return post

def call_method_dynamically(obj, method_name, **kwargs):
    if not hasattr(obj, method_name):
        raise AttributeError(f"对象没有方法 '{method_name}'")
    
    method = getattr(obj, method_name)
    if not callable(method):
        raise TypeError(f"'{method_name}' 不是可调用方法")
    
    return method(**kwargs)

# 使用示例
handler = APIHandler()

user_data = call_method_dynamically(
    handler, 
    "get_user", 
    user_id=123, 
    include_profile=True
)
print("用户数据:", user_data)

post_data = call_method_dynamically(
    handler,
    "create_post",
    title="Python反射",
    content="反射机制详解",
    tags=["python", "reflection"],
    published=False
)
print("文章数据:", post_data)

6.类级别反射

6.1.概念

类级别反射是指直接作用于类本身的反射操作，可以：

动态获取和修改类属性（包括类变量）
调用类方法和静态方法
检查类中的方法和属性是否存在
动态给类添加属性或方法

6.2.示例代码

class DatabaseModel:
    table_name = "users"
    connection_string = "sqlite:///database.db"
    
    def __init__(self, id, name):
        self.id = id
        self.name = name
    
    @classmethod
    def get_table_info(cls):
        return f"表名: {cls.table_name}, 连接: {cls.connection_string}"
    
    @staticmethod
    def validate_data(data):
        return isinstance(data, dict)
    
    def save(self):
        return f"保存 {self.name} 到数据库"

# 类级别的反射操作
print("类属性:")
print(f"table_name: {getattr(DatabaseModel, 'table_name')}")
print(f"connection_string: {getattr(DatabaseModel, 'connection_string')}")

# 动态修改类属性
setattr(DatabaseModel, 'table_name', 'customers')
print(f"修改后的表名: {DatabaseModel.table_name}")

# 调用类方法
class_method = getattr(DatabaseModel, 'get_table_info')
print(class_method())  # 表名: customers, 连接: sqlite:///database.db

# 调用静态方法
static_method = getattr(DatabaseModel, 'validate_data')
print(static_method({"key": "value"}))  # True

# 检查类成员
print("\n类成员检查:")
print(f"有table_name: {hasattr(DatabaseModel, 'table_name')}")
print(f"有get_table_info: {hasattr(DatabaseModel, 'get_table_info')}")
print(f"有save: {hasattr(DatabaseModel, 'save')}")  # True，但需要实例调用

7.模块级别反射

7.1.概念

模块级反射是指在运行时动态地检查、导入、操作模块及其内容的能力。它可以帮助我们开发出高度可配置、插件化、支持热加载的系统。

7.2.常用函数

函数	描述	示例
`importlib.import_module(name)`	按名称动态导入模块	`importlib.import_module("math")`
`getattr(module, name[, default])`	动态获取模块中的函数、类、变量	`getattr(math, "sqrt")`
`hasattr(module, name)`	判断模块是否有某个属性	`hasattr(math, "pow")`
`dir(module)`	获取模块内定义的名称列表	`dir(math)`
`importlib.reload(module)`	重新加载已加载的模块	`importlib.reload(math)`

7.3.基本示例

import importlib

# 动态导入模块
module_name = "math"
math_mod = importlib.import_module(module_name)

print(math_mod)  # <module 'math' (built-in)>
print(getattr(math_mod, "sqrt")(16))  # 4.0

# 检查模块是否有某个属性
if hasattr(math_mod, "pow"):
    pow_func = getattr(math_mod, "pow")
    print(pow_func(2, 8))  # 256.0

7.4.插件系统实现

import os
import importlib

class PluginManager:
    def __init__(self):
        self.plugins = {}
        self.load_plugins()
    
    def load_plugins(self):
        """动态加载所有插件"""
        plugins_dir = "plugins"
        
        for filename in os.listdir(plugins_dir):
            if filename.endswith('.py') and not filename.startswith('_'):
                module_name = filename[:-3]  # 移除 .py
                try:
                    # 动态导入模块
                    module = importlib.import_module(f"{plugins_dir}.{module_name}")
                    
                    # 查找插件类（约定：以Plugin结尾的类）
                    for attr_name in dir(module):
                        attr = getattr(module, attr_name)
                        if (isinstance(attr, type) and 
                            attr_name.endswith('Plugin') and 
                            attr_name != 'PluginManager'):
                            
                            plugin_instance = attr()
                            self.plugins[plugin_instance.name] = plugin_instance
                            print(f"加载插件: {plugin_instance.name} v{plugin_instance.version}")
                            
                except Exception as e:
                    print(f"加载插件 {module_name} 失败: {e}")
    
    def execute_plugin(self, plugin_name, *args, **kwargs):
        """执行插件"""
        if plugin_name in self.plugins:
            plugin = self.plugins[plugin_name]
            return plugin.execute(*args, **kwargs)
        else:
            return f"Error: Plugin '{plugin_name}' not found"
    
    def list_plugins(self):
        """列出所有可用插件"""
        return list(self.plugins.keys())

# 使用插件系统
manager = PluginManager()
print("可用插件:", manager.list_plugins())

# 动态调用插件
result1 = manager.execute_plugin("Basic Calculator", "add", 10, 5)
result2 = manager.execute_plugin("Greeter", "Alice", formal=True)

print("计算结果:", result1)  # 15
print("问候结果:", result2)  # Good day, Alice. How may I assist you?

8.魔术方法和反射

8.1.控制属性访问的魔术方法

方法	描述	触发时机
`__getattr__(self, name)`	访问不存在的属性时调用	`obj.nonexistent`
`__setattr__(self, name, value)`	所有属性赋值操作时调用	`obj.attr = value`
`__delattr__(self, name)`	删除属性时调用	`del obj.attr`
`__dir__(self)`	决定`dir(obj)`时返回的属性列表	`dir(obj)`

8.2.动态属性管理示例

class DynamicAttributes:
    """动态属性管理类"""
    
    def __init__(self):
        self._data = {}
        self._access_count = {}
    
    def __getattr__(self, name):
        """访问不存在的属性时调用"""
        if name in self._data:
            self._access_count[name] = self._access_count.get(name, 0) + 1
            return self._data[name]
        else:
            raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
    
    def __setattr__(self, name, value):
        """赋值属性时调用"""
        if name.startswith("_"):
            super().__setattr__(name, value)
        else:
            if "_data" in self.__dict__:
                self._data[name] = value
                self._access_count[name] = 0
            else:
                super().__setattr__(name, value)
    
    def __delattr__(self, name):
        """删除属性时调用"""
        if name in self._data:
            del self._data[name]
            del self._access_count[name]
        else:
            super().__delattr__(name)
    
    def __dir__(self):
        """自定义 dir() 输出"""
        base_attrs = super().__dir__()
        dynamic_attrs = list(self._data.keys())
        return base_attrs + dynamic_attrs
    
    def get_access_stats(self):
        """返回访问次数统计表"""
        return self._access_count.copy()

# 使用示例
obj = DynamicAttributes()
obj.foo = 123
obj.bar = "hello"

print(obj.foo)         # 123
print(obj.bar)         # hello
print(obj.foo)         # 再次访问

del obj.bar

print("所有属性：", dir(obj))
print("访问次数统计：", obj.get_access_stats())

9.实际应用场景

9.1.配置管理系统

class ConfigManager:
    def __init__(self, config_dict=None):
        self._config = config_dict or {}
        self._defaults = {
            'debug': False,
            'host': 'localhost',
            'port': 8080,
            'timeout': 30
        }
    
    def __getattr__(self, name):
        """通过属性访问配置"""
        if name in self._config:
            return self._config[name]
        elif name in self._defaults:
            return self._defaults[name]
        else:
            raise AttributeError(f"配置项 '{name}' 不存在")
    
    def set_config(self, **kwargs):
        """批量设置配置"""
        for key, value in kwargs.items():
            self._config[key] = value
    
    def show_config(self):
        """显示所有配置"""
        all_config = {**self._defaults, **self._config}
        for key, value in sorted(all_config.items()):
            source = "自定义" if key in self._config else "默认"
            print(f"{key:20} = {value:15} [{source}]")

# 使用示例
config = ConfigManager()

print(f"Host: {config.host}")        # localhost
print(f"Port: {config.port}")        # 8080
print(f"Debug: {config.debug}")      # False

config.set_config(host="127.0.0.1", port=9000, new_setting="custom")

print("\n所有配置:")
config.show_config()

9.2.API路由映射

class APIController:
    def get_users(self, user_id=None):
        if user_id:
            return f"获取用户 {user_id}"
        return "获取所有用户"
    
    def create_user(self, name, email):
        return f"创建用户: {name} ({email})"
    
    def update_user(self, user_id, **data):
        return f"更新用户 {user_id}: {data}"
    
    def delete_user(self, user_id):
        return f"删除用户 {user_id}"

class Router:
    def __init__(self):
        self.controller = APIController()
    
    def route(self, method, path, **params):
        """路由请求到对应方法"""
        path_parts = path.strip('/').split('/')
        
        if path_parts[0] == 'users':
            if method.upper() == 'GET':
                action = 'get_users'
                if len(path_parts) > 1:
                    params['user_id'] = path_parts[1]
            elif method.upper() == 'POST':
                action = 'create_user'
            elif method.upper() == 'PUT' and len(path_parts) > 1:
                action = 'update_user'
                params['user_id'] = path_parts[1]
            elif method.upper() == 'DELETE' and len(path_parts) > 1:
                action = 'delete_user'
                params['user_id'] = path_parts[1]
            else:
                return "Error: 无效的请求"
            
            if hasattr(self.controller, action):
                method_to_call = getattr(self.controller, action)
                return method_to_call(**params)
            else:
                return f"Error: 操作 {action} 不存在"
        
        return "Error: 路径不存在"

# 使用示例
router = Router()

requests = [
    ('GET', '/users'),
    ('GET', '/users/123'),
    ('POST', '/users', {'name': 'Alice', 'email': 'alice@example.com'}),
    ('PUT', '/users/123', {'name': 'Alice Smith'}),
    ('DELETE', '/users/123')
]

for method, path, *args in requests:
    params = args[0] if args else {}
    result = router.route(method, path, **params)
    print(f"{method} {path} -> {result}")

9.3.数据验证框架

class Validator:
    @staticmethod
    def validate_string(value, min_length=0, max_length=None):
        if not isinstance(value, str):
            return False, "必须为字符串"
        if len(value) < min_length:
            return False, f"长度不能小于{min_length}"
        if max_length and len(value) > max_length:
            return False, f"长度不能大于{max_length}"
        return True, "验证通过"
    
    @staticmethod
    def validate_number(value, min_val=None, max_val=None):
        if not isinstance(value, (int, float)):
            return False, "必须为数字"
        if min_val is not None and value < min_val:
            return False, f"不能小于{min_val}"
        if max_val is not None and value > max_val:
            return False, f"不能大于{max_val}"
        return True, "验证通过"
    
    @staticmethod
    def validate_email(value):
        if not isinstance(value, str) or '@' not in value:
            return False, "无效的邮箱格式"
        return True, "验证通过"

class DataModel:
    def __init__(self, **kwargs):
        self._validators = self._discover_validators()
        self._errors = {}
        for key, value in kwargs.items():
            setattr(self, key, value)
    
    def _discover_validators(self):
        """自动发现验证方法"""
        validators = {}
        for attr_name in dir(Validator):
            if attr_name.startswith('validate_'):
                field_name = attr_name[9:]  # 移除 'validate_' 前缀
                validators[field_name] = getattr(Validator, attr_name)
        return validators
    
    def validate(self):
        """验证所有字段"""
        self._errors = {}
        for field_name, validator in self._validators.items():
            if hasattr(self, field_name):
                value = getattr(self, field_name)
                is_valid, message = validator(value)
                if not is_valid:
                    self._errors[field_name] = message
        return len(self._errors) == 0
    
    def get_errors(self):
        return self._errors

# 使用示例
user_data = DataModel(
    name="Alice",
    age=25,
    email="alice@example.com"
)

user_data.password = "weak"  # 这个不会被验证，因为没有对应的验证器

if user_data.validate():
    print("数据验证通过!")
else:
    print("验证错误:", user_data.get_errors())

10.最佳实践和注意事项

10.1.最佳实践

10.1.1.1. 安全的属性访问

def safe_getattr(obj, attr_name, default=None):
    """安全地获取对象属性"""
    try:
        return getattr(obj, attr_name)
    except AttributeError:
        return default

# 使用示例
class Example:
    name = "test"
    value = 42

example = Example()
print(safe_getattr(example, 'name', 'default'))      # test
print(safe_getattr(example, 'nonexistent', 'default'))  # default

10.1.2.2. 类型检查

def call_if_callable(obj, method_name, *args, **kwargs):
    """只有当对象的方法可调用时才调用它"""
    if hasattr(obj, method_name):
        method = getattr(obj, method_name)
        if callable(method):
            return method(*args, **kwargs)
        else:
            raise TypeError(f"'{method_name}' 不是可调用方法")
    else:
        raise AttributeError(f"没有找到方法 '{method_name}'")

# 使用示例
class Example:
    def method(self):
        return "result"

example = Example()
result = call_if_callable(example, 'method')
print(result)  # result

10.1.3.3. 批量操作

def apply_to_all_attributes(obj, func):
    """将指定函数应用到对象的所有属性上"""
    results = {}
    for attr_name in dir(obj):
        if not attr_name.startswith('__'):
            try:
                attr_value = getattr(obj, attr_name)
                results[attr_name] = func(attr_value)
            except Exception as e:
                results[attr_name] = f"Error: {e}"
    return results

# 使用示例
class Example:
    name = "test"
    value = 42
    
    def method(self):
        return "result"

def describe_value(value):
    return f"{type(value).__name__}: {value}"

example = Example()
print(apply_to_all_attributes(example, describe_value))

10.2.注意事项

10.2.1.1. 性能考虑

import time

class Performance:
    def method(self):
        return "result"

obj = Performance()

# 测试直接方法调用与反射方法调用的性能差异
start = time.time()
for _ in range(100000):
    obj.method()
direct_time = time.time() - start

start = time.time()
for _ in range(100000):
    getattr(obj, 'method')()
reflect_time = time.time() - start

print(f"直接调用: {direct_time:.6f}s")
print(f"反射调用: {reflect_time:.6f}s")
print(f"性能差异: {reflect_time/direct_time:.2f}x")

10.2.2.2. 安全性考虑

class SafeReflection:
    @staticmethod
    def safe_reflect(obj, allowed_attrs):
        """只允许访问特定的属性"""
        def get_attr(attr_name):
            if attr_name in allowed_attrs:
                return getattr(obj, attr_name)
            else:
                raise AttributeError(f"不允许访问属性 '{attr_name}'")
        return get_attr

# 使用安全反射机制
class Example:
    def method(self):
        return "result"
    
    def secret_method(self):
        return "secret"

demo_obj = Example()
safe_getter = SafeReflection.safe_reflect(demo_obj, ['method'])

print(safe_getter('method'))  # 正常访问允许的属性
# print(safe_getter('secret_method'))  # 访问不允许的属性会报错

11.总结

11.1.反射的核心价值

动态性：运行时决定行为
通用性：编写可重用代码
灵活性：适应变化的需求
内省能力：了解对象结构

11.2.关键函数回顾

函数	描述
`hasattr(obj, name)`	检查属性是否存在
`getattr(obj, name[, default])`	获取属性值
`setattr(obj, name, value)`	设置属性值
`delattr(obj, name)`	删除属性
`dir(obj)`	获取所有属性和方法
`isinstance(obj, class)`	类型检查
`issubclass(cls, class)`	继承关系检查

11.3.适用场景

插件系统
配置管理
API路由
数据验证
测试框架
序列化/反序列化