123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091 |
- import functools
- import inspect
- import logging
- import os
- import sys
- import typing as t
- import weakref
- from datetime import timedelta
- from itertools import chain
- from threading import Lock
- from types import TracebackType
- from werkzeug.datastructures import Headers
- from werkzeug.datastructures import ImmutableDict
- from werkzeug.exceptions import BadRequest
- from werkzeug.exceptions import BadRequestKeyError
- from werkzeug.exceptions import HTTPException
- from werkzeug.exceptions import InternalServerError
- from werkzeug.local import ContextVar
- from werkzeug.routing import BuildError
- from werkzeug.routing import Map
- from werkzeug.routing import MapAdapter
- from werkzeug.routing import RequestRedirect
- from werkzeug.routing import RoutingException
- from werkzeug.routing import Rule
- from werkzeug.wrappers import Response as BaseResponse
- from . import cli
- from . import json
- from .config import Config
- from .config import ConfigAttribute
- from .ctx import _AppCtxGlobals
- from .ctx import AppContext
- from .ctx import RequestContext
- from .globals import _request_ctx_stack
- from .globals import g
- from .globals import request
- from .globals import session
- from .helpers import _split_blueprint_path
- from .helpers import get_debug_flag
- from .helpers import get_env
- from .helpers import get_flashed_messages
- from .helpers import get_load_dotenv
- from .helpers import locked_cached_property
- from .helpers import url_for
- from .json import jsonify
- from .logging import create_logger
- from .scaffold import _endpoint_from_view_func
- from .scaffold import _sentinel
- from .scaffold import find_package
- from .scaffold import Scaffold
- from .scaffold import setupmethod
- from .sessions import SecureCookieSessionInterface
- from .signals import appcontext_tearing_down
- from .signals import got_request_exception
- from .signals import request_finished
- from .signals import request_started
- from .signals import request_tearing_down
- from .templating import DispatchingJinjaLoader
- from .templating import Environment
- from .typing import BeforeFirstRequestCallable
- from .typing import ResponseReturnValue
- from .typing import TeardownCallable
- from .typing import TemplateFilterCallable
- from .typing import TemplateGlobalCallable
- from .typing import TemplateTestCallable
- from .wrappers import Request
- from .wrappers import Response
- if t.TYPE_CHECKING:
- import typing_extensions as te
- from .blueprints import Blueprint
- from .testing import FlaskClient
- from .testing import FlaskCliRunner
- from .typing import ErrorHandlerCallable
- if sys.version_info >= (3, 8):
- iscoroutinefunction = inspect.iscoroutinefunction
- else:
- def iscoroutinefunction(func: t.Any) -> bool:
- while inspect.ismethod(func):
- func = func.__func__
- while isinstance(func, functools.partial):
- func = func.func
- return inspect.iscoroutinefunction(func)
- def _make_timedelta(value: t.Optional[timedelta]) -> t.Optional[timedelta]:
- if value is None or isinstance(value, timedelta):
- return value
- return timedelta(seconds=value)
- class Flask(Scaffold):
- """The flask object implements a WSGI application and acts as the central
- object. It is passed the name of the module or package of the
- application. Once it is created it will act as a central registry for
- the view functions, the URL rules, template configuration and much more.
- The name of the package is used to resolve resources from inside the
- package or the folder the module is contained in depending on if the
- package parameter resolves to an actual python package (a folder with
- an :file:`__init__.py` file inside) or a standard module (just a ``.py`` file).
- For more information about resource loading, see :func:`open_resource`.
- Usually you create a :class:`Flask` instance in your main module or
- in the :file:`__init__.py` file of your package like this::
- from flask import Flask
- app = Flask(__name__)
- .. admonition:: About the First Parameter
- The idea of the first parameter is to give Flask an idea of what
- belongs to your application. This name is used to find resources
- on the filesystem, can be used by extensions to improve debugging
- information and a lot more.
- So it's important what you provide there. If you are using a single
- module, `__name__` is always the correct value. If you however are
- using a package, it's usually recommended to hardcode the name of
- your package there.
- For example if your application is defined in :file:`yourapplication/app.py`
- you should create it with one of the two versions below::
- app = Flask('yourapplication')
- app = Flask(__name__.split('.')[0])
- Why is that? The application will work even with `__name__`, thanks
- to how resources are looked up. However it will make debugging more
- painful. Certain extensions can make assumptions based on the
- import name of your application. For example the Flask-SQLAlchemy
- extension will look for the code in your application that triggered
- an SQL query in debug mode. If the import name is not properly set
- up, that debugging information is lost. (For example it would only
- pick up SQL queries in `yourapplication.app` and not
- `yourapplication.views.frontend`)
- .. versionadded:: 0.7
- The `static_url_path`, `static_folder`, and `template_folder`
- parameters were added.
- .. versionadded:: 0.8
- The `instance_path` and `instance_relative_config` parameters were
- added.
- .. versionadded:: 0.11
- The `root_path` parameter was added.
- .. versionadded:: 1.0
- The ``host_matching`` and ``static_host`` parameters were added.
- .. versionadded:: 1.0
- The ``subdomain_matching`` parameter was added. Subdomain
- matching needs to be enabled manually now. Setting
- :data:`SERVER_NAME` does not implicitly enable it.
- :param import_name: the name of the application package
- :param static_url_path: can be used to specify a different path for the
- static files on the web. Defaults to the name
- of the `static_folder` folder.
- :param static_folder: The folder with static files that is served at
- ``static_url_path``. Relative to the application ``root_path``
- or an absolute path. Defaults to ``'static'``.
- :param static_host: the host to use when adding the static route.
- Defaults to None. Required when using ``host_matching=True``
- with a ``static_folder`` configured.
- :param host_matching: set ``url_map.host_matching`` attribute.
- Defaults to False.
- :param subdomain_matching: consider the subdomain relative to
- :data:`SERVER_NAME` when matching routes. Defaults to False.
- :param template_folder: the folder that contains the templates that should
- be used by the application. Defaults to
- ``'templates'`` folder in the root path of the
- application.
- :param instance_path: An alternative instance path for the application.
- By default the folder ``'instance'`` next to the
- package or module is assumed to be the instance
- path.
- :param instance_relative_config: if set to ``True`` relative filenames
- for loading the config are assumed to
- be relative to the instance path instead
- of the application root.
- :param root_path: The path to the root of the application files.
- This should only be set manually when it can't be detected
- automatically, such as for namespace packages.
- """
- #: The class that is used for request objects. See :class:`~flask.Request`
- #: for more information.
- request_class = Request
- #: The class that is used for response objects. See
- #: :class:`~flask.Response` for more information.
- response_class = Response
- #: The class that is used for the Jinja environment.
- #:
- #: .. versionadded:: 0.11
- jinja_environment = Environment
- #: The class that is used for the :data:`~flask.g` instance.
- #:
- #: Example use cases for a custom class:
- #:
- #: 1. Store arbitrary attributes on flask.g.
- #: 2. Add a property for lazy per-request database connectors.
- #: 3. Return None instead of AttributeError on unexpected attributes.
- #: 4. Raise exception if an unexpected attr is set, a "controlled" flask.g.
- #:
- #: In Flask 0.9 this property was called `request_globals_class` but it
- #: was changed in 0.10 to :attr:`app_ctx_globals_class` because the
- #: flask.g object is now application context scoped.
- #:
- #: .. versionadded:: 0.10
- app_ctx_globals_class = _AppCtxGlobals
- #: The class that is used for the ``config`` attribute of this app.
- #: Defaults to :class:`~flask.Config`.
- #:
- #: Example use cases for a custom class:
- #:
- #: 1. Default values for certain config options.
- #: 2. Access to config values through attributes in addition to keys.
- #:
- #: .. versionadded:: 0.11
- config_class = Config
- #: The testing flag. Set this to ``True`` to enable the test mode of
- #: Flask extensions (and in the future probably also Flask itself).
- #: For example this might activate test helpers that have an
- #: additional runtime cost which should not be enabled by default.
- #:
- #: If this is enabled and PROPAGATE_EXCEPTIONS is not changed from the
- #: default it's implicitly enabled.
- #:
- #: This attribute can also be configured from the config with the
- #: ``TESTING`` configuration key. Defaults to ``False``.
- testing = ConfigAttribute("TESTING")
- #: If a secret key is set, cryptographic components can use this to
- #: sign cookies and other things. Set this to a complex random value
- #: when you want to use the secure cookie for instance.
- #:
- #: This attribute can also be configured from the config with the
- #: :data:`SECRET_KEY` configuration key. Defaults to ``None``.
- secret_key = ConfigAttribute("SECRET_KEY")
- #: The secure cookie uses this for the name of the session cookie.
- #:
- #: This attribute can also be configured from the config with the
- #: ``SESSION_COOKIE_NAME`` configuration key. Defaults to ``'session'``
- session_cookie_name = ConfigAttribute("SESSION_COOKIE_NAME")
- #: A :class:`~datetime.timedelta` which is used to set the expiration
- #: date of a permanent session. The default is 31 days which makes a
- #: permanent session survive for roughly one month.
- #:
- #: This attribute can also be configured from the config with the
- #: ``PERMANENT_SESSION_LIFETIME`` configuration key. Defaults to
- #: ``timedelta(days=31)``
- permanent_session_lifetime = ConfigAttribute(
- "PERMANENT_SESSION_LIFETIME", get_converter=_make_timedelta
- )
- #: A :class:`~datetime.timedelta` or number of seconds which is used
- #: as the default ``max_age`` for :func:`send_file`. The default is
- #: ``None``, which tells the browser to use conditional requests
- #: instead of a timed cache.
- #:
- #: Configured with the :data:`SEND_FILE_MAX_AGE_DEFAULT`
- #: configuration key.
- #:
- #: .. versionchanged:: 2.0
- #: Defaults to ``None`` instead of 12 hours.
- send_file_max_age_default = ConfigAttribute(
- "SEND_FILE_MAX_AGE_DEFAULT", get_converter=_make_timedelta
- )
- #: Enable this if you want to use the X-Sendfile feature. Keep in
- #: mind that the server has to support this. This only affects files
- #: sent with the :func:`send_file` method.
- #:
- #: .. versionadded:: 0.2
- #:
- #: This attribute can also be configured from the config with the
- #: ``USE_X_SENDFILE`` configuration key. Defaults to ``False``.
- use_x_sendfile = ConfigAttribute("USE_X_SENDFILE")
- #: The JSON encoder class to use. Defaults to :class:`~flask.json.JSONEncoder`.
- #:
- #: .. versionadded:: 0.10
- json_encoder = json.JSONEncoder
- #: The JSON decoder class to use. Defaults to :class:`~flask.json.JSONDecoder`.
- #:
- #: .. versionadded:: 0.10
- json_decoder = json.JSONDecoder
- #: Options that are passed to the Jinja environment in
- #: :meth:`create_jinja_environment`. Changing these options after
- #: the environment is created (accessing :attr:`jinja_env`) will
- #: have no effect.
- #:
- #: .. versionchanged:: 1.1.0
- #: This is a ``dict`` instead of an ``ImmutableDict`` to allow
- #: easier configuration.
- #:
- jinja_options: dict = {}
- #: Default configuration parameters.
- default_config = ImmutableDict(
- {
- "ENV": None,
- "DEBUG": None,
- "TESTING": False,
- "PROPAGATE_EXCEPTIONS": None,
- "PRESERVE_CONTEXT_ON_EXCEPTION": None,
- "SECRET_KEY": None,
- "PERMANENT_SESSION_LIFETIME": timedelta(days=31),
- "USE_X_SENDFILE": False,
- "SERVER_NAME": None,
- "APPLICATION_ROOT": "/",
- "SESSION_COOKIE_NAME": "session",
- "SESSION_COOKIE_DOMAIN": None,
- "SESSION_COOKIE_PATH": None,
- "SESSION_COOKIE_HTTPONLY": True,
- "SESSION_COOKIE_SECURE": False,
- "SESSION_COOKIE_SAMESITE": None,
- "SESSION_REFRESH_EACH_REQUEST": True,
- "MAX_CONTENT_LENGTH": None,
- "SEND_FILE_MAX_AGE_DEFAULT": None,
- "TRAP_BAD_REQUEST_ERRORS": None,
- "TRAP_HTTP_EXCEPTIONS": False,
- "EXPLAIN_TEMPLATE_LOADING": False,
- "PREFERRED_URL_SCHEME": "http",
- "JSON_AS_ASCII": True,
- "JSON_SORT_KEYS": True,
- "JSONIFY_PRETTYPRINT_REGULAR": False,
- "JSONIFY_MIMETYPE": "application/json",
- "TEMPLATES_AUTO_RELOAD": None,
- "MAX_COOKIE_SIZE": 4093,
- }
- )
- #: The rule object to use for URL rules created. This is used by
- #: :meth:`add_url_rule`. Defaults to :class:`werkzeug.routing.Rule`.
- #:
- #: .. versionadded:: 0.7
- url_rule_class = Rule
- #: The map object to use for storing the URL rules and routing
- #: configuration parameters. Defaults to :class:`werkzeug.routing.Map`.
- #:
- #: .. versionadded:: 1.1.0
- url_map_class = Map
- #: The :meth:`test_client` method creates an instance of this test
- #: client class. Defaults to :class:`~flask.testing.FlaskClient`.
- #:
- #: .. versionadded:: 0.7
- test_client_class: t.Optional[t.Type["FlaskClient"]] = None
- #: The :class:`~click.testing.CliRunner` subclass, by default
- #: :class:`~flask.testing.FlaskCliRunner` that is used by
- #: :meth:`test_cli_runner`. Its ``__init__`` method should take a
- #: Flask app object as the first argument.
- #:
- #: .. versionadded:: 1.0
- test_cli_runner_class: t.Optional[t.Type["FlaskCliRunner"]] = None
- #: the session interface to use. By default an instance of
- #: :class:`~flask.sessions.SecureCookieSessionInterface` is used here.
- #:
- #: .. versionadded:: 0.8
- session_interface = SecureCookieSessionInterface()
- def __init__(
- self,
- import_name: str,
- static_url_path: t.Optional[str] = None,
- static_folder: t.Optional[t.Union[str, os.PathLike]] = "static",
- static_host: t.Optional[str] = None,
- host_matching: bool = False,
- subdomain_matching: bool = False,
- template_folder: t.Optional[str] = "templates",
- instance_path: t.Optional[str] = None,
- instance_relative_config: bool = False,
- root_path: t.Optional[str] = None,
- ):
- super().__init__(
- import_name=import_name,
- static_folder=static_folder,
- static_url_path=static_url_path,
- template_folder=template_folder,
- root_path=root_path,
- )
- if instance_path is None:
- instance_path = self.auto_find_instance_path()
- elif not os.path.isabs(instance_path):
- raise ValueError(
- "If an instance path is provided it must be absolute."
- " A relative path was given instead."
- )
- #: Holds the path to the instance folder.
- #:
- #: .. versionadded:: 0.8
- self.instance_path = instance_path
- #: The configuration dictionary as :class:`Config`. This behaves
- #: exactly like a regular dictionary but supports additional methods
- #: to load a config from files.
- self.config = self.make_config(instance_relative_config)
- #: A list of functions that are called when :meth:`url_for` raises a
- #: :exc:`~werkzeug.routing.BuildError`. Each function registered here
- #: is called with `error`, `endpoint` and `values`. If a function
- #: returns ``None`` or raises a :exc:`BuildError` the next function is
- #: tried.
- #:
- #: .. versionadded:: 0.9
- self.url_build_error_handlers: t.List[
- t.Callable[[Exception, str, dict], str]
- ] = []
- #: A list of functions that will be called at the beginning of the
- #: first request to this instance. To register a function, use the
- #: :meth:`before_first_request` decorator.
- #:
- #: .. versionadded:: 0.8
- self.before_first_request_funcs: t.List[BeforeFirstRequestCallable] = []
- #: A list of functions that are called when the application context
- #: is destroyed. Since the application context is also torn down
- #: if the request ends this is the place to store code that disconnects
- #: from databases.
- #:
- #: .. versionadded:: 0.9
- self.teardown_appcontext_funcs: t.List[TeardownCallable] = []
- #: A list of shell context processor functions that should be run
- #: when a shell context is created.
- #:
- #: .. versionadded:: 0.11
- self.shell_context_processors: t.List[t.Callable[[], t.Dict[str, t.Any]]] = []
- #: Maps registered blueprint names to blueprint objects. The
- #: dict retains the order the blueprints were registered in.
- #: Blueprints can be registered multiple times, this dict does
- #: not track how often they were attached.
- #:
- #: .. versionadded:: 0.7
- self.blueprints: t.Dict[str, "Blueprint"] = {}
- #: a place where extensions can store application specific state. For
- #: example this is where an extension could store database engines and
- #: similar things.
- #:
- #: The key must match the name of the extension module. For example in
- #: case of a "Flask-Foo" extension in `flask_foo`, the key would be
- #: ``'foo'``.
- #:
- #: .. versionadded:: 0.7
- self.extensions: dict = {}
- #: The :class:`~werkzeug.routing.Map` for this instance. You can use
- #: this to change the routing converters after the class was created
- #: but before any routes are connected. Example::
- #:
- #: from werkzeug.routing import BaseConverter
- #:
- #: class ListConverter(BaseConverter):
- #: def to_python(self, value):
- #: return value.split(',')
- #: def to_url(self, values):
- #: return ','.join(super(ListConverter, self).to_url(value)
- #: for value in values)
- #:
- #: app = Flask(__name__)
- #: app.url_map.converters['list'] = ListConverter
- self.url_map = self.url_map_class()
- self.url_map.host_matching = host_matching
- self.subdomain_matching = subdomain_matching
- # tracks internally if the application already handled at least one
- # request.
- self._got_first_request = False
- self._before_request_lock = Lock()
- # Add a static route using the provided static_url_path, static_host,
- # and static_folder if there is a configured static_folder.
- # Note we do this without checking if static_folder exists.
- # For one, it might be created while the server is running (e.g. during
- # development). Also, Google App Engine stores static files somewhere
- if self.has_static_folder:
- assert (
- bool(static_host) == host_matching
- ), "Invalid static_host/host_matching combination"
- # Use a weakref to avoid creating a reference cycle between the app
- # and the view function (see #3761).
- self_ref = weakref.ref(self)
- self.add_url_rule(
- f"{self.static_url_path}/<path:filename>",
- endpoint="static",
- host=static_host,
- view_func=lambda **kw: self_ref().send_static_file(**kw), # type: ignore # noqa: B950
- )
- # Set the name of the Click group in case someone wants to add
- # the app's commands to another CLI tool.
- self.cli.name = self.name
- def _is_setup_finished(self) -> bool:
- return self.debug and self._got_first_request
- @locked_cached_property
- def name(self) -> str: # type: ignore
- """The name of the application. This is usually the import name
- with the difference that it's guessed from the run file if the
- import name is main. This name is used as a display name when
- Flask needs the name of the application. It can be set and overridden
- to change the value.
- .. versionadded:: 0.8
- """
- if self.import_name == "__main__":
- fn = getattr(sys.modules["__main__"], "__file__", None)
- if fn is None:
- return "__main__"
- return os.path.splitext(os.path.basename(fn))[0]
- return self.import_name
- @property
- def propagate_exceptions(self) -> bool:
- """Returns the value of the ``PROPAGATE_EXCEPTIONS`` configuration
- value in case it's set, otherwise a sensible default is returned.
- .. versionadded:: 0.7
- """
- rv = self.config["PROPAGATE_EXCEPTIONS"]
- if rv is not None:
- return rv
- return self.testing or self.debug
- @property
- def preserve_context_on_exception(self) -> bool:
- """Returns the value of the ``PRESERVE_CONTEXT_ON_EXCEPTION``
- configuration value in case it's set, otherwise a sensible default
- is returned.
- .. versionadded:: 0.7
- """
- rv = self.config["PRESERVE_CONTEXT_ON_EXCEPTION"]
- if rv is not None:
- return rv
- return self.debug
- @locked_cached_property
- def logger(self) -> logging.Logger:
- """A standard Python :class:`~logging.Logger` for the app, with
- the same name as :attr:`name`.
- In debug mode, the logger's :attr:`~logging.Logger.level` will
- be set to :data:`~logging.DEBUG`.
- If there are no handlers configured, a default handler will be
- added. See :doc:`/logging` for more information.
- .. versionchanged:: 1.1.0
- The logger takes the same name as :attr:`name` rather than
- hard-coding ``"flask.app"``.
- .. versionchanged:: 1.0.0
- Behavior was simplified. The logger is always named
- ``"flask.app"``. The level is only set during configuration,
- it doesn't check ``app.debug`` each time. Only one format is
- used, not different ones depending on ``app.debug``. No
- handlers are removed, and a handler is only added if no
- handlers are already configured.
- .. versionadded:: 0.3
- """
- return create_logger(self)
- @locked_cached_property
- def jinja_env(self) -> Environment:
- """The Jinja environment used to load templates.
- The environment is created the first time this property is
- accessed. Changing :attr:`jinja_options` after that will have no
- effect.
- """
- return self.create_jinja_environment()
- @property
- def got_first_request(self) -> bool:
- """This attribute is set to ``True`` if the application started
- handling the first request.
- .. versionadded:: 0.8
- """
- return self._got_first_request
- def make_config(self, instance_relative: bool = False) -> Config:
- """Used to create the config attribute by the Flask constructor.
- The `instance_relative` parameter is passed in from the constructor
- of Flask (there named `instance_relative_config`) and indicates if
- the config should be relative to the instance path or the root path
- of the application.
- .. versionadded:: 0.8
- """
- root_path = self.root_path
- if instance_relative:
- root_path = self.instance_path
- defaults = dict(self.default_config)
- defaults["ENV"] = get_env()
- defaults["DEBUG"] = get_debug_flag()
- return self.config_class(root_path, defaults)
- def auto_find_instance_path(self) -> str:
- """Tries to locate the instance path if it was not provided to the
- constructor of the application class. It will basically calculate
- the path to a folder named ``instance`` next to your main file or
- the package.
- .. versionadded:: 0.8
- """
- prefix, package_path = find_package(self.import_name)
- if prefix is None:
- return os.path.join(package_path, "instance")
- return os.path.join(prefix, "var", f"{self.name}-instance")
- def open_instance_resource(self, resource: str, mode: str = "rb") -> t.IO[t.AnyStr]:
- """Opens a resource from the application's instance folder
- (:attr:`instance_path`). Otherwise works like
- :meth:`open_resource`. Instance resources can also be opened for
- writing.
- :param resource: the name of the resource. To access resources within
- subfolders use forward slashes as separator.
- :param mode: resource file opening mode, default is 'rb'.
- """
- return open(os.path.join(self.instance_path, resource), mode)
- @property
- def templates_auto_reload(self) -> bool:
- """Reload templates when they are changed. Used by
- :meth:`create_jinja_environment`.
- This attribute can be configured with :data:`TEMPLATES_AUTO_RELOAD`. If
- not set, it will be enabled in debug mode.
- .. versionadded:: 1.0
- This property was added but the underlying config and behavior
- already existed.
- """
- rv = self.config["TEMPLATES_AUTO_RELOAD"]
- return rv if rv is not None else self.debug
- @templates_auto_reload.setter
- def templates_auto_reload(self, value: bool) -> None:
- self.config["TEMPLATES_AUTO_RELOAD"] = value
- def create_jinja_environment(self) -> Environment:
- """Create the Jinja environment based on :attr:`jinja_options`
- and the various Jinja-related methods of the app. Changing
- :attr:`jinja_options` after this will have no effect. Also adds
- Flask-related globals and filters to the environment.
- .. versionchanged:: 0.11
- ``Environment.auto_reload`` set in accordance with
- ``TEMPLATES_AUTO_RELOAD`` configuration option.
- .. versionadded:: 0.5
- """
- options = dict(self.jinja_options)
- if "autoescape" not in options:
- options["autoescape"] = self.select_jinja_autoescape
- if "auto_reload" not in options:
- options["auto_reload"] = self.templates_auto_reload
- rv = self.jinja_environment(self, **options)
- rv.globals.update(
- url_for=url_for,
- get_flashed_messages=get_flashed_messages,
- config=self.config,
- # request, session and g are normally added with the
- # context processor for efficiency reasons but for imported
- # templates we also want the proxies in there.
- request=request,
- session=session,
- g=g,
- )
- rv.policies["json.dumps_function"] = json.dumps
- return rv
- def create_global_jinja_loader(self) -> DispatchingJinjaLoader:
- """Creates the loader for the Jinja2 environment. Can be used to
- override just the loader and keeping the rest unchanged. It's
- discouraged to override this function. Instead one should override
- the :meth:`jinja_loader` function instead.
- The global loader dispatches between the loaders of the application
- and the individual blueprints.
- .. versionadded:: 0.7
- """
- return DispatchingJinjaLoader(self)
- def select_jinja_autoescape(self, filename: str) -> bool:
- """Returns ``True`` if autoescaping should be active for the given
- template name. If no template name is given, returns `True`.
- .. versionadded:: 0.5
- """
- if filename is None:
- return True
- return filename.endswith((".html", ".htm", ".xml", ".xhtml"))
- def update_template_context(self, context: dict) -> None:
- """Update the template context with some commonly used variables.
- This injects request, session, config and g into the template
- context as well as everything template context processors want
- to inject. Note that the as of Flask 0.6, the original values
- in the context will not be overridden if a context processor
- decides to return a value with the same key.
- :param context: the context as a dictionary that is updated in place
- to add extra variables.
- """
- names: t.Iterable[t.Optional[str]] = (None,)
- # A template may be rendered outside a request context.
- if request:
- names = chain(names, reversed(request.blueprints))
- # The values passed to render_template take precedence. Keep a
- # copy to re-apply after all context functions.
- orig_ctx = context.copy()
- for name in names:
- if name in self.template_context_processors:
- for func in self.template_context_processors[name]:
- context.update(func())
- context.update(orig_ctx)
- def make_shell_context(self) -> dict:
- """Returns the shell context for an interactive shell for this
- application. This runs all the registered shell context
- processors.
- .. versionadded:: 0.11
- """
- rv = {"app": self, "g": g}
- for processor in self.shell_context_processors:
- rv.update(processor())
- return rv
- #: What environment the app is running in. Flask and extensions may
- #: enable behaviors based on the environment, such as enabling debug
- #: mode. This maps to the :data:`ENV` config key. This is set by the
- #: :envvar:`FLASK_ENV` environment variable and may not behave as
- #: expected if set in code.
- #:
- #: **Do not enable development when deploying in production.**
- #:
- #: Default: ``'production'``
- env = ConfigAttribute("ENV")
- @property
- def debug(self) -> bool:
- """Whether debug mode is enabled. When using ``flask run`` to start
- the development server, an interactive debugger will be shown for
- unhandled exceptions, and the server will be reloaded when code
- changes. This maps to the :data:`DEBUG` config key. This is
- enabled when :attr:`env` is ``'development'`` and is overridden
- by the ``FLASK_DEBUG`` environment variable. It may not behave as
- expected if set in code.
- **Do not enable debug mode when deploying in production.**
- Default: ``True`` if :attr:`env` is ``'development'``, or
- ``False`` otherwise.
- """
- return self.config["DEBUG"]
- @debug.setter
- def debug(self, value: bool) -> None:
- self.config["DEBUG"] = value
- self.jinja_env.auto_reload = self.templates_auto_reload
- def run(
- self,
- host: t.Optional[str] = None,
- port: t.Optional[int] = None,
- debug: t.Optional[bool] = None,
- load_dotenv: bool = True,
- **options: t.Any,
- ) -> None:
- """Runs the application on a local development server.
- Do not use ``run()`` in a production setting. It is not intended to
- meet security and performance requirements for a production server.
- Instead, see :doc:`/deploying/index` for WSGI server recommendations.
- If the :attr:`debug` flag is set the server will automatically reload
- for code changes and show a debugger in case an exception happened.
- If you want to run the application in debug mode, but disable the
- code execution on the interactive debugger, you can pass
- ``use_evalex=False`` as parameter. This will keep the debugger's
- traceback screen active, but disable code execution.
- It is not recommended to use this function for development with
- automatic reloading as this is badly supported. Instead you should
- be using the :command:`flask` command line script's ``run`` support.
- .. admonition:: Keep in Mind
- Flask will suppress any server error with a generic error page
- unless it is in debug mode. As such to enable just the
- interactive debugger without the code reloading, you have to
- invoke :meth:`run` with ``debug=True`` and ``use_reloader=False``.
- Setting ``use_debugger`` to ``True`` without being in debug mode
- won't catch any exceptions because there won't be any to
- catch.
- :param host: the hostname to listen on. Set this to ``'0.0.0.0'`` to
- have the server available externally as well. Defaults to
- ``'127.0.0.1'`` or the host in the ``SERVER_NAME`` config variable
- if present.
- :param port: the port of the webserver. Defaults to ``5000`` or the
- port defined in the ``SERVER_NAME`` config variable if present.
- :param debug: if given, enable or disable debug mode. See
- :attr:`debug`.
- :param load_dotenv: Load the nearest :file:`.env` and :file:`.flaskenv`
- files to set environment variables. Will also change the working
- directory to the directory containing the first file found.
- :param options: the options to be forwarded to the underlying Werkzeug
- server. See :func:`werkzeug.serving.run_simple` for more
- information.
- .. versionchanged:: 1.0
- If installed, python-dotenv will be used to load environment
- variables from :file:`.env` and :file:`.flaskenv` files.
- If set, the :envvar:`FLASK_ENV` and :envvar:`FLASK_DEBUG`
- environment variables will override :attr:`env` and
- :attr:`debug`.
- Threaded mode is enabled by default.
- .. versionchanged:: 0.10
- The default port is now picked from the ``SERVER_NAME``
- variable.
- """
- # Change this into a no-op if the server is invoked from the
- # command line. Have a look at cli.py for more information.
- if os.environ.get("FLASK_RUN_FROM_CLI") == "true":
- from .debughelpers import explain_ignored_app_run
- explain_ignored_app_run()
- return
- if get_load_dotenv(load_dotenv):
- cli.load_dotenv()
- # if set, let env vars override previous values
- if "FLASK_ENV" in os.environ:
- self.env = get_env()
- self.debug = get_debug_flag()
- elif "FLASK_DEBUG" in os.environ:
- self.debug = get_debug_flag()
- # debug passed to method overrides all other sources
- if debug is not None:
- self.debug = bool(debug)
- server_name = self.config.get("SERVER_NAME")
- sn_host = sn_port = None
- if server_name:
- sn_host, _, sn_port = server_name.partition(":")
- if not host:
- if sn_host:
- host = sn_host
- else:
- host = "127.0.0.1"
- if port or port == 0:
- port = int(port)
- elif sn_port:
- port = int(sn_port)
- else:
- port = 5000
- options.setdefault("use_reloader", self.debug)
- options.setdefault("use_debugger", self.debug)
- options.setdefault("threaded", True)
- cli.show_server_banner(self.env, self.debug, self.name, False)
- from werkzeug.serving import run_simple
- try:
- run_simple(t.cast(str, host), port, self, **options)
- finally:
- # reset the first request information if the development server
- # reset normally. This makes it possible to restart the server
- # without reloader and that stuff from an interactive shell.
- self._got_first_request = False
- def test_client(self, use_cookies: bool = True, **kwargs: t.Any) -> "FlaskClient":
- """Creates a test client for this application. For information
- about unit testing head over to :doc:`/testing`.
- Note that if you are testing for assertions or exceptions in your
- application code, you must set ``app.testing = True`` in order for the
- exceptions to propagate to the test client. Otherwise, the exception
- will be handled by the application (not visible to the test client) and
- the only indication of an AssertionError or other exception will be a
- 500 status code response to the test client. See the :attr:`testing`
- attribute. For example::
- app.testing = True
- client = app.test_client()
- The test client can be used in a ``with`` block to defer the closing down
- of the context until the end of the ``with`` block. This is useful if
- you want to access the context locals for testing::
- with app.test_client() as c:
- rv = c.get('/?vodka=42')
- assert request.args['vodka'] == '42'
- Additionally, you may pass optional keyword arguments that will then
- be passed to the application's :attr:`test_client_class` constructor.
- For example::
- from flask.testing import FlaskClient
- class CustomClient(FlaskClient):
- def __init__(self, *args, **kwargs):
- self._authentication = kwargs.pop("authentication")
- super(CustomClient,self).__init__( *args, **kwargs)
- app.test_client_class = CustomClient
- client = app.test_client(authentication='Basic ....')
- See :class:`~flask.testing.FlaskClient` for more information.
- .. versionchanged:: 0.4
- added support for ``with`` block usage for the client.
- .. versionadded:: 0.7
- The `use_cookies` parameter was added as well as the ability
- to override the client to be used by setting the
- :attr:`test_client_class` attribute.
- .. versionchanged:: 0.11
- Added `**kwargs` to support passing additional keyword arguments to
- the constructor of :attr:`test_client_class`.
- """
- cls = self.test_client_class
- if cls is None:
- from .testing import FlaskClient as cls # type: ignore
- return cls( # type: ignore
- self, self.response_class, use_cookies=use_cookies, **kwargs
- )
- def test_cli_runner(self, **kwargs: t.Any) -> "FlaskCliRunner":
- """Create a CLI runner for testing CLI commands.
- See :ref:`testing-cli`.
- Returns an instance of :attr:`test_cli_runner_class`, by default
- :class:`~flask.testing.FlaskCliRunner`. The Flask app object is
- passed as the first argument.
- .. versionadded:: 1.0
- """
- cls = self.test_cli_runner_class
- if cls is None:
- from .testing import FlaskCliRunner as cls # type: ignore
- return cls(self, **kwargs) # type: ignore
- @setupmethod
- def register_blueprint(self, blueprint: "Blueprint", **options: t.Any) -> None:
- """Register a :class:`~flask.Blueprint` on the application. Keyword
- arguments passed to this method will override the defaults set on the
- blueprint.
- Calls the blueprint's :meth:`~flask.Blueprint.register` method after
- recording the blueprint in the application's :attr:`blueprints`.
- :param blueprint: The blueprint to register.
- :param url_prefix: Blueprint routes will be prefixed with this.
- :param subdomain: Blueprint routes will match on this subdomain.
- :param url_defaults: Blueprint routes will use these default values for
- view arguments.
- :param options: Additional keyword arguments are passed to
- :class:`~flask.blueprints.BlueprintSetupState`. They can be
- accessed in :meth:`~flask.Blueprint.record` callbacks.
- .. versionchanged:: 2.0.1
- The ``name`` option can be used to change the (pre-dotted)
- name the blueprint is registered with. This allows the same
- blueprint to be registered multiple times with unique names
- for ``url_for``.
- .. versionadded:: 0.7
- """
- blueprint.register(self, options)
- def iter_blueprints(self) -> t.ValuesView["Blueprint"]:
- """Iterates over all blueprints by the order they were registered.
- .. versionadded:: 0.11
- """
- return self.blueprints.values()
- @setupmethod
- def add_url_rule(
- self,
- rule: str,
- endpoint: t.Optional[str] = None,
- view_func: t.Optional[t.Callable] = None,
- provide_automatic_options: t.Optional[bool] = None,
- **options: t.Any,
- ) -> None:
- if endpoint is None:
- endpoint = _endpoint_from_view_func(view_func) # type: ignore
- options["endpoint"] = endpoint
- methods = options.pop("methods", None)
- # if the methods are not given and the view_func object knows its
- # methods we can use that instead. If neither exists, we go with
- # a tuple of only ``GET`` as default.
- if methods is None:
- methods = getattr(view_func, "methods", None) or ("GET",)
- if isinstance(methods, str):
- raise TypeError(
- "Allowed methods must be a list of strings, for"
- ' example: @app.route(..., methods=["POST"])'
- )
- methods = {item.upper() for item in methods}
- # Methods that should always be added
- required_methods = set(getattr(view_func, "required_methods", ()))
- # starting with Flask 0.8 the view_func object can disable and
- # force-enable the automatic options handling.
- if provide_automatic_options is None:
- provide_automatic_options = getattr(
- view_func, "provide_automatic_options", None
- )
- if provide_automatic_options is None:
- if "OPTIONS" not in methods:
- provide_automatic_options = True
- required_methods.add("OPTIONS")
- else:
- provide_automatic_options = False
- # Add the required methods now.
- methods |= required_methods
- rule = self.url_rule_class(rule, methods=methods, **options)
- rule.provide_automatic_options = provide_automatic_options # type: ignore
- self.url_map.add(rule)
- if view_func is not None:
- old_func = self.view_functions.get(endpoint)
- if old_func is not None and old_func != view_func:
- raise AssertionError(
- "View function mapping is overwriting an existing"
- f" endpoint function: {endpoint}"
- )
- self.view_functions[endpoint] = view_func
- @setupmethod
- def template_filter(
- self, name: t.Optional[str] = None
- ) -> t.Callable[[TemplateFilterCallable], TemplateFilterCallable]:
- """A decorator that is used to register custom template filter.
- You can specify a name for the filter, otherwise the function
- name will be used. Example::
- @app.template_filter()
- def reverse(s):
- return s[::-1]
- :param name: the optional name of the filter, otherwise the
- function name will be used.
- """
- def decorator(f: TemplateFilterCallable) -> TemplateFilterCallable:
- self.add_template_filter(f, name=name)
- return f
- return decorator
- @setupmethod
- def add_template_filter(
- self, f: TemplateFilterCallable, name: t.Optional[str] = None
- ) -> None:
- """Register a custom template filter. Works exactly like the
- :meth:`template_filter` decorator.
- :param name: the optional name of the filter, otherwise the
- function name will be used.
- """
- self.jinja_env.filters[name or f.__name__] = f
- @setupmethod
- def template_test(
- self, name: t.Optional[str] = None
- ) -> t.Callable[[TemplateTestCallable], TemplateTestCallable]:
- """A decorator that is used to register custom template test.
- You can specify a name for the test, otherwise the function
- name will be used. Example::
- @app.template_test()
- def is_prime(n):
- if n == 2:
- return True
- for i in range(2, int(math.ceil(math.sqrt(n))) + 1):
- if n % i == 0:
- return False
- return True
- .. versionadded:: 0.10
- :param name: the optional name of the test, otherwise the
- function name will be used.
- """
- def decorator(f: TemplateTestCallable) -> TemplateTestCallable:
- self.add_template_test(f, name=name)
- return f
- return decorator
- @setupmethod
- def add_template_test(
- self, f: TemplateTestCallable, name: t.Optional[str] = None
- ) -> None:
- """Register a custom template test. Works exactly like the
- :meth:`template_test` decorator.
- .. versionadded:: 0.10
- :param name: the optional name of the test, otherwise the
- function name will be used.
- """
- self.jinja_env.tests[name or f.__name__] = f
- @setupmethod
- def template_global(
- self, name: t.Optional[str] = None
- ) -> t.Callable[[TemplateGlobalCallable], TemplateGlobalCallable]:
- """A decorator that is used to register a custom template global function.
- You can specify a name for the global function, otherwise the function
- name will be used. Example::
- @app.template_global()
- def double(n):
- return 2 * n
- .. versionadded:: 0.10
- :param name: the optional name of the global function, otherwise the
- function name will be used.
- """
- def decorator(f: TemplateGlobalCallable) -> TemplateGlobalCallable:
- self.add_template_global(f, name=name)
- return f
- return decorator
- @setupmethod
- def add_template_global(
- self, f: TemplateGlobalCallable, name: t.Optional[str] = None
- ) -> None:
- """Register a custom template global function. Works exactly like the
- :meth:`template_global` decorator.
- .. versionadded:: 0.10
- :param name: the optional name of the global function, otherwise the
- function name will be used.
- """
- self.jinja_env.globals[name or f.__name__] = f
- @setupmethod
- def before_first_request(
- self, f: BeforeFirstRequestCallable
- ) -> BeforeFirstRequestCallable:
- """Registers a function to be run before the first request to this
- instance of the application.
- The function will be called without any arguments and its return
- value is ignored.
- .. versionadded:: 0.8
- """
- self.before_first_request_funcs.append(f)
- return f
- @setupmethod
- def teardown_appcontext(self, f: TeardownCallable) -> TeardownCallable:
- """Registers a function to be called when the application context
- ends. These functions are typically also called when the request
- context is popped.
- Example::
- ctx = app.app_context()
- ctx.push()
- ...
- ctx.pop()
- When ``ctx.pop()`` is executed in the above example, the teardown
- functions are called just before the app context moves from the
- stack of active contexts. This becomes relevant if you are using
- such constructs in tests.
- Since a request context typically also manages an application
- context it would also be called when you pop a request context.
- When a teardown function was called because of an unhandled exception
- it will be passed an error object. If an :meth:`errorhandler` is
- registered, it will handle the exception and the teardown will not
- receive it.
- The return values of teardown functions are ignored.
- .. versionadded:: 0.9
- """
- self.teardown_appcontext_funcs.append(f)
- return f
- @setupmethod
- def shell_context_processor(self, f: t.Callable) -> t.Callable:
- """Registers a shell context processor function.
- .. versionadded:: 0.11
- """
- self.shell_context_processors.append(f)
- return f
- def _find_error_handler(
- self, e: Exception
- ) -> t.Optional["ErrorHandlerCallable[Exception]"]:
- """Return a registered error handler for an exception in this order:
- blueprint handler for a specific code, app handler for a specific code,
- blueprint handler for an exception class, app handler for an exception
- class, or ``None`` if a suitable handler is not found.
- """
- exc_class, code = self._get_exc_class_and_code(type(e))
- names = (*request.blueprints, None)
- for c in (code, None) if code is not None else (None,):
- for name in names:
- handler_map = self.error_handler_spec[name][c]
- if not handler_map:
- continue
- for cls in exc_class.__mro__:
- handler = handler_map.get(cls)
- if handler is not None:
- return handler
- return None
- def handle_http_exception(
- self, e: HTTPException
- ) -> t.Union[HTTPException, ResponseReturnValue]:
- """Handles an HTTP exception. By default this will invoke the
- registered error handlers and fall back to returning the
- exception as response.
- .. versionchanged:: 1.0.3
- ``RoutingException``, used internally for actions such as
- slash redirects during routing, is not passed to error
- handlers.
- .. versionchanged:: 1.0
- Exceptions are looked up by code *and* by MRO, so
- ``HTTPException`` subclasses can be handled with a catch-all
- handler for the base ``HTTPException``.
- .. versionadded:: 0.3
- """
- # Proxy exceptions don't have error codes. We want to always return
- # those unchanged as errors
- if e.code is None:
- return e
- # RoutingExceptions are used internally to trigger routing
- # actions, such as slash redirects raising RequestRedirect. They
- # are not raised or handled in user code.
- if isinstance(e, RoutingException):
- return e
- handler = self._find_error_handler(e)
- if handler is None:
- return e
- return self.ensure_sync(handler)(e)
- def trap_http_exception(self, e: Exception) -> bool:
- """Checks if an HTTP exception should be trapped or not. By default
- this will return ``False`` for all exceptions except for a bad request
- key error if ``TRAP_BAD_REQUEST_ERRORS`` is set to ``True``. It
- also returns ``True`` if ``TRAP_HTTP_EXCEPTIONS`` is set to ``True``.
- This is called for all HTTP exceptions raised by a view function.
- If it returns ``True`` for any exception the error handler for this
- exception is not called and it shows up as regular exception in the
- traceback. This is helpful for debugging implicitly raised HTTP
- exceptions.
- .. versionchanged:: 1.0
- Bad request errors are not trapped by default in debug mode.
- .. versionadded:: 0.8
- """
- if self.config["TRAP_HTTP_EXCEPTIONS"]:
- return True
- trap_bad_request = self.config["TRAP_BAD_REQUEST_ERRORS"]
- # if unset, trap key errors in debug mode
- if (
- trap_bad_request is None
- and self.debug
- and isinstance(e, BadRequestKeyError)
- ):
- return True
- if trap_bad_request:
- return isinstance(e, BadRequest)
- return False
- def handle_user_exception(
- self, e: Exception
- ) -> t.Union[HTTPException, ResponseReturnValue]:
- """This method is called whenever an exception occurs that
- should be handled. A special case is :class:`~werkzeug
- .exceptions.HTTPException` which is forwarded to the
- :meth:`handle_http_exception` method. This function will either
- return a response value or reraise the exception with the same
- traceback.
- .. versionchanged:: 1.0
- Key errors raised from request data like ``form`` show the
- bad key in debug mode rather than a generic bad request
- message.
- .. versionadded:: 0.7
- """
- if isinstance(e, BadRequestKeyError) and (
- self.debug or self.config["TRAP_BAD_REQUEST_ERRORS"]
- ):
- e.show_exception = True
- if isinstance(e, HTTPException) and not self.trap_http_exception(e):
- return self.handle_http_exception(e)
- handler = self._find_error_handler(e)
- if handler is None:
- raise
- return self.ensure_sync(handler)(e)
- def handle_exception(self, e: Exception) -> Response:
- """Handle an exception that did not have an error handler
- associated with it, or that was raised from an error handler.
- This always causes a 500 ``InternalServerError``.
- Always sends the :data:`got_request_exception` signal.
- If :attr:`propagate_exceptions` is ``True``, such as in debug
- mode, the error will be re-raised so that the debugger can
- display it. Otherwise, the original exception is logged, and
- an :exc:`~werkzeug.exceptions.InternalServerError` is returned.
- If an error handler is registered for ``InternalServerError`` or
- ``500``, it will be used. For consistency, the handler will
- always receive the ``InternalServerError``. The original
- unhandled exception is available as ``e.original_exception``.
- .. versionchanged:: 1.1.0
- Always passes the ``InternalServerError`` instance to the
- handler, setting ``original_exception`` to the unhandled
- error.
- .. versionchanged:: 1.1.0
- ``after_request`` functions and other finalization is done
- even for the default 500 response when there is no handler.
- .. versionadded:: 0.3
- """
- exc_info = sys.exc_info()
- got_request_exception.send(self, exception=e)
- if self.propagate_exceptions:
- # Re-raise if called with an active exception, otherwise
- # raise the passed in exception.
- if exc_info[1] is e:
- raise
- raise e
- self.log_exception(exc_info)
- server_error: t.Union[InternalServerError, ResponseReturnValue]
- server_error = InternalServerError(original_exception=e)
- handler = self._find_error_handler(server_error)
- if handler is not None:
- server_error = self.ensure_sync(handler)(server_error)
- return self.finalize_request(server_error, from_error_handler=True)
- def log_exception(
- self,
- exc_info: t.Union[
- t.Tuple[type, BaseException, TracebackType], t.Tuple[None, None, None]
- ],
- ) -> None:
- """Logs an exception. This is called by :meth:`handle_exception`
- if debugging is disabled and right before the handler is called.
- The default implementation logs the exception as error on the
- :attr:`logger`.
- .. versionadded:: 0.8
- """
- self.logger.error(
- f"Exception on {request.path} [{request.method}]", exc_info=exc_info
- )
- def raise_routing_exception(self, request: Request) -> "te.NoReturn":
- """Exceptions that are recording during routing are reraised with
- this method. During debug we are not reraising redirect requests
- for non ``GET``, ``HEAD``, or ``OPTIONS`` requests and we're raising
- a different error instead to help debug situations.
- :internal:
- """
- if (
- not self.debug
- or not isinstance(request.routing_exception, RequestRedirect)
- or request.method in ("GET", "HEAD", "OPTIONS")
- ):
- raise request.routing_exception # type: ignore
- from .debughelpers import FormDataRoutingRedirect
- raise FormDataRoutingRedirect(request)
- def dispatch_request(self) -> ResponseReturnValue:
- """Does the request dispatching. Matches the URL and returns the
- return value of the view or error handler. This does not have to
- be a response object. In order to convert the return value to a
- proper response object, call :func:`make_response`.
- .. versionchanged:: 0.7
- This no longer does the exception handling, this code was
- moved to the new :meth:`full_dispatch_request`.
- """
- req = _request_ctx_stack.top.request
- if req.routing_exception is not None:
- self.raise_routing_exception(req)
- rule = req.url_rule
- # if we provide automatic options for this URL and the
- # request came with the OPTIONS method, reply automatically
- if (
- getattr(rule, "provide_automatic_options", False)
- and req.method == "OPTIONS"
- ):
- return self.make_default_options_response()
- # otherwise dispatch to the handler for that endpoint
- return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)
- def full_dispatch_request(self) -> Response:
- """Dispatches the request and on top of that performs request
- pre and postprocessing as well as HTTP exception catching and
- error handling.
- .. versionadded:: 0.7
- """
- self.try_trigger_before_first_request_functions()
- try:
- request_started.send(self)
- rv = self.preprocess_request()
- if rv is None:
- rv = self.dispatch_request()
- except Exception as e:
- rv = self.handle_user_exception(e)
- return self.finalize_request(rv)
- def finalize_request(
- self,
- rv: t.Union[ResponseReturnValue, HTTPException],
- from_error_handler: bool = False,
- ) -> Response:
- """Given the return value from a view function this finalizes
- the request by converting it into a response and invoking the
- postprocessing functions. This is invoked for both normal
- request dispatching as well as error handlers.
- Because this means that it might be called as a result of a
- failure a special safe mode is available which can be enabled
- with the `from_error_handler` flag. If enabled, failures in
- response processing will be logged and otherwise ignored.
- :internal:
- """
- response = self.make_response(rv)
- try:
- response = self.process_response(response)
- request_finished.send(self, response=response)
- except Exception:
- if not from_error_handler:
- raise
- self.logger.exception(
- "Request finalizing failed with an error while handling an error"
- )
- return response
- def try_trigger_before_first_request_functions(self) -> None:
- """Called before each request and will ensure that it triggers
- the :attr:`before_first_request_funcs` and only exactly once per
- application instance (which means process usually).
- :internal:
- """
- if self._got_first_request:
- return
- with self._before_request_lock:
- if self._got_first_request:
- return
- for func in self.before_first_request_funcs:
- self.ensure_sync(func)()
- self._got_first_request = True
- def make_default_options_response(self) -> Response:
- """This method is called to create the default ``OPTIONS`` response.
- This can be changed through subclassing to change the default
- behavior of ``OPTIONS`` responses.
- .. versionadded:: 0.7
- """
- adapter = _request_ctx_stack.top.url_adapter
- methods = adapter.allowed_methods()
- rv = self.response_class()
- rv.allow.update(methods)
- return rv
- def should_ignore_error(self, error: t.Optional[BaseException]) -> bool:
- """This is called to figure out if an error should be ignored
- or not as far as the teardown system is concerned. If this
- function returns ``True`` then the teardown handlers will not be
- passed the error.
- .. versionadded:: 0.10
- """
- return False
- def ensure_sync(self, func: t.Callable) -> t.Callable:
- """Ensure that the function is synchronous for WSGI workers.
- Plain ``def`` functions are returned as-is. ``async def``
- functions are wrapped to run and wait for the response.
- Override this method to change how the app runs async views.
- .. versionadded:: 2.0
- """
- if iscoroutinefunction(func):
- return self.async_to_sync(func)
- return func
- def async_to_sync(
- self, func: t.Callable[..., t.Coroutine]
- ) -> t.Callable[..., t.Any]:
- """Return a sync function that will run the coroutine function.
- .. code-block:: python
- result = app.async_to_sync(func)(*args, **kwargs)
- Override this method to change how the app converts async code
- to be synchronously callable.
- .. versionadded:: 2.0
- """
- try:
- from asgiref.sync import async_to_sync as asgiref_async_to_sync
- except ImportError:
- raise RuntimeError(
- "Install Flask with the 'async' extra in order to use async views."
- ) from None
- # Check that Werkzeug isn't using its fallback ContextVar class.
- if ContextVar.__module__ == "werkzeug.local":
- raise RuntimeError(
- "Async cannot be used with this combination of Python "
- "and Greenlet versions."
- )
- return asgiref_async_to_sync(func)
- def make_response(self, rv: ResponseReturnValue) -> Response:
- """Convert the return value from a view function to an instance of
- :attr:`response_class`.
- :param rv: the return value from the view function. The view function
- must return a response. Returning ``None``, or the view ending
- without returning, is not allowed. The following types are allowed
- for ``view_rv``:
- ``str``
- A response object is created with the string encoded to UTF-8
- as the body.
- ``bytes``
- A response object is created with the bytes as the body.
- ``dict``
- A dictionary that will be jsonify'd before being returned.
- ``tuple``
- Either ``(body, status, headers)``, ``(body, status)``, or
- ``(body, headers)``, where ``body`` is any of the other types
- allowed here, ``status`` is a string or an integer, and
- ``headers`` is a dictionary or a list of ``(key, value)``
- tuples. If ``body`` is a :attr:`response_class` instance,
- ``status`` overwrites the exiting value and ``headers`` are
- extended.
- :attr:`response_class`
- The object is returned unchanged.
- other :class:`~werkzeug.wrappers.Response` class
- The object is coerced to :attr:`response_class`.
- :func:`callable`
- The function is called as a WSGI application. The result is
- used to create a response object.
- .. versionchanged:: 0.9
- Previously a tuple was interpreted as the arguments for the
- response object.
- """
- status = headers = None
- # unpack tuple returns
- if isinstance(rv, tuple):
- len_rv = len(rv)
- # a 3-tuple is unpacked directly
- if len_rv == 3:
- rv, status, headers = rv
- # decide if a 2-tuple has status or headers
- elif len_rv == 2:
- if isinstance(rv[1], (Headers, dict, tuple, list)):
- rv, headers = rv
- else:
- rv, status = rv
- # other sized tuples are not allowed
- else:
- raise TypeError(
- "The view function did not return a valid response tuple."
- " The tuple must have the form (body, status, headers),"
- " (body, status), or (body, headers)."
- )
- # the body must not be None
- if rv is None:
- raise TypeError(
- f"The view function for {request.endpoint!r} did not"
- " return a valid response. The function either returned"
- " None or ended without a return statement."
- )
- # make sure the body is an instance of the response class
- if not isinstance(rv, self.response_class):
- if isinstance(rv, (str, bytes, bytearray)):
- # let the response class set the status and headers instead of
- # waiting to do it manually, so that the class can handle any
- # special logic
- rv = self.response_class(rv, status=status, headers=headers)
- status = headers = None
- elif isinstance(rv, dict):
- rv = jsonify(rv)
- elif isinstance(rv, BaseResponse) or callable(rv):
- # evaluate a WSGI callable, or coerce a different response
- # class to the correct type
- try:
- rv = self.response_class.force_type(rv, request.environ) # type: ignore # noqa: B950
- except TypeError as e:
- raise TypeError(
- f"{e}\nThe view function did not return a valid"
- " response. The return type must be a string,"
- " dict, tuple, Response instance, or WSGI"
- f" callable, but it was a {type(rv).__name__}."
- ).with_traceback(sys.exc_info()[2]) from None
- else:
- raise TypeError(
- "The view function did not return a valid"
- " response. The return type must be a string,"
- " dict, tuple, Response instance, or WSGI"
- f" callable, but it was a {type(rv).__name__}."
- )
- rv = t.cast(Response, rv)
- # prefer the status if it was provided
- if status is not None:
- if isinstance(status, (str, bytes, bytearray)):
- rv.status = status # type: ignore
- else:
- rv.status_code = status
- # extend existing headers with provided headers
- if headers:
- rv.headers.update(headers)
- return rv
- def create_url_adapter(
- self, request: t.Optional[Request]
- ) -> t.Optional[MapAdapter]:
- """Creates a URL adapter for the given request. The URL adapter
- is created at a point where the request context is not yet set
- up so the request is passed explicitly.
- .. versionadded:: 0.6
- .. versionchanged:: 0.9
- This can now also be called without a request object when the
- URL adapter is created for the application context.
- .. versionchanged:: 1.0
- :data:`SERVER_NAME` no longer implicitly enables subdomain
- matching. Use :attr:`subdomain_matching` instead.
- """
- if request is not None:
- # If subdomain matching is disabled (the default), use the
- # default subdomain in all cases. This should be the default
- # in Werkzeug but it currently does not have that feature.
- if not self.subdomain_matching:
- subdomain = self.url_map.default_subdomain or None
- else:
- subdomain = None
- return self.url_map.bind_to_environ(
- request.environ,
- server_name=self.config["SERVER_NAME"],
- subdomain=subdomain,
- )
- # We need at the very least the server name to be set for this
- # to work.
- if self.config["SERVER_NAME"] is not None:
- return self.url_map.bind(
- self.config["SERVER_NAME"],
- script_name=self.config["APPLICATION_ROOT"],
- url_scheme=self.config["PREFERRED_URL_SCHEME"],
- )
- return None
- def inject_url_defaults(self, endpoint: str, values: dict) -> None:
- """Injects the URL defaults for the given endpoint directly into
- the values dictionary passed. This is used internally and
- automatically called on URL building.
- .. versionadded:: 0.7
- """
- names: t.Iterable[t.Optional[str]] = (None,)
- # url_for may be called outside a request context, parse the
- # passed endpoint instead of using request.blueprints.
- if "." in endpoint:
- names = chain(
- names, reversed(_split_blueprint_path(endpoint.rpartition(".")[0]))
- )
- for name in names:
- if name in self.url_default_functions:
- for func in self.url_default_functions[name]:
- func(endpoint, values)
- def handle_url_build_error(
- self, error: Exception, endpoint: str, values: dict
- ) -> str:
- """Handle :class:`~werkzeug.routing.BuildError` on
- :meth:`url_for`.
- """
- for handler in self.url_build_error_handlers:
- try:
- rv = handler(error, endpoint, values)
- except BuildError as e:
- # make error available outside except block
- error = e
- else:
- if rv is not None:
- return rv
- # Re-raise if called with an active exception, otherwise raise
- # the passed in exception.
- if error is sys.exc_info()[1]:
- raise
- raise error
- def preprocess_request(self) -> t.Optional[ResponseReturnValue]:
- """Called before the request is dispatched. Calls
- :attr:`url_value_preprocessors` registered with the app and the
- current blueprint (if any). Then calls :attr:`before_request_funcs`
- registered with the app and the blueprint.
- If any :meth:`before_request` handler returns a non-None value, the
- value is handled as if it was the return value from the view, and
- further request handling is stopped.
- """
- names = (None, *reversed(request.blueprints))
- for name in names:
- if name in self.url_value_preprocessors:
- for url_func in self.url_value_preprocessors[name]:
- url_func(request.endpoint, request.view_args)
- for name in names:
- if name in self.before_request_funcs:
- for before_func in self.before_request_funcs[name]:
- rv = self.ensure_sync(before_func)()
- if rv is not None:
- return rv
- return None
- def process_response(self, response: Response) -> Response:
- """Can be overridden in order to modify the response object
- before it's sent to the WSGI server. By default this will
- call all the :meth:`after_request` decorated functions.
- .. versionchanged:: 0.5
- As of Flask 0.5 the functions registered for after request
- execution are called in reverse order of registration.
- :param response: a :attr:`response_class` object.
- :return: a new response object or the same, has to be an
- instance of :attr:`response_class`.
- """
- ctx = _request_ctx_stack.top
- for func in ctx._after_request_functions:
- response = self.ensure_sync(func)(response)
- for name in chain(request.blueprints, (None,)):
- if name in self.after_request_funcs:
- for func in reversed(self.after_request_funcs[name]):
- response = self.ensure_sync(func)(response)
- if not self.session_interface.is_null_session(ctx.session):
- self.session_interface.save_session(self, ctx.session, response)
- return response
- def do_teardown_request(
- self, exc: t.Optional[BaseException] = _sentinel # type: ignore
- ) -> None:
- """Called after the request is dispatched and the response is
- returned, right before the request context is popped.
- This calls all functions decorated with
- :meth:`teardown_request`, and :meth:`Blueprint.teardown_request`
- if a blueprint handled the request. Finally, the
- :data:`request_tearing_down` signal is sent.
- This is called by
- :meth:`RequestContext.pop() <flask.ctx.RequestContext.pop>`,
- which may be delayed during testing to maintain access to
- resources.
- :param exc: An unhandled exception raised while dispatching the
- request. Detected from the current exception information if
- not passed. Passed to each teardown function.
- .. versionchanged:: 0.9
- Added the ``exc`` argument.
- """
- if exc is _sentinel:
- exc = sys.exc_info()[1]
- for name in chain(request.blueprints, (None,)):
- if name in self.teardown_request_funcs:
- for func in reversed(self.teardown_request_funcs[name]):
- self.ensure_sync(func)(exc)
- request_tearing_down.send(self, exc=exc)
- def do_teardown_appcontext(
- self, exc: t.Optional[BaseException] = _sentinel # type: ignore
- ) -> None:
- """Called right before the application context is popped.
- When handling a request, the application context is popped
- after the request context. See :meth:`do_teardown_request`.
- This calls all functions decorated with
- :meth:`teardown_appcontext`. Then the
- :data:`appcontext_tearing_down` signal is sent.
- This is called by
- :meth:`AppContext.pop() <flask.ctx.AppContext.pop>`.
- .. versionadded:: 0.9
- """
- if exc is _sentinel:
- exc = sys.exc_info()[1]
- for func in reversed(self.teardown_appcontext_funcs):
- self.ensure_sync(func)(exc)
- appcontext_tearing_down.send(self, exc=exc)
- def app_context(self) -> AppContext:
- """Create an :class:`~flask.ctx.AppContext`. Use as a ``with``
- block to push the context, which will make :data:`current_app`
- point at this application.
- An application context is automatically pushed by
- :meth:`RequestContext.push() <flask.ctx.RequestContext.push>`
- when handling a request, and when running a CLI command. Use
- this to manually create a context outside of these situations.
- ::
- with app.app_context():
- init_db()
- See :doc:`/appcontext`.
- .. versionadded:: 0.9
- """
- return AppContext(self)
- def request_context(self, environ: dict) -> RequestContext:
- """Create a :class:`~flask.ctx.RequestContext` representing a
- WSGI environment. Use a ``with`` block to push the context,
- which will make :data:`request` point at this request.
- See :doc:`/reqcontext`.
- Typically you should not call this from your own code. A request
- context is automatically pushed by the :meth:`wsgi_app` when
- handling a request. Use :meth:`test_request_context` to create
- an environment and context instead of this method.
- :param environ: a WSGI environment
- """
- return RequestContext(self, environ)
- def test_request_context(self, *args: t.Any, **kwargs: t.Any) -> RequestContext:
- """Create a :class:`~flask.ctx.RequestContext` for a WSGI
- environment created from the given values. This is mostly useful
- during testing, where you may want to run a function that uses
- request data without dispatching a full request.
- See :doc:`/reqcontext`.
- Use a ``with`` block to push the context, which will make
- :data:`request` point at the request for the created
- environment. ::
- with test_request_context(...):
- generate_report()
- When using the shell, it may be easier to push and pop the
- context manually to avoid indentation. ::
- ctx = app.test_request_context(...)
- ctx.push()
- ...
- ctx.pop()
- Takes the same arguments as Werkzeug's
- :class:`~werkzeug.test.EnvironBuilder`, with some defaults from
- the application. See the linked Werkzeug docs for most of the
- available arguments. Flask-specific behavior is listed here.
- :param path: URL path being requested.
- :param base_url: Base URL where the app is being served, which
- ``path`` is relative to. If not given, built from
- :data:`PREFERRED_URL_SCHEME`, ``subdomain``,
- :data:`SERVER_NAME`, and :data:`APPLICATION_ROOT`.
- :param subdomain: Subdomain name to append to
- :data:`SERVER_NAME`.
- :param url_scheme: Scheme to use instead of
- :data:`PREFERRED_URL_SCHEME`.
- :param data: The request body, either as a string or a dict of
- form keys and values.
- :param json: If given, this is serialized as JSON and passed as
- ``data``. Also defaults ``content_type`` to
- ``application/json``.
- :param args: other positional arguments passed to
- :class:`~werkzeug.test.EnvironBuilder`.
- :param kwargs: other keyword arguments passed to
- :class:`~werkzeug.test.EnvironBuilder`.
- """
- from .testing import EnvironBuilder
- builder = EnvironBuilder(self, *args, **kwargs)
- try:
- return self.request_context(builder.get_environ())
- finally:
- builder.close()
- def wsgi_app(self, environ: dict, start_response: t.Callable) -> t.Any:
- """The actual WSGI application. This is not implemented in
- :meth:`__call__` so that middlewares can be applied without
- losing a reference to the app object. Instead of doing this::
- app = MyMiddleware(app)
- It's a better idea to do this instead::
- app.wsgi_app = MyMiddleware(app.wsgi_app)
- Then you still have the original application object around and
- can continue to call methods on it.
- .. versionchanged:: 0.7
- Teardown events for the request and app contexts are called
- even if an unhandled error occurs. Other events may not be
- called depending on when an error occurs during dispatch.
- See :ref:`callbacks-and-errors`.
- :param environ: A WSGI environment.
- :param start_response: A callable accepting a status code,
- a list of headers, and an optional exception context to
- start the response.
- """
- ctx = self.request_context(environ)
- error: t.Optional[BaseException] = None
- try:
- try:
- ctx.push()
- response = self.full_dispatch_request()
- except Exception as e:
- error = e
- response = self.handle_exception(e)
- except: # noqa: B001
- error = sys.exc_info()[1]
- raise
- return response(environ, start_response)
- finally:
- if self.should_ignore_error(error):
- error = None
- ctx.auto_pop(error)
- def __call__(self, environ: dict, start_response: t.Callable) -> t.Any:
- """The WSGI server calls the Flask application object as the
- WSGI application. This calls :meth:`wsgi_app`, which can be
- wrapped to apply middleware.
- """
- return self.wsgi_app(environ, start_response)
|