def __enter__()

in src/openai/resources/beta/realtime/realtime.py [0:0]

• 35 lines of code
• 4 McCabe index (conditional complexity)

    def __enter__(self) -> RealtimeConnection:
        """
        👋 If your application doesn't work well with the context manager approach then you
        can call this method directly to initiate a connection.

        **Warning**: You must remember to close the connection with `.close()`.

        ```py
        connection = client.beta.realtime.connect(...).enter()
        # ...
        connection.close()
        ```
        """
        try:
            from websockets.sync.client import connect
        except ImportError as exc:
            raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc

        extra_query = self.__extra_query
        auth_headers = self.__client.auth_headers
        if is_azure_client(self.__client):
            url, auth_headers = self.__client._configure_realtime(self.__model, extra_query)
        else:
            url = self._prepare_url().copy_with(
                params={
                    **self.__client.base_url.params,
                    "model": self.__model,
                    **extra_query,
                },
            )
        log.debug("Connecting to %s", url)
        if self.__websocket_connection_options:
            log.debug("Connection options: %s", self.__websocket_connection_options)

        self.__connection = RealtimeConnection(
            connect(
                str(url),
                user_agent_header=self.__client.user_agent,
                additional_headers=_merge_mappings(
                    {
                        **auth_headers,
                        "OpenAI-Beta": "realtime=v1",
                    },
                    self.__extra_headers,
                ),
                **self.__websocket_connection_options,
            )
        )

        return self.__connection