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 

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 for key, value in list(path_params.items()): 

99 if "{" + key + "}" in path: 

100 convertor = param_convertors[key] 

101 value = convertor.to_string(value) 

102 path = path.replace("{" + key + "}", value) 

103 path_params.pop(key) 

104 return path, path_params 

105 

106 

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

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

109 

110 

111def compile_path( 

112 path: str, 

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

114 """ 

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

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

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

118 

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

120 format: "/{username}" 

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

122 """ 

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

124 

125 path_regex = "^" 

126 path_format = "" 

127 duplicated_params: set[str] = set() 

128 

129 idx = 0 

130 param_convertors = {} 

131 for match in PARAM_REGEX.finditer(path): 

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

133 convertor_type = convertor_type.lstrip(":") 

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

135 convertor = CONVERTOR_TYPES[convertor_type] 

136 

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

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

139 

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

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

142 

143 if param_name in param_convertors: 

144 duplicated_params.add(param_name) 

145 

146 param_convertors[param_name] = convertor 

147 

148 idx = match.end() 

149 

150 if duplicated_params: 

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

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

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

154 

155 if is_host: 

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

157 hostname = path[idx:].split(":")[0] 

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

159 else: 

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

161 

162 path_format += path[idx:] 

163 

164 return re.compile(path_regex), path_format, param_convertors 

165 

166 

167class BaseRoute: 

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

169 raise NotImplementedError() # pragma: no cover 

170 

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

172 raise NotImplementedError() # pragma: no cover 

173 

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

175 raise NotImplementedError() # pragma: no cover 

176 

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

178 """ 

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

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

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

182 """ 

183 match, child_scope = self.matches(scope) 

184 if match == Match.NONE: 

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

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

187 await response(scope, receive, send) 

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

189 websocket_close = WebSocketClose() 

190 await websocket_close(scope, receive, send) 

191 return 

192 

193 scope.update(child_scope) 

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

195 

196 

197class Route(BaseRoute): 

198 def __init__( 

199 self, 

200 path: str, 

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

202 *, 

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

204 name: str | None = None, 

205 include_in_schema: bool = True, 

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

207 max_body_size: int | None = None, 

208 ) -> None: 

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

210 self.path = path 

211 self.endpoint = endpoint 

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

213 self.include_in_schema = include_in_schema 

214 

215 endpoint_handler = endpoint 

216 while isinstance(endpoint_handler, functools.partial): 

217 endpoint_handler = endpoint_handler.func 

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

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

220 self.app = request_response(endpoint) 

221 if methods is None: 

222 methods = ["GET"] 

223 else: 

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

225 self.app = endpoint 

226 

227 if middleware is not None: 

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

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

230 if max_body_size is not None: 

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

232 

233 if methods is None: 

234 self.methods = None 

235 else: 

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

237 if "GET" in self.methods: 

238 self.methods.add("HEAD") 

239 

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

241 

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

243 path_params: dict[str, Any] 

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

245 route_path = get_route_path(scope) 

246 match = self.path_regex.match(route_path) 

247 if match: 

248 matched_params = match.groupdict() 

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

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

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

252 path_params.update(matched_params) 

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

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

255 return Match.PARTIAL, child_scope 

256 else: 

257 return Match.FULL, child_scope 

258 return Match.NONE, {} 

259 

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

261 seen_params = set(path_params.keys()) 

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

263 

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

265 raise NoMatchFound(name, path_params) 

266 

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

268 assert not remaining_params 

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

270 

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

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

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

274 if "app" in scope: 

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

276 else: 

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

278 await response(scope, receive, send) 

279 else: 

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

281 

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

283 return ( 

284 isinstance(other, Route) 

285 and self.path == other.path 

286 and self.endpoint == other.endpoint 

287 and self.methods == other.methods 

288 ) 

289 

290 def __repr__(self) -> str: 

291 class_name = self.__class__.__name__ 

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

293 path, name = self.path, self.name 

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

295 

296 

297class WebSocketRoute(BaseRoute): 

298 def __init__( 

299 self, 

300 path: str, 

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

302 *, 

303 name: str | None = None, 

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

305 ) -> None: 

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

307 self.path = path 

308 self.endpoint = endpoint 

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

310 

311 endpoint_handler = endpoint 

312 while isinstance(endpoint_handler, functools.partial): 

313 endpoint_handler = endpoint_handler.func 

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

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

316 self.app = websocket_session(endpoint) 

317 else: 

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

319 self.app = endpoint 

320 

321 if middleware is not None: 

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

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

324 

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

326 

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

328 path_params: dict[str, Any] 

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

330 route_path = get_route_path(scope) 

331 match = self.path_regex.match(route_path) 

332 if match: 

333 matched_params = match.groupdict() 

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

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

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

337 path_params.update(matched_params) 

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

339 return Match.FULL, child_scope 

340 return Match.NONE, {} 

341 

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

343 seen_params = set(path_params.keys()) 

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

345 

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

347 raise NoMatchFound(name, path_params) 

348 

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

350 assert not remaining_params 

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

352 

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

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

355 

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

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

358 

359 def __repr__(self) -> str: 

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

361 

362 

363class Mount(BaseRoute): 

364 def __init__( 

365 self, 

366 path: str, 

367 app: ASGIApp | None = None, 

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

369 name: str | None = None, 

370 *, 

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

372 max_body_size: int | None = None, 

373 ) -> None: 

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

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

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

377 if app is not None: 

378 self._base_app: ASGIApp = app 

379 else: 

380 self._base_app = Router(routes=routes) 

381 self.app = self._base_app 

382 if middleware is not None: 

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

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

385 if max_body_size is not None: 

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

387 self.name = name 

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

389 

390 @property 

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

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

393 

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

395 path_params: dict[str, Any] 

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

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

398 route_path = get_route_path(scope) 

399 match = self.path_regex.match(route_path) 

400 if match: 

401 matched_params = match.groupdict() 

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

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

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

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

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

407 path_params.update(matched_params) 

408 child_scope = { 

409 "path_params": path_params, 

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

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

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

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

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

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

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

417 # just inherit that app_root_path default value stored in the 

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

419 # uses the app_root_path to build the URL path. 

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

421 "root_path": root_path + matched_path, 

422 "endpoint": self.app, 

423 } 

424 return Match.FULL, child_scope 

425 return Match.NONE, {} 

426 

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

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

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

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

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

432 if not remaining_params: 

433 return URLPath(path=path) 

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

435 if self.name is None: 

436 # No mount name. 

437 remaining_name = name 

438 else: 

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

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

441 path_kwarg = path_params.get("path") 

442 path_params["path"] = "" 

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

444 if path_kwarg is not None: 

445 remaining_params["path"] = path_kwarg 

446 for route in self.routes or []: 

447 try: 

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

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

450 except NoMatchFound: 

451 pass 

452 raise NoMatchFound(name, path_params) 

453 

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

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

456 

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

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

459 

460 def __repr__(self) -> str: 

461 class_name = self.__class__.__name__ 

462 name = self.name or "" 

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

464 

465 

466class Host(BaseRoute): 

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

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

469 self.host = host 

470 self.app = app 

471 self.name = name 

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

473 

474 @property 

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

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

477 

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

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

480 headers = Headers(scope=scope) 

481 host = headers.get("host", "").split(":")[0] 

482 match = self.host_regex.match(host) 

483 if match: 

484 matched_params = match.groupdict() 

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

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

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

488 path_params.update(matched_params) 

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

490 return Match.FULL, child_scope 

491 return Match.NONE, {} 

492 

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

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

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

496 path = path_params.pop("path") 

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

498 if not remaining_params: 

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

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

501 if self.name is None: 

502 # No mount name. 

503 remaining_name = name 

504 else: 

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

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

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

508 for route in self.routes or []: 

509 try: 

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

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

512 except NoMatchFound: 

513 pass 

514 raise NoMatchFound(name, path_params) 

515 

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

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

518 

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

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

521 

522 def __repr__(self) -> str: 

523 class_name = self.__class__.__name__ 

524 name = self.name or "" 

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

526 

527 

528_T = TypeVar("_T") 

529 

530 

531class _AsyncLiftContextManager(AbstractAsyncContextManager[_T]): 

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

533 self._cm = cm 

534 

535 async def __aenter__(self) -> _T: 

536 return self._cm.__enter__() 

537 

538 async def __aexit__( 

539 self, 

540 exc_type: type[BaseException] | None, 

541 exc_value: BaseException | None, 

542 traceback: types.TracebackType | None, 

543 ) -> bool | None: 

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

545 

546 

547def _wrap_gen_lifespan_context( 

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

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

550 cmgr = contextlib.contextmanager(lifespan_context) 

551 

552 @functools.wraps(cmgr) 

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

554 return _AsyncLiftContextManager(cmgr(app)) 

555 

556 return wrapper 

557 

558 

559class _DefaultLifespan: 

560 def __init__(self, router: Router): 

561 self._router = router 

562 

563 async def __aenter__(self) -> None: 

564 pass 

565 

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

567 pass 

568 

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

570 return self 

571 

572 

573class Router: 

574 def __init__( 

575 self, 

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

577 redirect_slashes: bool = True, 

578 default: ASGIApp | None = None, 

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

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

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

582 *, 

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

584 max_body_size: int | None = None, 

585 ) -> None: 

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

587 self.redirect_slashes = redirect_slashes 

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

589 

590 if lifespan is None: 

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

592 

593 elif inspect.isasyncgenfunction(lifespan): 

594 warnings.warn( 

595 "async generator function lifespans are deprecated, " 

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

597 StarletteDeprecationWarning, 

598 ) 

599 self.lifespan_context = asynccontextmanager(lifespan) 

600 elif inspect.isgeneratorfunction(lifespan): 

601 warnings.warn( 

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

603 StarletteDeprecationWarning, 

604 ) 

605 self.lifespan_context = _wrap_gen_lifespan_context(lifespan) 

606 else: 

607 self.lifespan_context = lifespan 

608 

609 self.middleware_stack = self.app 

610 if middleware: 

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

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

613 if max_body_size is not None: 

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

615 

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

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

618 websocket_close = WebSocketClose() 

619 await websocket_close(scope, receive, send) 

620 return 

621 

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

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

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

625 if "app" in scope: 

626 raise HTTPException(status_code=404) 

627 else: 

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

629 await response(scope, receive, send) 

630 

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

632 for route in self.routes: 

633 try: 

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

635 except NoMatchFound: 

636 pass 

637 raise NoMatchFound(name, path_params) 

638 

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

640 """ 

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

642 startup and shutdown events. 

643 """ 

644 started = False 

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

646 await receive() 

647 try: 

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

649 if maybe_state is not None: 

650 if "state" not in scope: 

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

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

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

654 started = True 

655 await receive() 

656 except BaseException: 

657 exc_text = traceback.format_exc() 

658 if started: 

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

660 else: 

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

662 raise 

663 else: 

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

665 

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

667 """ 

668 The main entry point to the Router class. 

669 """ 

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

671 

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

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

674 

675 if "router" not in scope: 

676 scope["router"] = self 

677 

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

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

680 return 

681 

682 partial = None 

683 

684 for route in self.routes: 

685 # Determine if any route matches the incoming scope, 

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

687 match, child_scope = route.matches(scope) 

688 if match == Match.FULL: 

689 scope.update(child_scope) 

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

691 return 

692 elif match == Match.PARTIAL and partial is None: 

693 partial = route 

694 partial_scope = child_scope 

695 

696 if partial is not None: 

697 #  Handle partial matches. These are cases where an endpoint is 

698 # able to handle the request, but is not a preferred option. 

699 # We use this in particular to deal with "405 Method Not Allowed". 

700 scope.update(partial_scope) 

701 await partial.handle(scope, receive, send) 

702 return 

703 

704 route_path = get_route_path(scope) 

705 if scope["type"] == "http" and self.redirect_slashes and route_path != "/":