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 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:] 

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

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

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

161 else: 

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

163 

164 path_format += path[idx:] 

165 

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

167 

168 

169class BaseRoute: 

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

171 raise NotImplementedError() # pragma: no cover 

172 

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

174 raise NotImplementedError() # pragma: no cover 

175 

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

177 raise NotImplementedError() # pragma: no cover 

178 

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

180 """ 

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

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

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

184 """ 

185 match, child_scope = self.matches(scope) 

186 if match == Match.NONE: 

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

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

189 await response(scope, receive, send) 

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

191 websocket_close = WebSocketClose() 

192 await websocket_close(scope, receive, send) 

193 return 

194 

195 scope.update(child_scope) 

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

197 

198 

199class Route(BaseRoute): 

200 def __init__( 

201 self, 

202 path: str, 

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

204 *, 

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

206 name: str | None = None, 

207 include_in_schema: bool = True, 

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

209 max_body_size: int | None = None, 

210 ) -> None: 

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

212 self.path = path 

213 self.endpoint = endpoint 

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

215 self.include_in_schema = include_in_schema 

216 

217 endpoint_handler = endpoint 

218 while isinstance(endpoint_handler, functools.partial): 

219 endpoint_handler = endpoint_handler.func 

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

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

222 self.app = request_response(endpoint) 

223 if methods is None: 

224 methods = ["GET"] 

225 else: 

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

227 self.app = endpoint 

228 

229 if middleware is not None: 

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

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

232 if max_body_size is not None: 

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

234 

235 if methods is None: 

236 self.methods = None 

237 else: 

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

239 if "GET" in self.methods: 

240 self.methods.add("HEAD") 

241 

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

243 

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

245 path_params: dict[str, Any] 

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

247 route_path = get_route_path(scope) 

248 match = self.path_regex.match(route_path) 

249 if match: 

250 matched_params = match.groupdict() 

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

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

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

254 path_params.update(matched_params) 

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

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

257 return Match.PARTIAL, child_scope 

258 else: 

259 return Match.FULL, child_scope 

260 return Match.NONE, {} 

261 

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

263 seen_params = set(path_params.keys()) 

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

265 

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

267 raise NoMatchFound(name, path_params) 

268 

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

270 assert not remaining_params 

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

272 

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

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

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

276 if "app" in scope: 

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

278 else: 

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

280 await response(scope, receive, send) 

281 else: 

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

283 

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

285 return ( 

286 isinstance(other, Route) 

287 and self.path == other.path 

288 and self.endpoint == other.endpoint 

289 and self.methods == other.methods 

290 ) 

291 

292 def __repr__(self) -> str: 

293 class_name = self.__class__.__name__ 

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

295 path, name = self.path, self.name 

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

297 

298 

299class WebSocketRoute(BaseRoute): 

300 def __init__( 

301 self, 

302 path: str, 

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

304 *, 

305 name: str | None = None, 

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

307 ) -> None: 

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

309 self.path = path 

310 self.endpoint = endpoint 

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

312 

313 endpoint_handler = endpoint 

314 while isinstance(endpoint_handler, functools.partial): 

315 endpoint_handler = endpoint_handler.func 

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

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

318 self.app = websocket_session(endpoint) 

319 else: 

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

321 self.app = endpoint 

322 

323 if middleware is not None: 

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

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

326 

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

328 

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

330 path_params: dict[str, Any] 

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

332 route_path = get_route_path(scope) 

333 match = self.path_regex.match(route_path) 

334 if match: 

335 matched_params = match.groupdict() 

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

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

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

339 path_params.update(matched_params) 

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

341 return Match.FULL, child_scope 

342 return Match.NONE, {} 

343 

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

345 seen_params = set(path_params.keys()) 

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

347 

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

349 raise NoMatchFound(name, path_params) 

350 

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

352 assert not remaining_params 

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

354 

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

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

357 

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

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

360 

361 def __repr__(self) -> str: 

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

363 

364 

365class Mount(BaseRoute): 

366 def __init__( 

367 self, 

368 path: str, 

369 app: ASGIApp | None = None, 

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

371 name: str | None = None, 

372 *, 

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

374 max_body_size: int | None = None, 

375 ) -> None: 

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

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

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

379 if app is not None: 

380 self._base_app: ASGIApp = app 

381 else: 

382 self._base_app = Router(routes=routes) 

383 self.app = self._base_app 

384 if middleware is not None: 

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

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

387 if max_body_size is not None: 

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

389 self.name = name 

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

391 

392 @property 

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

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

395 

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

397 path_params: dict[str, Any] 

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

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

400 route_path = get_route_path(scope) 

401 match = self.path_regex.match(route_path) 

402 if match: 

403 matched_params = match.groupdict() 

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

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

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

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

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

409 path_params.update(matched_params) 

410 child_scope = { 

411 "path_params": path_params, 

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

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

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

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

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

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

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

419 # just inherit that app_root_path default value stored in the 

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

421 # uses the app_root_path to build the URL path. 

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

423 "root_path": root_path + matched_path, 

424 "endpoint": self.app, 

425 } 

426 return Match.FULL, child_scope 

427 return Match.NONE, {} 

428 

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

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

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

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

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

434 if not remaining_params: 

435 return URLPath(path=path) 

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

