py.lib

Yohn Y. 2022-08-27 Parent:4186c3b229fa Child:d9a3784f681b

41:483727ff89c4 Go to Latest

py.lib/webapp/bottle_urils.py

+ Возможность удалить cookie при формировании штатного ответа.

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 remove_cookies: Optional[List[str]] = None
154 ) -> bottle.HTTPResponse:
155 """\
156 Формирует ``API`` ответ.
157 """
158
159 if data is None:
160 data = {}
161
162 res = bottle.HTTPResponse(body=json.dumps(json_type_sanitizer(data)), status=status)
163 res.content_type = 'application/json'
164
165 if cookies is not None:
166 for cookie in cookies:
167 cookie.response_add(res)
168
169 if remove_cookies is not None and remove_cookies:
170 for cookie in remove_cookies:
171 res.delete_cookie(cookie)
172
173 return res
174
175
176 def get_request_json() -> Optional[dict]:
177 """\
178 Если в запросе, который мы обрабатываем в текущий момент, использовалась посылка данных JSON, получить их.
179 """
180
181 res = bottle.request.json
182 if res is None:
183 raise ValueError('Не обнаружено JSON в запросе')
184
185 return res
186
187
188 def get_cookie(name: str) -> Optional[str]:
189 """\
190 Получить значение ``cookie`` с заданным именем из текущего запроса.
191 """
192
193 return bottle.request.get_cookie(name)
194
195
196 def get_param(name: str, param_type: Union[type, str] = str,
197 default=None, postprocess: bool = True, param_source: str = None):
198 """\
199 Получить из обрабатываемого на данный момент запроса значения установленного параметра. Обрабатываются как
200 ``GET``, так и ``POST`` параметры, если иное не указано конкретно.
201
202 :param name: Имя параметра
203 :param param_type: Тип, в который нужно преобразовать полученный параметр
204 :param default: Значение, отдаваемое, если параметр не установлен в запросе
205 :param postprocess: Производить ли постобработку параметра
206 :param param_source: Источник параметра, [``get``, ``post``]
207 :returns Полученное из запроса значение
208 """
209
210 if param_source is None:
211 res = bottle.request.params.getunicode(name)
212 elif param_source.lower() == 'get':
213 res = bottle.request.GET.getunicode(name)
214 elif param_source.lower() == 'post':
215 res = bottle.request.POST.getunicode(name)
216 else:
217 raise ValueError(f'Unknown parameter source "{param_source}" for "{name}"')
218
219 if res is None:
220 res = default
221
222 if isinstance(param_type, str):
223 if param_type.lower() == 'json':
224 if not res:
225 return None
226
227 else:
228 try:
229 return json.loads(res)
230
231 except json.JSONDecodeError as e:
232 raise ValueError(f'Unable to get JSON from from parameter "{name}": param="{res}" error="{e}"')
233 else:
234 try:
235 return variable_prep(res, new_value_type=param_type, postprocess=postprocess)
236
237 except (TypeError, ValueError) as e:
238 raise ValueError(f'Error parsing parameter "{name}": {e}')
239
240
241 def dict_filter(data: dict, need_keys: Union[List[str], Tuple[str]]) -> dict:
242 """\
243 Создать новый словарь на основе существующего, добавив в него только нужные ключи
244 """
245
246 return dict((key, val) for key, val in data.items() if key in need_keys)
247
248
249 def json_type_sanitizer(val):
250 """\
251 Преобразует значение ``val`` в пригодное для преобразования в json значение.
252 """
253
254 val_t = type(val)
255
256 if is_dataclass(val):
257 return json_type_sanitizer(asdict(val))
258
259 elif val_t in (int, float, str, bool) or val is None:
260 return val
261
262 elif val_t in (list, tuple):
263 return list(map(json_type_sanitizer, val))
264
265 elif val_t == dict:
266 return dict((key, json_type_sanitizer(d_val)) for key, d_val in val.items())
267
268 else:
269 return str(val)
270
271
272 def make_log_ident(user: Optional[str] = None) -> str:
273 """\
274 Создаём строку, идентифицирующую данный запрос для журналирования операций.
275
276 :param user: Если известен пользователь, задаём его здесь
277 """
278
279 ip = get_client_ip()
280 url_path = bottle.request.fullpath
281 conn_id = bottle.request.get_header(CONN_ID_HEADER)
282
283 if conn_id is not None:
284 ip = f'{ip} | {conn_id}'
285
286 if user is not None:
287 ip = f'_NOID_[{ip}]'
288
289 else:
290 ip = f'{user}[{ip}]'
291
292 return f'{ip} - {url_path}'
293
294
295 def make_error_response(
296 code: int, err: Union[Exception, str],
297 msg: str = None,
298 remove_cookies: Optional[List[str]] = None,
299 cookies: Optional[List[Cookie]] = None
300 ) -> bottle.HTTPResponse:
301 """\
302 Создание сообщения об ошибке для JSON REST API сервисов
303
304 :param code: HTTP-код ответа
305 :param err: Либо объект исключения, либо название ошибки
306 :param msg: Если не ``None`` - сообщение об ошибке. В противном случае, если ``err`` является исключением,
307 сообщение берётся из ``str(err)`` если нет, становится пустой строкой.
308 :param remove_cookies: Список cookie, которые должны быть удалены с клиента.
309 :param cookies: Набор cookies вставляемый в конструктор ответа без изменений. Читать в соответствующем
310 параметре ``make_response``
311 """
312
313 if isinstance(err, Exception):
314 err_type = type(err).__name__
315 err_msg = str(err) if msg is None else msg
316
317 else:
318 err_type = err
319 err_msg = msg if msg is not None else ''
320
321 res = make_response({
322 'error': err_type,
323 'msg': err_msg,
324 }, code, cookies)
325
326 if remove_cookies is not None:
327 for cookie_name in remove_cookies:
328 res.delete_cookie(cookie_name)
329
330 return res