Интроспекция и рефлексия в Python: мощные инструменты метапрограммирования

class DataAnalyzer:
def __init__(self, data):
self.data = data
self._cache = {}

def process(self):
# Обработка данных
pass

analyzer = DataAnalyzer([1, 2, 3])
attrs = dir(analyzer)
# Фильтрация пользовательских атрибутов
user_attrs = [attr for attr in attrs if not attr.startswith('__')]
print(user_attrs) # ['_cache', 'data', 'process']

# Безопасное получение атрибута
value = getattr(analyzer, 'data', None)

# Динамическое создание и модификация атрибутов
if not hasattr(analyzer, 'metadata'):
setattr(analyzer, 'metadata', {'created': '2023-10-15'})

# Полиморфное вызывание методов
method_name = 'process'
if hasattr(analyzer, method_name) and callable(getattr(analyzer, method_name)):
getattr(analyzer, method_name)()

def describe_object(obj):
type_name = type(obj).__name__
mro = [base.__name__ for base in type(obj).__mro__] if hasattr(type(obj), '__mro__') else []

if isinstance(obj, (int, float, complex)):
category = "Числовой тип"
elif isinstance(obj, (str, bytes)):
category = "Строковый тип"
elif isinstance(obj, (list, tuple, set)):
category = "Последовательность"
elif isinstance(obj, dict):
category = "Словарь"
elif callable(obj):
category = "Вызываемый объект"
else:
category = "Другой тип"

return {
"type": type_name,
"category": category,
"inheritance": mro
}

print(describe_object(analyzer))

import inspect

def process_data(data, normalize=True, callback=None, *, output_format='json'):
"""Process the input data with optional normalization.

Args:
data: The data to process
normalize: Whether to normalize the data
callback: Optional callback function
output_format: Output format (json, xml, csv)

Returns:
Processed data in the specified format
"""
pass

# Получение сигнатуры функции
sig = inspect.signature(process_data)
print(f"Параметры: {sig}")

# Анализ параметров
for name, param in sig.parameters.items():
print(f"{name}: {param.kind}, default={param.default}")

# Получение документации
doc = inspect.getdoc(process_data)
print(f"Документация: {doc}")

# Получение исходного кода
source = inspect.getsource(process_data)
print(f"Исходный код: {source}")

def get_caller_info():
# Получаем текущий фрейм
frame = inspect.currentframe()

# Получаем информацию о вызывающем фрейме
caller_frame = inspect.getouterframes(frame)[1]

return {
'function': caller_frame.function,
'filename': caller_frame.filename,
'lineno': caller_frame.lineno,
'context': caller_frame.code_context
}

def some_function():
caller = get_caller_info()
print(f"Вызвано из: {caller['function']} в {caller['filename']}:{caller['lineno']}")

def main():
some_function()

class Base:
def method1(self):
pass

class Derived(Base):
def method2(self):
pass

def __private_method(self):
pass

# Получение членов класса
members = inspect.getmembers(Derived)
methods = [name for name, obj in members if inspect.isfunction(obj)]
print(f"Методы: {methods}")

# Проверка наследования
print(f"Является ли подклассом: {inspect.issubclass(Derived, Base)}")

# Получение MRO (порядка разрешения методов)
print(f"MRO: {inspect.getmro(Derived)}")

# Получение исходного файла
print(f"Определено в: {inspect.getsourcefile(Derived)}")

# Создание класса динамически
def method_generator(prefix):
def method(self, x):
return f"{prefix}: {x}"
return method

# Создаем класс с динамически сгенерированными методами
methods = {
'process': method_generator("Processing"),
'analyze': method_generator("Analyzing"),
'transform': method_generator("Transforming")
}

DynamicClass = type(
'DynamicClass', # Имя класса
(object,), # Базовые классы
{ # Словарь атрибутов
**methods,
'description': 'Динамически созданный класс'
}
)

# Использование созданного класса
instance = DynamicClass()
print(instance.process(42)) # Output: Processing: 42
print(instance.description) # Output: Динамически созданный класс

def create_validator_class(validation_rules):
"""Создает класс валидатора на основе правил валидации."""

def validate(self, data):
errors = {}
for field, rules in self.validation_rules.items():
if field not in data:
if 'required' in rules:
errors[field] = 'Required field is missing'
continue

value = data[field]

if 'type' in rules and not isinstance(value, rules['type']):
errors[field] = f'Expected type {rules["type"].__name__}'

