py.lib.aw_web_tools

Yohn Y. 2025-10-19 Parent:2d0e1f161f26

16:c53a61e27af8 Go to Latest

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

* Ошибка в написании метода

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