py.lib.aw_web_tools

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

14:0920ae304dfd Go to Latest

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

.. 1.202510.1 + Режим фековой авторизации для адаптера authelia. Режим требуется для отладки, поскольку вряд ли на машине разработчика будет развёрнуто это ПО на ранних стадиях разработки (преальфа). Возможно режим будет полезен при поиска проблем в приложении при авторизации.

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
11 from .cookie import Cookie
14 VARIABLE_PREFIX = 'ru.a0fs.app' # Префикс переменных и иных структур, которые сохраняются во внутренних
15 # структурах ``bottle``
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' # Идентификатор текущего соединения.
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
32 else:
33 return request.remote_addr
36 def get_session_fingerprint(*add_params, request: bottle.BaseRequest = bottle.request) -> str:
37 """\
38 Конструируем некий отпечаток пользовательской сессии.
40 В качестве параметра принимается всё, что должно участвовать в создании отпечатка.
41 Это всё превращается в строки и подмешивается в хэш
43 При необходимости задать объект запроса, следует использовать форму ``request=<REQOBJ>``
44 """
45 ua = request.get_header('user-agent')
46 add_params_mixin = ':'.join(map(str, add_params))
48 return sha512(f'{ua}:{get_client_ip()}:{add_params_mixin}'.encode('utf-8')).hexdigest().lower()
51 def variable_prep(value, new_value_type: type = str, postprocess: bool = True) -> Any:
52 """\
53 Более интеллектуальная процедура преобразования базовых типов
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
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__}"')
72 res = new_value_type(value)
74 if isinstance(res, str) and postprocess:
75 res = res.strip()
77 return res
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]
88 else:
89 return default
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
99 def json_type_sanitizer(val):
100 """\
101 Преобразует значение ``val`` в пригодное для преобразования в json значение.
102 """
103 val_t = type(val)
105 if is_dataclass(val):
106 return json_type_sanitizer(asdict(val))
108 elif val_t in (int, float, str, bool) or val is None:
109 return val
111 elif val_t in (list, tuple):
112 return list(map(json_type_sanitizer, val))
114 elif val_t == dict:
115 return dict((key, json_type_sanitizer(d_val)) for key, d_val in val.items())
117 else:
118 return str(val)
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 = {}
131 res = bottle.HTTPResponse(body=json.dumps(json_type_sanitizer(data)), status=status)
132 res.content_type = 'application/json'
134 if cookies is not None:
135 for cookie in cookies:
136 cookie.response_add(res)
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)
143 else:
144 res.delete_cookie(cookie)
146 return res
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 в запросе')
157 return res
160 def get_cookie(name: str, request: bottle.BaseRequest = bottle.request) -> Optional[str]:
161 """\
162 Получить значение ``cookie`` с заданным именем из текущего запроса.
163 """
164 return request.get_cookie(name)
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`` параметры, если иное не указано конкретно.
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}"')
193 if res is None:
194 res = default
196 if isinstance(param_type, str):
197 if param_type.lower() == 'json':
198 if not res:
199 return None
201 else:
202 try:
203 return json.loads(res)
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)
211 except (TypeError, ValueError) as e:
212 raise ValueError(f'Error parsing parameter "{name}": {e}')
215 def make_log_topic(
216 user: Optional[str] = None,
217 request: bottle.BaseRequest = bottle.request
218 ) -> str:
219 """\
220 Создаём строку, идентифицирующую данный запрос для журналирования операций.
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)
230 if req_id is not None:
231 ip = f'{ip} | {req_id}'
233 elif conn_id is not None:
234 ip = f'{ip} | {conn_id}'
236 if user is None:
237 ip = f'_NOUID_[{ip}]'
239 else:
240 ip = f'{user}[{ip}]'
242 return f'{ip} - {url_path}'
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 сервисов
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
266 else:
267 err_type = err
268 err_msg = msg if msg is not None else ''
270 res = make_response({
271 'error': err_type,
272 'msg': err_msg,
273 }, code, cookies)
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)
280 else:
281 res.delete_cookie(cookie)
283 return res