Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/starlette/routing.py: 21%

1from __future__ import annotations 

2 

3import contextlib 

4import functools 

5import inspect 

6import re 

7import traceback 

8import types 

9import warnings 

10from collections.abc import Awaitable, Callable, Collection, Generator, Sequence 

11from contextlib import AbstractAsyncContextManager, AbstractContextManager, asynccontextmanager 

12from enum import Enum 

13from re import Pattern 

14from typing import Any, TypeVar 

15 

16from starlette._exception_handler import wrap_app_handling_exceptions 

17from starlette._utils import get_route_path, is_async_callable, parse_host_header 

18from starlette.concurrency import run_in_threadpool 

19from starlette.convertors import CONVERTOR_TYPES, Convertor 

20from starlette.datastructures import URL, Headers, URLPath 

21from starlette.exceptions import HTTPException, StarletteDeprecationWarning 

22from starlette.middleware import Middleware 

23from starlette.middleware.body_limit import RequestBodyLimitMiddleware 

24from starlette.requests import Request 

25from starlette.responses import PlainTextResponse, RedirectResponse, Response 

26from starlette.types import ASGIApp, Lifespan, Receive, Scope, Send 

27from starlette.websockets import WebSocket, WebSocketClose 

28 

29 

30class NoMatchFound(Exception): 

31 """ 

32 Raised by `.url_for(name, **path_params)` and `.url_path_for(name, **path_params)` 

33 if no matching route exists. 

34 """ 

35 

36 def __init__(self, name: str, path_params: dict[str, Any]) -> None: 

37 params = ", ".join(list(path_params.keys())) 

38 super().__init__(f'No route exists for name "{name}" and params "{params}".') 

39 

40 

41class Match(Enum): 

42 NONE = 0 

43 PARTIAL = 1 

44 FULL = 2 

45 

46 

47def request_response( 

48 func: Callable[[Request], Awaitable[Response] | Response], 

49) -> ASGIApp: 

50 """ 

51 Takes a function or coroutine `func(request) -> response`, 

52 and returns an ASGI application. 

53 """ 

54 f: Callable[[Request], Awaitable[Response]] = ( 

55 func if is_async_callable(func) else functools.partial(run_in_threadpool, func) # type: ignore[assignment, call-arg] 

56 ) 

57 

58 async def app(scope: Scope, receive: Receive, send: Send) -> None: 

59 request = Request(scope, receive, send) 

60 

61 async def app(scope: Scope, receive: Receive, send: Send) -> None: 

62 response = await f(request) 

63 await response(scope, receive, send) 

64 

65 await wrap_app_handling_exceptions(app, request)(scope, receive, send) 

66 

67 return app 

68 

69 

70def websocket_session( 

71 func: Callable[[WebSocket], Awaitable[None]], 

72) -> ASGIApp: 

73 """ 

74 Takes a coroutine `func(session)`, and returns an ASGI application. 

75 """ 

76 # assert asyncio.iscoroutinefunction(func), "WebSocket endpoints must be async" 

77 

78 async def app(scope: Scope, receive: Receive, send: Send) -> None: 

79 session = WebSocket(scope, receive=receive, send=send) 

80 

81 async def app(scope: Scope, receive: Receive, send: Send) -> None: 

82 await func(session) 

83 

84 await wrap_app_handling_exceptions(app, session)(scope, receive, send) 

85 

86 return app 

87 

88 

89def get_name(endpoint: Callable[..., Any]) -> str: 

90 return getattr(endpoint, "__name__", endpoint.__class__.__name__) 

91 

92 

93def replace_params( 

94 path: str, 

95 param_convertors: dict[str, Convertor[Any]], 

96 path_params: dict[str, str], 

97) -> tuple[str, dict[str, str]]: 

98 if not path_params: 

99 return path, path_params 

100 parts = PARAM_FORMAT_REGEX.split(path) 

101 for i in range(1, len(parts), 2): 

102 key = parts[i] 

103 if key in path_params: 

104 parts[i] = param_convertors[key].to_string(path_params.pop(key)) 

