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 

22from starlette.middleware import Middleware 

23from starlette.requests import Request 

24from starlette.responses import PlainTextResponse, RedirectResponse, Response 

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

26from starlette.websockets import WebSocket, WebSocketClose 

27 

28 

29class NoMatchFound(Exception): 

30 """ 

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

32 if no matching route exists. 

33 """ 

34 

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

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

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

38 

39 

40class Match(Enum): 

41 NONE = 0 

42 PARTIAL = 1 

43 FULL = 2 

44 

45 

46def request_response( 

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

48) -> ASGIApp: 

49 """ 

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

51 and returns an ASGI application. 

52 """ 

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

54 func if is_async_callable(func) else functools.partial(run_in_threadpool, func) 

55 ) 

56 

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

58 request = Request(scope, receive, send) 

59 

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

61 response = await f(request) 

62 await response(scope, receive, send) 

63 

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

65 

66 return app 

67 

68 

69def websocket_session( 

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

71) -> ASGIApp: 

72 """ 

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

74 """ 

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

76 

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

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

79 

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

81 await func(session) 

82 

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

84 

85 return app 

86 

87 

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

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

90 

91 

92def replace_params( 

93 path: str, 

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

95 path_params: dict[str, str], 

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

97 for key, value in list(path_params.items()): 

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

99 convertor = param_convertors[key] 

100 value = convertor.to_string(value) 

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

102 path_params.pop(key) 

103 return path, path_params 

104 

105 

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

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

108 

109 

110def compile_path( 

111 path: str, 

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

113 """ 

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

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

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

117 

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

119 format: "/{username}" 

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

121 """ 

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

123 

124 path_regex = "^" 

125 path_format = "" 

126 duplicated_params: set[str] = set() 

127 

128 idx = 0 

129 param_convertors = {} 

130 for match in PARAM_REGEX.finditer(path): 

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

132 convertor_type = convertor_type.lstrip(":") 

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

134 convertor = CONVERTOR_TYPES[convertor_type] 

135 

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

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

138 

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

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

141 

142 if param_name in param_convertors: 

143 duplicated_params.add(param_name) 

144 

145 param_convertors[param_name] = convertor 

146 

147 idx = match.end() 

148 

149 if duplicated_params: 

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

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

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

153 

154 if is_host: 

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

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

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

158 else: 

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

160 

161 path_format += path[idx:] 

162 

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

164 

165 

166class BaseRoute: 

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

168 raise NotImplementedError() # pragma: no cover 

169 

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

171 raise NotImplementedError() # pragma: no cover 

172 

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

174 raise NotImplementedError() # pragma: no cover 

175 

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

177 """ 

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

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

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

181 """ 

182 match, child_scope = self.matches(scope) 

183 if match == Match.NONE: 

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

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

186 await response(scope, receive, send) 

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

188 websocket_close = WebSocketClose() 

189 await websocket_close(scope, receive, send) 

190 return 

191 

192 scope.update(child_scope) 

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

194 

195 

196class Route(BaseRoute): 

197 def __init__( 

198 self, 

199 path: str, 

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

201 *, 

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

203 name: str | None = None, 

204 include_in_schema: bool = True, 

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

206 ) -> None: 

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

208 self.path = path 

209 self.endpoint = endpoint 

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

211 self.include_in_schema = include_in_schema 

212 

213 endpoint_handler = endpoint 

214 while isinstance(endpoint_handler, functools.partial): 

215 endpoint_handler = endpoint_handler.func 

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

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

218 self.app = request_response(endpoint) 

219 if methods is None: 

220 methods = ["GET"] 

221 else: 

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

223 self.app = endpoint 

224 

225 if middleware is not None: 

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

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

228 

229 if methods is None: 

230 self.methods = None 

231 else: 

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

233 if "GET" in self.methods: 

234 self.methods.add("HEAD") 

235 

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

237 

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

239 path_params: dict[str, Any] 

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

241 route_path = get_route_path(scope) 

242 match = self.path_regex.match(route_path) 

243 if match: 

244 matched_params = match.groupdict() 

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

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

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

248 path_params.update(matched_params) 

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

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

251 return Match.PARTIAL, child_scope 

252 else: 

253 return Match.FULL, child_scope 

254 return Match.NONE, {} 

255 

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

257 seen_params = set(path_params.keys()) 

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

259 

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

261 raise NoMatchFound(name, path_params) 

262 

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

264 assert not remaining_params 

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

266 

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

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

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

270 if "app" in scope: 

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

272 else: 

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

274 await response(scope, receive, send) 

275 else: 

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

277 

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

279 return ( 

280 isinstance(other, Route) 

281 and self.path == other.path 

282 and self.endpoint == other.endpoint 

283 and self.methods == other.methods 

284 ) 

285 

286 def __repr__(self) -> str: 

287 class_name = self.__class__.__name__ 

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

289 path, name = self.path, self.name 

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

291 

292 

293class WebSocketRoute(BaseRoute): 

294 def __init__( 

295 self, 

296 path: str, 

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

298 *, 

299 name: str | None = None, 

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

301 ) -> None: 

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

303 self.path = path 

304 self.endpoint = endpoint 

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

306 

307 endpoint_handler = endpoint 

308 while isinstance(endpoint_handler, functools.partial): 

309 endpoint_handler = endpoint_handler.func 

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

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

312 self.app = websocket_session(endpoint) 

313 else: 

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

315 self.app = endpoint 

316 

317 if middleware is not None: 

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

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

320 

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

322 

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

324 path_params: dict[str, Any] 

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

326 route_path = get_route_path(scope) 

327 match = self.path_regex.match(route_path) 

328 if match: 

329 matched_params = match.groupdict() 

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

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

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

333 path_params.update(matched_params) 

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

335 return Match.FULL, child_scope 

336 return Match.NONE, {} 

337 

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

339 seen_params = set(path_params.keys()) 

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

341 

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

343 raise NoMatchFound(name, path_params) 

344 

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

346 assert not remaining_params 

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

348 

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

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

351 

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

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

354 

355 def __repr__(self) -> str: 

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

357 

358 

359class Mount(BaseRoute): 

360 def __init__( 

361 self, 

362 path: str, 

363 app: ASGIApp | None = None, 

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

365 name: str | None = None, 

366 *, 

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

368 ) -> None: 

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

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

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

372 if app is not None: 

373 self._base_app: ASGIApp = app 

374 else: 

375 self._base_app = Router(routes=routes) 

376 self.app = self._base_app 

377 if middleware is not None: 

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

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

380 self.name = name 

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

382 

383 @property 

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

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

386 

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

388 path_params: dict[str, Any] 

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

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

391 route_path = get_route_path(scope) 

392 match = self.path_regex.match(route_path) 

393 if match: 

394 matched_params = match.groupdict() 

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

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

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

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

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

400 path_params.update(matched_params) 

401 child_scope = { 

402 "path_params": path_params, 

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

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

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

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

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

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

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

410 # just inherit that app_root_path default value stored in the 

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

412 # uses the app_root_path to build the URL path. 

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

414 "root_path": root_path + matched_path, 

415 "endpoint": self.app, 

416 } 

417 return Match.FULL, child_scope 

418 return Match.NONE, {} 

419 

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

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

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

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

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

425 if not remaining_params: 

426 return URLPath(path=path) 

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

428 if self.name is None: 

429 # No mount name. 

430 remaining_name = name 

431 else: 

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

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

434 path_kwarg = path_params.get("path") 

435 path_params["path"] = "" 

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

437 if path_kwarg is not None: 

438 remaining_params["path"] = path_kwarg 

439 for route in self.routes or []: 

440 try: 

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

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

443 except NoMatchFound: 

444 pass 

445 raise NoMatchFound(name, path_params) 

446 

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

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

449 

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

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

452 

453 def __repr__(self) -> str: 

454 class_name = self.__class__.__name__ 

455 name = self.name or "" 

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

457 

458 

459class Host(BaseRoute): 

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

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

462 self.host = host 

463 self.app = app 

464 self.name = name 

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

466 

467 @property 

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

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

470 

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

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

473 headers = Headers(scope=scope) 

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

475 match = self.host_regex.match(host) 

476 if match: 

477 matched_params = match.groupdict() 

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

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

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

481 path_params.update(matched_params) 

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

483 return Match.FULL, child_scope 

484 return Match.NONE, {} 

485 

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

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

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

489 path = path_params.pop("path") 

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

491 if not remaining_params: 

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

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

494 if self.name is None: 

495 # No mount name. 

496 remaining_name = name 

497 else: 

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

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

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

501 for route in self.routes or []: 

502 try: 

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

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

505 except NoMatchFound: 

506 pass 

507 raise NoMatchFound(name, path_params) 

508 

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

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

511 

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

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

514 

515 def __repr__(self) -> str: 

516 class_name = self.__class__.__name__ 

517 name = self.name or "" 

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

519 

520 

521_T = TypeVar("_T") 

522 

523 

524class _AsyncLiftContextManager(AbstractAsyncContextManager[_T]): 

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

526 self._cm = cm 

527 

528 async def __aenter__(self) -> _T: 

529 return self._cm.__enter__() 

530 

531 async def __aexit__( 

532 self, 

533 exc_type: type[BaseException] | None, 

534 exc_value: BaseException | None, 

535 traceback: types.TracebackType | None, 

536 ) -> bool | None: 

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

538 

539 

540def _wrap_gen_lifespan_context( 

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

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

543 cmgr = contextlib.contextmanager(lifespan_context) 

544 

545 @functools.wraps(cmgr) 

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

547 return _AsyncLiftContextManager(cmgr(app)) 

548 

549 return wrapper 

550 

551 

552class _DefaultLifespan: 

553 def __init__(self, router: Router): 

554 self._router = router 

555 

556 async def __aenter__(self) -> None: 

557 pass 

558 

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

560 pass 

561 

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

563 return self 

564 

565 

566class Router: 

567 def __init__( 

568 self, 

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

570 redirect_slashes: bool = True, 

571 default: ASGIApp | None = None, 

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

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

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

575 *, 

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

577 ) -> None: 

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

579 self.redirect_slashes = redirect_slashes 

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

581 

582 if lifespan is None: 

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

584 

585 elif inspect.isasyncgenfunction(lifespan): 

586 warnings.warn( 

587 "async generator function lifespans are deprecated, " 

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

589 DeprecationWarning, 

590 ) 

591 self.lifespan_context = asynccontextmanager(lifespan) 

592 elif inspect.isgeneratorfunction(lifespan): 

593 warnings.warn( 

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

595 DeprecationWarning, 

596 ) 

597 self.lifespan_context = _wrap_gen_lifespan_context(lifespan) 

598 else: 

599 self.lifespan_context = lifespan 

600 

601 self.middleware_stack = self.app 

602 if middleware: 

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

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

605 

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

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

608 websocket_close = WebSocketClose() 

609 await websocket_close(scope, receive, send) 

610 return 

611 

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

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

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

615 if "app" in scope: 

616 raise HTTPException(status_code=404) 

617 else: 

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

619 await response(scope, receive, send) 

620 

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

622 for route in self.routes: 

623 try: 

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

625 except NoMatchFound: 

626 pass 

627 raise NoMatchFound(name, path_params) 

628 

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

630 """ 

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

632 startup and shutdown events. 

633 """ 

634 started = False 

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

636 await receive() 

637 try: 

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

639 if maybe_state is not None: 

640 if "state" not in scope: 

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

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

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

644 started = True 

645 await receive() 

646 except BaseException: 

647 exc_text = traceback.format_exc() 

648 if started: 

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

650 else: 

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

652 raise 

653 else: 

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

655 

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

657 """ 

658 The main entry point to the Router class. 

659 """ 

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

661 

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

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

664 

665 if "router" not in scope: 

666 scope["router"] = self 

667 

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

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

670 return 

671 

672 partial = None 

673 

674 for route in self.routes: 

675 # Determine if any route matches the incoming scope, 

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

677 match, child_scope = route.matches(scope) 

678 if match == Match.FULL: 

679 scope.update(child_scope) 

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

681 return 

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

683 partial = route 

684 partial_scope = child_scope 

685 

686 if partial is not None: 

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

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

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

690 scope.update(partial_scope) 

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

692 return 

693 

694 route_path = get_route_path(scope) 

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

696 redirect_scope = dict(scope) 

697 if route_path.endswith("/"): 

698 redirect_scope["path"] = redirect_scope["path"].rstrip("/") 

699 else: 

700 redirect_scope["path"] = redirect_scope["path"] + "/" 

701 

702 for route in self.routes: 

703 match, child_scope = route.matches(redirect_scope) 

704 if match != Match.NONE: 

705 redirect_url = URL(scope=redirect_scope) 

706 response = RedirectResponse(url=str(redirect_url))