py.lib

Yohn Y. 2023-01-28 Parent:d9a3784f681b

43:6f8bea109183 Go to Latest

py.lib/webapp/bottle_urils.py

. Наведение порядка в коде логирования

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