diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/starlette/applications.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/starlette/applications.py | 249 |
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 |