Skip to content

RedisAckMixin

faststream.redis.message.RedisAckMixin #

Bases: StreamMessage[MsgType]

A class to represent a Redis ACK mixin.

body instance-attribute #

body: Union[bytes, Any]

committed class-attribute instance-attribute #

committed: bool = field(default=False, init=False)

content_type class-attribute instance-attribute #

content_type: Optional[str] = None

correlation_id class-attribute instance-attribute #

correlation_id: str = field(
    default_factory=lambda: str(uuid4())
)

decoded_body class-attribute instance-attribute #

decoded_body: Optional[DecodedMessage] = None

headers class-attribute instance-attribute #

headers: AnyDict = field(default_factory=dict)

message_id class-attribute instance-attribute #

message_id: str = field(
    default_factory=lambda: str(uuid4())
)

path class-attribute instance-attribute #

path: AnyDict = field(default_factory=dict)

processed class-attribute instance-attribute #

processed: bool = field(default=False, init=False)

raw_message instance-attribute #

raw_message: Msg

reply_to class-attribute instance-attribute #

reply_to: str = ''

ack async #

ack(redis: Redis[bytes], **kwargs: Any) -> None
Source code in faststream/redis/message.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@override
async def ack(  # type: ignore[override]
    self,
    redis: "Redis[bytes]",
    **kwargs: Any,
) -> None:
    handler: Optional["Handler"]
    if (
        not self.committed
        and (ids := self.raw_message.get("message_ids"))
        and (handler := context.get_local("handler_")) is not None
        and (stream := handler.stream_sub)
        and (group := stream.group)
    ):
        await redis.xack(self.raw_message["channel"], group, *ids)  # type: ignore[no-untyped-call]
        await super().ack()

nack async #

nack(**kwargs: Any) -> None
Source code in faststream/broker/message.py 
64
65
async def nack(self, **kwargs: Any) -> None:
    self.committed = True

reject async #

reject(**kwargs: Any) -> None
Source code in faststream/broker/message.py 
67
68
async def reject(self, **kwargs: Any) -> None:
    self.committed = True