12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552 |
- import functools
- import inspect
- import json
- import logging
- import os
- import sys
- import typing as t
- import weakref
- from collections.abc import Iterator as _abc_Iterator
- from datetime import timedelta
- from itertools import chain
- from threading import Lock
- from types import TracebackType
- from urllib.parse import quote as _url_quote
- import click
- from werkzeug.datastructures import Headers
- from werkzeug.datastructures import ImmutableDict
- from werkzeug.exceptions import Aborter
- from werkzeug.exceptions import BadRequest
- from werkzeug.exceptions import BadRequestKeyError
- from werkzeug.exceptions import HTTPException
- from werkzeug.exceptions import InternalServerError
- 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.serving import is_running_from_reloader
- from werkzeug.utils import redirect as _wz_redirect
- from werkzeug.wrappers import Response as BaseResponse
- from . import cli
- from . import typing as ft
- from .config import Config
- from .config import ConfigAttribute
- from .ctx import _AppCtxGlobals
- from .ctx import AppContext
- from .ctx import RequestContext
- from .globals import _cv_app
- from .globals import _cv_request
- from .globals import g
- from .globals import request
- from .globals import request_ctx
- from .globals import session
- from .helpers import _split_blueprint_path
- from .helpers import get_debug_flag
- from .helpers import get_flashed_messages
- from .helpers import get_load_dotenv
- from .helpers import locked_cached_property
- from .json.provider import DefaultJSONProvider
- from .json.provider import JSONProvider
- 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 .sessions import SessionInterface
- 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 .wrappers import Request
- from .wrappers import Response
- if t.TYPE_CHECKING: # pragma: no cover
- import typing_extensions as te
- from .blueprints import Blueprint
- from .testing import FlaskClient
- from .testing import FlaskCliRunner
- T_before_first_request = t.TypeVar(
- "T_before_first_request", bound=ft.BeforeFirstRequestCallable
- )
- T_shell_context_processor = t.TypeVar(
- "T_shell_context_processor", bound=ft.ShellContextProcessorCallable
- )
- T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable)
- T_template_filter = t.TypeVar("T_template_filter", bound=ft.TemplateFilterCallable)
- T_template_global = t.TypeVar("T_template_global", bound=ft.TemplateGlobalCallable)
- T_template_test = t.TypeVar("T_template_test", bound=ft.TemplateTestCallable)
- 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.Union[timedelta, int, None]) -> 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 of the object assigned to :attr:`aborter`, created by
- #: :meth:`create_aborter`. That object is called by
- #: :func:`flask.abort` to raise HTTP errors, and can be
- #: called directly as well.
- #:
- #: Defaults to :class:`werkzeug.exceptions.Aborter`.
- #:
- #: .. versionadded:: 2.2
- aborter_class = Aborter
- #: 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")
- @property
- def session_cookie_name(self) -> str:
- """The name of the cookie set by the session interface.
- .. deprecated:: 2.2
- Will be removed in Flask 2.3. Use ``app.config["SESSION_COOKIE_NAME"]``
- instead.
- """
- import warnings
- warnings.warn(
- "'session_cookie_name' is deprecated and will be removed in Flask 2.3. Use"
- " 'SESSION_COOKIE_NAME' in 'app.config' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- return self.config["SESSION_COOKIE_NAME"]
- @session_cookie_name.setter
- def session_cookie_name(self, value: str) -> None:
- import warnings
- warnings.warn(
- "'session_cookie_name' is deprecated and will be removed in Flask 2.3. Use"
- " 'SESSION_COOKIE_NAME' in 'app.config' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- self.config["SESSION_COOKIE_NAME"] = value
- #: 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
- )
- @property
- def send_file_max_age_default(self) -> t.Optional[timedelta]:
- """The default value for ``max_age`` for :func:`~flask.send_file`. The default
- is ``None``, which tells the browser to use conditional requests instead of a
- timed cache.
- .. deprecated:: 2.2
- Will be removed in Flask 2.3. Use
- ``app.config["SEND_FILE_MAX_AGE_DEFAULT"]`` instead.
- .. versionchanged:: 2.0
- Defaults to ``None`` instead of 12 hours.
- """
- import warnings
- warnings.warn(
- "'send_file_max_age_default' is deprecated and will be removed in Flask"
- " 2.3. Use 'SEND_FILE_MAX_AGE_DEFAULT' in 'app.config' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- return _make_timedelta(self.config["SEND_FILE_MAX_AGE_DEFAULT"])
- @send_file_max_age_default.setter
- def send_file_max_age_default(self, value: t.Union[int, timedelta, None]) -> None:
- import warnings
- warnings.warn(
- "'send_file_max_age_default' is deprecated and will be removed in Flask"
- " 2.3. Use 'SEND_FILE_MAX_AGE_DEFAULT' in 'app.config' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- self.config["SEND_FILE_MAX_AGE_DEFAULT"] = _make_timedelta(value)
- @property
- def use_x_sendfile(self) -> bool:
- """Enable this to use the ``X-Sendfile`` feature, assuming the server supports
- it, from :func:`~flask.send_file`.
- .. deprecated:: 2.2
- Will be removed in Flask 2.3. Use ``app.config["USE_X_SENDFILE"]`` instead.
- """
- import warnings
- warnings.warn(
- "'use_x_sendfile' is deprecated and will be removed in Flask 2.3. Use"
- " 'USE_X_SENDFILE' in 'app.config' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- return self.config["USE_X_SENDFILE"]
- @use_x_sendfile.setter
- def use_x_sendfile(self, value: bool) -> None:
- import warnings
- warnings.warn(
- "'use_x_sendfile' is deprecated and will be removed in Flask 2.3. Use"
- " 'USE_X_SENDFILE' in 'app.config' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- self.config["USE_X_SENDFILE"] = value
- _json_encoder: t.Union[t.Type[json.JSONEncoder], None] = None
- _json_decoder: t.Union[t.Type[json.JSONDecoder], None] = None
- @property # type: ignore[override]
- def json_encoder(self) -> t.Type[json.JSONEncoder]:
- """The JSON encoder class to use. Defaults to
- :class:`~flask.json.JSONEncoder`.
- .. deprecated:: 2.2
- Will be removed in Flask 2.3. Customize
- :attr:`json_provider_class` instead.
- .. versionadded:: 0.10
- """
- import warnings
- warnings.warn(
- "'app.json_encoder' is deprecated and will be removed in Flask 2.3."
- " Customize 'app.json_provider_class' or 'app.json' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- if self._json_encoder is None:
- from . import json
- return json.JSONEncoder
- return self._json_encoder
- @json_encoder.setter
- def json_encoder(self, value: t.Type[json.JSONEncoder]) -> None:
- import warnings
- warnings.warn(
- "'app.json_encoder' is deprecated and will be removed in Flask 2.3."
- " Customize 'app.json_provider_class' or 'app.json' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- self._json_encoder = value
- @property # type: ignore[override]
- def json_decoder(self) -> t.Type[json.JSONDecoder]:
- """The JSON decoder class to use. Defaults to
- :class:`~flask.json.JSONDecoder`.
- .. deprecated:: 2.2
- Will be removed in Flask 2.3. Customize
- :attr:`json_provider_class` instead.
- .. versionadded:: 0.10
- """
- import warnings
- warnings.warn(
- "'app.json_decoder' is deprecated and will be removed in Flask 2.3."
- " Customize 'app.json_provider_class' or 'app.json' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- if self._json_decoder is None:
- from . import json
- return json.JSONDecoder
- return self._json_decoder
- @json_decoder.setter
- def json_decoder(self, value: t.Type[json.JSONDecoder]) -> None:
- import warnings
- warnings.warn(
- "'app.json_decoder' is deprecated and will be removed in Flask 2.3."
- " Customize 'app.json_provider_class' or 'app.json' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- self._json_decoder = value
- json_provider_class: t.Type[JSONProvider] = DefaultJSONProvider
- """A subclass of :class:`~flask.json.provider.JSONProvider`. An
- instance is created and assigned to :attr:`app.json` when creating
- the app.
- The default, :class:`~flask.json.provider.DefaultJSONProvider`, uses
- Python's built-in :mod:`json` library. A different provider can use
- a different JSON library.
- .. versionadded:: 2.2
- """
- #: 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,
- "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": None,
- "JSON_SORT_KEYS": None,
- "JSONIFY_PRETTYPRINT_REGULAR": None,
- "JSONIFY_MIMETYPE": None,
- "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: SessionInterface = 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[t.Union[str, os.PathLike]] = "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)
- #: An instance of :attr:`aborter_class` created by
- #: :meth:`make_aborter`. This is called by :func:`flask.abort`
- #: to raise HTTP errors, and can be called directly as well.
- #:
- #: .. versionadded:: 2.2
- #: Moved from ``flask.abort``, which calls this object.
- self.aborter = self.make_aborter()
- self.json: JSONProvider = self.json_provider_class(self)
- """Provides access to JSON methods. Functions in ``flask.json``
- will call methods on this provider when the application context
- is active. Used for handling JSON requests and responses.
- An instance of :attr:`json_provider_class`. Can be customized by
- changing that attribute on a subclass, or by assigning to this
- attribute afterwards.
- The default, :class:`~flask.json.provider.DefaultJSONProvider`,
- uses Python's built-in :mod:`json` library. A different provider
- can use a different JSON library.
- .. versionadded:: 2.2
- """
- #: A list of functions that are called by
- #: :meth:`handle_url_build_error` when :meth:`.url_for` raises a
- #: :exc:`~werkzeug.routing.BuildError`. Each function is called
- #: with ``error``, ``endpoint`` and ``values``. If a function
- #: returns ``None`` or raises a ``BuildError``, it is skipped.
- #: Otherwise, its return value is returned by ``url_for``.
- #:
- #: .. versionadded:: 0.9
- self.url_build_error_handlers: t.List[
- t.Callable[[Exception, str, t.Dict[str, t.Any]], 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.
- #:
- #: .. deprecated:: 2.2
- #: Will be removed in Flask 2.3. Run setup code when
- #: creating the application instead.
- #:
- #: .. versionadded:: 0.8
- self.before_first_request_funcs: t.List[ft.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[ft.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[ft.ShellContextProcessorCallable] = []
- #: 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 _check_setup_finished(self, f_name: str) -> None:
- if self._got_first_request:
- raise AssertionError(
- f"The setup method '{f_name}' can no longer be called"
- " on the application. It has already handled its first"
- " request, any changes will not be applied"
- " consistently.\n"
- "Make sure all imports, decorators, functions, etc."
- " needed to set up the application are done before"
- " running it."
- )
- @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.
- .. deprecated:: 2.2
- Will be removed in Flask 2.3.
- .. versionadded:: 0.7
- """
- import warnings
- warnings.warn(
- "'propagate_exceptions' is deprecated and will be removed in Flask 2.3.",
- DeprecationWarning,
- stacklevel=2,
- )
- rv = self.config["PROPAGATE_EXCEPTIONS"]
- if rv is not None:
- return rv
- return self.testing or 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"] = os.environ.get("FLASK_ENV") or "production"
- defaults["DEBUG"] = get_debug_flag()
- return self.config_class(root_path, defaults)
- def make_aborter(self) -> Aborter:
- """Create the object to assign to :attr:`aborter`. That object
- is called by :func:`flask.abort` to raise HTTP errors, and can
- be called directly as well.
- By default, this creates an instance of :attr:`aborter_class`,
- which defaults to :class:`werkzeug.exceptions.Aborter`.
- .. versionadded:: 2.2
- """
- return self.aborter_class()
- 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`. It is enabled by default in debug mode.
- .. deprecated:: 2.2
- Will be removed in Flask 2.3. Use ``app.config["TEMPLATES_AUTO_RELOAD"]``
- instead.
- .. versionadded:: 1.0
- This property was added but the underlying config and behavior
- already existed.
- """
- import warnings
- warnings.warn(
- "'templates_auto_reload' is deprecated and will be removed in Flask 2.3."
- " Use 'TEMPLATES_AUTO_RELOAD' in 'app.config' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- 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:
- import warnings
- warnings.warn(
- "'templates_auto_reload' is deprecated and will be removed in Flask 2.3."
- " Use 'TEMPLATES_AUTO_RELOAD' in 'app.config' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- 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:
- auto_reload = self.config["TEMPLATES_AUTO_RELOAD"]
- if auto_reload is None:
- auto_reload = self.debug
- options["auto_reload"] = auto_reload
- rv = self.jinja_environment(self, **options)
- rv.globals.update(
- url_for=self.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"] = self.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`.
- .. versionchanged:: 2.2
- Autoescaping is now enabled by default for ``.svg`` files.
- .. versionadded:: 0.5
- """
- if filename is None:
- return True
- return filename.endswith((".html", ".htm", ".xml", ".xhtml", ".svg"))
- 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
- @property
- def env(self) -> str:
- """What environment the app is running in. This maps to the :data:`ENV` config
- key.
- **Do not enable development when deploying in production.**
- Default: ``'production'``
- .. deprecated:: 2.2
- Will be removed in Flask 2.3.
- """
- import warnings
- warnings.warn(
- "'app.env' is deprecated and will be removed in Flask 2.3."
- " Use 'app.debug' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- return self.config["ENV"]
- @env.setter
- def env(self, value: str) -> None:
- import warnings
- warnings.warn(
- "'app.env' is deprecated and will be removed in Flask 2.3."
- " Use 'app.debug' instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- self.config["ENV"] = value
- @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. It may not behave as expected if set late.
- **Do not enable debug mode when deploying in production.**
- Default: ``False``
- """
- return self.config["DEBUG"]
- @debug.setter
- def debug(self, value: bool) -> None:
- self.config["DEBUG"] = value
- if self.config["TEMPLATES_AUTO_RELOAD"] is None:
- self.jinja_env.auto_reload = value
- 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.
- The :envvar:`FLASK_DEBUG` environment variable will override :attr:`debug`.
- Threaded mode is enabled by default.
- .. versionchanged:: 0.10
- The default port is now picked from the ``SERVER_NAME``
- variable.
- """
- # Ignore this call so that it doesn't start another server if
- # the 'flask run' command is used.
- if os.environ.get("FLASK_RUN_FROM_CLI") == "true":
- if not is_running_from_reloader():
- click.secho(
- " * Ignoring a call to 'app.run()' that would block"
- " the current 'flask' CLI command.\n"
- " Only call 'app.run()' in an 'if __name__ =="
- ' "__main__"\' guard.',
- fg="red",
- )
- return
- if get_load_dotenv(load_dotenv):
- cli.load_dotenv()
- # if set, let env vars override previous values
- if "FLASK_ENV" in os.environ:
- print(
- "'FLASK_ENV' is deprecated and will not be used in"
- " Flask 2.3. Use 'FLASK_DEBUG' instead.",
- file=sys.stderr,
- )
- self.config["ENV"] = os.environ.get("FLASK_ENV") or "production"
- 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.debug, self.name)
- 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
- 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
- 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[ft.RouteCallable] = 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[[T_template_filter], T_template_filter]:
- """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: T_template_filter) -> T_template_filter:
- self.add_template_filter(f, name=name)
- return f
- return decorator
- @setupmethod
- def add_template_filter(
- self, f: ft.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[[T_template_test], T_template_test]:
- """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: T_template_test) -> T_template_test:
- self.add_template_test(f, name=name)
- return f
- return decorator
- @setupmethod
- def add_template_test(
- self, f: ft.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[[T_template_global], T_template_global]:
- """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: T_template_global) -> T_template_global:
- self.add_template_global(f, name=name)
- return f
- return decorator
- @setupmethod
- def add_template_global(
- self, f: ft.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: T_before_first_request) -> T_before_first_request:
- """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.
- .. deprecated:: 2.2
- Will be removed in Flask 2.3. Run setup code when creating
- the application instead.
- .. versionadded:: 0.8
- """
- import warnings
- warnings.warn(
- "'before_first_request' is deprecated and will be removed"
- " in Flask 2.3. Run setup code while creating the"
- " application instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- self.before_first_request_funcs.append(f)
- return f
- @setupmethod
- def teardown_appcontext(self, f: T_teardown) -> T_teardown:
- """Registers a function to be called when the application
- context is popped. The application context is typically popped
- after the request context for each request, at the end of CLI
- commands, or after a manually pushed context ends.
- .. code-block:: python
- with app.app_context():
- ...
- When the ``with`` block exits (or ``ctx.pop()`` is called), the
- teardown functions are called just before the app context is
- made inactive. 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.
- Teardown functions must avoid raising exceptions. If they
- execute code that might fail they must surround that code with a
- ``try``/``except`` block and log any errors.
- 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_shell_context_processor
- ) -> T_shell_context_processor:
- """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[ft.ErrorHandlerCallable]:
- """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, ft.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, ft.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)
- propagate = self.config["PROPAGATE_EXCEPTIONS"]
- if propagate is None:
- propagate = self.testing or self.debug
- if propagate:
- # 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, ft.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":
- """Intercept routing exceptions and possibly do something else.
- In debug mode, intercept a routing redirect and replace it with
- an error if the body will be discarded.
- With modern Werkzeug this shouldn't occur, since it now uses a
- 308 status which tells the browser to resend the method and
- body.
- .. versionchanged:: 2.1
- Don't intercept 307 and 308 redirects.
- :meta private:
- :internal:
- """
- if (
- not self.debug
- or not isinstance(request.routing_exception, RequestRedirect)
- or request.routing_exception.code in {307, 308}
- or request.method in {"GET", "HEAD", "OPTIONS"}
- ):
- raise request.routing_exception # type: ignore
- from .debughelpers import FormDataRoutingRedirect
- raise FormDataRoutingRedirect(request)
- def dispatch_request(self) -> ft.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.request
- if req.routing_exception is not None:
- self.raise_routing_exception(req)
- rule: Rule = req.url_rule # type: ignore[assignment]
- # 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
- view_args: t.Dict[str, t.Any] = req.view_args # type: ignore[assignment]
- return self.ensure_sync(self.view_functions[rule.endpoint])(**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
- """
- # Run before_first_request functions if this is the thread's first request.
- # Inlined to avoid a method call on subsequent requests.
- # This is deprecated, will be removed in Flask 2.3.
- if not self._got_first_request:
- with self._before_request_lock:
- if not self._got_first_request:
- for func in self.before_first_request_funcs:
- self.ensure_sync(func)()
- self._got_first_request = True
- 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[ft.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 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.url_adapter
- methods = adapter.allowed_methods() # type: ignore[union-attr]
- 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
- return asgiref_async_to_sync(func)
- def url_for(
- self,
- endpoint: str,
- *,
- _anchor: t.Optional[str] = None,
- _method: t.Optional[str] = None,
- _scheme: t.Optional[str] = None,
- _external: t.Optional[bool] = None,
- **values: t.Any,
- ) -> str:
- """Generate a URL to the given endpoint with the given values.
- This is called by :func:`flask.url_for`, and can be called
- directly as well.
- An *endpoint* is the name of a URL rule, usually added with
- :meth:`@app.route() <route>`, and usually the same name as the
- view function. A route defined in a :class:`~flask.Blueprint`
- will prepend the blueprint's name separated by a ``.`` to the
- endpoint.
- In some cases, such as email messages, you want URLs to include
- the scheme and domain, like ``https://example.com/hello``. When
- not in an active request, URLs will be external by default, but
- this requires setting :data:`SERVER_NAME` so Flask knows what
- domain to use. :data:`APPLICATION_ROOT` and
- :data:`PREFERRED_URL_SCHEME` should also be configured as
- needed. This config is only used when not in an active request.
- Functions can be decorated with :meth:`url_defaults` to modify
- keyword arguments before the URL is built.
- If building fails for some reason, such as an unknown endpoint
- or incorrect values, the app's :meth:`handle_url_build_error`
- method is called. If that returns a string, that is returned,
- otherwise a :exc:`~werkzeug.routing.BuildError` is raised.
- :param endpoint: The endpoint name associated with the URL to
- generate. If this starts with a ``.``, the current blueprint
- name (if any) will be used.
- :param _anchor: If given, append this as ``#anchor`` to the URL.
- :param _method: If given, generate the URL associated with this
- method for the endpoint.
- :param _scheme: If given, the URL will have this scheme if it
- is external.
- :param _external: If given, prefer the URL to be internal
- (False) or require it to be external (True). External URLs
- include the scheme and domain. When not in an active
- request, URLs are external by default.
- :param values: Values to use for the variable parts of the URL
- rule. Unknown keys are appended as query string arguments,
- like ``?a=b&c=d``.
- .. versionadded:: 2.2
- Moved from ``flask.url_for``, which calls this method.
- """
- req_ctx = _cv_request.get(None)
- if req_ctx is not None:
- url_adapter = req_ctx.url_adapter
- blueprint_name = req_ctx.request.blueprint
- # If the endpoint starts with "." and the request matches a
- # blueprint, the endpoint is relative to the blueprint.
- if endpoint[:1] == ".":
- if blueprint_name is not None:
- endpoint = f"{blueprint_name}{endpoint}"
- else:
- endpoint = endpoint[1:]
- # When in a request, generate a URL without scheme and
- # domain by default, unless a scheme is given.
- if _external is None:
- _external = _scheme is not None
- else:
- app_ctx = _cv_app.get(None)
- # If called by helpers.url_for, an app context is active,
- # use its url_adapter. Otherwise, app.url_for was called
- # directly, build an adapter.
- if app_ctx is not None:
- url_adapter = app_ctx.url_adapter
- else:
- url_adapter = self.create_url_adapter(None)
- if url_adapter is None:
- raise RuntimeError(
- "Unable to build URLs outside an active request"
- " without 'SERVER_NAME' configured. Also configure"
- " 'APPLICATION_ROOT' and 'PREFERRED_URL_SCHEME' as"
- " needed."
- )
- # When outside a request, generate a URL with scheme and
- # domain by default.
- if _external is None:
- _external = True
- # It is an error to set _scheme when _external=False, in order
- # to avoid accidental insecure URLs.
- if _scheme is not None and not _external:
- raise ValueError("When specifying '_scheme', '_external' must be True.")
- self.inject_url_defaults(endpoint, values)
- try:
- rv = url_adapter.build( # type: ignore[union-attr]
- endpoint,
- values,
- method=_method,
- url_scheme=_scheme,
- force_external=_external,
- )
- except BuildError as error:
- values.update(
- _anchor=_anchor, _method=_method, _scheme=_scheme, _external=_external
- )
- return self.handle_url_build_error(error, endpoint, values)
- if _anchor is not None:
- _anchor = _url_quote(_anchor, safe="%!#$&'()*+,/:;=?@")
- rv = f"{rv}#{_anchor}"
- return rv
- def redirect(self, location: str, code: int = 302) -> BaseResponse:
- """Create a redirect response object.
- This is called by :func:`flask.redirect`, and can be called
- directly as well.
- :param location: The URL to redirect to.
- :param code: The status code for the redirect.
- .. versionadded:: 2.2
- Moved from ``flask.redirect``, which calls this method.
- """
- return _wz_redirect(location, code=code, Response=self.response_class)
- def make_response(self, rv: ft.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.
- ``list``
- A list that will be jsonify'd before being returned.
- ``generator`` or ``iterator``
- A generator that returns ``str`` or ``bytes`` to be
- streamed as the response.
- ``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:: 2.2
- A generator will be converted to a streaming response.
- A list will be converted to a JSON response.
- .. versionchanged:: 1.1
- A dict will be converted to a JSON response.
- .. 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 # type: ignore[misc]
- # 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 # type: ignore[assignment,misc]
- # 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)) or isinstance(rv, _abc_Iterator):
- # 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, # type: ignore[arg-type]
- )
- status = headers = None
- elif isinstance(rv, (dict, list)):
- rv = self.json.response(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[arg-type]
- )
- 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, list, tuple with headers or status,"
- " Response instance, or WSGI callable, but it"
- f" 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, list, tuple with headers or status,"
- " Response instance, or WSGI callable, but it was a"
- f" {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
- else:
- rv.status_code = status
- # extend existing headers with provided headers
- if headers:
- rv.headers.update(headers) # type: ignore[arg-type]
- 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: BuildError, endpoint: str, values: t.Dict[str, t.Any]
- ) -> str:
- """Called by :meth:`.url_for` if a
- :exc:`~werkzeug.routing.BuildError` was raised. If this returns
- a value, it will be returned by ``url_for``, otherwise the error
- will be re-raised.
- Each function in :attr:`url_build_error_handlers` is called with
- ``error``, ``endpoint`` and ``values``. If a function returns
- ``None`` or raises a ``BuildError``, it is skipped. Otherwise,
- its return value is returned by ``url_for``.
- :param error: The active ``BuildError`` being handled.
- :param endpoint: The endpoint being built.
- :param values: The keyword arguments passed to ``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[ft.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._get_current_object() # type: ignore[attr-defined]
- 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 "werkzeug.debug.preserve_context" in environ:
- environ["werkzeug.debug.preserve_context"](_cv_app.get())
- environ["werkzeug.debug.preserve_context"](_cv_request.get())
- if error is not None and self.should_ignore_error(error):
- error = None
- ctx.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)
|