authentication

frequenz.client.base.authentication

An Interceptor that adds the API key to a gRPC call.

Classes

frequenz.client.base.authentication.AuthenticationInterceptorUnaryStream

Bases: UnaryStreamClientInterceptor

An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call.

Source code in src/frequenz/client/base/authentication.py

class AuthenticationInterceptorUnaryStream(UnaryStreamClientInterceptor):  # type: ignore[type-arg]
    """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""

    def __init__(self, options: AuthenticationOptions):
        """Create an instance of the interceptor.

        Args:
            options: The options for authenticating to the endpoint.
        """
        self._key = options.api_key

    async def intercept_unary_stream(
        self,
        continuation: Callable[
            [ClientCallDetails, object], UnaryStreamCall[object, object]
        ],
        client_call_details: ClientCallDetails,
        request: object,
    ) -> AsyncIterable[object] | UnaryStreamCall[object, object]:
        """Intercept the call to add HMAC authentication to the metadata fields.

        This is a known method from the base class that is overridden.

        Args:
            continuation: The next interceptor in the chain.
            client_call_details: The call details.
            request: The request object.

        Returns:
            The response object (this implementation does not modify the response).
        """
        _add_auth_header(self._key, client_call_details)

        return await continuation(client_call_details, request)  # type: ignore

Functions

__init__

__init__(options: AuthenticationOptions)

Create an instance of the interceptor.

PARAMETER	DESCRIPTION
options	The options for authenticating to the endpoint. TYPE: AuthenticationOptions

Source code in src/frequenz/client/base/authentication.py

def __init__(self, options: AuthenticationOptions):
    """Create an instance of the interceptor.

    Args:
        options: The options for authenticating to the endpoint.
    """
    self._key = options.api_key

intercept_unary_stream async

intercept_unary_stream(
    continuation: Callable[
        [ClientCallDetails, object],
        UnaryStreamCall[object, object],
    ],
    client_call_details: ClientCallDetails,
    request: object,
) -> (
    AsyncIterable[object] | UnaryStreamCall[object, object]
)

Intercept the call to add HMAC authentication to the metadata fields.

This is a known method from the base class that is overridden.

PARAMETER	DESCRIPTION
continuation	The next interceptor in the chain. TYPE: Callable[[ClientCallDetails, object], UnaryStreamCall[object, object]]
client_call_details	The call details. TYPE: ClientCallDetails
request	The request object. TYPE: object

RETURNS	DESCRIPTION
AsyncIterable[object] | UnaryStreamCall[object, object]	The response object (this implementation does not modify the response).

Source code in src/frequenz/client/base/authentication.py

async def intercept_unary_stream(
    self,
    continuation: Callable[
        [ClientCallDetails, object], UnaryStreamCall[object, object]
    ],
    client_call_details: ClientCallDetails,
    request: object,
) -> AsyncIterable[object] | UnaryStreamCall[object, object]:
    """Intercept the call to add HMAC authentication to the metadata fields.

    This is a known method from the base class that is overridden.

    Args:
        continuation: The next interceptor in the chain.
        client_call_details: The call details.
        request: The request object.

    Returns:
        The response object (this implementation does not modify the response).
    """
    _add_auth_header(self._key, client_call_details)

    return await continuation(client_call_details, request)  # type: ignore

frequenz.client.base.authentication.AuthenticationInterceptorUnaryUnary

Bases: UnaryUnaryClientInterceptor

An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call.

Source code in src/frequenz/client/base/authentication.py

class AuthenticationInterceptorUnaryUnary(UnaryUnaryClientInterceptor):  # type: ignore[type-arg]
    """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""

    def __init__(self, options: AuthenticationOptions):
        """Create an instance of the interceptor.

        Args:
            options: The options for authenticating to the endpoint.
        """
        self._key = options.api_key

    async def intercept_unary_unary(
        self,
        continuation: Callable[
            [ClientCallDetails, object], UnaryUnaryCall[object, object]
        ],
        client_call_details: ClientCallDetails,
        request: object,
    ) -> object:
        """Intercept the call to add HMAC authentication to the metadata fields.

        This is a known method from the base class that is overridden.

        Args:
            continuation: The next interceptor in the chain.
            client_call_details: The call details.
            request: The request object.

        Returns:
            The response object (this implementation does not modify the response).
        """
        _add_auth_header(self._key, client_call_details)
        return await continuation(client_call_details, request)

Functions

__init__

__init__(options: AuthenticationOptions)

Create an instance of the interceptor.

PARAMETER	DESCRIPTION
options	The options for authenticating to the endpoint. TYPE: AuthenticationOptions

Source code in src/frequenz/client/base/authentication.py

def __init__(self, options: AuthenticationOptions):
    """Create an instance of the interceptor.

    Args:
        options: The options for authenticating to the endpoint.
    """
    self._key = options.api_key

intercept_unary_unary async

intercept_unary_unary(
    continuation: Callable[
        [ClientCallDetails, object],
        UnaryUnaryCall[object, object],
    ],
    client_call_details: ClientCallDetails,
    request: object,
) -> object

Intercept the call to add HMAC authentication to the metadata fields.

This is a known method from the base class that is overridden.

PARAMETER	DESCRIPTION
continuation	The next interceptor in the chain. TYPE: Callable[[ClientCallDetails, object], UnaryUnaryCall[object, object]]
client_call_details	The call details. TYPE: ClientCallDetails
request	The request object. TYPE: object

RETURNS	DESCRIPTION
object	The response object (this implementation does not modify the response).

Source code in src/frequenz/client/base/authentication.py

async def intercept_unary_unary(
    self,
    continuation: Callable[
        [ClientCallDetails, object], UnaryUnaryCall[object, object]
    ],
    client_call_details: ClientCallDetails,
    request: object,
) -> object:
    """Intercept the call to add HMAC authentication to the metadata fields.

    This is a known method from the base class that is overridden.

    Args:
        continuation: The next interceptor in the chain.
        client_call_details: The call details.
        request: The request object.

    Returns:
        The response object (this implementation does not modify the response).
    """
    _add_auth_header(self._key, client_call_details)
    return await continuation(client_call_details, request)

frequenz.client.base.authentication.AuthenticationOptions dataclass

Options for authenticating to the endpoint.

Source code in src/frequenz/client/base/authentication.py

@dataclasses.dataclass(frozen=True)
class AuthenticationOptions:
    """Options for authenticating to the endpoint."""

    api_key: str
    """The API key to authenticate with."""

Attributes

api_key instance-attribute

api_key: str

The API key to authenticate with.