1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
from __future__ import annotations
import collections
import inspect
from typing import Any, Iterator
from twisted.python.modules import PythonAttribute, PythonModule, getModule
from automat import MethodicalMachine
from ._typed import TypeMachine, InputProtocol, Core
def isOriginalLocation(attr: PythonAttribute | PythonModule) -> bool:
"""
Attempt to discover if this appearance of a PythonAttribute
representing a class refers to the module where that class was
defined.
"""
sourceModule = inspect.getmodule(attr.load())
if sourceModule is None:
return False
currentModule = attr
while not isinstance(currentModule, PythonModule):
currentModule = currentModule.onObject
return currentModule.name == sourceModule.__name__
def findMachinesViaWrapper(
within: PythonModule | PythonAttribute,
) -> Iterator[tuple[str, MethodicalMachine | TypeMachine[InputProtocol, Core]]]:
"""
Recursively yield L{MethodicalMachine}s and their FQPNs within a
L{PythonModule} or a L{twisted.python.modules.PythonAttribute}
wrapper object.
Note that L{PythonModule}s may refer to packages, as well.
The discovery heuristic considers L{MethodicalMachine} instances
that are module-level attributes or class-level attributes
accessible from module scope. Machines inside nested classes will
be discovered, but those returned from functions or methods will not be.
@type within: L{PythonModule} or L{twisted.python.modules.PythonAttribute}
@param within: Where to start the search.
@return: a generator which yields FQPN, L{MethodicalMachine} pairs.
"""
queue = collections.deque([within])
visited: set[
PythonModule
| PythonAttribute
| MethodicalMachine
| TypeMachine[InputProtocol, Core]
| type[Any]
] = set()
while queue:
attr = queue.pop()
value = attr.load()
if (
isinstance(value, MethodicalMachine) or isinstance(value, TypeMachine)
) and value not in visited:
visited.add(value)
yield attr.name, value
elif (
inspect.isclass(value) and isOriginalLocation(attr) and value not in visited
):
visited.add(value)
queue.extendleft(attr.iterAttributes())
elif isinstance(attr, PythonModule) and value not in visited:
visited.add(value)
queue.extendleft(attr.iterAttributes())
queue.extendleft(attr.iterModules())
class InvalidFQPN(Exception):
"""
The given FQPN was not a dot-separated list of Python objects.
"""
class NoModule(InvalidFQPN):
"""
A prefix of the FQPN was not an importable module or package.
"""
class NoObject(InvalidFQPN):
"""
A suffix of the FQPN was not an accessible object
"""
def wrapFQPN(fqpn: str) -> PythonModule | PythonAttribute:
"""
Given an FQPN, retrieve the object via the global Python module
namespace and wrap it with a L{PythonModule} or a
L{twisted.python.modules.PythonAttribute}.
"""
# largely cribbed from t.p.reflect.namedAny
if not fqpn:
raise InvalidFQPN("FQPN was empty")
components = collections.deque(fqpn.split("."))
if "" in components:
raise InvalidFQPN(
"name must be a string giving a '.'-separated list of Python "
"identifiers, not %r" % (fqpn,)
)
component = components.popleft()
try:
module = getModule(component)
except KeyError:
raise NoModule(component)
# find the bottom-most module
while components:
component = components.popleft()
try:
module = module[component]
except KeyError:
components.appendleft(component)
break
else:
module.load()
else:
return module
# find the bottom-most attribute
attribute = module
for component in components:
try:
attribute = next(
child
for child in attribute.iterAttributes()
if child.name.rsplit(".", 1)[-1] == component
)
except StopIteration:
raise NoObject("{}.{}".format(attribute.name, component))
return attribute
def findMachines(
fqpn: str,
) -> Iterator[tuple[str, MethodicalMachine | TypeMachine[InputProtocol, Core]]]:
"""
Recursively yield L{MethodicalMachine}s and their FQPNs in and under the a
Python object specified by an FQPN.
The discovery heuristic considers L{MethodicalMachine} instances that are
module-level attributes or class-level attributes accessible from module
scope. Machines inside nested classes will be discovered, but those
returned from functions or methods will not be.
@param fqpn: a fully-qualified Python identifier (i.e. the dotted
identifier of an object defined at module or class scope, including the
package and modele names); where to start the search.
@return: a generator which yields (C{FQPN}, L{MethodicalMachine}) pairs.
"""
return findMachinesViaWrapper(wrapFQPN(fqpn))
|
