FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
python-webuntis/webuntis/utils/remote.py at master · mrtncode/python-webuntis · GitHub
mrtncode
/
python-webuntis
Public
forked from
python-webuntis/python-webuntis
Notifications
You must be signed in to change notification settings
Fork
0
Star
0
Code
Pull requests
0
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
python-webuntis
/
webuntis
/
utils
/
remote.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
executable file
·
165 lines (131 loc) · 4.9 KB
Breadcrumbs
python-webuntis
/
webuntis
/
utils
/
remote.py
Copy path
File metadata and controls
executable file
·
165 lines (131 loc) · 4.9 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
"""
This file is part of python-webuntis
:copyright: (c) 2012 by Markus Unterwaditzer.
:license: BSD, see LICENSE for more details.
"""
from
webuntis
import
errors
from
webuntis
.
utils
import
log
from
webuntis
.
utils
.
userinput
import
unicode_string
,
bytestring
from
webuntis
.
utils
.
third_party
import
json
import
datetime
import
requests
_errorcodes
=
{
-
32601
:
errors
.
MethodNotFoundError
,
-
8504
:
errors
.
BadCredentialsError
,
-
8520
:
errors
.
NotLoggedInError
,
-
7004
:
errors
.
DateNotAllowed
,
-
8507
:
errors
.
DateNotAllowed
,
}
'''The API-errorcodes python-webuntis is able to interpret, together with the
exception that will be thrown.'''
def
rpc_request
(
config
,
method
,
params
):
"""
A method for sending a JSON-RPC request.
:param config: A dictionary containing ``useragent``, ``server``,
``school``, ``username`` and ``password``
:type config: dict or FilterDict
:param method: The JSON-RPC method to be executed
:type method: str
:param params: JSON-RPC parameters to the method (should be JSON
serializable)
:type params: dict
"""
server
=
config
[
'server'
]
school
=
config
[
'school'
]
useragent
=
config
[
'useragent'
]
assert
isinstance
(
method
,
unicode_string
)
assert
isinstance
(
server
,
unicode_string
)
assert
isinstance
(
school
,
unicode_string
)
assert
isinstance
(
useragent
,
unicode_string
)
for
v
in
params
.
values
():
assert
not
isinstance
(
v
,
bytestring
)
url
=
server
+
u'?school='
+
school
headers
=
{
u'User-Agent'
:
useragent
,
u'Content-Type'
:
u'application/json'
}
request_body
=
{
u'id'
:
_request_getid
(),
u'method'
:
method
,
u'params'
:
params
,
u'jsonrpc'
:
u'2.0'
}
if
method
!=
u'authenticate'
:
if
'jsessionid'
in
config
:
assert
isinstance
(
config
[
'jsessionid'
],
unicode_string
)
headers
[
'Cookie'
]
=
u'JSESSIONID='
+
config
[
'jsessionid'
]
else
:
raise
errors
.
NotLoggedInError
(
'Don
\'
t have JSESSIONID. Did you already log out?'
)
log
(
'debug'
,
'Making new request:'
)
log
(
'debug'
,
'URL: '
+
url
)
if
method
!=
u'authenticate'
:
# user credentials will not be logged - fixing #14
log
(
'debug'
,
'DATA: '
+
str
(
request_body
))
if
'_http_session'
not
in
config
:
config
[
'_http_session'
]
=
requests
.
session
()
http_session
=
config
[
'_http_session'
]
result_body
=
_send_request
(
url
,
request_body
,
headers
,
http_session
)
return
_parse_result
(
request_body
,
result_body
)
def
_request_getid
():
"""
calculate the id field for the request -- use current date
If you want to get a fixed id for tests:
import webuntis.utils.remote
webuntis.utils.remote._request_getid = lambda: "12345678910"
:return: id field for request
:rtype: str
"""
return
str
(
datetime
.
datetime
.
today
())
def
_parse_result
(
request_body
,
result_body
):
"""A subfunction of rpc_request, that, given the decoded JSON result,
handles the error codes or, if everything went well, returns the result
attribute of it. The request data has to be given too for logging and
ID validation.
:param request_body: The not-yet-encoded body of the request sent.
:param result_body: The decoded body of the result received.
"""
if
request_body
[
u'id'
]
!=
result_body
[
u'id'
]:
raise
errors
.
RemoteError
(
'Request ID was not the same one as returned. %s -- %s'
%
(
request_body
[
u'id'
],
result_body
[
u'id'
]))
try
:
return
result_body
[
u'result'
]
except
KeyError
:
_parse_error_code
(
request_body
,
result_body
)
def
_parse_error_code
(
request_body
,
result_body
):
"""A helper function for handling JSON error codes."""
log
(
'error'
,
result_body
)
try
:
error
=
result_body
[
u'error'
]
code
,
message
=
error
[
u'code'
],
error
[
u'message'
]
except
KeyError
:
code
=
None
# None doesn't exist in_errorcodes
message
=
(
'Some error happened and there is no information provided '
'what went wrong.'
)
exc
=
_errorcodes
.
get
(
code
,
errors
.
RemoteError
)(
message
)
exc
.
request
=
request_body
exc
.
result
=
result_body
exc
.
code
=
code
raise
exc
def
_send_request
(
url
,
data
,
headers
,
http_session
=
None
):
"""Sends a POST request given the endpoint URL, JSON-encodable data,
a dictionary with headers and, optionally, a session object for requests.
"""
if
http_session
is
None
:
http_session
=
requests
.
session
()
r
=
http_session
.
post
(
url
,
data
=
json
.
dumps
(
data
),
headers
=
headers
)
result
=
r
.
text
# this will eventually raise errors, e.g. on timeout
try
:
result_data
=
json
.
loads
(
result
, )
log
(
'debug'
,
'Valid JSON found'
)
log
(
'debug'
,
' Got data'
+
str
(
result
)[:
100
])
except
ValueError
:
raise
errors
.
RemoteError
(
'Invalid JSON'
,
result
)
else
:
return
result_data
Back
|
FazBrowse Home
|
New Git URL