py.lib

Yohn Y. 2022-08-18 Parent:4186c3b229fa Child:483727ff89c4

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