py.lib.aw_web_tools

Yohn Y. 2024-11-09 Parent:2d0e1f161f26

9:9c734271dcf2 Go to Latest

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

.. 1.202411.2 + Поддержка Authelia

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