-
Notifications
You must be signed in to change notification settings - Fork 4
/
context.py
129 lines (89 loc) · 3.53 KB
/
context.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
from typing import Dict
from entities import Session, GeneratedWindow, Tag, WindowType
from exceptions import WindowNotFoundError, InvalidSelectedOptionError, ProcessorNotFoundError, \
ProcessorInvalidReturnError, SessionAlreadyFinishedError
class LottusContext:
"""
"""
def __init__(self, initial_window: str, current_session: Session, processors: Dict) -> None:
"""
:param initial_window:
:param current_session:
:param processors:
"""
if not initial_window or not isinstance(initial_window, str):
raise ValueError("initial window cannot be empty nor null")
if not current_session or not isinstance(current_session, Session):
raise ValueError("current session cannot be null")
if not processors or not isinstance(processors, Dict):
raise ValueError("processors cannot be null")
if initial_window not in processors:
raise WindowNotFoundError(None, f"Window {initial_window} not found in the processors list")
self.__processors = processors
self.__currentSession = current_session
self.__initial_window = initial_window
def set_current_window(self, current_window: GeneratedWindow) -> None:
"""
:param current_window:
:return:
"""
self.__currentSession.current_window = current_window
@property
def current_window(self) -> GeneratedWindow:
"""
:return:
"""
return self.__currentSession.current_window
def add_tag(self, name, value=None) -> None:
"""
:param name:
:param value:
:return:
"""
if not name:
raise ValueError("Invalid argument 'name'")
self.__currentSession.add_tag(Tag(name, value))
@property
def current_session(self) -> Session:
"""
:return:
"""
return self.__currentSession
def call_processor(self, processor_name: str, command: str) -> GeneratedWindow:
"""
:param command:
:param processor_name:
:return:
"""
if processor_name in self.__processors:
processor = self.__processors[processor_name]
else:
raise ProcessorNotFoundError(None, f"No processor found for window {processor_name}")
return processor(processor_name, command)
def process_command(self, command: str) -> GeneratedWindow:
"""
:param command:
:return:
"""
current_window = self.current_session.current_window
if not current_window:
processor = self.__processors.get(self.__initial_window)
else:
if current_window.window_type == WindowType.MESSAGE:
raise SessionAlreadyFinishedError(None, "Current session already finished")
processor = self.__processors.get(current_window.name)
if not processor:
raise ProcessorNotFoundError(None, f"Processor for window {current_window.name} not found")
generated_window = processor(self, command)
if not generated_window or not isinstance(generated_window, GeneratedWindow):
raise ProcessorInvalidReturnError(
None,
f"Processor {processor} for window {current_window.name} returned an invalid response")
generated_window.previous_window = current_window
return generated_window
@property
def processors(self) -> Dict:
"""
:return:
"""
return self.__processors