py.lib.aw_web_tools

Yohn Y. 2024-02-27 Parent:06f00ec09030 Child:2d0e1f161f26

2:774c6728b7ff Go to Latest

py.lib.aw_web_tools/src/aw_web_tools/btle_tools.py

Added tag 0.202402.2 for changeset cb0dd8d1c0e9

Download raw file
View source Diff to previous Annotate
History
1 # coding: utf-8
2 """\
3 Набор инструментов, для более удобного взаимодействия с фреймворком ``bottle``
4 """
5 import bottle
6 import json
7 from typing import Union, List, Optional, Any
8 from hashlib import sha512
9 from dataclasses import is_dataclass, asdict
10
11 from .cookie import Cookie
12
13
14 VARIABLE_PREFIX = 'ru.a0fs.app' # Префикс переменных и иных структур, которые сохраняются во внутренних
15 # структурах ``bottle``
16
17 # Имена заголовков взяты из собственного стандарта на конфигурарование nginx. При не совпадении нужно сменить.
18 IP_HEADER = 'X-Real-IP' # Заголовок, в котором реверс-прокси хранит реальный IP клиента.
19 REQ_ID_HEADER = 'X-Request-Id' # Заголовок запроса, в котором может храниться уникальный ID запроса,
20 # выставленного реверс-прокси.
21 CONN_ID_HEADER = 'X-Conn-ID' # Идентификатор текущего соединения.
22
23
24 def get_client_ip() -> str:
25 """\
26 Получение реального адреса клиента
27 """
28 ip = bottle.request.get_header(IP_HEADER)
29 if ip:
30 return ip
31
32 else:
33 return bottle.request.remote_addr
34
35
36 def get_session_fingerprint(*add_params) -> str:
37 """\
38 Конструируем некий отпечаток пользовательской сессии.
39
40 В качестве параметра принимается всё, что должно участвовать в создании отпечатка.
41 Это всё превращается в строки и подмешивается в хэш
42 """
43 ua = bottle.request.get_header('user-agent')
44 add_params_mixin = ':'.join(map(str, add_params))
45
46 return sha512(f'{ua}:{get_client_ip()}:{add_params_mixin}'.encode('utf-8')).hexdigest().lower()
47
48
49 def variable_prep(value, new_value_type: type = str, postprocess: bool = True) -> Any:
50 """\
51 Более интеллектуальная процедура преобразования базовых типов
52
53 :param value: Значение, которое необходимо преобразовать
54 :param new_value_type: Тип, в который необходимо преобразовать значение
55 :param postprocess: Применять ли последующую обработку полученного результата
56 :returns: Значение ``value`` преобразованное к типу ``value_type``
57 """
58 if value is None:
59 return value
60
61 if issubclass(new_value_type, bool):
62 if isinstance(value, str):
63 if value.lower() in ('on', 'yes', '1', 'true',):
64 return True
65 elif value.lower() in ('off', 'no', '0', 'false',):
66 return False
67 else:
68 raise ValueError(f'Unknown method translate "{value}" to "{new_value_type.__name__}"')
69
70 res = new_value_type(value)
71
72 if isinstance(res, str) and postprocess:
73 res = res.strip()
74
75 return res
76
77
78 def get_env(name: str, default=None):
79 """\
80 Возвращает значения из хранилища контекста запроса bottle
81 """
82 _name = f'{VARIABLE_PREFIX}.{name}'
83 if _name in bottle.request.environ:
84 return bottle.request.environ[_name]
85
86 else:
87 return default
88
89
90 def set_env(name: str, value):
91 """\
92 Устанавливает значения в хранилище контекста запроса bottle
93 """
94 bottle.request.environ[f'{VARIABLE_PREFIX}.{name}'] = value
95
96
97 def json_type_sanitizer(val):
98 """\
99 Преобразует значение ``val`` в пригодное для преобразования в json значение.
100 """
101 val_t = type(val)
102
103 if is_dataclass(val):
104 return json_type_sanitizer(asdict(val))
105
106 elif val_t in (int, float, str, bool) or val is None:
107 return val
108
109 elif val_t in (list, tuple):
110 return list(map(json_type_sanitizer, val))
111
112 elif val_t == dict:
113 return dict((key, json_type_sanitizer(d_val)) for key, d_val in val.items())
114
115 else:
116 return str(val)
117
118
119 def make_response(
120 data=None, status=200, cookies: Optional[List[Cookie]] = None,
121 remove_cookies: Optional[List[Union[str, Cookie]]] = None
122 ) -> bottle.HTTPResponse:
123 """\
124 Формирует ``API`` ответ.
125 """
126 if data is None:
127 data = {}
128
129 res = bottle.HTTPResponse(body=json.dumps(json_type_sanitizer(data)), status=status)
130 res.content_type = 'application/json'
131
132 if cookies is not None:
133 for cookie in cookies:
134 cookie.response_add(res)
135
136 if remove_cookies is not None and remove_cookies:
137 for cookie in remove_cookies:
138 if isinstance(cookie, Cookie):
139 cookie.response_delete(res)
140
141 else:
142 res.delete_cookie(cookie)
143
144 return res
145
146
147 def get_request_json() -> Optional[dict]:
148 """\
149 Если в запросе, который мы обрабатываем в текущий момент, использовалась посылка данных JSON, получить их.
150 """
151 res = bottle.request.json
152 if res is None:
153 raise ValueError('Не обнаружено JSON в запросе')
154
155 return res
156
157
158 def get_cookie(name: str) -> Optional[str]:
159 """\
160 Получить значение ``cookie`` с заданным именем из текущего запроса.
161 """
162 return bottle.request.get_cookie(name)
163
164
165 def get_param(name: str, param_type: Union[type, str] = str,
166 default=None, postprocess: bool = True, param_source: str = None):
167 """\
168 Получить из обрабатываемого на данный момент запроса значения установленного параметра. Обрабатываются как
169 ``GET``, так и ``POST`` параметры, если иное не указано конкретно.
170
171 :param name: Имя параметра
172 :param param_type: Тип, в который нужно преобразовать полученный параметр
173 :param default: Значение, отдаваемое, если параметр не установлен в запросе
174 :param postprocess: Производить ли постобработку параметра
175 :param param_source: Источник параметра, [``get``, ``post``]
176 :returns Полученное из запроса значение
177 """
178 if param_source is None:
179 res = bottle.request.params.getunicode(name)
180 elif param_source.lower() == 'get':
181 res = bottle.request.GET.getunicode(name)
182 elif param_source.lower() == 'post':
183 res = bottle.request.POST.getunicode(name)
184 else:
185 raise ValueError(f'Unknown parameter source "{param_source}" for "{name}"')
186
187 if res is None:
188 res = default
189
190 if isinstance(param_type, str):
191 if param_type.lower() == 'json':
192 if not res:
193 return None
194
195 else:
196 try:
197 return json.loads(res)
198
199 except json.JSONDecodeError as e:
200 raise ValueError(f'Unable to get JSON from from parameter "{name}": param="{res}" error="{e}"')
201 else:
202 try:
203 return variable_prep(res, new_value_type=param_type, postprocess=postprocess)
204
205 except (TypeError, ValueError) as e:
206 raise ValueError(f'Error parsing parameter "{name}": {e}')
207
208
209 def make_log_topic(user: Optional[str] = None) -> str:
210 """\
211 Создаём строку, идентифицирующую данный запрос для журналирования операций.
212
213 :param user: Если известен пользователь, задаём его здесь
214 """
215 ip = get_client_ip()
216 url_path = bottle.request.fullpath
217 conn_id = bottle.request.get_header(CONN_ID_HEADER)
218 req_id = bottle.request.get_header(REQ_ID_HEADER)
219
220 if req_id is not None:
221 ip = f'{ip} | {req_id}'
222
223 elif conn_id is not None:
224 ip = f'{ip} | {conn_id}'
225
226 if user is None:
227 ip = f'_NOUID_[{ip}]'
228
229 else:
230 ip = f'{user}[{ip}]'
231
232 return f'{ip} - {url_path}'
233
234
235 def make_error_response(
236 code: int, err: Union[Exception, str],
237 msg: str = None,
238 remove_cookies: Optional[List[Union[str, Cookie]]] = None,
239 cookies: Optional[List[Cookie]] = None
240 ) -> bottle.HTTPResponse:
241 """\
242 Создание сообщения об ошибке для JSON REST API сервисов
243
244 :param code: HTTP-код ответа
245 :param err: Либо объект исключения, либо название ошибки
246 :param msg: Если не ``None`` - сообщение об ошибке. В противном случае, если ``err`` является исключением,
247 сообщение берётся из ``str(err)`` если нет, становится пустой строкой.
248 :param remove_cookies: Список cookie, которые должны быть удалены с клиента.
249 :param cookies: Набор cookies вставляемый в конструктор ответа без изменений. Читать в соответствующем
250 параметре ``make_response``
251 """
252 if isinstance(err, Exception):
253 err_type = type(err).__name__
254 err_msg = str(err) if msg is None else msg
255
256 else:
257 err_type = err
258 err_msg = msg if msg is not None else ''
259
260 res = make_response({
261 'error': err_type,
262 'msg': err_msg,
263 }, code, cookies)
264
265 if remove_cookies is not None and remove_cookies:
266 for cookie in remove_cookies:
267 if isinstance(cookie, Cookie):
268 cookie.response_delete(res)
269
270 else:
271 res.delete_cookie(cookie)
272
273 return res