-
Notifications
You must be signed in to change notification settings - Fork 21
/
__init__.py
95 lines (89 loc) · 3.07 KB
/
__init__.py
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
# -*- coding: utf-8 -*-
import ast
import logging
from types import ModuleType
from typing import TYPE_CHECKING, Any, Optional
# force handler registration by exec()ing the handler modules here
import ipyflow.tracing.external_calls.base_handlers # noqa: F401
import ipyflow.tracing.external_calls.list_handlers # noqa: F401
from ipyflow.singletons import flow
from ipyflow.tracing.external_calls.base_handlers import (
REGISTERED_HANDLER_BY_FUNCTION,
REGISTERED_HANDLER_BY_METHOD,
ExternalCallHandler,
MutatingMethodEventNotYetImplemented,
NoopCallHandler,
StandardMutation,
)
if TYPE_CHECKING:
from ipyflow.data_model.symbol import Symbol
def resolve_external_call(
module: Optional[ModuleType],
caller_self: Optional[Any],
function_or_method: Optional[Any],
method: Optional[str],
call_node: Optional[ast.Call] = None,
use_standard_default: bool = True,
calling_symbol: Optional["Symbol"] = None,
) -> Optional[ExternalCallHandler]:
if hasattr(function_or_method, "__self__") and hasattr(
function_or_method, "__name__"
):
function_or_method = getattr(
function_or_method.__self__.__class__,
function_or_method.__name__,
function_or_method,
)
if caller_self is not None and isinstance(caller_self, ModuleType):
if module is None:
module = caller_self
caller_self = None
if (
module is logging
or getattr(module, "__name__", None) == "__main__"
or function_or_method == print
):
return None
if caller_self is logging or isinstance(caller_self, logging.Logger):
return None
elif caller_self is not None and any(
sym.is_class for sym in flow().aliases.get(id(type(caller_self)), [])
):
return None
# TODO: handle case where it's a function defined in-notebook
elif caller_self is None:
pass
elif method is None:
return None
if isinstance(caller_self, ModuleType):
caller_self = None
try:
external_call_type = REGISTERED_HANDLER_BY_FUNCTION.get(function_or_method)
except TypeError:
return None
if (
external_call_type is None
and caller_self is not None
and not isinstance(caller_self, type)
):
for cls in caller_self.__class__.mro():
external_call_type = REGISTERED_HANDLER_BY_METHOD.get((cls, method))
if external_call_type is not None:
module = getattr(cls, "__module__", module)
break
if external_call_type is None:
if use_standard_default:
external_call_type = StandardMutation
else:
return None
elif external_call_type is NoopCallHandler:
return None
elif external_call_type is MutatingMethodEventNotYetImplemented:
external_call_type = StandardMutation
return external_call_type.create(
module=module,
caller_self=caller_self,
function_or_method=function_or_method,
call_node=call_node,
calling_symbol=calling_symbol,
)