kopia lustrzana https://github.com/Yakifo/amqtt
display 'default_factory' in a field as the actual value
rodzic
1be3a7c848
commit
38443f034e
|
@ -31,6 +31,8 @@ class ListenerType(StrEnum):
|
||||||
TCP = 'tcp'
|
TCP = 'tcp'
|
||||||
WS = 'ws'
|
WS = 'ws'
|
||||||
|
|
||||||
|
def __repr__(self) -> str:
|
||||||
|
return f'"{str(self.value)}"'
|
||||||
|
|
||||||
class Dictable:
|
class Dictable:
|
||||||
def __getitem__(self, key):
|
def __getitem__(self, key):
|
||||||
|
|
|
@ -0,0 +1,116 @@
|
||||||
|
import ast
|
||||||
|
import json
|
||||||
|
import pprint
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import griffe
|
||||||
|
from _griffe.agents.inspector import Inspector
|
||||||
|
from _griffe.agents.nodes.runtime import ObjectNode
|
||||||
|
from _griffe.agents.visitor import Visitor
|
||||||
|
from _griffe.models import Attribute
|
||||||
|
|
||||||
|
from amqtt.contexts import default_listeners, default_broker_plugins, default_client_plugins
|
||||||
|
|
||||||
|
default_factory_map = {
|
||||||
|
'default_listeners': default_listeners(),
|
||||||
|
'default_broker_plugins': default_broker_plugins(),
|
||||||
|
'default_client_plugins': default_client_plugins()
|
||||||
|
}
|
||||||
|
|
||||||
|
def get_qualified_name(node: ast.AST) -> str | None:
|
||||||
|
"""Recursively build the qualified name from an AST node."""
|
||||||
|
if isinstance(node, ast.Name):
|
||||||
|
return node.id
|
||||||
|
elif isinstance(node, ast.Attribute):
|
||||||
|
parent = get_qualified_name(node.value)
|
||||||
|
if parent:
|
||||||
|
return f"{parent}.{node.attr}"
|
||||||
|
return node.attr
|
||||||
|
elif isinstance(node, ast.Call):
|
||||||
|
# e.g., uuid.uuid4()
|
||||||
|
return get_qualified_name(node.func)
|
||||||
|
return None
|
||||||
|
|
||||||
|
def get_fully_qualified_name(call_node):
|
||||||
|
"""
|
||||||
|
Extracts the fully qualified name from an ast.Call node.
|
||||||
|
"""
|
||||||
|
if isinstance(call_node.func, ast.Name):
|
||||||
|
# Direct function call (e.g., "my_function(arg)")
|
||||||
|
return call_node.func.id
|
||||||
|
elif isinstance(call_node.func, ast.Attribute):
|
||||||
|
# Method call or qualified name (e.g., "obj.method(arg)" or "module.submodule.function(arg)")
|
||||||
|
parts = []
|
||||||
|
current = call_node.func
|
||||||
|
while isinstance(current, ast.Attribute):
|
||||||
|
parts.append(current.attr)
|
||||||
|
current = current.value
|
||||||
|
if isinstance(current, ast.Name):
|
||||||
|
parts.append(current.id)
|
||||||
|
return ".".join(reversed(parts))
|
||||||
|
else:
|
||||||
|
# Handle other potential cases (e.g., ast.Subscript) if necessary
|
||||||
|
return None
|
||||||
|
|
||||||
|
def get_callable_name(node):
|
||||||
|
if isinstance(node, ast.Name):
|
||||||
|
return node.id
|
||||||
|
elif isinstance(node, ast.Attribute):
|
||||||
|
return f"{get_callable_name(node.value)}.{node.attr}"
|
||||||
|
return None
|
||||||
|
|
||||||
|
def evaluate_callable_node(node):
|
||||||
|
try:
|
||||||
|
# Wrap the node in an Expression so it can be compiled
|
||||||
|
expr = ast.Expression(body=node)
|
||||||
|
compiled = compile(expr, filename="<ast>", mode="eval")
|
||||||
|
return eval(compiled, {"__builtins__": __builtins__, "list": list, "dict": dict})
|
||||||
|
except Exception as e:
|
||||||
|
return f"<unresolvable: {e}>"
|
||||||
|
|
||||||
|
class MyExtension(griffe.Extension):
|
||||||
|
def on_instance(
|
||||||
|
self,
|
||||||
|
node: ast.AST | griffe.ObjectNode,
|
||||||
|
obj: griffe.Object,
|
||||||
|
agent: griffe.Visitor | griffe.Inspector,
|
||||||
|
**kwargs,
|
||||||
|
) -> None:
|
||||||
|
"""Do something with `node` and/or `obj`."""
|
||||||
|
if obj.kind == griffe.Kind.FUNCTION and obj.name == 'default_broker_plugins':
|
||||||
|
print(f"my extension on instance {node} > {obj}")
|
||||||
|
|
||||||
|
def on_attribute_instance(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
node: ast.AST | ObjectNode,
|
||||||
|
attr: Attribute,
|
||||||
|
agent: Visitor | Inspector,
|
||||||
|
**kwargs: Any,
|
||||||
|
) -> None:
|
||||||
|
if not hasattr(node, "value"):
|
||||||
|
return
|
||||||
|
if isinstance(node.value, ast.Call):
|
||||||
|
for kw in node.value.keywords:
|
||||||
|
if kw.arg == "default_factory":
|
||||||
|
if get_callable_name(kw.value) != "dict":
|
||||||
|
callable_name = get_callable_name(kw.value)
|
||||||
|
if callable_name in default_factory_map:
|
||||||
|
print(default_factory_map[callable_name])
|
||||||
|
if "factory" not in attr.extra:
|
||||||
|
attr.extra["factory"] = {}
|
||||||
|
attr.extra["factory"]["has_badge"] = True
|
||||||
|
pprint.pprint(default_factory_map[callable_name])
|
||||||
|
f = f"{pprint.pformat(default_factory_map[callable_name], indent=4, width=80, sort_dicts=False)}"
|
||||||
|
attr.extra["factory"]["eval"] = f
|
||||||
|
else:
|
||||||
|
attr.extra["factory"]["has_badge"] = True
|
||||||
|
attr.extra["factory"]["eval"] = "{}"
|
||||||
|
|
||||||
|
# print(f'callable name: {get_callable_name(kw.value)}')
|
||||||
|
# # print(evaluate_callable_node(kw.value))
|
||||||
|
# print(f'type of node {type(kw.value)}')
|
||||||
|
# print(f'qualified name: {get_qualified_name(kw.value)}')
|
||||||
|
# # print(f'fully qualified name: {get_fully_qualified_name(node.value)}')
|
||||||
|
# factory = evaluate_callable_node(kw.value)
|
||||||
|
# attribute.extensions["default_factory"] = factory
|
|
@ -3,5 +3,3 @@
|
||||||
{% block logs scoped %}
|
{% block logs scoped %}
|
||||||
<p style="color:red">attribute.html.jinja</p>
|
<p style="color:red">attribute.html.jinja</p>
|
||||||
{% endblock logs %}
|
{% endblock logs %}
|
||||||
|
|
||||||
|
|
|
@ -1,8 +1 @@
|
||||||
{% extends "_base/expression.html.jinja" %}
|
{% if 'default_factory' in expression.__str__() %}{{ obj.extra.factory.eval | safe }}{% else %}{{ expression | safe }}{% endif %}
|
||||||
|
|
||||||
{% block logs scoped %}
|
|
||||||
|
|
||||||
<< expression.html.jinja >>
|
|
||||||
{% endblock logs %}
|
|
||||||
|
|
||||||
|
|
|
@ -1,7 +1,6 @@
|
||||||
{% extends "_base/labels.html.jinja" %}
|
{% extends "_base/labels.html.jinja" %}
|
||||||
|
|
||||||
{% block logs scoped %}
|
{% block logs scoped %}
|
||||||
<p style="color:red">labels.html.jinja</p>
|
<p style="color:red">labels.html.jinja</p>
|
||||||
{% endblock logs %}
|
{% endblock logs %}
|
||||||
|
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
{% extends "_base/signature.html.jinja" %}
|
{% extends "_base/signature.html.jinja" %}
|
||||||
|
|
||||||
{% block logs scoped %}
|
{# {% block logs scoped %}#}
|
||||||
<< signature.html.jinja >>
|
{# << signature.html.jinja >>#}
|
||||||
{% endblock logs %}
|
{#{% endblock logs %}#}
|
||||||
|
|
||||||
|
|
|
@ -151,7 +151,8 @@ plugins:
|
||||||
signature_crossrefs: true
|
signature_crossrefs: true
|
||||||
summary: true
|
summary: true
|
||||||
extensions:
|
extensions:
|
||||||
|
- docs/scripts/exts.py:MyExtension:
|
||||||
|
paths: [mypkg.mymod.myobj]
|
||||||
- git-revision-date-localized:
|
- git-revision-date-localized:
|
||||||
enabled: !ENV [DEPLOY, false]
|
enabled: !ENV [DEPLOY, false]
|
||||||
enable_creation_date: true
|
enable_creation_date: true
|
||||||
|
|
Ładowanie…
Reference in New Issue