if 'min_length' in rules and len(value) < rules['min_length']:
errors[field] = f'Length must be at least {rules["min_length"]}'

if 'pattern' in rules and not rules['pattern'].match(value):
errors[field] = 'Value does not match required pattern'

return errors

return type(
'DynamicValidator', 
(object,), 
{
'validation_rules': validation_rules,
'validate': validate
}
)

# Использование фабрики для создания валидатора
import re

user_validator = create_validator_class({
'username': {
'required': True, 
'type': str, 
'min_length': 3,
'pattern': re.compile(r'^[a-zA-Z0-9_]+$')
},
'email': {
'required': True, 
'type': str,
'pattern': re.compile(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$')
},
'age': {'type': int}
})

# Валидация данных
validator = user_validator()
errors = validator.validate({
'username': 'user_123',
'email': 'invalid-email',
'age': '25' # Строка вместо ожидаемого int
})

print(errors)

def enhance_object(obj, **attributes):
"""Добавляет новые атрибуты и методы к существующему объекту."""

# Добавление атрибутов
for name, value in attributes.items():
if not hasattr(obj, name):
setattr(obj, name, value)

# Добавление динамического метода
def add_method(obj, method_name, method):
setattr(obj, method_name, method.__get__(obj, obj.__class__))

# Пример добавления метода для вывода всех атрибутов
def describe_self(self):
attrs = {
name: value for name, value in vars(self).items() 
if not name.startswith('_')
}
return f"{self.__class__.__name__} with attributes: {attrs}"

add_method(obj, 'describe', describe_self)

return obj

# Использование для расширения объекта
class SimpleClass:
def __init__(self, name):
self.name = name

obj = SimpleClass("Original")
enhanced_obj = enhance_object(
obj, 
version="1.0", 
tags=["dynamic", "enhanced"]
)

print(enhanced_obj.describe()) # SimpleClass with attributes: {'name': 'Original', 'version': '1.0', 'tags': ['dynamic', 'enhanced']}

class Validator:
def __init__(self, name, validation_function, error_msg):
self.name = name
self.validation_function = validation_function
self.error_msg = error_msg

def __get__(self, instance, owner):
if instance is None:
return self
return instance.__dict__.get(self.name)

def __set__(self, instance, value):
if not self.validation_function(value):
raise ValueError(f'{self.name}: {self.error_msg}')
instance.__dict__[self.name] = value

def __delete__(self, instance):
del instance.__dict__[self.name]

# Использование дескрипторов для валидации атрибутов
class User:
name = Validator('name', 
lambda x: isinstance(x, str) and 3 <= len(x) <= 50, 
'Name must be a string with 3-50 characters')

age = Validator('age', 
lambda x: isinstance(x, int) and 18 <= x <= 120, 
'Age must be an integer between 18 and 120')

email = Validator('email', 
lambda x: isinstance(x, str) and '@' in x, 
'Invalid email format')

def __init__(self, name, age, email):
self.name = name
self.age = age
self.email = email

# Корректное создание объекта
user = User("John Doe", 30, "john@example.com")

# Вызовет ошибку валидации
try:
user.age = 15
except ValueError as e:
print(f"Validation error: {e}")

# Вызовет ошибку валидации
try:
user.email = "invalid_email"
except ValueError as e:
print(f"Validation error: {e}")

class LazyProperty:
"""Дескриптор для ленивой инициализации ресурсоемких свойств."""

def __init__(self, function):
self.function = function
self.name = function.__name__

def __get__(self, instance, owner):
if instance is None:
return self

# Вычисляем значение при первом доступе
value = self.function(instance)

# Сохраняем вычисленное значение в экземпляре
setattr(instance, self.name, value)

return value

class ObservableProperty:
"""Дескриптор для отслеживания изменений атрибута."""

def __init__(self, initial_value=None):
self.value = initial_value
self.observers = []

def __get__(self, instance, owner):
if instance is None:
return self
return self.value

def __set__(self, instance, value):
old_value = self.value
self.value = value

# Уведомляем всех наблюдателей об изменении
for observer in self.observers:
observer(instance, old_value, value)

def add_observer(self, observer):
self.observers.append(observer)
return self

# Пример использования
class DataProcessor:
def __init__(self, raw_data):
self.raw_data = raw_data
self.processed = False

@LazyProperty
def processed_data(self):
print("Processing data (expensive operation)...")
# Имитация сложной обработки данных
import time
time.sleep(2)
return [x * 2 for x in self.raw_data]

# Наблюдаемый атрибут с логированием изменений
status = ObservableProperty("idle")

def start_processing(self):
self.status = "processing"
result = self.processed_data # Ленивая инициализация
self.status = "completed"
self.processed = True
return result

# Добавляем наблюдателя для логирования изменений статуса
def log_status_change(instance, old_value, new_value):
print(f"Status changed: {old_value} -> {new_value}")

DataProcessor.status.add_observer(log_status_change)

# Использование
processor = DataProcessor([1, 2, 3, 4, 5])
print("Created processor, accessing properties:")

# Первый доступ вызовет вычисление
processor.start_processing()

# Последующие доступы используют кэшированное значение
print("Cached data:", processor.processed_data)

class ValidationMeta(type):
"""Метакласс для автоматической валидации атрибутов."""

def __new__(mcs, name, bases, namespace):
# Создаем новый класс
cls = super().__new__(mcs, name, bases, namespace)

# Получаем аннотации типов из класса
annotations = namespace.get('__annotations__', {})

# Для каждой аннотации создаем валидатор
for attr_name, attr_type in annotations.items():
if attr_name.startswith('_'):
continue

# Заменяем обычный атрибут на дескриптор для валидации типа
setattr(cls, attr_name, 
Validator(attr_name, 
lambda x, t=attr_type: isinstance(x, t),
f'must be of type {attr_type.__name__}'))

return cls

# Использование метакласса
class TypedUser(metaclass=ValidationMeta):
name: str
age: int
email: str

def __init__(self, name, age, email):
self.name = name
self.age = age
self.email = email

# Работает корректно
user = TypedUser("John", 30, "john@example.com")

# Вызовет ошибку типизации
try:
user.age = "thirty" # age должен быть int
except ValueError as e:
print(f"Type error: {e}")

class SingletonMeta(type):
"""Метакласс для реализации паттерна Singleton."""

_instances = {}

def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]