105 else: 

106 parts[i] = "{" + key + "}" 

107 return "".join(parts), path_params 

108 

109 

110# Match parameters in URL paths, eg. '{param}', and '{param:int}' 

111PARAM_REGEX = re.compile("{([a-zA-Z_][a-zA-Z0-9_]*)(:[a-zA-Z_][a-zA-Z0-9_]*)?}") 

112# Match parameters in path formats, eg. '{param}' 

113PARAM_FORMAT_REGEX = re.compile("{([a-zA-Z_][a-zA-Z0-9_]*)}") 

114 

115 

116def compile_path( 

117 path: str, 

118) -> tuple[Pattern[str], str, dict[str, Convertor[Any]]]: 

119 """ 

120 Given a path string, like: "/{username:str}", 

121 or a host string, like: "{subdomain}.mydomain.org", return a three-tuple 

122 of (regex, format, {param_name:convertor}). 

123 

124 regex: "/(?P<username>[^/]+)" 

125 format: "/{username}" 

126 convertors: {"username": StringConvertor()} 

127 """ 

128 is_host = not path.startswith("/") 

129 

130 path_regex = "^" 

131 path_format = "" 

132 duplicated_params: set[str] = set() 

133 

134 idx = 0 

135 param_convertors = {} 

136 for match in PARAM_REGEX.finditer(path): 

137 param_name, convertor_type = match.groups("str") 

138 convertor_type = convertor_type.lstrip(":") 

139 assert convertor_type in CONVERTOR_TYPES, f"Unknown path convertor '{convertor_type}'" 

140 convertor = CONVERTOR_TYPES[convertor_type] 

141 

142 path_regex += re.escape(path[idx : match.start()]) 

143 path_regex += f"(?P<{param_name}>{convertor.regex})" 

144 

145 path_format += path[idx : match.start()] 

146 path_format += "{%s}" % param_name 

147 

148 if param_name in param_convertors: 

149 duplicated_params.add(param_name) 

150 

151 param_convertors[param_name] = convertor 

152 

153 idx = match.end() 

154 

155 if duplicated_params: 

156 names = ", ".join(sorted(duplicated_params)) 

157 ending = "s" if len(duplicated_params) > 1 else "" 

158 raise ValueError(f"Duplicated param name{ending} {names} at path {path}") 

159 

160 if is_host: 

161 # Align with `Host.matches()` behavior, which ignores port. 

162 hostname = path[idx:] 

163 if not hostname.endswith("]"): 

164 hostname = hostname.rsplit(":", 1)[0] 

165 path_regex += re.escape(hostname) + "$" 

166 else: 

167 path_regex += re.escape(path[idx:]) + "$" 

168 

169 path_format += path[idx:] 

170 

171 # Host names are case-insensitive (RFC 9110, section 4.2.3). 

172 flags = re.IGNORECASE if is_host else 0 

173 return re.compile(path_regex, flags), path_format, param_convertors 

174 

175 

176class BaseRoute: 

177 def matches(self, scope: Scope) -> tuple[Match, Scope]: 

178 raise NotImplementedError() # pragma: no cover 

179 

180 def url_path_for(self, name: str, /, **path_params: Any) -> URLPath: 

181 raise NotImplementedError() # pragma: no cover 

182 

183 async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: 

184 raise NotImplementedError() # pragma: no cover 

185 

186 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

187 """ 

188 A route may be used in isolation as a stand-alone ASGI app. 

189 This is a somewhat contrived case, as they'll almost always be used 

190 within a Router, but could be useful for some tooling and minimal apps. 

191 """ 

192 match, child_scope = self.matches(scope) 

193 if match is Match.NONE: 

194 if scope["type"] == "http": 

195 response = PlainTextResponse("Not Found", status_code=404) 

196 await response(scope, receive, send) 

197 elif scope["type"] == "websocket": # pragma: no branch 

198 websocket_close = WebSocketClose() 

199 await websocket_close(scope, receive, send) 

200 return 

201 

202 scope.update(child_scope) 

203 await self.handle(scope, receive, send) 

