py.lib

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

37:ae0107755941 Go to Latest

py.lib/webapp/bottle_urils.py

. Исправление ошибок и рефакторинг

Download raw file

View source Diff to previous Annotate

1 # coding: utf-8

2 """\

3 Набор инструментов, для более удобного взаимодействия с фреймворком ``bottle``

4 """

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

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 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 }

63 if self.samesite:

64 res['samesite'] = 'strict'

66 for k in ('max_age', 'expires', 'path', 'domain'):

67 if getattr(self, k) is not None:

68 res[k] = getattr(self, k)

70 resp.set_cookie(**res)

73 def get_client_ip() -> str:

74 """\

75 Получение реального адреса клиента

76 """

78 ip = bottle.request.get_header(IP_HEADER)

79 if ip:

80 return ip

82 else:

83 return bottle.request.remote_addr

86 def get_session_fingerprint(*add_params) -> str:

87 """\

88 Конструируем некий отпечаток пользовательской сессии.

90 В качестве параметра принимается всё, что должно участвовать в создании отпечатка.

91 Это всё превращается в строки и подмешивается в хэш

92 """

94 ua = bottle.request.get_header('user-agent')

95 add_params_mixin = ':'.join(map(str, add_params))

97 return sha512(f'{ua}:{get_client_ip()}:{add_params_mixin}'.encode('utf-8')).hexdigest().lower()

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