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