204 

205 

206class Route(BaseRoute): 

207 def __init__( 

208 self, 

209 path: str, 

210 endpoint: Callable[..., Any], 

211 *, 

212 methods: Collection[str] | None = None, 

213 name: str | None = None, 

214 include_in_schema: bool = True, 

215 middleware: Sequence[Middleware] | None = None, 

216 max_body_size: int | None = None, 

217 ) -> None: 

218 assert path.startswith("/"), "Routed paths must start with '/'" 

219 self.path = path 

220 self.endpoint = endpoint 

221 self.name = get_name(endpoint) if name is None else name 

222 self.include_in_schema = include_in_schema 

223 

224 endpoint_handler = endpoint 

225 while isinstance(endpoint_handler, functools.partial): 

226 endpoint_handler = endpoint_handler.func 

227 if inspect.isfunction(endpoint_handler) or inspect.ismethod(endpoint_handler): 

228 # Endpoint is function or method. Treat it as `func(request) -> response`. 

229 self.app = request_response(endpoint) 

230 if methods is None: 

231 methods = ["GET"] 

232 else: 

233 # Endpoint is a class. Treat it as ASGI. 

234 self.app = endpoint 

235 

236 if middleware is not None: 

237 for cls, args, kwargs in reversed(middleware): 

238 self.app = cls(self.app, *args, **kwargs) 

239 if max_body_size is not None: 

240 self.app = RequestBodyLimitMiddleware(self.app, max_body_size=max_body_size) 

241 

242 if methods is None: 

243 self.methods = None 

244 else: 

245 self.methods = {method.upper() for method in methods} 

246 if "GET" in self.methods: 

247 self.methods.add("HEAD") 

248 

249 self.path_regex, self.path_format, self.param_convertors = compile_path(path) 

250 

251 def matches(self, scope: Scope) -> tuple[Match, Scope]: 

252 path_params: dict[str, Any] 

253 if scope["type"] == "http": 

254 route_path = get_route_path(scope) 

255 match = self.path_regex.match(route_path) 

256 if match: 

257 matched_params = match.groupdict() 

258 for key, value in matched_params.items(): 

259 matched_params[key] = self.param_convertors[key].convert(value) 

260 path_params = dict(scope.get("path_params", {})) 

261 path_params.update(matched_params) 

262 child_scope = {"endpoint": self.endpoint, "path_params": path_params} 

263 if self.methods and scope["method"] not in self.methods: 

264 return Match.PARTIAL, child_scope 

265 else: 

266 return Match.FULL, child_scope 

267 return Match.NONE, {} 

268 

269 def url_path_for(self, name: str, /, **path_params: Any) -> URLPath: 

270 seen_params = set(path_params.keys()) 

271 expected_params = set(self.param_convertors.keys()) 

272 

273 if name != self.name or seen_params != expected_params: 

274 raise NoMatchFound(name, path_params) 

275 

276 path, remaining_params = replace_params(self.path_format, self.param_convertors, path_params) 

277 assert not remaining_params 

278 return URLPath(path=path, protocol="http") 

279 

280 async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: 

281 if self.methods and scope["method"] not in self.methods: 

282 headers = {"Allow": ", ".join(self.methods)} 

283 if "app" in scope: 

284 raise HTTPException(status_code=405, headers=headers) 

285 else: 

286 response = PlainTextResponse("Method Not Allowed", status_code=405, headers=headers) 

287 await response(scope, receive, send) 

288 else: 

289 await self.app(scope, receive, send) 

290 

291 def __eq__(self, other: Any) -> bool: 

292 return ( 

293 isinstance(other, Route) 

294 and self.path == other.path 

295 and self.endpoint == other.endpoint 

296 and self.methods == other.methods 

297 ) 

298 

299 def __repr__(self) -> str: 

300 class_name = self.__class__.__name__ 

301 methods = sorted(self.methods or []) 

302 path, name = self.path, self.name 

303 return f"{class_name}(path={path!r}, name={name!r}, methods={methods!r})" 

304 

305 

