Authentication internals

The authentication subsystem lets applications inspect or modify the initialized backend stack before mapper authentication. See the authentication guide for a complete typed example.

AuthFlow

Bases: BaseMaxObject, Generic[M, P, T]

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

token class-attribute instance-attribute

token: str | None = None

mapper instance-attribute

mapper: M

protocol instance-attribute

protocol: P

transport instance-attribute

transport: T

AuthMiddlewareManager

Bases: MiddlewareManager

Source code in src/pyromax/dispatcher/middlewares/manager.py

def __init__(self) -> None:
    """Initialize the middleware manager.
    """
    self._middlewares: list[MiddlewareType[MaxObject]] = []

BaseAuthMiddleware

Bases: AbstractMiddleware[AuthFlow[Any, Any, Any], AuthFlow[Any, Any, Any]]

__call__ abstractmethod async

__call__(
    handler: Callable[
        [
            AuthFlow[Any, Any, Any],
            dict[type[Any] | str, Any],
        ],
        Awaitable[Any],
    ],
    event: AuthFlow[Any, Any, Any],
    data: dict[type[Any] | str, Any],
) -> Any

Process an event through the base auth middleware.

Parameters:

Name	Type	Description	Default
handler	Callable[[AuthFlow[Any, Any, Any], dict[type[Any] | str, Any]], Awaitable[Any]]	Handler to invoke.	required
event	AuthFlow[Any, Any, Any]	Incoming event to process.	required
data	dict[type[Any] | str, Any]	Contextual data passed through the processing pipeline.	required

Returns:

Type	Description
Any	The value returned by the wrapped callable or backend.

Source code in src/pyromax/auth/middleware.py

@abstractmethod
async def __call__(
    self,
    handler: Callable[
        [AuthFlow[Any, Any, Any], dict[type[Any] | str, Any]], Awaitable[Any]
    ],
    event: AuthFlow[Any, Any, Any],
    data: dict[type[Any] | str, Any],
) -> Any:
    """Process an event through the base auth middleware.

    :param handler: Handler to invoke.
    :type handler: Callable[[AuthFlow[Any, Any, Any], dict[type[Any] | str, Any]], Awaitable[Any]]
    :param event: Incoming event to process.
    :type event: AuthFlow[Any, Any, Any]
    :param data: Contextual data passed through the processing pipeline.
    :type data: dict[type[Any] | str, Any]
    :returns: The value returned by the wrapped callable or backend.
    :rtype: Any
    """
    ...