py.lib

Yohn Y. 2022-08-05 Child:483727ff89c4

31:4186c3b229fa Go to Latest

py.lib/webapp/bottle_urils.py

+ Модуль работы с датаклассами и их наполнения из ORM + Утилиты Bottle + Утилиты JWT + Помошник в парсинге конфигурационных файлов

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