Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/flask/ctx.py: 28%

1from __future__ import annotations 

2 

3import contextvars 

4import sys 

5import typing as t 

6from functools import update_wrapper 

7from types import TracebackType 

8 

9from werkzeug.exceptions import HTTPException 

10 

11from . import typing as ft 

12from .globals import _cv_app 

13from .globals import _cv_request 

14from .signals import appcontext_popped 

15from .signals import appcontext_pushed 

16 

17if t.TYPE_CHECKING: # pragma: no cover 

18 from _typeshed.wsgi import WSGIEnvironment 

19 

20 from .app import Flask 

21 from .sessions import SessionMixin 

22 from .wrappers import Request 

23 

24 

25# a singleton sentinel value for parameter defaults 

26_sentinel = object() 

27 

28 

29class _AppCtxGlobals: 

30 """A plain object. Used as a namespace for storing data during an 

31 application context. 

32 

33 Creating an app context automatically creates this object, which is 

34 made available as the :data:`g` proxy. 

35 

36 .. describe:: 'key' in g 

37 

38 Check whether an attribute is present. 

39 

40 .. versionadded:: 0.10 

41 

42 .. describe:: iter(g) 

43 

44 Return an iterator over the attribute names. 

45 

46 .. versionadded:: 0.10 

47 """ 

48 

49 # Define attr methods to let mypy know this is a namespace object 

50 # that has arbitrary attributes. 

51 

52 def __getattr__(self, name: str) -> t.Any: 

53 try: 

54 return self.__dict__[name] 

55 except KeyError: 

56 raise AttributeError(name) from None 

57 

58 def __setattr__(self, name: str, value: t.Any) -> None: 

59 self.__dict__[name] = value 

60 

61 def __delattr__(self, name: str) -> None: 

62 try: 

63 del self.__dict__[name] 

64 except KeyError: 

65 raise AttributeError(name) from None 

66 

67 def get(self, name: str, default: t.Any | None = None) -> t.Any: 

68 """Get an attribute by name, or a default value. Like 

69 :meth:`dict.get`. 

70 

71 :param name: Name of attribute to get. 

72 :param default: Value to return if the attribute is not present. 

73 

74 .. versionadded:: 0.10 

75 """ 

76 return self.__dict__.get(name, default) 

77 

78 def pop(self, name: str, default: t.Any = _sentinel) -> t.Any: 

79 """Get and remove an attribute by name. Like :meth:`dict.pop`. 

80 

81 :param name: Name of attribute to pop. 

82 :param default: Value to return if the attribute is not present, 

83 instead of raising a ``KeyError``. 

84 

85 .. versionadded:: 0.11 

86 """ 

87 if default is _sentinel: 

88 return self.__dict__.pop(name) 

89 else: 

90 return self.__dict__.pop(name, default) 

91 

92 def setdefault(self, name: str, default: t.Any = None) -> t.Any: 

93 """Get the value of an attribute if it is present, otherwise 

94 set and return a default value. Like :meth:`dict.setdefault`. 

95 

96 :param name: Name of attribute to get. 

97 :param default: Value to set and return if the attribute is not 

98 present. 

99 

100 .. versionadded:: 0.11 

101 """ 

102 return self.__dict__.setdefault(name, default) 

103 

104 def __contains__(self, item: str) -> bool: 

105 return item in self.__dict__ 

106 

107 def __iter__(self) -> t.Iterator[str]: 

108 return iter(self.__dict__) 

109 

110 def __repr__(self) -> str: 

111 ctx = _cv_app.get(None) 

112 if ctx is not None: 

113 return f"<flask.g of '{ctx.app.name}'>" 

114 return object.__repr__(self) 

115 

116 

117def after_this_request( 

118 f: ft.AfterRequestCallable[t.Any], 

119) -> ft.AfterRequestCallable[t.Any]: 

120 """Executes a function after this request. This is useful to modify 

121 response objects. The function is passed the response object and has 

122 to return the same or a new one. 

123 

124 Example:: 

125 

126 @app.route('/') 

127 def index(): 

128 @after_this_request 

129 def add_header(response): 

130 response.headers['X-Foo'] = 'Parachute' 

131 return response 

132 return 'Hello World!' 

133 

134 This is more useful if a function other than the view function wants to 

135 modify a response. For instance think of a decorator that wants to add 

136 some headers without converting the return value into a response object. 

137 

138 .. versionadded:: 0.9 

139 """ 

140 ctx = _cv_request.get(None) 

141 

142 if ctx is None: 

143 raise RuntimeError( 

144 "'after_this_request' can only be used when a request" 

145 " context is active, such as in a view function." 

146 ) 

147 

148 ctx._after_request_functions.append(f) 

149 return f 

150 

151 

152F = t.TypeVar("F", bound=t.Callable[..., t.Any]) 

153 

154 

155def copy_current_request_context(f: F) -> F: 

