applications.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118
  1. from __future__ import annotations
  2. from collections.abc import Awaitable, Callable, Mapping, Sequence
  3. from typing import Any, ParamSpec, TypeVar
  4. from starlette.datastructures import State, URLPath
  5. from starlette.middleware import Middleware, _MiddlewareFactory
  6. from starlette.middleware.errors import ServerErrorMiddleware
  7. from starlette.middleware.exceptions import ExceptionMiddleware
  8. from starlette.requests import Request
  9. from starlette.responses import Response
  10. from starlette.routing import BaseRoute, Router
  11. from starlette.types import ASGIApp, ExceptionHandler, Lifespan, Receive, Scope, Send
  12. AppType = TypeVar("AppType", bound="Starlette")
  13. P = ParamSpec("P")
  14. class Starlette:
  15. """Creates an Starlette application."""
  16. def __init__(
  17. self: AppType,
  18. debug: bool = False,
  19. routes: Sequence[BaseRoute] | None = None,
  20. middleware: Sequence[Middleware] | None = None,
  21. exception_handlers: Mapping[Any, ExceptionHandler] | None = None,
  22. lifespan: Lifespan[AppType] | None = None,
  23. ) -> None:
  24. """Initializes the application.
  25. Parameters:
  26. debug: Boolean indicating if debug tracebacks should be returned on errors.
  27. routes: A list of routes to serve incoming HTTP and WebSocket requests.
  28. middleware: A list of middleware to run for every request. A starlette
  29. application will always automatically include two middleware classes.
  30. `ServerErrorMiddleware` is added as the very outermost middleware, to handle
  31. any uncaught errors occurring anywhere in the entire stack.
  32. `ExceptionMiddleware` is added as the very innermost middleware, to deal
  33. with handled exception cases occurring in the routing or endpoints.
  34. exception_handlers: A mapping of either integer status codes,
  35. or exception class types onto callables which handle the exceptions.
  36. Exception handler callables should be of the form
  37. `handler(request, exc) -> response` and may be either standard functions, or
  38. async functions.
  39. lifespan: A lifespan context function, which can be used to perform
  40. startup and shutdown tasks. This is a newer style that replaces the
  41. `on_startup` and `on_shutdown` handlers. Use one or the other, not both.
  42. """
  43. self.debug = debug
  44. self.state = State()
  45. self.router = Router(routes, lifespan=lifespan)
  46. self.exception_handlers = {} if exception_handlers is None else dict(exception_handlers)
  47. self.user_middleware = [] if middleware is None else list(middleware)
  48. self.middleware_stack: ASGIApp | None = None
  49. def build_middleware_stack(self) -> ASGIApp:
  50. debug = self.debug
  51. error_handler = None
  52. exception_handlers: dict[Any, ExceptionHandler] = {}
  53. for key, value in self.exception_handlers.items():
  54. if key in (500, Exception):
  55. error_handler = value
  56. else:
  57. exception_handlers[key] = value
  58. middleware = (
  59. [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
  60. + self.user_middleware
  61. + [Middleware(ExceptionMiddleware, handlers=exception_handlers, debug=debug)]
  62. )
  63. app = self.router
  64. for cls, args, kwargs in reversed(middleware):
  65. app = cls(app, *args, **kwargs)
  66. return app
  67. @property
  68. def routes(self) -> list[BaseRoute]:
  69. return self.router.routes
  70. def url_path_for(self, name: str, /, **path_params: Any) -> URLPath:
  71. return self.router.url_path_for(name, **path_params)
  72. async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
  73. scope["app"] = self
  74. if self.middleware_stack is None:
  75. self.middleware_stack = self.build_middleware_stack()
  76. await self.middleware_stack(scope, receive, send)
  77. def mount(self, path: str, app: ASGIApp, name: str | None = None) -> None:
  78. self.router.mount(path, app=app, name=name) # pragma: no cover
  79. def host(self, host: str, app: ASGIApp, name: str | None = None) -> None:
  80. self.router.host(host, app=app, name=name) # pragma: no cover
  81. def add_middleware(self, middleware_class: _MiddlewareFactory[P], *args: P.args, **kwargs: P.kwargs) -> None:
  82. if self.middleware_stack is not None: # pragma: no cover
  83. raise RuntimeError("Cannot add middleware after an application has started")
  84. self.user_middleware.insert(0, Middleware(middleware_class, *args, **kwargs))
  85. def add_exception_handler(
  86. self,
  87. exc_class_or_status_code: int | type[Exception],
  88. handler: ExceptionHandler,
  89. ) -> None: # pragma: no cover
  90. self.exception_handlers[exc_class_or_status_code] = handler
  91. def add_route(
  92. self,
  93. path: str,
  94. route: Callable[[Request], Awaitable[Response] | Response],
  95. methods: list[str] | None = None,
  96. name: str | None = None,
  97. include_in_schema: bool = True,
  98. ) -> None: # pragma: no cover
  99. self.router.add_route(path, route, methods=methods, name=name, include_in_schema=include_in_schema)