class RegistryMeta(type):
"""Метакласс для автоматической регистрации классов."""

registry = {}

def __new__(mcs, name, bases, namespace):
cls = super().__new__(mcs, name, bases, namespace)

# Регистрируем класс, если он не является абстрактным
if not namespace.get('__abstract__', False):
key = namespace.get('registry_key', name)
mcs.registry[key] = cls

return cls

@classmethod
def get_class(mcs, key):
return mcs.registry.get(key)

# Примеры использования
class DatabaseConnection(metaclass=SingletonMeta):
def __init__(self, connection_string):
print(f"Establishing connection to {connection_string}")
self.connection_string = connection_string

# Оба объекта будут ссылаться на одну и ту же инстанцию
db1 = DatabaseConnection("postgresql://localhost/db1")
db2 = DatabaseConnection("postgresql://localhost/db2")
print(f"Same instance: {db1 is db2}")
print(f"Connection string: {db2.connection_string}") # Будет показана строка из первого вызова

# Использование реестра классов
class APIEndpoint(metaclass=RegistryMeta):
__abstract__ = True # Не регистрируем базовый класс

class UserAPI(APIEndpoint):
registry_key = "users"

def get_all(self):
return ["user1", "user2"]

class ProductAPI(APIEndpoint):
registry_key = "products"

def get_all(self):
return ["product1", "product2"]

# Получение классов из реестра
users_api_cls = RegistryMeta.get_class("users")
products_api_cls = RegistryMeta.get_class("products")

users_api = users_api_cls()
print(f"Users: {users_api.get_all()}")

import inspect
import datetime
import uuid

class AutoSerializer:
"""Автоматический сериализатор объектов на основе интроспекции."""

PRIMITIVE_TYPES = (int, float, str, bool, type(None))
COLLECTION_TYPES = (list, tuple, set, frozenset)
SPECIAL_TYPES = {
datetime.datetime: lambda dt: dt.isoformat(),
datetime.date: lambda d: d.isoformat(),
uuid.UUID: lambda u: str(u)
}

@classmethod
def serialize(cls, obj):
"""Сериализует объект в структуру данных JSON."""

# Примитивные типы
if isinstance(obj, cls.PRIMITIVE_TYPES):
return obj

# Обработка специальных типов
for special_type, converter in cls.SPECIAL_TYPES.items():
if isinstance(obj, special_type):
return converter(obj)

# Обработка словарей
if isinstance(obj, dict):
return {str(k): cls.serialize(v) for k, v in obj.items()}

