def _patch()

in instrumentation/elastic-opentelemetry-instrumentation-openai/src/opentelemetry/instrumentation/openai/__init__.py [0:0]

• 34 lines of code
• 3 McCabe index (conditional complexity)

    def _patch(self, module):
        version = tuple([int(x) for x in getattr(getattr(module, "version"), "VERSION").split(".")])
        self.beta_chat_available = version >= (1, 40, 0)
        wrap_function_wrapper(
            "openai.resources.chat.completions",
            "Completions.create",
            self._chat_completion_wrapper,
        )
        wrap_function_wrapper(
            "openai.resources.chat.completions",
            "AsyncCompletions.create",
            self._async_chat_completion_wrapper,
        )
        if self.beta_chat_available:
            wrap_function_wrapper(
                "openai.resources.beta.chat.completions",
                "Completions.parse",
                self._chat_completion_wrapper,
            )
            wrap_function_wrapper(
                "openai.resources.beta.chat.completions",
                "AsyncCompletions.parse",
                self._async_chat_completion_wrapper,
            )
        wrap_function_wrapper(
            "openai.resources.embeddings",
            "Embeddings.create",
            self._embeddings_wrapper,
        )
        wrap_function_wrapper(
            "openai.resources.embeddings",
            "AsyncEmbeddings.create",
            self._async_embeddings_wrapper,
        )