156 """A helper function that decorates a function to retain the current 

157 request context. This is useful when working with greenlets. The moment 

158 the function is decorated a copy of the request context is created and 

159 then pushed when the function is called. The current session is also 

160 included in the copied request context. 

161 

162 Example:: 

163 

164 import gevent 

165 from flask import copy_current_request_context 

166 

167 @app.route('/') 

168 def index(): 

169 @copy_current_request_context 

170 def do_some_work(): 

171 # do some work here, it can access flask.request or 

172 # flask.session like you would otherwise in the view function. 

173 ... 

174 gevent.spawn(do_some_work) 

175 return 'Regular response' 

176 

177 .. versionadded:: 0.10 

178 """ 

179 ctx = _cv_request.get(None) 

180 

181 if ctx is None: 

182 raise RuntimeError( 

183 "'copy_current_request_context' can only be used when a" 

184 " request context is active, such as in a view function." 

185 ) 

186 

187 ctx = ctx.copy() 

188 

189 def wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any: 

190 with ctx: # type: ignore[union-attr] 

191 return ctx.app.ensure_sync(f)(*args, **kwargs) # type: ignore[union-attr] 

192 

193 return update_wrapper(wrapper, f) # type: ignore[return-value] 

194 

195 

196def has_request_context() -> bool: 

197 """If you have code that wants to test if a request context is there or 

198 not this function can be used. For instance, you may want to take advantage 

199 of request information if the request object is available, but fail 

200 silently if it is unavailable. 

201 

202 :: 

203 

204 class User(db.Model): 

205 

206 def __init__(self, username, remote_addr=None): 

207 self.username = username 

208 if remote_addr is None and has_request_context(): 

209 remote_addr = request.remote_addr 

210 self.remote_addr = remote_addr 

211 

212 Alternatively you can also just test any of the context bound objects 

213 (such as :class:`request` or :class:`g`) for truthness:: 

214 

215 class User(db.Model): 

216 

217 def __init__(self, username, remote_addr=None): 

218 self.username = username 

219 if remote_addr is None and request: 

220 remote_addr = request.remote_addr 

221 self.remote_addr = remote_addr 

222 

223 .. versionadded:: 0.7 

224 """ 

225 return _cv_request.get(None) is not None 

226 

227 

228def has_app_context() -> bool: 

229 """Works like :func:`has_request_context` but for the application 

230 context. You can also just do a boolean check on the 

231 :data:`current_app` object instead. 

232 

233 .. versionadded:: 0.9 

234 """ 

235 return _cv_app.get(None) is not None 

236 

237 

238class AppContext: 

239 """The app context contains application-specific information. An app 

240 context is created and pushed at the beginning of each request if 

241 one is not already active. An app context is also pushed when 

242 running CLI commands. 

243 """ 

244 

245 def __init__(self, app: Flask) -> None: 

246 self.app = app 

247 self.url_adapter = app.create_url_adapter(None) 

248 self.g: _AppCtxGlobals = app.app_ctx_globals_class() 

249 self._cv_tokens: list[contextvars.Token[AppContext]] = [] 

250 

251 def push(self) -> None: 

252 """Binds the app context to the current context.""" 

253 self._cv_tokens.append(_cv_app.set(self)) 

254 appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync) 

255 

256 def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore 

257 """Pops the app context.""" 

258 try: 

259 if len(self._cv_tokens) == 1: 

260 if exc is _sentinel: 

261 exc = sys.exc_info()[1] 

262 self.app.do_teardown_appcontext(exc) 

263 finally: 

264 ctx = _cv_app.get() 

265 _cv_app.reset(self._cv_tokens.pop()) 

266 

267 if ctx is not self: 

268 raise AssertionError( 

269 f"Popped wrong app context. ({ctx!r} instead of {self!r})" 

270 ) 

271 

272 appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync) 

273 

274 def __enter__(self) -> AppContext: 

275 self.push() 

276 return self 

277 

278 def __exit__( 

279 self, 

280 exc_type: type | None, 

281 exc_value: BaseException | None, 

282 tb: TracebackType | None, 

283 ) -> None: 

284 self.pop(exc_value) 

285 

286 

287class RequestContext: 

288 """The request context contains per-request information. The Flask 

289 app creates and pushes it at the beginning of the request, then pops 

290 it at the end of the request. It will create the URL adapter and 

291 request object for the WSGI environment provided. 

292 

293 Do not attempt to use this class directly, instead use 

294 :meth:`~flask.Flask.test_request_context` and 

295 :meth:`~flask.Flask.request_context` to create this object. 

296 

297 When the request context is popped, it will evaluate all the 

298 functions registered on the application for teardown execution 

299 (:meth:`~flask.Flask.teardown_request`). 

300 

301 The request context is automatically popped at the end of the 

302 request. When using the interactive debugger, the context will be 

303 restored so ``request`` is still accessible. Similarly, the test 

304 client can preserve the context after the request ends. However, 

305 teardown functions may already have closed some resources such as 

306 database connections. 

307 """ 