437 if self.name is None: 

438 # No mount name. 

439 remaining_name = name 

440 else: 

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

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

443 path_kwarg = path_params.get("path") 

444 path_params["path"] = "" 

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

446 if path_kwarg is not None: 

447 remaining_params["path"] = path_kwarg 

448 for route in self.routes or []: 

449 try: 

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

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

452 except NoMatchFound: 

453 pass 

454 raise NoMatchFound(name, path_params) 

455 

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

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

458 

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

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

461 

462 def __repr__(self) -> str: 

463 class_name = self.__class__.__name__ 

464 name = self.name or "" 

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

466 

467 

468class Host(BaseRoute): 

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

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

471 self.host = host 

472 self.app = app 

473 self.name = name 

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

475 

476 @property 

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

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

479 

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

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

482 headers = Headers(scope=scope) 

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

484 if parsed_host is None: 

485 return Match.NONE, {} 

486 host = parsed_host.host 

487 

488 match = self.host_regex.match(host) 

489 if match: 

490 matched_params = match.groupdict() 

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

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

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

494 path_params.update(matched_params) 

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

496 return Match.FULL, child_scope 

497 return Match.NONE, {} 

498 

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

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

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

502 path = path_params.pop("path") 

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

504 if not remaining_params: 

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

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

507 if self.name is None: 

508 # No mount name. 

509 remaining_name = name 

510 else: 

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

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

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

514 for route in self.routes or []: 

515 try: 

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

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

518 except NoMatchFound: 

519 pass 

520 raise NoMatchFound(name, path_params) 

521 

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

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

524 

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

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

527 

528 def __repr__(self) -> str: 

529 class_name = self.__class__.__name__ 

530 name = self.name or "" 

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

532 

533 

534_T = TypeVar("_T") 

535 

536 

537class _AsyncLiftContextManager(AbstractAsyncContextManager[_T]): 

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

539 self._cm = cm 

540 

541 async def __aenter__(self) -> _T: 

542 return self._cm.__enter__() 

543 

544 async def __aexit__( 

545 self, 

546 exc_type: type[BaseException] | None, 

547 exc_value: BaseException | None, 

548 traceback: types.TracebackType | None, 

549 ) -> bool | None: 

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

551 

552 

553def _wrap_gen_lifespan_context( 

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

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

556 cmgr = contextlib.contextmanager(lifespan_context) 

557 

558 @functools.wraps(cmgr) 

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

560 return _AsyncLiftContextManager(cmgr(app)) 

561 

562 return wrapper 

563 

564 

565class _DefaultLifespan: 

566 def __init__(self, router: Router): 

567 self._router = router 

568 

569 async def __aenter__(self) -> None: 

570 pass 

571 

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

573 pass 

574 

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

576 return self 

577 

578 

579class Router: 

580 def __init__( 

581 self, 

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

583 redirect_slashes: bool = True, 

584 default: ASGIApp | None = None, 

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

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

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

588 *, 

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

590 max_body_size: int | None = None, 

591 ) -> None: 

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

593 self.redirect_slashes = redirect_slashes 

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

595 

596 if lifespan is None: 

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

598 

599 elif inspect.isasyncgenfunction(lifespan): 

600 warnings.warn( 

601 "async generator function lifespans are deprecated, " 

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

603 StarletteDeprecationWarning, 

604 ) 

605 self.lifespan_context = asynccontextmanager(lifespan) 

606 elif inspect.isgeneratorfunction(lifespan): 

607 warnings.warn( 

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

609 StarletteDeprecationWarning, 

610 ) 

611 self.lifespan_context = _wrap_gen_lifespan_context(lifespan) 

612 else: 

613 self.lifespan_context = lifespan 

614 

615 self.middleware_stack = self.app 

616 if middleware: 

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

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

619 if max_body_size is not None: 

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

621 

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

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

624 websocket_close = WebSocketClose() 

625 await websocket_close(scope, receive, send) 

626 return 

627 

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

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

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

631 if "app" in scope: 

632 raise HTTPException(status_code=404) 

633 else: 

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

635 await response(scope, receive, send) 

636 

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

638 for route in self.routes: 

639 try: 

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

641 except NoMatchFound: 

642 pass 

643 raise NoMatchFound(name, path_params) 

644 

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

646 """ 

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

648 startup and shutdown events. 

649 """ 

650 started = False 

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

652 await receive() 

653 try: 

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

655 if maybe_state is not None: 

656 if "state" not in scope: 

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

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

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

660 started = True 

661 await receive() 

662 except BaseException: 

663 exc_text = traceback.format_exc() 

664 if started: 

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

666 else: 

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

668 raise 

669 else: 

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

671 

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

673 """ 

674 The main entry point to the Router class. 

675 """ 

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

677 

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

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

680 

681 if "router" not in scope: 

682 scope["router"] = self 

683 

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

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

686 return 

687 

688 partial = None 

689 

690 for route in self.routes: 

691 # Determine if any route matches the incoming scope, 

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

693 match, child_scope = route.matches(scope) 

694 if match == Match.FULL: 

695 scope["route"] = route 

696 scope.update(child_scope) 

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

698 return 

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

700 partial = route 

701 partial_scope = child_scope 

702 

703 if partial is not None: 

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

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