16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154 | class UtilsInspectStats:
"""Утилиты для инспекции статистики."""
label_width = 26
def _app_parser(self, app: "QueueTasks", json: bool = False):
"""Парсер для информации о приложении.
Args:
app (QueueTasks): Экземпляр приложения.
Returns:
str: Информация о приложении.
"""
lines = []
plugins_sum = (
len(app.plugins)
+ len(app.broker.plugins)
+ len(app.worker.plugins)
+ (len(app.starter.plugins) if app.starter else 0)
+ (len(app.broker.storage.plugins) if app.broker.storage else 0)
+ (len(app.broker.storage.global_config.plugins) if app.broker.storage.global_config else 0)
)
task_info = {
"Имя": app.name,
"Метод": app._method,
"Версия": app.version,
"Конфигурация": str(app.config),
"Количество задач": len(app.tasks),
"Количество роутеров": len(app.routers),
"Количество плагинов": plugins_sum,
"Количество инициализаций": sum(len(inits) for inits in app.events.on._events.values()),
"Брокер": app.broker.__class__.__name__,
"Воркер": app.worker.__class__.__name__,
"Стартер": app.starter.__class__.__name__ if app.starter else "—",
"Хранилище": app.broker.storage.__class__.__name__,
"GlobalConfig": app.broker.storage.global_config.__class__.__name__ if app.broker.storage.global_config else "—",
"Лог": app.log.__class__.__name__,
}
if json:
return self._parser_json(task_info)
# Форматируем словарь
task_block = "\n".join(
f"{label:<{self.label_width}}: {value}" for label, value in task_info.items()
)
lines.append(task_block)
lines.append("-" * 50)
return "\n".join(lines)
def _parser_json(self, data: Union[Any, Tuple[Any]]) -> str:
def formatter(d):
if is_dataclass(d):
return asdict(d)
return d
data = [formatter(d) for d in data] if isinstance(data, (tuple, list, ValuesView)) else formatter(data)
return json.dumps(data, ensure_ascii=False, indent=2, default=str)
def _tasks_parser(self, tasks: Tuple[TaskExecSchema]) -> str:
"""Форматированный вывод всех зарегистрированных задач."""
lines = []
for task in tasks:
args, kwargs = self._task_get_args_kwargs(task.func)
task_info = {
"Имя задачи": task.name,
"Приоритет": task.priority,
"Описание": task.description or "—",
"Теги": ', '.join(task.tags) if task.tags else "—",
"Асинхронность": task.awaiting,
"Генерация": task.generating,
"Self перед задачей": task.echo,
"Args": ', '.join(args) if args else "—",
"Kwargs": ', '.join(f"{k}={v}" for k, v in kwargs.items()) if kwargs else "—",
}
if task.retry is not None:
task_info["Повторов"] = task.retry
if task.retry_on_exc:
task_info["Искл. для повторов"] = pformat(task.retry_on_exc)
if task.decode:
task_info["Декодирование"] = str(task.decode)
if task.generate_handler:
task_info["Генератор"] = str(task.generate_handler)
if task.executor:
task_info["Исполнитель"] = str(task.executor)
if task.middlewares_before:
task_info["Мидлвари до"] = pformat(task.middlewares_before)
if task.middlewares_after:
task_info["Мидлвари после"] = pformat(task.middlewares_after)
if task.extra:
# Вставляем многострочное значение с отступом
extra_lines = "\n" + "\n".join(f" * {k}: {v}" for k, v in task.extra.items())
task_info["Дополнительно"] = extra_lines
# Форматируем словарь
task_block = "\n".join(
f"{label:<{self.label_width}}: {value}" for label, value in task_info.items()
)
lines.append(task_block)
lines.append("-" * 50)
return "\n".join(lines) or "Нет зарегистрированных задач."
def _task_get_args_kwargs(self, func):
"""Получение позиционных и ключевых аргументов функции задачи.
Args:
func (Callable): Функция задачи.
Returns:
tuple: Позиционные и ключевые аргументы.
"""
sig = signature(func)
positional_args = []
keyword_args = {}
for name, param in sig.parameters.items():
annotation = param.annotation if param.annotation is not _empty else None
type_str = f": {annotation.__name__}" if isinstance(annotation, type) else f": {annotation}" if annotation else ""
if param.kind in (param.POSITIONAL_ONLY, param.POSITIONAL_OR_KEYWORD):
if param.default is param.empty:
positional_args.append(f"{name}{type_str}")
else:
keyword_args[f"{name}{type_str}"] = param.default
elif param.kind == param.KEYWORD_ONLY:
type_str = type_str or ""
keyword_args[f"{name}{type_str}"] = param.default if param.default is not param.empty else "required"
elif param.kind == param.VAR_POSITIONAL:
positional_args.append(f"*{name}")
elif param.kind == param.VAR_KEYWORD:
keyword_args[f"**{name}"] = "..."
return positional_args, keyword_args
|