306class WebSocketRoute(BaseRoute): 

307 def __init__( 

308 self, 

309 path: str, 

310 endpoint: Callable[..., Any], 

311 *, 

312 name: str | None = None, 

313 middleware: Sequence[Middleware] | None = None, 

314 ) -> None: 

315 assert path.startswith("/"), "Routed paths must start with '/'" 

316 self.path = path 

317 self.endpoint = endpoint 

318 self.name = get_name(endpoint) if name is None else name 

319 

320 endpoint_handler = endpoint 

321 while isinstance(endpoint_handler, functools.partial): 

322 endpoint_handler = endpoint_handler.func 

323 if inspect.isfunction(endpoint_handler) or inspect.ismethod(endpoint_handler): 

324 # Endpoint is function or method. Treat it as `func(websocket)`. 

325 self.app = websocket_session(endpoint) 

326 else: 

327 # Endpoint is a class. Treat it as ASGI. 

328 self.app = endpoint 

329 

330 if middleware is not None: 

331 for cls, args, kwargs in reversed(middleware): 

332 self.app = cls(self.app, *args, **kwargs) 

333 

334 self.path_regex, self.path_format, self.param_convertors = compile_path(path) 

335 

336 def matches(self, scope: Scope) -> tuple[Match, Scope]: 

337 path_params: dict[str, Any] 

338 if scope["type"] == "websocket": 

339 route_path = get_route_path(scope) 

340 match = self.path_regex.match(route_path) 

341 if match: 

342 matched_params = match.groupdict() 

343 for key, value in matched_params.items(): 

344 matched_params[key] = self.param_convertors[key].convert(value) 

345 path_params = dict(scope.get("path_params", {})) 

346 path_params.update(matched_params) 

347 child_scope = {"endpoint": self.endpoint, "path_params": path_params} 

348 return Match.FULL, child_scope 

349 return Match.NONE, {} 

350 

351 def url_path_for(self, name: str, /, **path_params: Any) -> URLPath: 

352 seen_params = set(path_params.keys()) 

353 expected_params = set(self.param_convertors.keys()) 

354 

355 if name != self.name or seen_params != expected_params: 

356 raise NoMatchFound(name, path_params) 

357 

358 path, remaining_params = replace_params(self.path_format, self.param_convertors, path_params) 

359 assert not remaining_params 

360 return URLPath(path=path, protocol="websocket") 

361 

362 async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: 

363 await self.app(scope, receive, send) 

364 

365 def __eq__(self, other: Any) -> bool: 

366 return isinstance(other, WebSocketRoute) and self.path == other.path and self.endpoint == other.endpoint 

367 

368 def __repr__(self) -> str: 

369 return f"{self.__class__.__name__}(path={self.path!r}, name={self.name!r})" 

370 

371 

372class Mount(BaseRoute): 

373 def __init__( 

374 self, 

375 path: str, 

376 app: ASGIApp | None = None, 

377 routes: Sequence[BaseRoute] | None = None, 

378 name: str | None = None, 

379 *, 

380 middleware: Sequence[Middleware] | None = None, 

381 max_body_size: int | None = None, 

382 ) -> None: 

383 assert path == "" or path.startswith("/"), "Routed paths must start with '/'" 

384 assert app is not None or routes is not None, "Either 'app=...', or 'routes=' must be specified" 

385 self.path = path.rstrip("/") 

386 if app is not None: 

387 self._base_app: ASGIApp = app 

388 else: 

389 self._base_app = Router(routes=routes) 

390 self.app = self._base_app 

391 if middleware is not None: 

392 for cls, args, kwargs in reversed(middleware): 

393 self.app = cls(self.app, *args, **kwargs) 

394 if max_body_size is not None: 

395 self.app = RequestBodyLimitMiddleware(self.app, max_body_size=max_body_size) 

396 self.name = name 

397 self.path_regex, self.path_format, self.param_convertors = compile_path(self.path + "/{path:path}") 

398 

399 @property 

400 def routes(self) -> list[BaseRoute]: 

401 return getattr(self._base_app, "routes", []) 

