(self, model: str, extra_query: Query)
| 719 | return options |
| 720 | |
| 721 | async def _configure_realtime(self, model: str, extra_query: Query) -> tuple[httpx.URL, dict[str, str]]: |
| 722 | auth_headers = {} |
| 723 | query = { |
| 724 | **extra_query, |
| 725 | "api-version": self._api_version, |
| 726 | "deployment": self._azure_deployment or model, |
| 727 | } |
| 728 | if self.api_key and self.api_key != "<missing API key>": |
| 729 | auth_headers = {"api-key": self.api_key} |
| 730 | else: |
| 731 | token = await self._get_azure_ad_token() |
| 732 | if token: |
| 733 | auth_headers = {"Authorization": f"Bearer {token}"} |
| 734 | |
| 735 | if self.websocket_base_url is not None: |
| 736 | base_url = httpx.URL(self.websocket_base_url) |
| 737 | merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/realtime" |
| 738 | realtime_url = base_url.copy_with(raw_path=merge_raw_path) |
| 739 | else: |
| 740 | base_url = self._prepare_url("/realtime") |
| 741 | realtime_url = base_url.copy_with(scheme="wss") |
| 742 | |
| 743 | url = realtime_url.copy_with(params={**query}) |
| 744 | return url, auth_headers |
nothing calls this directly
no test coverage detected