Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/flask/helpers.py: 30%
143 statements
« prev ^ index » next coverage.py v7.0.1, created at 2022-12-25 06:11 +0000
1import os
2import pkgutil
3import socket
4import sys
5import typing as t
6from datetime import datetime
7from functools import lru_cache
8from functools import update_wrapper
9from threading import RLock
11import werkzeug.utils
12from werkzeug.exceptions import abort as _wz_abort
13from werkzeug.utils import redirect as _wz_redirect
15from .globals import _cv_request
16from .globals import current_app
17from .globals import request
18from .globals import request_ctx
19from .globals import session
20from .signals import message_flashed
22if t.TYPE_CHECKING: # pragma: no cover
23 from werkzeug.wrappers import Response as BaseResponse
24 from .wrappers import Response
25 import typing_extensions as te
28def get_env() -> str:
29 """Get the environment the app is running in, indicated by the
30 :envvar:`FLASK_ENV` environment variable. The default is
31 ``'production'``.
33 .. deprecated:: 2.2
34 Will be removed in Flask 2.3.
35 """
36 import warnings
38 warnings.warn(
39 "'FLASK_ENV' and 'get_env' are deprecated and will be removed"
40 " in Flask 2.3. Use 'FLASK_DEBUG' instead.",
41 DeprecationWarning,
42 stacklevel=2,
43 )
44 return os.environ.get("FLASK_ENV") or "production"
47def get_debug_flag() -> bool:
48 """Get whether debug mode should be enabled for the app, indicated by the
49 :envvar:`FLASK_DEBUG` environment variable. The default is ``False``.
50 """
51 val = os.environ.get("FLASK_DEBUG")
53 if not val:
54 env = os.environ.get("FLASK_ENV")
56 if env is not None:
57 print(
58 "'FLASK_ENV' is deprecated and will not be used in"
59 " Flask 2.3. Use 'FLASK_DEBUG' instead.",
60 file=sys.stderr,
61 )
62 return env == "development"
64 return False
66 return val.lower() not in {"0", "false", "no"}
69def get_load_dotenv(default: bool = True) -> bool:
70 """Get whether the user has disabled loading default dotenv files by
71 setting :envvar:`FLASK_SKIP_DOTENV`. The default is ``True``, load
72 the files.
74 :param default: What to return if the env var isn't set.
75 """
76 val = os.environ.get("FLASK_SKIP_DOTENV")
78 if not val:
79 return default
81 return val.lower() in ("0", "false", "no")
84def stream_with_context(
85 generator_or_function: t.Union[
86 t.Iterator[t.AnyStr], t.Callable[..., t.Iterator[t.AnyStr]]
87 ]
88) -> t.Iterator[t.AnyStr]:
89 """Request contexts disappear when the response is started on the server.
90 This is done for efficiency reasons and to make it less likely to encounter
91 memory leaks with badly written WSGI middlewares. The downside is that if
92 you are using streamed responses, the generator cannot access request bound
93 information any more.
95 This function however can help you keep the context around for longer::
97 from flask import stream_with_context, request, Response
99 @app.route('/stream')
100 def streamed_response():
101 @stream_with_context
102 def generate():
103 yield 'Hello '
104 yield request.args['name']
105 yield '!'
106 return Response(generate())
108 Alternatively it can also be used around a specific generator::
110 from flask import stream_with_context, request, Response
112 @app.route('/stream')
113 def streamed_response():
114 def generate():
115 yield 'Hello '
116 yield request.args['name']
117 yield '!'
118 return Response(stream_with_context(generate()))
120 .. versionadded:: 0.9
121 """
122 try:
123 gen = iter(generator_or_function) # type: ignore
124 except TypeError:
126 def decorator(*args: t.Any, **kwargs: t.Any) -> t.Any:
127 gen = generator_or_function(*args, **kwargs) # type: ignore
128 return stream_with_context(gen)
130 return update_wrapper(decorator, generator_or_function) # type: ignore
132 def generator() -> t.Generator:
133 ctx = _cv_request.get(None)
134 if ctx is None:
135 raise RuntimeError(
136 "'stream_with_context' can only be used when a request"
137 " context is active, such as in a view function."
138 )
139 with ctx:
140 # Dummy sentinel. Has to be inside the context block or we're
141 # not actually keeping the context around.
142 yield None
144 # The try/finally is here so that if someone passes a WSGI level
145 # iterator in we're still running the cleanup logic. Generators
146 # don't need that because they are closed on their destruction
147 # automatically.
148 try:
149 yield from gen
150 finally:
151 if hasattr(gen, "close"):
152 gen.close() # type: ignore
154 # The trick is to start the generator. Then the code execution runs until
155 # the first dummy None is yielded at which point the context was already
156 # pushed. This item is discarded. Then when the iteration continues the
157 # real generator is executed.
158 wrapped_g = generator()
159 next(wrapped_g)
160 return wrapped_g
163def make_response(*args: t.Any) -> "Response":
164 """Sometimes it is necessary to set additional headers in a view. Because
165 views do not have to return response objects but can return a value that
166 is converted into a response object by Flask itself, it becomes tricky to
167 add headers to it. This function can be called instead of using a return
168 and you will get a response object which you can use to attach headers.
170 If view looked like this and you want to add a new header::
172 def index():
173 return render_template('index.html', foo=42)
175 You can now do something like this::
177 def index():
178 response = make_response(render_template('index.html', foo=42))
179 response.headers['X-Parachutes'] = 'parachutes are cool'
180 return response
182 This function accepts the very same arguments you can return from a
183 view function. This for example creates a response with a 404 error
184 code::
186 response = make_response(render_template('not_found.html'), 404)
188 The other use case of this function is to force the return value of a
189 view function into a response which is helpful with view
190 decorators::
192 response = make_response(view_function())
193 response.headers['X-Parachutes'] = 'parachutes are cool'
195 Internally this function does the following things:
197 - if no arguments are passed, it creates a new response argument
198 - if one argument is passed, :meth:`flask.Flask.make_response`
199 is invoked with it.
200 - if more than one argument is passed, the arguments are passed
201 to the :meth:`flask.Flask.make_response` function as tuple.
203 .. versionadded:: 0.6
204 """
205 if not args:
206 return current_app.response_class()
207 if len(args) == 1:
208 args = args[0]
209 return current_app.make_response(args) # type: ignore
212def url_for(
213 endpoint: str,
214 *,
215 _anchor: t.Optional[str] = None,
216 _method: t.Optional[str] = None,
217 _scheme: t.Optional[str] = None,
218 _external: t.Optional[bool] = None,
219 **values: t.Any,
220) -> str:
221 """Generate a URL to the given endpoint with the given values.
223 This requires an active request or application context, and calls
224 :meth:`current_app.url_for() <flask.Flask.url_for>`. See that method
225 for full documentation.
227 :param endpoint: The endpoint name associated with the URL to
228 generate. If this starts with a ``.``, the current blueprint
229 name (if any) will be used.
230 :param _anchor: If given, append this as ``#anchor`` to the URL.
231 :param _method: If given, generate the URL associated with this
232 method for the endpoint.
233 :param _scheme: If given, the URL will have this scheme if it is
234 external.
235 :param _external: If given, prefer the URL to be internal (False) or
236 require it to be external (True). External URLs include the
237 scheme and domain. When not in an active request, URLs are
238 external by default.
239 :param values: Values to use for the variable parts of the URL rule.
240 Unknown keys are appended as query string arguments, like
241 ``?a=b&c=d``.
243 .. versionchanged:: 2.2
244 Calls ``current_app.url_for``, allowing an app to override the
245 behavior.
247 .. versionchanged:: 0.10
248 The ``_scheme`` parameter was added.
250 .. versionchanged:: 0.9
251 The ``_anchor`` and ``_method`` parameters were added.
253 .. versionchanged:: 0.9
254 Calls ``app.handle_url_build_error`` on build errors.
255 """
256 return current_app.url_for(
257 endpoint,
258 _anchor=_anchor,
259 _method=_method,
260 _scheme=_scheme,
261 _external=_external,
262 **values,
263 )
266def redirect(
267 location: str, code: int = 302, Response: t.Optional[t.Type["BaseResponse"]] = None
268) -> "BaseResponse":
269 """Create a redirect response object.
271 If :data:`~flask.current_app` is available, it will use its
272 :meth:`~flask.Flask.redirect` method, otherwise it will use
273 :func:`werkzeug.utils.redirect`.
275 :param location: The URL to redirect to.
276 :param code: The status code for the redirect.
277 :param Response: The response class to use. Not used when
278 ``current_app`` is active, which uses ``app.response_class``.
280 .. versionadded:: 2.2
281 Calls ``current_app.redirect`` if available instead of always
282 using Werkzeug's default ``redirect``.
283 """
284 if current_app:
285 return current_app.redirect(location, code=code)
287 return _wz_redirect(location, code=code, Response=Response)
290def abort( # type: ignore[misc]
291 code: t.Union[int, "BaseResponse"], *args: t.Any, **kwargs: t.Any
292) -> "te.NoReturn":
293 """Raise an :exc:`~werkzeug.exceptions.HTTPException` for the given
294 status code.
296 If :data:`~flask.current_app` is available, it will call its
297 :attr:`~flask.Flask.aborter` object, otherwise it will use
298 :func:`werkzeug.exceptions.abort`.
300 :param code: The status code for the exception, which must be
301 registered in ``app.aborter``.
302 :param args: Passed to the exception.
303 :param kwargs: Passed to the exception.
305 .. versionadded:: 2.2
306 Calls ``current_app.aborter`` if available instead of always
307 using Werkzeug's default ``abort``.
308 """
309 if current_app:
310 current_app.aborter(code, *args, **kwargs)
312 _wz_abort(code, *args, **kwargs)
315def get_template_attribute(template_name: str, attribute: str) -> t.Any:
316 """Loads a macro (or variable) a template exports. This can be used to
317 invoke a macro from within Python code. If you for example have a
318 template named :file:`_cider.html` with the following contents:
320 .. sourcecode:: html+jinja
322 {% macro hello(name) %}Hello {{ name }}!{% endmacro %}
324 You can access this from Python code like this::
326 hello = get_template_attribute('_cider.html', 'hello')
327 return hello('World')
329 .. versionadded:: 0.2
331 :param template_name: the name of the template
332 :param attribute: the name of the variable of macro to access
333 """
334 return getattr(current_app.jinja_env.get_template(template_name).module, attribute)
337def flash(message: str, category: str = "message") -> None:
338 """Flashes a message to the next request. In order to remove the
339 flashed message from the session and to display it to the user,
340 the template has to call :func:`get_flashed_messages`.
342 .. versionchanged:: 0.3
343 `category` parameter added.
345 :param message: the message to be flashed.
346 :param category: the category for the message. The following values
347 are recommended: ``'message'`` for any kind of message,
348 ``'error'`` for errors, ``'info'`` for information
349 messages and ``'warning'`` for warnings. However any
350 kind of string can be used as category.
351 """
352 # Original implementation:
353 #
354 # session.setdefault('_flashes', []).append((category, message))
355 #
356 # This assumed that changes made to mutable structures in the session are
357 # always in sync with the session object, which is not true for session
358 # implementations that use external storage for keeping their keys/values.
359 flashes = session.get("_flashes", [])
360 flashes.append((category, message))
361 session["_flashes"] = flashes
362 message_flashed.send(
363 current_app._get_current_object(), # type: ignore
364 message=message,
365 category=category,
366 )
369def get_flashed_messages(
370 with_categories: bool = False, category_filter: t.Iterable[str] = ()
371) -> t.Union[t.List[str], t.List[t.Tuple[str, str]]]:
372 """Pulls all flashed messages from the session and returns them.
373 Further calls in the same request to the function will return
374 the same messages. By default just the messages are returned,
375 but when `with_categories` is set to ``True``, the return value will
376 be a list of tuples in the form ``(category, message)`` instead.
378 Filter the flashed messages to one or more categories by providing those
379 categories in `category_filter`. This allows rendering categories in
380 separate html blocks. The `with_categories` and `category_filter`
381 arguments are distinct:
383 * `with_categories` controls whether categories are returned with message
384 text (``True`` gives a tuple, where ``False`` gives just the message text).
385 * `category_filter` filters the messages down to only those matching the
386 provided categories.
388 See :doc:`/patterns/flashing` for examples.
390 .. versionchanged:: 0.3
391 `with_categories` parameter added.
393 .. versionchanged:: 0.9
394 `category_filter` parameter added.
396 :param with_categories: set to ``True`` to also receive categories.
397 :param category_filter: filter of categories to limit return values. Only
398 categories in the list will be returned.
399 """
400 flashes = request_ctx.flashes
401 if flashes is None:
402 flashes = session.pop("_flashes") if "_flashes" in session else []
403 request_ctx.flashes = flashes
404 if category_filter:
405 flashes = list(filter(lambda f: f[0] in category_filter, flashes))
406 if not with_categories:
407 return [x[1] for x in flashes]
408 return flashes
411def _prepare_send_file_kwargs(**kwargs: t.Any) -> t.Dict[str, t.Any]:
412 if kwargs.get("max_age") is None:
413 kwargs["max_age"] = current_app.get_send_file_max_age
415 kwargs.update(
416 environ=request.environ,
417 use_x_sendfile=current_app.config["USE_X_SENDFILE"],
418 response_class=current_app.response_class,
419 _root_path=current_app.root_path, # type: ignore
420 )
421 return kwargs
424def send_file(
425 path_or_file: t.Union[os.PathLike, str, t.BinaryIO],
426 mimetype: t.Optional[str] = None,
427 as_attachment: bool = False,
428 download_name: t.Optional[str] = None,
429 conditional: bool = True,
430 etag: t.Union[bool, str] = True,
431 last_modified: t.Optional[t.Union[datetime, int, float]] = None,
432 max_age: t.Optional[
433 t.Union[int, t.Callable[[t.Optional[str]], t.Optional[int]]]
434 ] = None,
435) -> "Response":
436 """Send the contents of a file to the client.
438 The first argument can be a file path or a file-like object. Paths
439 are preferred in most cases because Werkzeug can manage the file and
440 get extra information from the path. Passing a file-like object
441 requires that the file is opened in binary mode, and is mostly
442 useful when building a file in memory with :class:`io.BytesIO`.
444 Never pass file paths provided by a user. The path is assumed to be
445 trusted, so a user could craft a path to access a file you didn't
446 intend. Use :func:`send_from_directory` to safely serve
447 user-requested paths from within a directory.
449 If the WSGI server sets a ``file_wrapper`` in ``environ``, it is
450 used, otherwise Werkzeug's built-in wrapper is used. Alternatively,
451 if the HTTP server supports ``X-Sendfile``, configuring Flask with
452 ``USE_X_SENDFILE = True`` will tell the server to send the given
453 path, which is much more efficient than reading it in Python.
455 :param path_or_file: The path to the file to send, relative to the
456 current working directory if a relative path is given.
457 Alternatively, a file-like object opened in binary mode. Make
458 sure the file pointer is seeked to the start of the data.
459 :param mimetype: The MIME type to send for the file. If not
460 provided, it will try to detect it from the file name.
461 :param as_attachment: Indicate to a browser that it should offer to
462 save the file instead of displaying it.
463 :param download_name: The default name browsers will use when saving
464 the file. Defaults to the passed file name.
465 :param conditional: Enable conditional and range responses based on
466 request headers. Requires passing a file path and ``environ``.
467 :param etag: Calculate an ETag for the file, which requires passing
468 a file path. Can also be a string to use instead.
469 :param last_modified: The last modified time to send for the file,
470 in seconds. If not provided, it will try to detect it from the
471 file path.
472 :param max_age: How long the client should cache the file, in
473 seconds. If set, ``Cache-Control`` will be ``public``, otherwise
474 it will be ``no-cache`` to prefer conditional caching.
476 .. versionchanged:: 2.0
477 ``download_name`` replaces the ``attachment_filename``
478 parameter. If ``as_attachment=False``, it is passed with
479 ``Content-Disposition: inline`` instead.
481 .. versionchanged:: 2.0
482 ``max_age`` replaces the ``cache_timeout`` parameter.
483 ``conditional`` is enabled and ``max_age`` is not set by
484 default.
486 .. versionchanged:: 2.0
487 ``etag`` replaces the ``add_etags`` parameter. It can be a
488 string to use instead of generating one.
490 .. versionchanged:: 2.0
491 Passing a file-like object that inherits from
492 :class:`~io.TextIOBase` will raise a :exc:`ValueError` rather
493 than sending an empty file.
495 .. versionadded:: 2.0
496 Moved the implementation to Werkzeug. This is now a wrapper to
497 pass some Flask-specific arguments.
499 .. versionchanged:: 1.1
500 ``filename`` may be a :class:`~os.PathLike` object.
502 .. versionchanged:: 1.1
503 Passing a :class:`~io.BytesIO` object supports range requests.
505 .. versionchanged:: 1.0.3
506 Filenames are encoded with ASCII instead of Latin-1 for broader
507 compatibility with WSGI servers.
509 .. versionchanged:: 1.0
510 UTF-8 filenames as specified in :rfc:`2231` are supported.
512 .. versionchanged:: 0.12
513 The filename is no longer automatically inferred from file
514 objects. If you want to use automatic MIME and etag support,
515 pass a filename via ``filename_or_fp`` or
516 ``attachment_filename``.
518 .. versionchanged:: 0.12
519 ``attachment_filename`` is preferred over ``filename`` for MIME
520 detection.
522 .. versionchanged:: 0.9
523 ``cache_timeout`` defaults to
524 :meth:`Flask.get_send_file_max_age`.
526 .. versionchanged:: 0.7
527 MIME guessing and etag support for file-like objects was
528 deprecated because it was unreliable. Pass a filename if you are
529 able to, otherwise attach an etag yourself.
531 .. versionchanged:: 0.5
532 The ``add_etags``, ``cache_timeout`` and ``conditional``
533 parameters were added. The default behavior is to add etags.
535 .. versionadded:: 0.2
536 """
537 return werkzeug.utils.send_file( # type: ignore[return-value]
538 **_prepare_send_file_kwargs(
539 path_or_file=path_or_file,
540 environ=request.environ,
541 mimetype=mimetype,
542 as_attachment=as_attachment,
543 download_name=download_name,
544 conditional=conditional,
545 etag=etag,
546 last_modified=last_modified,
547 max_age=max_age,
548 )
549 )
552def send_from_directory(
553 directory: t.Union[os.PathLike, str],
554 path: t.Union[os.PathLike, str],
555 **kwargs: t.Any,
556) -> "Response":
557 """Send a file from within a directory using :func:`send_file`.
559 .. code-block:: python
561 @app.route("/uploads/<path:name>")
562 def download_file(name):
563 return send_from_directory(
564 app.config['UPLOAD_FOLDER'], name, as_attachment=True
565 )
567 This is a secure way to serve files from a folder, such as static
568 files or uploads. Uses :func:`~werkzeug.security.safe_join` to
569 ensure the path coming from the client is not maliciously crafted to
570 point outside the specified directory.
572 If the final path does not point to an existing regular file,
573 raises a 404 :exc:`~werkzeug.exceptions.NotFound` error.
575 :param directory: The directory that ``path`` must be located under,
576 relative to the current application's root path.
577 :param path: The path to the file to send, relative to
578 ``directory``.
579 :param kwargs: Arguments to pass to :func:`send_file`.
581 .. versionchanged:: 2.0
582 ``path`` replaces the ``filename`` parameter.
584 .. versionadded:: 2.0
585 Moved the implementation to Werkzeug. This is now a wrapper to
586 pass some Flask-specific arguments.
588 .. versionadded:: 0.5
589 """
590 return werkzeug.utils.send_from_directory( # type: ignore[return-value]
591 directory, path, **_prepare_send_file_kwargs(**kwargs)
592 )
595def get_root_path(import_name: str) -> str:
596 """Find the root path of a package, or the path that contains a
597 module. If it cannot be found, returns the current working
598 directory.
600 Not to be confused with the value returned by :func:`find_package`.
602 :meta private:
603 """
604 # Module already imported and has a file attribute. Use that first.
605 mod = sys.modules.get(import_name)
607 if mod is not None and hasattr(mod, "__file__") and mod.__file__ is not None:
608 return os.path.dirname(os.path.abspath(mod.__file__))
610 # Next attempt: check the loader.
611 loader = pkgutil.get_loader(import_name)
613 # Loader does not exist or we're referring to an unloaded main
614 # module or a main module without path (interactive sessions), go
615 # with the current working directory.
616 if loader is None or import_name == "__main__":
617 return os.getcwd()
619 if hasattr(loader, "get_filename"):
620 filepath = loader.get_filename(import_name) # type: ignore
621 else:
622 # Fall back to imports.
623 __import__(import_name)
624 mod = sys.modules[import_name]
625 filepath = getattr(mod, "__file__", None)
627 # If we don't have a file path it might be because it is a
628 # namespace package. In this case pick the root path from the
629 # first module that is contained in the package.
630 if filepath is None:
631 raise RuntimeError(
632 "No root path can be found for the provided module"
633 f" {import_name!r}. This can happen because the module"
634 " came from an import hook that does not provide file"
635 " name information or because it's a namespace package."
636 " In this case the root path needs to be explicitly"
637 " provided."
638 )
640 # filepath is import_name.py for a module, or __init__.py for a package.
641 return os.path.dirname(os.path.abspath(filepath))
644class locked_cached_property(werkzeug.utils.cached_property):
645 """A :func:`property` that is only evaluated once. Like
646 :class:`werkzeug.utils.cached_property` except access uses a lock
647 for thread safety.
649 .. versionchanged:: 2.0
650 Inherits from Werkzeug's ``cached_property`` (and ``property``).
651 """
653 def __init__(
654 self,
655 fget: t.Callable[[t.Any], t.Any],
656 name: t.Optional[str] = None,
657 doc: t.Optional[str] = None,
658 ) -> None:
659 super().__init__(fget, name=name, doc=doc)
660 self.lock = RLock()
662 def __get__(self, obj: object, type: type = None) -> t.Any: # type: ignore
663 if obj is None:
664 return self
666 with self.lock:
667 return super().__get__(obj, type=type)
669 def __set__(self, obj: object, value: t.Any) -> None:
670 with self.lock:
671 super().__set__(obj, value)
673 def __delete__(self, obj: object) -> None:
674 with self.lock:
675 super().__delete__(obj)
678def is_ip(value: str) -> bool:
679 """Determine if the given string is an IP address.
681 :param value: value to check
682 :type value: str
684 :return: True if string is an IP address
685 :rtype: bool
686 """
687 for family in (socket.AF_INET, socket.AF_INET6):
688 try:
689 socket.inet_pton(family, value)
690 except OSError:
691 pass
692 else:
693 return True
695 return False
698@lru_cache(maxsize=None)
699def _split_blueprint_path(name: str) -> t.List[str]:
700 out: t.List[str] = [name]
702 if "." in name:
703 out.extend(_split_blueprint_path(name.rpartition(".")[0]))
705 return out