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