py.lib

Yohn Y. 2022-08-27 Parent:483727ff89c4

42:d9a3784f681b Go to Latest

py.lib/webapp/bottle_urils.py

+ Возможность удаления cookie по полному объекту (полезно, когда выставляются особые параметры на cookie, и их не удаётся удалить по имени)

History
1 # coding: utf-8
2 """\
3 Набор инструментов, для более удобного взаимодействия с фреймворком ``bottle``
4 """
6 import bottle
7 import json
8 from typing import Union, List, Optional, Tuple, Any, Dict
9 from datetime import datetime
10 from hashlib import sha512
11 from dataclasses import is_dataclass, asdict
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' # Идентификатор текущего соединения.
22 class Cookie(object):
23 """\
24 Класс хранящий ``cookie`` и способный их устанавливать в объекты http-ответов ``bottle``
25 """
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 """
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
55 def get_add_params(self) -> Dict[str, Any]:
56 """\
57 Подготавливает параметры, которые можно передать в процедуру ``set_cookie``
58 объекта ``Response``
59 """
60 res = {
61 'name': self.name,
62 'value': self.value,
63 'secure': self.secure,
64 'httponly': self.httponly,
65 }
67 if self.samesite:
68 res['samesite'] = 'strict'
70 for k in ('max_age', 'expires', 'path', 'domain'):
71 if getattr(self, k) is not None:
72 res[k] = getattr(self, k)
74 return res
76 def get_remove_param(self) -> Dict[str, Any]:
77 """\
78 Подготавливает параметры, которые можно передать в процедуру ``delete_cookie``
79 объекта ``Response``
80 """
81 res = self.get_add_params()
82 res['key'] = res['name']
83 del res['value']
84 del res['name']
85 return res
87 def response_add(self, resp: bottle.BaseResponse):
88 resp.set_cookie(**self.get_add_params())
90 def response_delete(self, resp: bottle.BaseResponse):
91 resp.delete_cookie(**self.get_remove_param())
94 def get_client_ip() -> str:
95 """\
96 Получение реального адреса клиента
97 """
99 ip = bottle.request.get_header(IP_HEADER)
100 if ip:
101 return ip
103 else:
104 return bottle.request.remote_addr
107 def get_session_fingerprint(*add_params) -> str:
108 """\
109 Конструируем некий отпечаток пользовательской сессии.
111 В качестве параметра принимается всё, что должно участвовать в создании отпечатка.
112 Это всё превращается в строки и подмешивается в хэш
113 """
115 ua = bottle.request.get_header('user-agent')
116 add_params_mixin = ':'.join(map(str, add_params))
118 return sha512(f'{ua}:{get_client_ip()}:{add_params_mixin}'.encode('utf-8')).hexdigest().lower()
121 def variable_prep(value, new_value_type: type = str, postprocess: bool = True) -> Any:
122 """\
123 Более интеллектуальная процедура преобразования базовых типов
125 :param value: Значение, которое необходимо преобразовать
126 :param new_value_type: Тип в который необходимо преобразовать значение
127 :param postprocess: Применять ли последующую обработку полученного результата
128 :returns: Значение ``value`` преобразованное к типу ``value_type``
129 """
131 if value is None:
132 return value
134 if issubclass(new_value_type, bool):
135 if isinstance(value, str):
136 if value.lower() in ('on', 'yes', '1', 'true',):
137 return True
138 elif value.lower() in ('off', 'no', '0', 'false',):
139 return False
140 else:
141 raise ValueError(f'Unknown method translate "{value}" to "{new_value_type.__name__}"')
143 res = new_value_type(value)
145 if isinstance(res, str) and postprocess:
146 res = res.strip()
148 return res
151 def get_variable(name: str, default=None):
152 """\
153 Возвращает значения из хранилища контекста запроса bottle
154 """
156 _name = f'{VARIABLE_PREFIX}.{name}'
157 if _name in bottle.request.environ:
158 return bottle.request.environ[_name]
160 else:
161 return default
164 def set_variable(name: str, value):
165 """\
166 Устанавливает значения в хранилище контекста запроса bottle
167 """
169 bottle.request.environ[f'{VARIABLE_PREFIX}.{name}'] = value
172 def make_response(
173 data=None, status=200, cookies: Optional[List[Cookie]] = None,
174 remove_cookies: Optional[List[Union[str, Cookie]]] = None
175 ) -> bottle.HTTPResponse:
176 """\
177 Формирует ``API`` ответ.
178 """
180 if data is None:
181 data = {}
183 res = bottle.HTTPResponse(body=json.dumps(json_type_sanitizer(data)), status=status)
184 res.content_type = 'application/json'
186 if cookies is not None:
187 for cookie in cookies:
188 cookie.response_add(res)
190 if remove_cookies is not None and remove_cookies:
191 for cookie in remove_cookies:
192 if isinstance(cookie, Cookie):
193 cookie.response_delete(res)
195 else:
196 res.delete_cookie(cookie)
198 return res
201 def get_request_json() -> Optional[dict]:
202 """\
203 Если в запросе, который мы обрабатываем в текущий момент, использовалась посылка данных JSON, получить их.
204 """
206 res = bottle.request.json
207 if res is None:
208 raise ValueError('Не обнаружено JSON в запросе')
210 return res
213 def get_cookie(name: str) -> Optional[str]:
214 """\
215 Получить значение ``cookie`` с заданным именем из текущего запроса.
216 """
218 return bottle.request.get_cookie(name)
221 def get_param(name: str, param_type: Union[type, str] = str,
222 default=None, postprocess: bool = True, param_source: str = None):
223 """\
224 Получить из обрабатываемого на данный момент запроса значения установленного параметра. Обрабатываются как
225 ``GET``, так и ``POST`` параметры, если иное не указано конкретно.
227 :param name: Имя параметра
228 :param param_type: Тип, в который нужно преобразовать полученный параметр
229 :param default: Значение, отдаваемое, если параметр не установлен в запросе
230 :param postprocess: Производить ли постобработку параметра
231 :param param_source: Источник параметра, [``get``, ``post``]
232 :returns Полученное из запроса значение
233 """
235 if param_source is None:
236 res = bottle.request.params.getunicode(name)
237 elif param_source.lower() == 'get':
238 res = bottle.request.GET.getunicode(name)
239 elif param_source.lower() == 'post':
240 res = bottle.request.POST.getunicode(name)
241 else:
242 raise ValueError(f'Unknown parameter source "{param_source}" for "{name}"')
244 if res is None:
245 res = default
247 if isinstance(param_type, str):
248 if param_type.lower() == 'json':
249 if not res:
250 return None
252 else:
253 try:
254 return json.loads(res)
256 except json.JSONDecodeError as e:
257 raise ValueError(f'Unable to get JSON from from parameter "{name}": param="{res}" error="{e}"')
258 else:
259 try:
260 return variable_prep(res, new_value_type=param_type, postprocess=postprocess)
262 except (TypeError, ValueError) as e:
263 raise ValueError(f'Error parsing parameter "{name}": {e}')
266 def dict_filter(data: dict, need_keys: Union[List[str], Tuple[str]]) -> dict:
267 """\
268 Создать новый словарь на основе существующего, добавив в него только нужные ключи
269 """
271 return dict((key, val) for key, val in data.items() if key in need_keys)
274 def json_type_sanitizer(val):
275 """\
276 Преобразует значение ``val`` в пригодное для преобразования в json значение.
277 """
279 val_t = type(val)
281 if is_dataclass(val):
282 return json_type_sanitizer(asdict(val))
284 elif val_t in (int, float, str, bool) or val is None:
285 return val
287 elif val_t in (list, tuple):
288 return list(map(json_type_sanitizer, val))
290 elif val_t == dict:
291 return dict((key, json_type_sanitizer(d_val)) for key, d_val in val.items())
293 else:
294 return str(val)
297 def make_log_ident(user: Optional[str] = None) -> str:
298 """\
299 Создаём строку, идентифицирующую данный запрос для журналирования операций.
301 :param user: Если известен пользователь, задаём его здесь
302 """
304 ip = get_client_ip()
305 url_path = bottle.request.fullpath
306 conn_id = bottle.request.get_header(CONN_ID_HEADER)
308 if conn_id is not None:
309 ip = f'{ip} | {conn_id}'
311 if user is None:
312 ip = f'_NOID_[{ip}]'
314 else:
315 ip = f'{user}[{ip}]'
317 return f'{ip} - {url_path}'
320 def make_error_response(
321 code: int, err: Union[Exception, str],
322 msg: str = None,
323 remove_cookies: Optional[List[Union[str, Cookie]]] = None,
324 cookies: Optional[List[Cookie]] = None
325 ) -> bottle.HTTPResponse:
326 """\
327 Создание сообщения об ошибке для JSON REST API сервисов
329 :param code: HTTP-код ответа
330 :param err: Либо объект исключения, либо название ошибки
331 :param msg: Если не ``None`` - сообщение об ошибке. В противном случае, если ``err`` является исключением,
332 сообщение берётся из ``str(err)`` если нет, становится пустой строкой.
333 :param remove_cookies: Список cookie, которые должны быть удалены с клиента.
334 :param cookies: Набор cookies вставляемый в конструктор ответа без изменений. Читать в соответствующем
335 параметре ``make_response``
336 """
338 if isinstance(err, Exception):
339 err_type = type(err).__name__
340 err_msg = str(err) if msg is None else msg
342 else:
343 err_type = err
344 err_msg = msg if msg is not None else ''
346 res = make_response({
347 'error': err_type,
348 'msg': err_msg,
349 }, code, cookies)
351 if remove_cookies is not None and remove_cookies:
352 for cookie in remove_cookies:
353 if isinstance(cookie, Cookie):
354 cookie.response_delete(res)
356 else:
357 res.delete_cookie(cookie)
359 return res