py.lib

Yohn Y. 2023-01-30 Parent:d9a3784f681b

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

102

103 else:

104 return bottle.request.remote_addr

105

106

107 def get_session_fingerprint(*add_params) -> str:

108 """\

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

110

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

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

113 """

114

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

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

117

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

119

120

121 def variable_prep(value, new_value_type: type = str, postprocess: bool = True) -> Any:

122 """\

123 Более интеллектуальная процедура преобразования базовых типов

124

125 :param value: Значение, которое необходимо преобразовать

126 :param new_value_type: Тип в который необходимо преобразовать значение

127 :param postprocess: Применять ли последующую обработку полученного результата

128 :returns: Значение ``value`` преобразованное к типу ``value_type``

129 """

130

131 if value is None:

132 return value

133

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__}"')

142

143 res = new_value_type(value)

144

145 if isinstance(res, str) and postprocess:

146 res = res.strip()

147

148 return res

149

150

151 def get_variable(name: str, default=None):

152 """\

153 Возвращает значения из хранилища контекста запроса bottle

154 """

155

156 _name = f'{VARIABLE_PREFIX}.{name}'

157 if _name in bottle.request.environ:

158 return bottle.request.environ[_name]

159

160 else:

161 return default

162

163

164 def set_variable(name: str, value):

165 """\

166 Устанавливает значения в хранилище контекста запроса bottle

167 """

168

169 bottle.request.environ[f'{VARIABLE_PREFIX}.{name}'] = value

170

171

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

179

180 if data is None:

181 data = {}

182

183 res = bottle.HTTPResponse(body=json.dumps(json_type_sanitizer(data)), status=status)

184 res.content_type = 'application/json'

185

186 if cookies is not None:

187 for cookie in cookies:

188 cookie.response_add(res)

189

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)

194

195 else:

196 res.delete_cookie(cookie)

197

198 return res

199

200

201 def get_request_json() -> Optional[dict]:

202 """\

203 Если в запросе, который мы обрабатываем в текущий момент, использовалась посылка данных JSON, получить их.

204 """

205

206 res = bottle.request.json

207 if res is None:

208 raise ValueError('Не обнаружено JSON в запросе')

209

210 return res

211

212

213 def get_cookie(name: str) -> Optional[str]:

214 """\

215 Получить значение ``cookie`` с заданным именем из текущего запроса.

216 """

217

218 return bottle.request.get_cookie(name)

219

220

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`` параметры, если иное не указано конкретно.

226

227 :param name: Имя параметра

228 :param param_type: Тип, в который нужно преобразовать полученный параметр

229 :param default: Значение, отдаваемое, если параметр не установлен в запросе

230 :param postprocess: Производить ли постобработку параметра

231 :param param_source: Источник параметра, [``get``, ``post``]

232 :returns Полученное из запроса значение

233 """

234

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}"')

243

244 if res is None:

245 res = default

246

247 if isinstance(param_type, str):

248 if param_type.lower() == 'json':

249 if not res:

250 return None

251

252 else:

253 try:

254 return json.loads(res)

255

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)

261

262 except (TypeError, ValueError) as e:

263 raise ValueError(f'Error parsing parameter "{name}": {e}')

264

265

266 def dict_filter(data: dict, need_keys: Union[List[str], Tuple[str]]) -> dict:

267 """\

268 Создать новый словарь на основе существующего, добавив в него только нужные ключи

269 """

270

271 return dict((key, val) for key, val in data.items() if key in need_keys)

272

273

274 def json_type_sanitizer(val):

275 """\

276 Преобразует значение ``val`` в пригодное для преобразования в json значение.

277 """

278

279 val_t = type(val)

280

281 if is_dataclass(val):

282 return json_type_sanitizer(asdict(val))

283

284 elif val_t in (int, float, str, bool) or val is None:

285 return val

286

287 elif val_t in (list, tuple):

288 return list(map(json_type_sanitizer, val))

289

290 elif val_t == dict:

291 return dict((key, json_type_sanitizer(d_val)) for key, d_val in val.items())

292

293 else:

294 return str(val)

295

296

297 def make_log_ident(user: Optional[str] = None) -> str:

298 """\

299 Создаём строку, идентифицирующую данный запрос для журналирования операций.

300

301 :param user: Если известен пользователь, задаём его здесь

302 """

303

304 ip = get_client_ip()

305 url_path = bottle.request.fullpath

306 conn_id = bottle.request.get_header(CONN_ID_HEADER)

307

308 if conn_id is not None:

309 ip = f'{ip} | {conn_id}'

310

311 if user is None:

312 ip = f'_NOID_[{ip}]'

313

314 else:

315 ip = f'{user}[{ip}]'

316

317 return f'{ip} - {url_path}'

318

319

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 сервисов

328

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

337

338 if isinstance(err, Exception):

339 err_type = type(err).__name__

340 err_msg = str(err) if msg is None else msg

341

342 else:

343 err_type = err

344 err_msg = msg if msg is not None else ''

345

346 res = make_response({

347 'error': err_type,

348 'msg': err_msg,

349 }, code, cookies)

350

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)

355

356 else:

357 res.delete_cookie(cookie)

358

359 return res