Source code for aiocometd.extensions

"""Extension classes"""
from abc import ABC, abstractmethod
from typing import Optional

from aiocometd.typing import Payload, Headers

[docs]class Extension(ABC): """Defines operations supported by extensions"""
[docs] @abstractmethod async def outgoing(self, payload: Payload, headers: Headers) -> None: """Process outgoing *payload* and *headers* Called just before a payload is sent :param payload: List of outgoing messages :param headers: Headers to send """
[docs] @abstractmethod async def incoming(self, payload: Payload, headers: Optional[Headers] = None) -> None: """Process incoming *payload* and *headers* Called just after a payload is received :param payload: List of incoming messages :param headers: Headers to send """
[docs]class AuthExtension(Extension): # pylint: disable=abstract-method """Extension with support for authentication"""
[docs] async def authenticate(self) -> None: """Called after a failed authentication attempt For authentication schemes where the credentials are static it doesn't makes much sense to reimplement this function. However for schemes where the credentials can expire (like OAuth, JWT...) this method can be reimplemented to update those credentials """