from __future__ import annotations

import collections.abc as cabc
import os
import sys
import typing as t
import weakref
from datetime import timedelta
from inspect import iscoroutinefunction
from itertools import chain
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 BadRequestKeyError
from werkzeug.exceptions import HTTPException
from werkzeug.exceptions import InternalServerError
from werkzeug.routing import BuildError
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.wrappers import Response as BaseResponse

The snippet above establishes the foundational imports for the module. It enables postponed evaluation of annotations for forward references and imports standard library modules for system operations, typing, and weak references. It also integrates Werkzeug components for HTTP handling, routing, and exceptions, alongside Click for CLI capabilities.

from . import cli
from . import typing as ft
from .ctx import AppContext
from .ctx import RequestContext
from .globals import _cv_app
from .globals import _cv_request
from .globals import current_app
from .globals import g
from .globals import request
from .globals import request_ctx
from .globals import session
from .helpers import get_debug_flag
from .helpers import get_flashed_messages
from .helpers import get_load_dotenv
from .helpers import send_from_directory
from .sansio.app import App
from .sansio.scaffold import _sentinel
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 Environment
from .wrappers import Request
from .wrappers import Response

if t.TYPE_CHECKING:  # pragma: no cover
    from _typeshed.wsgi import StartResponse
    from _typeshed.wsgi import WSGIEnvironment

    from .testing import FlaskClient
    from .testing import FlaskCliRunner

Internal Flask components are imported here, covering context management, session handling, and templating. The TYPE_CHECKING block ensures that WSGI types and testing utilities are available for static analysis without impacting runtime performance.

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)


def _convert_to_timedelta(val: timedelta | int | None) -> timedelta | None:
    if val is None or isinstance(val, timedelta):
        return val

    return timedelta(seconds=val)

Generic type variables are defined to enforce specific callable signatures for decorators like teardown and template filters. The helper function _convert_to_timedelta normalizes input values, converting integer seconds into timedelta objects if necessary.

class Flask(App):
    default_settings = ImmutableDict(
        {
            "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",
            "TEMPLATES_AUTO_RELOAD": None,
            "MAX_COOKIE_SIZE": 4093,
        }
    )

    #: The class that is used for request objects.
    req_class: type[Request] = Request

    #: The class that is used for response objects.
    resp_class: type[Response] = Response

    #: The session interface to use.
    session_handler: SessionInterface = SecureCookieSessionInterface()

The Flask class inherits from App. It defines a frozen dictionary of default_settings coverign debugging, session lifetimes, and cookie security. It also assigns the specific classes to handle incoming requests (req_class) and outgoing responses (resp_class), alongside initializing the session handler.

    def __init__(
        self,
        module_name: str,
        static_url: str | None = None,
        static_dir: str | os.PathLike[str] | None = "static",
        static_host: str | None = None,
        match_host: bool = False,
        match_subdomain: bool = False,
        template_dir: str | os.PathLike[str] | None = "templates",
        instance_path: str | None = None,
        instance_relative: bool = False,
        base_path: str | None = None,
    ):
        super().__init__(
            import_name=module_name,
            static_url_path=static_url,
            static_folder=static_dir,
            static_host=static_host,
            host_matching=match_host,
            subdomain_matching=match_subdomain,
            template_folder=template_dir,
            instance_path=instance_path,
            instance_relative_config=instance_relative,
            root_path=base_path,
        )

        self.cli = cli.AppGroup()
        self.cli.name = self.name

        if self.has_static_folder:
            assert (
                bool(static_host) == match_host
            ), "Invalid static_host/match_host combination"
            app_weak_ref = weakref.ref(self)
            self.add_url_rule(
                f"{self.static_url_path}/<path:fname>",
                endpoint="static",
                host=static_host,
                view_func=lambda **kwargs: app_weak_ref().send_static_file(**kwargs),
            )

