-
Notifications
You must be signed in to change notification settings - Fork 12
/
hooks.py
105 lines (90 loc) · 3.61 KB
/
hooks.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
import abc
import logging
import typing
from . import log
if typing.TYPE_CHECKING:
from . import state # noqa: F401
class BaseHook(abc.ABC):
"""
Interface that must be implemented to satisfy naz's hooks.
User implementations should inherit this class and
implement the :func:`to_smsc <BaseHook.to_smsc>` and :func:`from_smsc <BaseHook.from_smsc>` methods with the type signatures shown.
A hook is class with methods that are called just before sending data to SMSC and just after receiving data from SMSC.
"""
@abc.abstractmethod
async def to_smsc(self, smpp_command: str, log_id: str, hook_metadata: str, pdu: bytes) -> None:
"""
called before sending data to SMSC.
Parameters:
smpp_command: any one of the SMSC commands eg submit_sm
log_id: an ID that a user's application had previously supplied to naz to track/correlate different messages.
hook_metadata: a string that a user's application had previously supplied to naz that it may want to be correlated with the log_id.
pdu: the full PDU as sent to SMSC
"""
raise NotImplementedError("to_smsc method must be implemented.")
@abc.abstractmethod
async def from_smsc(
self,
smpp_command: str,
log_id: str,
hook_metadata: str,
status: "state.CommandStatus",
pdu: bytes,
) -> None:
"""
called after receiving data from SMSC.
Parameters:
smpp_command: any one of the SMSC commands eg submit_sm_resp
log_id: an ID that a user's application had previously supplied to naz to track/correlate different messages.
hook_metadata: a string that a user's application had previously supplied to naz that it may want to be correlated with the log_id.
status: the state of request/response from SMSC.
pdu: the full PDU as received from SMSC
"""
raise NotImplementedError("from_smsc method must be implemented.")
class SimpleHook(BaseHook):
"""
This is an implementation of BaseHook.
When this class is called by naz, it just logs the request or response.
"""
def __init__(self, logger: typing.Union[None, logging.Logger] = None) -> None:
if not isinstance(logger, (type(None), logging.Logger)):
raise ValueError(
"`logger` should be of type:: `None` or `logging.Logger` You entered: {0}".format(
type(logger)
)
)
if logger is not None:
self.logger = logger
else:
self.logger = log.SimpleLogger("naz.SimpleHook")
async def to_smsc(self, smpp_command: str, log_id: str, hook_metadata: str, pdu: bytes) -> None:
self.logger.log(
logging.NOTSET,
{
"event": "naz.SimpleHook.to_smsc",
"stage": "start",
"smpp_command": smpp_command,
"log_id": log_id,
"hook_metadata": hook_metadata,
# SimpleLogger cant log bytes(pdu)
},
)
async def from_smsc(
self,
smpp_command: str,
log_id: str,
hook_metadata: str,
status: "state.CommandStatus",
pdu: bytes,
) -> None:
self.logger.log(
logging.NOTSET,
{
"event": "naz.SimpleHook.from_smsc",
"stage": "start",
"smpp_command": smpp_command,
"log_id": log_id,
"hook_metadata": hook_metadata,
"status": status.description,
},
)