## /utils/system/introspection.py
import importlib.util
import pkgutil
import sys
from time import perf_counter, sleep
from pathlib import Path
from typing import Any, overload
from collections.abc import Callable
from functools import wraps
from types import ModuleType
from cProfile import Profile
from pstats import Stats, SortKey
from loguru import logger
__all__ = [
'measure_time',
'run_on_import',
'retry',
'profile_function',
'import_module_from_path',
'get_class_hierarchy',
'print_class_hierarchy',
'scan_package_classes',
]
[документация]
def import_module_from_path(file_path: str | Path) -> ModuleType:
"""
Динамически импортирует и регистрирует в системе Python-модуль (скрипт)
по его абсолютному или относительному пути в файловой системе.
Полезно для построения систем плагинов и аддонов.
:param file_path: Имя файла или путь к .py скрипту.
:return: Объект импортированного модуля (ModuleType).
:raises FileNotFoundError: Если модуля по указанному пути не существует.
:raises ImportError: Если загрузка модуля не получилась.
"""
path = Path(file_path).resolve()
if not path.exists():
raise FileNotFoundError(f"Failed to import module: file {path} does not exist.")
# Извлекаем чистое имя модуля для системного реестра (например, 'my_plugin')
module_name = path.stem
try:
# Сценарий 1: Если модуль уже загружен ранее, возвращаем его из кэша sys.modules
if module_name in sys.modules:
return sys.modules[module_name]
# Сценарий 2: Пошаговая низкоуровневая загрузка через импортлиб спецификации
spec = importlib.util.spec_from_file_location(module_name, str(path))
if spec is None or spec.loader is None:
raise ImportError(f"Failed to create loader specification for {path}")
module = importlib.util.module_from_spec(spec)
# Записываем в кэш ДО выполнения кода, чтобы избежать циклических импортов в плагинах
sys.modules[module_name] = module
spec.loader.exec_module(module)
logger.info(f"Module '{module_name}' successfully imported dynamically from {path.name}")
return module
except Exception as e:
if module_name in sys.modules:
del sys.modules[module_name]
logger.error(f"Failure in dynamic import module from path {path}: {e}")
raise
[документация]
def get_class_hierarchy(base_class: type) -> dict[str, Any]:
"""
Рекурсивно строит дерево иерархии классов (наследников) для указанного базового класса.
Помогает отслеживать и инспектировать структуру всех зарегистрированных компонентов.
Пример вывода:
{
'WidgetContainer': {
'Form': {},
'Layout': {},
'Tabview': {}
}
}
:param base_class: Базовый класс (например, Widget, Event, Window).
:return: Словарь, представляющий дерево наследования.
"""
hierarchy: dict[str, Any] = {}
try:
subclasses = base_class.__subclasses__()
except (TypeError, AttributeError):
return hierarchy
for sub in subclasses:
hierarchy[sub.__name__] = get_class_hierarchy(sub)
return hierarchy
[документация]
def print_class_hierarchy(hierarchy: dict[str, Any], indent: str = "", project_only: bool = True) -> None:
"""
Красиво выводит древовидный словарь иерархии классов в терминал с отступами-ветвями.
Пример вывода:
├── WidgetContainer
│ ├── Form
│ └── Layout
└── Button
:param hierarchy: Словарь предстваляющий иерархию пректа, собранную через
get_class_hierarchy или scan_package_classes
:param indent: разделитель между ветвями
:param project_only: Если True, скрывает системные модули Python,
выводя только классы вашего фреймворка/приложения.
"""
items = list(hierarchy.items())
for idx, (class_name, sub_tree) in enumerate(items):
# Если включен фильтр, отсекаем системный мусор встроенных библиотек Python
if project_only and class_name.startswith(('_', 'BaseException', 'Exception', 'type', 'str', 'int', 'dict', 'list')):
continue
# Определяем, является ли элемент последним на данном уровне для правильного символа ветки
is_last = (idx == len(items) - 1)
branch = "└── " if is_last else "├── "
# Печатаем имя класса с текущим отступом и символом ветвления
print(f"{indent}{branch}{class_name}")
# Если у класса есть наследники, уходим в рекурсию, наращивая символы отступа
if sub_tree:
next_indent = indent + (" " if is_last else "│ ")
print_class_hierarchy(sub_tree, next_indent)
[документация]
def scan_package_classes(package_name: str, base_class: type = object) -> dict[str, Any]:
"""
Сканирует все подмодули указанного пакета, используя встроенный механизм pkgutil.
Автоматически и корректно разрешает относительные импорты (relative imports),
загружая модули с учетом их полного родительского контекста.
Сканирует все .py файлы в указанной директории пакета, принудительно импортирует их
в память Python, после чего строит и возвращает полную иерархию классов для base_class.
:param package_name: Имя импортированного пакета.
:param base_class: Базовый класс, иерархию которого нужно собрать.
:return: Словарь дерева иерархии классов.
:raises ImportError: При ошибке импорта корневого пакета.
:raises ValueError: Если указанный пакет не является пакетом.
"""
# 1. Сначала пытаемся штатно импортировать сам корневой пакет
try:
root_package = importlib.import_module(package_name)
except ImportError as e:
logger.error(f"Failed to import root packege '{package_name}': {e}")
raise
# 2. Извлекаем физические пути расположения пакета на диске
package_path = getattr(root_package, "__path__", None)
if not package_path:
raise ValueError(f"Module '{package_name}' is not a packege (attribute __path__ is missing).")
# Добавляем корень проекта в пути поиска, если его там нет
root_dir = str(Path(package_path[0]).parent.resolve())
if root_dir not in sys.path:
sys.path.insert(0, root_dir)
logger.info(f"beginning automatic packege scanning '{package_name}'...")
# 3. Используем walk_packages для глубокого рекурсивного обхода всех вложенных подмодулей.
# walk_packages автоматически вычисляет правильные полные имена (например, 'ctk_extender.event.listener')
for module_info in pkgutil.walk_packages(package_path, prefix=f"{package_name}."):
try:
# Импортируем модуль по его полному системному имени.
importlib.import_module(module_info.name)
logger.debug(f"Module '{module_info.name}' successfully loaded in context.")
except Exception as e:
# Игнорируем битые тест-скрипты, если они вызвали ошибку при загрузке
logger.warning(f"Skipping module '{module_info.name}' import error: {e}")
continue
# 4. Когда граф типов полностью заполнен, строим дерево иерархии
return get_class_hierarchy(base_class)
# Сигнатура для вызова без скобок: @measure_time
@overload
def measure_time[F: Callable[..., Any]](func: F, /) -> F: ...
# Сигнатура для вызова со скобками: @measure_time()
@overload
def measure_time() -> Callable[[Any], Any]: ...
[документация]
def measure_time(func: Any = None, /) -> Any:
"""
Декоратор для измерения времени выполнения функции с использованием loguru.
Логирует результат на уровне DEBUG, подменяя контекст на вызываемую функцию.
Поддерживает вызов как со скобками, так и без них:
@measure_time
@measure_time()
:param func: Функция время выполнения которой нужно измерить
или None, в случае вызова как декоратора со скобками.
:return: Декоратор ввыполняющий измерение времени выполнения указанной функции.
"""
def decorator[F: Callable[..., Any]](target_func: F) -> F:
"""
Выполняет измерение времени выполнения функкции.
:param target_func: Функция, время выполнения которой нужно замерить.
:return: Результат выполнения функции.
"""
@wraps(target_func)
def wrapper(*args: object, **kwargs: object) -> Any:
"""
Обертка, выполняющая измерение времени выполнения функкции.
:param args: Аргументы.
:param kwargs: Ключевые аргументы.
:return: Результат выполнения функции.
"""
t1 = perf_counter()
result = target_func(*args, **kwargs)
t2 = perf_counter()
measurement_string = f'Function {target_func.__name__} took {t2 - t1:.6f} seconds to run'
# Подменяем метаданные лога, чтобы loguru вывел имя файла и модуль целевой функции
try:
caller_frame = sys._getframe(0)
caller_module = caller_frame.__module__
caller_file = Path(caller_frame.__code__.co_filename).name #type ignore[attr-defined]
caller_line = caller_frame.__code__.co_firstlineno #type ignore[attr-defined]
logger.patch(lambda record: record.update( #type ignore[call-arg]
name=caller_module, #type ignore[call-arg]
file=caller_file, #type ignore[call-arg]
line=caller_line, #type ignore[call-arg]
function=target_func.__name__ #type ignore[call-arg]
)).debug(measurement_string)
except Exception:
logger.debug(measurement_string)
return result
return wrapper # type: ignore[return-value]
# Если первый аргумент — это функция, значит вызвали без скобок: @measure_time
if func is not None and callable(func):
return decorator(func)
# Если декоратор вызван со скобками: @measure_time()
return decorator
# Сигнатура для вызова без скобок: @profile_function
@overload
def profile_function[F: Callable[..., Any]](file_name: F, /) -> F: ...
# Сигнатура для вызова со скобками: @profile_function("profile.prof")
@overload
def profile_function[F: Callable[..., Any]](file_name: str | None = None, /) -> Callable[[F], F]: ...
[документация]
def profile_function(file_name: Any = None, /) -> Any:
"""
Декоратор для создания профиля функции.
Записывает профиль в файл, или выводит в консоль, если файл не задан.
Поддерживает вызов как со скобками, так и без них:
@profile_function
@profile_function("profile.prof")
Просмотреть профиль можно выполнив tuna file_name в консоли
:param file_name: Имя файла для сохранения профиля функции.
:return: Декоратор выполняющий профилирование указанной функции.
"""
def decorator[F: Callable[..., Any]](func: F) -> F:
"""
Выполняет измерение времени выполнения функкции.
:param func: Функция, профилирование которой нужно произвести.
:return: Результат выполнения функции.
"""
@wraps(func)
def wrapper(*args: object, **kwargs: object) -> Any:
"""
Обертка, выполняющая измерение времени выполнения функкции.
:param args: Аргументы.
:param kwargs: Ключевые аргументы.
:return: Результат выполнения функции.
"""
with Profile() as profile:
result = func(*args, **kwargs)
stats_results = Stats(profile)
stats_results.sort_stats(SortKey.TIME)
if isinstance(file_name, str) and file_name:
stats_results.dump_stats(file_name) #tuna file_name
else:
stats_results.print_stats()
return result
return wrapper # type: ignore[return-value]
# Если первый аргумент — это функция, значит вызвали без скобок: @profile_function
if callable(file_name):
actual_func = file_name
file_name = None
return decorator(actual_func)
# Если передан файл или ничего не передано, возвращаем сам декоратор
return decorator
[документация]
def retry[**P, R](
*exceptions: type[BaseException],
tries: int = 3,
delay: float = 0.1
) -> Callable[[Callable[P, R]], Callable[P, R]] | Callable[[Callable[P, R]], Callable[P, R]]:
"""
Декоратор для повторного выполнения функции при возникновении ошибок.
Можно использовать как @retry, @retry(ValueError) или @retry(ValueError, TypeError, tries=5).
:param exceptions: Перечень исключений, при которых нужен повтор.
Если не передано, перехватывает любое исключение (Exception).
:param tries: Количество попыток выполнения функции, по умолчанию 3.
:param delay: Задержка между попытками в секундах, по умолчанию 0.1.
:return: Декорированная функция с поддержкой повторных попыток.
:raises BaseException: Последнее пойманное исключение, если все попытки исчерпаны.
"""
# Если декоратор вызван без скобок: @retry вместо @retry()
# В таком случае первым аргументом (в *exceptions) прилетит сама функция
if len(exceptions) == 1 and callable(exceptions[0]):
func: Callable[P, R] = exceptions[0]
return retry()(func)
# Если исключения не переданы, перехватываем стандартный Exception
caught_exceptions = exceptions if exceptions else (Exception,)
def decorator(func: Callable[P, R]) -> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
for attempt in range(1, tries + 1):
try:
return func(*args, **kwargs)
except caught_exceptions as e:
if attempt == tries:
raise e
sleep(delay)
# Код ниже недостижим, но необходим для корректного статического анализа
raise RuntimeError("Unexpected retry exhaustion")
return wrapper
return decorator
[документация]
def run_on_import[**P, R](func: Callable[P, R]) -> Callable[P, R]:
"""
Декоратор для автоматического запуска функции в момент импорта модуля.
Функция вызывается без аргументов сразу при инициализации файла.
:param func: Декорируемая функция, не должна требовать обязательных аргументов.
:return: Исходная функция для возможности последующих ручных вызовов.
"""
# Автоматический запуск функции без параметров при импорте
func() # type: ignore[call-arg]
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
return func(*args, **kwargs)
return wrapper