308 

309 def __init__( 

310 self, 

311 app: Flask, 

312 environ: WSGIEnvironment, 

313 request: Request | None = None, 

314 session: SessionMixin | None = None, 

315 ) -> None: 

316 self.app = app 

317 if request is None: 

318 request = app.request_class(environ) 

319 request.json_module = app.json 

320 self.request: Request = request 

321 self.url_adapter = None 

322 try: 

323 self.url_adapter = app.create_url_adapter(self.request) 

324 except HTTPException as e: 

325 self.request.routing_exception = e 

326 self.flashes: list[tuple[str, str]] | None = None 

327 self.session: SessionMixin | None = session 

328 # Functions that should be executed after the request on the response 

329 # object. These will be called before the regular "after_request" 

330 # functions. 

331 self._after_request_functions: list[ft.AfterRequestCallable[t.Any]] = [] 

332 

333 self._cv_tokens: list[ 

334 tuple[contextvars.Token[RequestContext], AppContext | None] 

335 ] = [] 

336 

337 def copy(self) -> RequestContext: 

338 """Creates a copy of this request context with the same request object. 

339 This can be used to move a request context to a different greenlet. 

340 Because the actual request object is the same this cannot be used to 

341 move a request context to a different thread unless access to the 

342 request object is locked. 

343 

344 .. versionadded:: 0.10 

345 

346 .. versionchanged:: 1.1 

347 The current session object is used instead of reloading the original 

348 data. This prevents `flask.session` pointing to an out-of-date object. 

349 """ 

350 return self.__class__( 

351 self.app, 

352 environ=self.request.environ, 

353 request=self.request, 

354 session=self.session, 

355 ) 

356 

357 def match_request(self) -> None: 

358 """Can be overridden by a subclass to hook into the matching 

359 of the request. 

360 """ 

361 try: 

362 result = self.url_adapter.match(return_rule=True) # type: ignore 

363 self.request.url_rule, self.request.view_args = result # type: ignore 

364 except HTTPException as e: 

365 self.request.routing_exception = e 

366 

367 def push(self) -> None: 

368 # Before we push the request context we have to ensure that there 

369 # is an application context. 

370 app_ctx = _cv_app.get(None) 

371 

372 if app_ctx is None or app_ctx.app is not self.app: 

373 app_ctx = self.app.app_context() 

374 app_ctx.push() 

375 else: 

376 app_ctx = None 

377 

378 self._cv_tokens.append((_cv_request.set(self), app_ctx)) 

379 

380 # Open the session at the moment that the request context is available. 

381 # This allows a custom open_session method to use the request context. 

382 # Only open a new session if this is the first time the request was 

383 # pushed, otherwise stream_with_context loses the session. 

384 if self.session is None: 

385 session_interface = self.app.session_interface 

386 self.session = session_interface.open_session(self.app, self.request) 

387 

388 if self.session is None: 

389 self.session = session_interface.make_null_session(self.app) 

390 

391 # Match the request URL after loading the session, so that the 

392 # session is available in custom URL converters. 

393 if self.url_adapter is not None: 

394 self.match_request() 

395 

396 def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore 

397 """Pops the request context and unbinds it by doing that. This will 

398 also trigger the execution of functions registered by the 

399 :meth:`~flask.Flask.teardown_request` decorator. 

400 

401 .. versionchanged:: 0.9 

402 Added the `exc` argument. 

403 """ 

404 clear_request = len(self._cv_tokens) == 1 

405 

406 try: 

407 if clear_request: 

408 if exc is _sentinel: 

409 exc = sys.exc_info()[1] 

410 self.app.do_teardown_request(exc) 

411 

412 request_close = getattr(self.request, "close", None) 

413 if request_close is not None: 

414 request_close() 

415 finally: 

416 ctx = _cv_request.get() 

417 token, app_ctx = self._cv_tokens.pop() 

418 _cv_request.reset(token) 

419 

420 # get rid of circular dependencies at the end of the request 

421 # so that we don't require the GC to be active. 

422 if clear_request: 

423 ctx.request.environ["werkzeug.request"] = None 

424 

425 if app_ctx is not None: 

426 app_ctx.pop(exc) 

427 

428 if ctx is not self: 

429 raise AssertionError( 

430 f"Popped wrong request context. ({ctx!r} instead of {self!r})" 

431 ) 

432 

433 def __enter__(self) -> RequestContext: 

434 self.push() 

435 return self 

436 

437 def __exit__( 

438 self, 

439 exc_type: type | None, 

440 exc_value: BaseException | None, 

441 tb: TracebackType | None, 

442 ) -> None: 

443 self.pop(exc_value) 

444 

445 def __repr__(self) -> str: 

446 return ( 

447 f"<{type(self).__name__} {self.request.url!r}" 

448 f" [{self.request.method}] of {self.app.name}>" 

449 )