-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathnew_container_call.py
More file actions
179 lines (146 loc) · 6.36 KB
/
new_container_call.py
File metadata and controls
179 lines (146 loc) · 6.36 KB
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
169
170
171
172
173
174
175
176
177
178
179
"""
New container call implementation that works with the new injection system.
"""
import inspect
from typing import get_type_hints, get_origin, get_args
from bevy.injection_types import (
InjectionStrategy, TypeMatchingStrategy,
extract_injection_info, is_optional_type, get_non_none_type
)
from bevy.injections import get_injection_info, analyze_function_signature
def new_call_function(container, func, args, kwargs):
"""
New implementation of container function calling with new injection system.
Args:
container: Container instance
func: Function to call with injection
args: Positional arguments passed to function
kwargs: Keyword arguments passed to function
Returns:
Result of function call with injected dependencies
"""
# Get function signature
sig = inspect.signature(func)
# Check if function has injection metadata from @injectable decorator
injection_info = get_injection_info(func)
if injection_info:
# Use metadata from @injectable decorator
injection_params = injection_info['params']
type_matching = injection_info['type_matching']
strict_mode = injection_info['strict_mode']
debug_mode = injection_info['debug_mode']
else:
# Analyze function dynamically using ANY_NOT_PASSED strategy
injection_params = analyze_function_signature(func, InjectionStrategy.ANY_NOT_PASSED)
type_matching = TypeMatchingStrategy.SUBCLASS
strict_mode = True
debug_mode = False
# Bind provided arguments
bound_args = sig.bind_partial(*args, **kwargs)
bound_args.apply_defaults()
# Inject missing dependencies
for param_name, (param_type, options) in injection_params.items():
if param_name not in bound_args.arguments:
# This parameter needs injection
try:
injected_value = _resolve_dependency(
container, param_type, options, type_matching, strict_mode, debug_mode
)
bound_args.arguments[param_name] = injected_value
if debug_mode:
print(f"[BEVY DEBUG] Injected {param_name}: {param_type} = {injected_value}")
except Exception as e:
if strict_mode:
# Check if this is an optional type
if is_optional_type(param_type):
bound_args.arguments[param_name] = None
if debug_mode:
print(f"[BEVY DEBUG] Optional dependency {param_name} not found, using None")
else:
raise
else:
# Non-strict mode: inject None for missing dependencies
bound_args.arguments[param_name] = None
if debug_mode:
print(f"[BEVY DEBUG] Non-strict mode: {param_name} not found, using None")
# Call function with resolved arguments
return func(*bound_args.args, **bound_args.kwargs)
def _resolve_dependency(container, param_type, options, type_matching, strict_mode, debug_mode):
"""
Resolve a single dependency from the container.
Args:
container: Container instance
param_type: Type to resolve
options: Options object with qualifier, etc.
type_matching: Type matching strategy
strict_mode: Whether to raise errors or return None
debug_mode: Whether to log debug info
Returns:
Resolved dependency instance
Raises:
Exception if dependency cannot be resolved and strict_mode is True
"""
# Handle optional types (Type | None)
if is_optional_type(param_type):
actual_type = get_non_none_type(param_type)
try:
return _resolve_single_type(container, actual_type, options, type_matching, debug_mode)
except Exception:
# Optional dependency not found
return None
else:
return _resolve_single_type(container, param_type, options, type_matching, debug_mode)
def _resolve_single_type(container, param_type, options, type_matching, debug_mode):
"""
Resolve a single non-optional type from the container.
Args:
container: Container instance
param_type: Type to resolve
options: Options object
type_matching: Type matching strategy
debug_mode: Whether to log debug info
Returns:
Resolved instance
Raises:
Exception if type cannot be resolved
"""
if debug_mode:
print(f"[BEVY DEBUG] Resolving {param_type} with options {options}")
# Handle qualified dependencies
if options and options.qualifier:
return _resolve_qualified_dependency(container, param_type, options.qualifier, debug_mode)
# Handle configuration binding
if options and options.from_config:
return _resolve_from_config(container, param_type, options.from_config, debug_mode)
# Handle default factory
if options and options.default_factory:
try:
return container.get(param_type)
except Exception:
if debug_mode:
print(f"[BEVY DEBUG] Using default factory for {param_type}")
return options.default_factory()
# Standard resolution using container.get()
return container.get(param_type)
def _resolve_qualified_dependency(container, param_type, qualifier, debug_mode):
"""
Resolve a qualified dependency.
TODO: This will need to be implemented when qualifier support is added to Container.
For now, raise an informative error.
"""
raise NotImplementedError(
f"Qualified dependencies not yet implemented. "
f"Cannot resolve {param_type} with qualifier '{qualifier}'. "
f"This feature will be added in a future update."
)
def _resolve_from_config(container, param_type, config_key, debug_mode):
"""
Resolve a dependency from configuration.
TODO: This will need to be implemented when config binding is added.
For now, raise an informative error.
"""
raise NotImplementedError(
f"Configuration binding not yet implemented. "
f"Cannot resolve {param_type} from config key '{config_key}'. "
f"This feature will be added in a future update."
)