402 

403 def matches(self, scope: Scope) -> tuple[Match, Scope]: 

404 path_params: dict[str, Any] 

405 if scope["type"] in ("http", "websocket"): # pragma: no branch 

406 root_path = scope.get("root_path", "") 

407 route_path = get_route_path(scope) 

408 match = self.path_regex.match(route_path) 

409 if match: 

410 matched_params = match.groupdict() 

411 for key, value in matched_params.items(): 

412 matched_params[key] = self.param_convertors[key].convert(value) 

413 remaining_path = "/" + matched_params.pop("path") 

414 matched_path = route_path[: -len(remaining_path)] 

415 path_params = dict(scope.get("path_params", {})) 

416 path_params.update(matched_params) 

417 child_scope = { 

418 "path_params": path_params, 

419 # app_root_path will only be set at the top level scope, 

420 # initialized with the (optional) value of a root_path 

421 # set above/before Starlette. And even though any 

422 # mount will have its own child scope with its own respective 

423 # root_path, the app_root_path will always be available in all 

424 # the child scopes with the same top level value because it's 

425 # set only once here with a default, any other child scope will 

426 # just inherit that app_root_path default value stored in the 

427 # scope. All this is needed to support Request.url_for(), as it 

428 # uses the app_root_path to build the URL path. 

429 "app_root_path": scope.get("app_root_path", root_path), 

430 "root_path": root_path + matched_path, 

431 "endpoint": self.app, 

432 } 

433 return Match.FULL, child_scope 

434 return Match.NONE, {} 

435 

436 def url_path_for(self, name: str, /, **path_params: Any) -> URLPath: 

437 if self.name is not None and name == self.name and "path" in path_params: 

438 # 'name' matches "<mount_name>". 

439 path_params["path"] = path_params["path"].lstrip("/") 

440 path, remaining_params = replace_params(self.path_format, self.param_convertors, path_params) 

441 if not remaining_params: 

442 return URLPath(path=path) 

443 elif self.name is None or name.startswith(self.name + ":"): 

444 if self.name is None: 

445 # No mount name. 

446 remaining_name = name 

447 else: 

448 # 'name' matches "<mount_name>:<child_name>". 

449 remaining_name = name[len(self.name) + 1 :] 

450 path_kwarg = path_params.get("path") 

451 path_params["path"] = "" 

452 path_prefix, remaining_params = replace_params(self.path_format, self.param_convertors, path_params) 

453 if path_kwarg is not None: 

454 remaining_params["path"] = path_kwarg 

455 for route in self.routes or []: 

456 try: 

457 url = route.url_path_for(remaining_name, **remaining_params) 

458 return URLPath(path=path_prefix.rstrip("/") + str(url), protocol=url.protocol) 

459 except NoMatchFound: 

460 pass 

461 raise NoMatchFound(name, path_params) 

462 

463 async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: 

464 await self.app(scope, receive, send) 

465 

466 def __eq__(self, other: Any) -> bool: 

467 return isinstance(other, Mount) and self.path == other.path and self.app == other.app 

468 

469 def __repr__(self) -> str: 

470 class_name = self.__class__.__name__ 

471 name = self.name or "" 

472 return f"{class_name}(path={self.path!r}, name={name!r}, app={self.app!r})" 

473 

474 

475class Host(BaseRoute): 

476 def __init__(self, host: str, app: ASGIApp, name: str | None = None) -> None: 

477 assert not host.startswith("/"), "Host must not start with '/'" 

478 self.host = host 

479 self.app = app 

480 self.name = name 

481 self.host_regex, self.host_format, self.param_convertors = compile_path(host) 

482 

483 @property 

484 def routes(self) -> list[BaseRoute]: 

485 return getattr(self.app, "routes", []) 

486 

487 def matches(self, scope: Scope) -> tuple[Match, Scope]: 

488 if scope["type"] in ("http", "websocket"): # pragma:no branch 

489 headers = Headers(scope=scope) 

490 parsed_host = parse_host_header(headers.get("host")) 

491 if parsed_host is None: 