The constructor initializes the application with paths for static files and templates. It sets up the CLI command group and registers a URL rule for serving static assets. A weak reference (app_weak_ref) is used for the static file view function to prevent circular references between the app instance and the function.

    def get_cache_timeout(self, filepath: str | None) -> int | None:
        """Determines the max_age for send_file based on SEND_FILE_MAX_AGE_DEFAULT."""
        val = current_app.config["SEND_FILE_MAX_AGE_DEFAULT"]

        if val is None:
            return None

        if isinstance(val, timedelta):
            return int(val.total_seconds())

        return val

This method resolves the cache control max_age for static files served by the application, checking configuration values and converting timedelta objects to seconds.

    def send_static_file(self, filepath: str) -> Response:
        """Serves files from the static_folder."""
        if not self.has_static_folder:
            raise RuntimeError("'static_folder' must be set to serve static_files.")

        cache_time = self.get_cache_timeout(filepath)
        return send_from_directory(
            t.cast(str, self.static_folder), filepath, max_age=cache_time
        )

A view function that handles the actual delivery of static files. It calculates the appropriate cache headers and uses send_from_directory to stream the file content.

    def open_app_resource(self, res: str, mode: str = "rb") -> t.IO[t.AnyStr]:
        """Opens a file relative to root_path for reading."""
        if mode not in {"r", "rt", "rb"}:
            raise ValueError("Resources can only be opened for reading.")

        return open(os.path.join(self.root_path, res), mode)

    def open_instance_resource(self, res: str, mode: str = "rb") -> t.IO[t.AnyStr]:
        """Opens a file relative to instance_path."""
        return open(os.path.join(self.instance_path, res), mode)

These methods provide file I/O access. open_app_resource is restricted to reading application-level resources, while open_instance_resource allows broader access to the instance folder, which is often used for configuration or data files not tracked in version control.

    def init_jinja_engine(self) -> Environment:
        """Creates and configures the Jinja2 environment."""
        opts = dict(self.jinja_options)

        if "autoescape" not in opts:
            opts["autoescape"] = self.select_jinja_autoescape

        if "auto_reload" not in opts:
            reload_templates = self.config["TEMPLATES_AUTO_RELOAD"]
            if reload_templates is None:
                reload_templates = self.debug
            opts["auto_reload"] = reload_templates

        env = self.jinja_environment(self, **opts)
        env.globals.update(
            url_for=self.url_for,
            get_flashed_messages=get_flashed_messages,
            config=self.config,
            request=request,
            session=session,
            g=g,
        )
        env.policies["json.dumps_function"] = self.json.dumps
        return env

This function bootstraps the Jinja2 templating environment. It applies autoescape and reload policies based on the app's configuration and debug state. It injects global variables (like request and session) and sets the JSON encoder policy.

    def make_url_adapter(self, req: Request | None) -> MapAdapter | None:
        """Creates a MapAdapter for URL generation/matching."""
        if req is not None:
            subdomain = None
            if not self.subdomain_matching:
                subdomain = self.url_map.default_subdomain or None

            return self.url_map.bind_to_environ(
                req.environ,
                server_name=self.config["SERVER_NAME"],
                subdomain=subdomain,
            )

        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

Responsible for creating the URL adapter, which maps incoming requests to endpoints. It handles both request-bound contexts (using WSGI environ) and application-bound contexts (using SERVER_NAME).

    def raise_routing_exception(self, req: Request) -> t.NoReturn:
        """Handles routing exceptions, potentially modifying behavior in debug mode."""
        if (
            not self.debug
            or not isinstance(req.routing_exception, RequestRedirect)
            or req.routing_exception.code in {307, 308}
            or req.method in {"GET", "HEAD", "OPTIONS"}
        ):
            raise req.routing_exception

        from .debughelpers import FormDataRoutingRedirect

        raise FormDataRoutingRedirect(req)

This method processes routing errors. In production, it raises the standard exception. In debug mode, it intercepts specific redirects that might drop form data and raises a more descriptive FormDataRoutingRedirect exception instead.