qdrant-client/qdrant_client/auth/bearer_auth.py

import asyncio
from typing import Awaitable, Callable

import httpx


class BearerAuth(httpx.Auth):
    def __init__(
        self,
        auth_token_provider: Callable[[], str] | Callable[[], Awaitable[str]],
    ):
        self.async_token: Callable[[], Awaitable[str]] | None = None
        self.sync_token: Callable[[], str] | None = None

        if asyncio.iscoroutinefunction(auth_token_provider):
            self.async_token = auth_token_provider
        else:
            if callable(auth_token_provider):
                self.sync_token = auth_token_provider  # type: ignore
            else:
                raise ValueError("auth_token_provider must be a callable or awaitable")

    def _sync_get_token(self) -> str:
        if self.sync_token is None:
            raise ValueError("Synchronous token provider is not set.")
        return self.sync_token()

    def sync_auth_flow(self, request: httpx.Request) -> httpx.Request:
        token = self._sync_get_token()
        request.headers["Authorization"] = f"Bearer {token}"
        yield request

    async def _async_get_token(self) -> str:
        if self.async_token is not None:
            return await self.async_token()  # type: ignore
        # Fallback to synchronous token if asynchronous token is not available
        return self._sync_get_token()

    async def async_auth_flow(self, request: httpx.Request) -> httpx.Request:
        token = await self._async_get_token()
        request.headers["Authorization"] = f"Bearer {token}"
        yield request