492 return Match.NONE, {} 

493 host = parsed_host.host.lower() 

494 

495 match = self.host_regex.match(host) 

496 if match: 

497 matched_params = match.groupdict() 

498 for key, value in matched_params.items(): 

499 matched_params[key] = self.param_convertors[key].convert(value) 

500 path_params = dict(scope.get("path_params", {})) 

501 path_params.update(matched_params) 

502 child_scope = {"path_params": path_params, "endpoint": self.app} 

503 return Match.FULL, child_scope 

504 return Match.NONE, {} 

505 

506 def url_path_for(self, name: str, /, **path_params: Any) -> URLPath: 

507 if self.name is not None and name == self.name and "path" in path_params: 

508 # 'name' matches "<mount_name>". 

509 path = path_params.pop("path") 

510 host, remaining_params = replace_params(self.host_format, self.param_convertors, path_params) 

511 if not remaining_params: 

512 return URLPath(path=path, host=host) 

513 elif self.name is None or name.startswith(self.name + ":"): 

514 if self.name is None: 

515 # No mount name. 

516 remaining_name = name 

517 else: 

518 # 'name' matches "<mount_name>:<child_name>". 

519 remaining_name = name[len(self.name) + 1 :] 

520 host, remaining_params = replace_params(self.host_format, self.param_convertors, path_params) 

521 for route in self.routes or []: 

522 try: 

523 url = route.url_path_for(remaining_name, **remaining_params) 

524 return URLPath(path=str(url), protocol=url.protocol, host=host) 

525 except NoMatchFound: 

526 pass 

527 raise NoMatchFound(name, path_params) 

528 

529 async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: 

530 await self.app(scope, receive, send) 

531 

532 def __eq__(self, other: Any) -> bool: 

533 return isinstance(other, Host) and self.host == other.host and self.app == other.app 

534 

535 def __repr__(self) -> str: 

536 class_name = self.__class__.__name__ 

537 name = self.name or "" 

538 return f"{class_name}(host={self.host!r}, name={name!r}, app={self.app!r})" 

539 

540 

541_T = TypeVar("_T") 

542 

543 

544class _AsyncLiftContextManager(AbstractAsyncContextManager[_T]): 

545 def __init__(self, cm: AbstractContextManager[_T]): 

546 self._cm = cm 

547 

548 async def __aenter__(self) -> _T: 

549 return self._cm.__enter__() 

550 

551 async def __aexit__( 

552 self, 

553 exc_type: type[BaseException] | None, 

554 exc_value: BaseException | None, 

555 traceback: types.TracebackType | None, 

556 ) -> bool | None: 

557 return self._cm.__exit__(exc_type, exc_value, traceback) 

558 

559 

560def _wrap_gen_lifespan_context( 

561 lifespan_context: Callable[[Any], Generator[Any, Any, Any]], 

562) -> Callable[[Any], AbstractAsyncContextManager[Any]]: 

563 cmgr = contextlib.contextmanager(lifespan_context) 

564 

565 @functools.wraps(cmgr) 

566 def wrapper(app: Any) -> _AsyncLiftContextManager[Any]: 

567 return _AsyncLiftContextManager(cmgr(app)) 

568 

569 return wrapper 

570 

571 

572class _DefaultLifespan: 

573 def __init__(self, router: Router): 

574 self._router = router 

575 

576 async def __aenter__(self) -> None: 

577 pass 

578 

579 async def __aexit__(self, *exc_info: object) -> None: 

580 pass 

581 

582 def __call__(self: _T, app: object) -> _T: 

583 return self 

584 

585 

586class Router: 

587 def __init__( 

588 self, 

589 routes: Sequence[BaseRoute] | None = None, 

590 redirect_slashes: bool = True, 

591 default: ASGIApp | None = None, 

592 # the generic to Lifespan[AppType] is the type of the top level application 

593 # which the router cannot know statically, so we use Any 

594 lifespan: Lifespan[Any] | None = None, 

595 *, 

596 middleware: Sequence[Middleware] | None = None, 

597 max_body_size: int | None = None, 

598 ) -> None: 

