| 457 | |
| 458 | |
| 459 | class 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__ |
no outgoing calls