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
"""