| 783 | |
| 784 | |
| 785 | class APIWebSocketRoute(routing.WebSocketRoute): |
| 786 | def __init__( |
| 787 | self, |
| 788 | path: str, |
| 789 | endpoint: Callable[..., Any], |
| 790 | *, |
| 791 | name: str | None = None, |
| 792 | dependencies: Sequence[params.Depends] | None = None, |
| 793 | dependency_overrides_provider: Any | None = None, |
| 794 | ) -> None: |
| 795 | self.path = path |
| 796 | self.endpoint = endpoint |
| 797 | self.name = get_name(endpoint) if name is None else name |
| 798 | self.dependencies = list(dependencies or []) |
| 799 | self.path_regex, self.path_format, self.param_convertors = compile_path(path) |
| 800 | ( |
| 801 | self.dependant, |
| 802 | self._flat_dependant, |
| 803 | self._embed_body_fields, |
| 804 | ) = _build_dependant_with_parameterless_dependencies( |
| 805 | path=self.path_format, |
| 806 | call=self.endpoint, |
| 807 | dependencies=self.dependencies, |
| 808 | ) |
| 809 | self.app = websocket_session( |
| 810 | get_websocket_app( |
| 811 | dependant=self.dependant, |
| 812 | dependency_overrides_provider=dependency_overrides_provider, |
| 813 | embed_body_fields=self._embed_body_fields, |
| 814 | ) |
| 815 | ) |
| 816 | |
| 817 | def matches(self, scope: Scope) -> tuple[Match, Scope]: |
| 818 | match, child_scope = super().matches(scope) |
| 819 | if match != Match.NONE: |
| 820 | child_scope["route"] = self |
| 821 | return match, child_scope |
| 822 | |
| 823 | |
| 824 | _FASTAPI_SCOPE_KEY = "fastapi" |
no outgoing calls
no test coverage detected
searching dependent graphs…