Skip to content

BaseMiddleware class

fastpubsub.BaseMiddleware 

BaseMiddleware(next_call)

Base class for middlewares.

Your middlewares should extend this class if you want to implement your own middleware.

Initializes the BaseMiddleware.

PARAMETER DESCRIPTION
next_call

The next middleware or command in the chain.

TYPE: Union[BaseMiddleware, None]

Source code in fastpubsub/middlewares/base.py 
17
18
19
20
21
22
23
def __init__(self, next_call: Union["BaseMiddleware", None]):
    """Initializes the BaseMiddleware.

    Args:
        next_call: The next middleware or command in the chain.
    """
    self.next_call = next_call

next_call instance-attribute 

next_call = next_call

on_message async 

on_message(message)

Handles a message.

When extending this methods, you should always call await super().on_message(...) to continue the chain.

PARAMETER DESCRIPTION
message

The message to handle.

TYPE: Message

Source code in fastpubsub/middlewares/base.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
async def on_message(self, message: "Message") -> Any:
    """Handles a message.

    When extending this methods, you should always call
    `await super().on_message(...)` to continue the chain.

    Args:
        message: The message to handle.
    """
    if not self.next_call:
        return

    return await self.next_call.on_message(message)

on_publish async 

on_publish(data, ordering_key, attributes)

Handles a publish event.

When extending this methods, you should always call await super().on_publish(...) to continue the chain.

PARAMETER DESCRIPTION
data

The message data.

TYPE: bytes

ordering_key

The ordering key for the message.

TYPE: str

attributes

A dictionary of message attributes.

TYPE: dict[str, str] | None

Source code in fastpubsub/middlewares/base.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
async def on_publish(
    self, data: bytes, ordering_key: str, attributes: dict[str, str] | None
) -> Any:
    """Handles a publish event.

    When extending this methods, you should always call
    `await super().on_publish(...)` to continue the chain.

    Args:
        data: The message data.
        ordering_key: The ordering key for the message.
        attributes: A dictionary of message attributes.
    """
    if not self.next_call:
        return

    return await self.next_call.on_publish(data, ordering_key, attributes)

Middleware class

fastpubsub.Middleware 

Middleware(cls, *args, **kwargs)

Wrapper class for middlewares.

You should only use this class to create middlewares on class constructors. Its purpose is to only store the middleware information for delayed initiatization.

Initializes the Middleware.

PARAMETER DESCRIPTION
cls

The middleware class you want to initialize later.

TYPE: type[BaseMiddleware]

args

The middleware class positional arguments.

TYPE: Any DEFAULT: ()

kwargs

The middleware class keyword arguments.

TYPE: Any DEFAULT: {}

Source code in fastpubsub/middlewares/base.py 
65
66
67
68
69
70
71
72
73
74
75
def __init__(self, cls: type[BaseMiddleware], *args: Any, **kwargs: Any) -> None:
    """Initializes the Middleware.

    Args:
        cls: The middleware class you want to initialize later.
        args: The middleware class positional arguments.
        kwargs: The middleware class keyword arguments.
    """
    self.cls = cls
    self.args = args
    self.kwargs = kwargs

cls instance-attribute 

cls = cls

args instance-attribute 

args = args

kwargs instance-attribute 

kwargs = kwargs