aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/starlette/applications.py
diff options
context:
space:
mode:
Diffstat (limited to '.venv/lib/python3.12/site-packages/starlette/applications.py')
-rw-r--r--.venv/lib/python3.12/site-packages/starlette/applications.py249
1 files changed, 249 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/starlette/applications.py b/.venv/lib/python3.12/site-packages/starlette/applications.py
new file mode 100644
index 00000000..6df5a707
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/starlette/applications.py
@@ -0,0 +1,249 @@
+from __future__ import annotations
+
+import sys
+import typing
+import warnings
+
+if sys.version_info >= (3, 10): # pragma: no cover
+ from typing import ParamSpec
+else: # pragma: no cover
+ from typing_extensions import ParamSpec
+
+from starlette.datastructures import State, URLPath
+from starlette.middleware import Middleware, _MiddlewareFactory
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.middleware.errors import ServerErrorMiddleware
+from starlette.middleware.exceptions import ExceptionMiddleware
+from starlette.requests import Request
+from starlette.responses import Response
+from starlette.routing import BaseRoute, Router
+from starlette.types import ASGIApp, ExceptionHandler, Lifespan, Receive, Scope, Send
+from starlette.websockets import WebSocket
+
+AppType = typing.TypeVar("AppType", bound="Starlette")
+P = ParamSpec("P")
+
+
+class Starlette:
+ """Creates an Starlette application."""
+
+ def __init__(
+ self: AppType,
+ debug: bool = False,
+ routes: typing.Sequence[BaseRoute] | None = None,
+ middleware: typing.Sequence[Middleware] | None = None,
+ exception_handlers: typing.Mapping[typing.Any, ExceptionHandler] | None = None,
+ on_startup: typing.Sequence[typing.Callable[[], typing.Any]] | None = None,
+ on_shutdown: typing.Sequence[typing.Callable[[], typing.Any]] | None = None,
+ lifespan: Lifespan[AppType] | None = None,
+ ) -> None:
+ """Initializes the application.
+
+ Parameters:
+ debug: Boolean indicating if debug tracebacks should be returned on errors.
+ routes: A list of routes to serve incoming HTTP and WebSocket requests.
+ middleware: A list of middleware to run for every request. A starlette
+ application will always automatically include two middleware classes.
+ `ServerErrorMiddleware` is added as the very outermost middleware, to handle
+ any uncaught errors occurring anywhere in the entire stack.
+ `ExceptionMiddleware` is added as the very innermost middleware, to deal
+ with handled exception cases occurring in the routing or endpoints.
+ exception_handlers: A mapping of either integer status codes,
+ or exception class types onto callables which handle the exceptions.
+ Exception handler callables should be of the form
+ `handler(request, exc) -> response` and may be either standard functions, or
+ async functions.
+ on_startup: A list of callables to run on application startup.
+ Startup handler callables do not take any arguments, and may be either
+ standard functions, or async functions.
+ on_shutdown: A list of callables to run on application shutdown.
+ Shutdown handler callables do not take any arguments, and may be either
+ standard functions, or async functions.
+ lifespan: A lifespan context function, which can be used to perform
+ startup and shutdown tasks. This is a newer style that replaces the
+ `on_startup` and `on_shutdown` handlers. Use one or the other, not both.
+ """
+ # The lifespan context function is a newer style that replaces
+ # on_startup / on_shutdown handlers. Use one or the other, not both.
+ assert lifespan is None or (on_startup is None and on_shutdown is None), (
+ "Use either 'lifespan' or 'on_startup'/'on_shutdown', not both."
+ )
+
+ self.debug = debug
+ self.state = State()
+ self.router = Router(routes, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan)
+ self.exception_handlers = {} if exception_handlers is None else dict(exception_handlers)
+ self.user_middleware = [] if middleware is None else list(middleware)
+ self.middleware_stack: ASGIApp | None = None
+
+ def build_middleware_stack(self) -> ASGIApp:
+ debug = self.debug
+ error_handler = None
+ exception_handlers: dict[typing.Any, typing.Callable[[Request, Exception], Response]] = {}
+
+ for key, value in self.exception_handlers.items():
+ if key in (500, Exception):
+ error_handler = value
+ else:
+ exception_handlers[key] = value
+
+ middleware = (
+ [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
+ + self.user_middleware
+ + [Middleware(ExceptionMiddleware, handlers=exception_handlers, debug=debug)]
+ )
+
+ app = self.router
+ for cls, args, kwargs in reversed(middleware):
+ app = cls(app, *args, **kwargs)
+ return app
+
+ @property
+ def routes(self) -> list[BaseRoute]:
+ return self.router.routes
+
+ def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath:
+ return self.router.url_path_for(name, **path_params)
+
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ scope["app"] = self
+ if self.middleware_stack is None:
+ self.middleware_stack = self.build_middleware_stack()
+ await self.middleware_stack(scope, receive, send)
+
+ def on_event(self, event_type: str) -> typing.Callable: # type: ignore[type-arg]
+ return self.router.on_event(event_type) # pragma: no cover
+
+ def mount(self, path: str, app: ASGIApp, name: str | None = None) -> None:
+ self.router.mount(path, app=app, name=name) # pragma: no cover
+
+ def host(self, host: str, app: ASGIApp, name: str | None = None) -> None:
+ self.router.host(host, app=app, name=name) # pragma: no cover
+
+ def add_middleware(
+ self,
+ middleware_class: _MiddlewareFactory[P],
+ *args: P.args,
+ **kwargs: P.kwargs,
+ ) -> None:
+ if self.middleware_stack is not None: # pragma: no cover
+ raise RuntimeError("Cannot add middleware after an application has started")
+ self.user_middleware.insert(0, Middleware(middleware_class, *args, **kwargs))
+
+ def add_exception_handler(
+ self,
+ exc_class_or_status_code: int | type[Exception],
+ handler: ExceptionHandler,
+ ) -> None: # pragma: no cover
+ self.exception_handlers[exc_class_or_status_code] = handler
+
+ def add_event_handler(
+ self,
+ event_type: str,
+ func: typing.Callable, # type: ignore[type-arg]
+ ) -> None: # pragma: no cover
+ self.router.add_event_handler(event_type, func)
+
+ def add_route(
+ self,
+ path: str,
+ route: typing.Callable[[Request], typing.Awaitable[Response] | Response],
+ methods: list[str] | None = None,
+ name: str | None = None,
+ include_in_schema: bool = True,
+ ) -> None: # pragma: no cover
+ self.router.add_route(path, route, methods=methods, name=name, include_in_schema=include_in_schema)
+
+ def add_websocket_route(
+ self,
+ path: str,
+ route: typing.Callable[[WebSocket], typing.Awaitable[None]],
+ name: str | None = None,
+ ) -> None: # pragma: no cover
+ self.router.add_websocket_route(path, route, name=name)
+
+ def exception_handler(self, exc_class_or_status_code: int | type[Exception]) -> typing.Callable: # type: ignore[type-arg]
+ warnings.warn(
+ "The `exception_handler` decorator is deprecated, and will be removed in version 1.0.0. "
+ "Refer to https://www.starlette.io/exceptions/ for the recommended approach.",
+ DeprecationWarning,
+ )
+
+ def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg]
+ self.add_exception_handler(exc_class_or_status_code, func)
+ return func
+
+ return decorator
+
+ def route(
+ self,
+ path: str,
+ methods: list[str] | None = None,
+ name: str | None = None,
+ include_in_schema: bool = True,
+ ) -> typing.Callable: # type: ignore[type-arg]
+ """
+ We no longer document this decorator style API, and its usage is discouraged.
+ Instead you should use the following approach:
+
+ >>> routes = [Route(path, endpoint=...), ...]
+ >>> app = Starlette(routes=routes)
+ """
+ warnings.warn(
+ "The `route` decorator is deprecated, and will be removed in version 1.0.0. "
+ "Refer to https://www.starlette.io/routing/ for the recommended approach.",
+ DeprecationWarning,
+ )
+
+ def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg]
+ self.router.add_route(
+ path,
+ func,
+ methods=methods,
+ name=name,
+ include_in_schema=include_in_schema,
+ )
+ return func
+
+ return decorator
+
+ def websocket_route(self, path: str, name: str | None = None) -> typing.Callable: # type: ignore[type-arg]
+ """
+ We no longer document this decorator style API, and its usage is discouraged.
+ Instead you should use the following approach:
+
+ >>> routes = [WebSocketRoute(path, endpoint=...), ...]
+ >>> app = Starlette(routes=routes)
+ """
+ warnings.warn(
+ "The `websocket_route` decorator is deprecated, and will be removed in version 1.0.0. "
+ "Refer to https://www.starlette.io/routing/#websocket-routing for the recommended approach.",
+ DeprecationWarning,
+ )
+
+ def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg]
+ self.router.add_websocket_route(path, func, name=name)
+ return func
+
+ return decorator
+
+ def middleware(self, middleware_type: str) -> typing.Callable: # type: ignore[type-arg]
+ """
+ We no longer document this decorator style API, and its usage is discouraged.
+ Instead you should use the following approach:
+
+ >>> middleware = [Middleware(...), ...]
+ >>> app = Starlette(middleware=middleware)
+ """
+ warnings.warn(
+ "The `middleware` decorator is deprecated, and will be removed in version 1.0.0. "
+ "Refer to https://www.starlette.io/middleware/#using-middleware for recommended approach.",
+ DeprecationWarning,
+ )
+ assert middleware_type == "http", 'Currently only middleware("http") is supported.'
+
+ def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg]
+ self.add_middleware(BaseHTTPMiddleware, dispatch=func)
+ return func
+
+ return decorator