# Обработка коллекций
if isinstance(obj, cls.COLLECTION_TYPES):
return [cls.serialize(item) for item in obj]

# Получение атрибутов объекта через интроспекцию
result = {}

# Обрабатываем только пользовательские атрибуты (не служебные)
attrs = inspect.getmembers(obj, lambda a: not inspect.isroutine(a))
for key, value in attrs:
if not key.startswith('_'): # Пропускаем приватные атрибуты
result[key] = cls.serialize(value)

return result

@classmethod
def deserialize(cls, data, target_class=None):
"""Десериализует данные в объекты."""

if data is None or isinstance(data, cls.PRIMITIVE_TYPES):
return data

if isinstance(data, list):
return [cls.deserialize(item) for item in data]

if isinstance(data, dict) and target_class:
# Создаем экземпляр целевого класса
instance = target_class()

# Заполняем атрибуты
for key, value in data.items():
if hasattr(instance, key):
# Определяем тип атрибута для корректной десериализации
attr_type = type(getattr(instance, key, None))
if attr_type in cls.SPECIAL_TYPES.keys():
# Обработка специальных типов
if attr_type == datetime.datetime:
setattr(instance, key, datetime.datetime.fromisoformat(value))
elif attr_type == datetime.date:
setattr(instance, key, datetime.date.fromisoformat(value))
elif attr_type == uuid.UUID:
setattr(instance, key, uuid.UUID(value))
else:
# Обычные типы
setattr(instance, key, cls.deserialize(value))

return instance

return data

# Пример использования
class User:
def __init__(self):
self.id = uuid.uuid4()
self.name = ""
self.email = ""
self.created_at = datetime.datetime.now()
self.roles = []

def __str__(self):
return f"User(id={self.id}, name={self.name}, email={self.email})"

# Создаем пользователя
user = User()
user.name = "John Doe"
user.email = "john@example.com"
user.roles = ["admin", "editor"]

# Сериализуем
serialized = AutoSerializer.serialize(user)
print("Serialized:", serialized)

# Десериализуем
deserialized = AutoSerializer.deserialize(serialized, User)
print("Deserialized:", deserialized)
print(f"Same data: {user.name == deserialized.name}")
print(f"Same roles: {user.roles == deserialized.roles}")

class Registry:
"""Реестр для автоматической регистрации классов и функций."""

_registry = {}

@classmethod
def register(cls, key=None):
"""Декоратор для регистрации классов и функций."""
def decorator(obj):
# Определяем ключ регистрации
registry_key = key
if registry_key is None:
registry_key = obj.__name__

# Регистрируем объект
cls._registry[registry_key] = obj
return obj
return decorator

@classmethod
def get(cls, key):
"""Получает зарегистрированный объект по ключу."""
return cls._registry.get(key)

@classmethod
def get_all(cls):
"""Возвращает все зарегистрированные объекты."""
return cls._registry.copy()

# Регистрация команд в CLI-приложении
@Registry.register(key="help")
def show_help(args):
print("Available commands:")
for cmd_name, cmd_func in Registry.get_all().items():
if callable(cmd_func):
doc = inspect.getdoc(cmd_func) or "No description"
print(f" {cmd_name}: {doc.splitlines()[0]}")

@Registry.register(key="list")
def list_items(args):
"""List all items in the system."""
print("Listing items...")
# Здесь был бы код для вывода элементов

@Registry.register(key="add")
def add_item(args):
"""Add a new item to the system.

Usage: add <name> [<description>]
"""
if not args:
print("Error: Name is required")
return
print(f"Adding item: {args[0]}")

# Простой CLI-интерпретатор на основе реестра
def cli_interpreter():
while True:
cmd_line = input("> ")
if cmd_line == "exit":
break

parts = cmd_line.split()
if not parts:
continue

cmd_name = parts[0]
args = parts[1:]

cmd_func = Registry.get(cmd_name)
if cmd_func:
cmd_func(args)
else:
print(f"Unknown command: {cmd_name}")
print("Type 'help' for available commands")

# Запуск интерпретатора
# cli_interpreter()

import inspect
from typing import get_type_hints, List, Dict, Any, Optional

class SelfDocumenting:
"""Миксин для создания самодокументирующихся классов."""

def get_schema(self):
"""Генерирует схему класса на основе аннотаций типов."""

hints = get_type_hints(self.__class__)
schema = {
"class_name": self.__class__.__name__,
"description": inspect.getdoc(self.__class__) or "",
"properti