599 self.routes = [] if routes is None else list(routes) 

600 self.redirect_slashes = redirect_slashes 

601 self.default = self.not_found if default is None else default 

602 

603 if lifespan is None: 

604 self.lifespan_context: Lifespan[Any] = _DefaultLifespan(self) 

605 

606 elif inspect.isasyncgenfunction(lifespan): 

607 warnings.warn( 

608 "async generator function lifespans are deprecated, " 

609 "use an @contextlib.asynccontextmanager function instead", 

610 StarletteDeprecationWarning, 

611 ) 

612 self.lifespan_context = asynccontextmanager(lifespan) 

613 elif inspect.isgeneratorfunction(lifespan): 

614 warnings.warn( 

615 "generator function lifespans are deprecated, use an @contextlib.asynccontextmanager function instead", 

616 StarletteDeprecationWarning, 

617 ) 

618 self.lifespan_context = _wrap_gen_lifespan_context(lifespan) 

619 else: 

620 self.lifespan_context = lifespan 

621 

622 self.middleware_stack = self.app 

623 if middleware: 

624 for cls, args, kwargs in reversed(middleware): 

625 self.middleware_stack = cls(self.middleware_stack, *args, **kwargs) 

626 if max_body_size is not None: 

627 self.middleware_stack = RequestBodyLimitMiddleware(self.middleware_stack, max_body_size=max_body_size) 

628 

629 async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None: 

630 if scope["type"] == "websocket": 

631 websocket_close = WebSocketClose() 

632 await websocket_close(scope, receive, send) 

633 return 

634 

635 # If we're running inside a starlette application then raise an 

636 # exception, so that the configurable exception handler can deal with 

637 # returning the response. For plain ASGI apps, just return the response. 

638 if "app" in scope: 

639 raise HTTPException(status_code=404) 

640 else: 

641 response = PlainTextResponse("Not Found", status_code=404) 

642 await response(scope, receive, send) 

643 

644 def url_path_for(self, name: str, /, **path_params: Any) -> URLPath: 

645 for route in self.routes: 

646 try: 

647 return route.url_path_for(name, **path_params) 

648 except NoMatchFound: 

649 pass 

650 raise NoMatchFound(name, path_params) 

651 

652 async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None: 

653 """ 

654 Handle ASGI lifespan messages, which allows us to manage application 

655 startup and shutdown events. 

656 """ 

657 started = False 

658 app: Any = scope.get("app") 

659 await receive() 

660 try: 

661 async with self.lifespan_context(app) as maybe_state: 

662 if maybe_state is not None: 

663 if "state" not in scope: 

664 raise RuntimeError('The server does not support "state" in the lifespan scope.') 

665 scope["state"].update(maybe_state) 

666 await send({"type": "lifespan.startup.complete"}) 

667 started = True 

668 await receive() 

669 except BaseException: 

670 exc_text = traceback.format_exc() 

671 if started: 

672 await send({"type": "lifespan.shutdown.failed", "message": exc_text}) 

673 else: 

674 await send({"type": "lifespan.startup.failed", "message": exc_text}) 

675 raise 

676 else: 

677 await send({"type": "lifespan.shutdown.complete"}) 

678 

679 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

680 """ 

681 The main entry point to the Router class. 

682 """ 

683 await self.middleware_stack(scope, receive, send) 

684 

685 async def app(self, scope: Scope, receive: Receive, send: Send) -> None: 

686 assert scope["type"] in ("http", "websocket", "lifespan") 

687 

688 if "router" not in scope: 

689 scope["router"] = self 

690 

691 if scope["type"] == "lifespan": 

692 await self.lifespan(scope, receive, send) 

693 return 

694 

695 partial = None 

696 

697 for route in self.routes: 

698 # Determine if any route matches the incoming scope, 

699 # and hand over to the matching route if found. 

700 match, child_scope = route.matches(scope) 

701 if match is Match.FULL: 

702 scope["route"] = route 

703 scope.update(child_scope) 

704 await route.handle(scope, receive, send)