-
Notifications
You must be signed in to change notification settings - Fork 0
/
ServiceInvocator.py
154 lines (139 loc) · 6.09 KB
/
ServiceInvocator.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
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
from __future__ import annotations
from argrelay.custom_integ.ServiceArgType import ServiceArgType
from argrelay.custom_integ.value_constants import (
goto_host_funct_,
goto_service_funct_,
list_service_func_,
list_host_func_,
)
from argrelay.enum_desc.ArgSource import ArgSource
from argrelay.plugin_invocator.AbstractInvocator import (
AbstractInvocator,
get_data_envelopes,
get_func_name_from_container,
get_func_name_from_envelope,
)
from argrelay.plugin_invocator.ErrorInvocator import ErrorInvocator
from argrelay.plugin_invocator.InvocationInput import InvocationInput
from argrelay.relay_server.LocalServer import LocalServer
from argrelay.relay_server.QueryEngine import populate_query_dict
from argrelay.runtime_context.EnvelopeContainer import EnvelopeContainer
from argrelay.runtime_context.InterpContext import InterpContext
from argrelay.runtime_data.AssignedValue import AssignedValue
host_envelope_ipos_ = 1
service_envelope_ipos_ = 1
access_envelope_ipos_ = 2
def set_default_to(arg_type, arg_val, envelope_container):
if arg_type in envelope_container.search_control.types_to_keys_dict.keys():
if arg_type not in envelope_container.assigned_types_to_values.keys():
if arg_type in envelope_container.remaining_types_to_values:
if arg_val in envelope_container.remaining_types_to_values[arg_type]:
del envelope_container.remaining_types_to_values[arg_type]
envelope_container.assigned_types_to_values[arg_type] = AssignedValue(
arg_val,
ArgSource.DefaultValue,
)
else:
# Setting `ArgSource.DefaultValue`: it cannot override any, right? No point to handle assigned case:
pass
def redirect_to_error(interp_ctx, server_config):
# Redirect to `ErrorInvocator`:
invocator_plugin_instance_id = ErrorInvocator.__name__
invocation_input = InvocationInput(
invocator_plugin_entry = server_config.plugin_dict[invocator_plugin_instance_id],
data_envelopes = get_data_envelopes(interp_ctx),
custom_plugin_data = {},
)
return invocation_input
class ServiceInvocator(AbstractInvocator):
def __init__(
self,
plugin_instance_id: str,
config_dict: dict,
):
super().__init__(
plugin_instance_id,
config_dict,
)
def run_fill_control(
self,
interp_ctx: "InterpContext",
):
func_name = get_func_name_from_container(interp_ctx)
if func_name in [
goto_host_funct_,
goto_service_funct_,
]:
assert host_envelope_ipos_ == service_envelope_ipos_
object_envelope_ipos_ = host_envelope_ipos_
if interp_ctx.curr_container_ipos == interp_ctx.curr_interp.base_envelope_ipos + access_envelope_ipos_:
object_envelope = interp_ctx.envelope_containers[interp_ctx.curr_interp.base_envelope_ipos + object_envelope_ipos_].data_envelope
access_container = interp_ctx.envelope_containers[interp_ctx.curr_interp.base_envelope_ipos + access_envelope_ipos_]
code_arg_type = ServiceArgType.CodeMaturity.name
if code_arg_type in object_envelope:
code_arg_val = object_envelope[code_arg_type]
if code_arg_val == "prod":
set_default_to(ServiceArgType.AccessType.name, "ro", access_container)
else:
set_default_to(ServiceArgType.AccessType.name, "rw", access_container)
elif func_name in [
list_host_func_,
list_service_func_,
]:
pass
else:
raise RuntimeError
def run_invoke_control(
self,
interp_ctx: InterpContext,
local_server: LocalServer,
) -> InvocationInput:
assert interp_ctx.is_funct_found(), "the (first) function envelope must be found"
func_name = get_func_name_from_container(interp_ctx)
if func_name in [
goto_host_funct_,
goto_service_funct_,
]:
return redirect_to_error(
interp_ctx,
local_server.server_config,
)
elif func_name in [
list_host_func_,
list_service_func_,
]:
vararg_data_envelope_ipos = host_envelope_ipos_
assert vararg_data_envelope_ipos == host_envelope_ipos_ == service_envelope_ipos_
if interp_ctx.curr_container_ipos >= host_envelope_ipos_:
query_dict = populate_query_dict(interp_ctx.envelope_containers[vararg_data_envelope_ipos])
invocator_plugin_instance_id = ServiceInvocator.__name__
invocation_input = InvocationInput(
invocator_plugin_entry = local_server.server_config.plugin_dict[invocator_plugin_instance_id],
data_envelopes = (
# existing envelopes (until vararg one):
get_data_envelopes(interp_ctx)[:vararg_data_envelope_ipos]
+
# all envelopes in vararg set:
local_server.get_query_engine().query_data_envelopes(query_dict)
),
custom_plugin_data = {},
)
return invocation_input
else:
return redirect_to_error(
interp_ctx,
local_server.server_config,
)
else:
raise RuntimeError
@staticmethod
def invoke_action(invocation_input: InvocationInput):
func_name = get_func_name_from_envelope(invocation_input.data_envelopes)
if func_name == list_host_func_:
for data_envelope in invocation_input.data_envelopes[host_envelope_ipos_:]:
print(data_envelope)
if func_name == list_service_func_:
for data_envelope in invocation_input.data_envelopes[service_envelope_ipos_:]:
print(data_envelope)
else:
raise RuntimeError