approck-messaging

Pydantic models and thin FastStream helpers for delivering Telegram-style messages over Redis streams. Used as the shared message contract between backend services and bots (for example with approck-aiogram-utils).

Python: 3.10+

Install

From PyPI:

uv add approck-messaging

Or with pip:

pip install approck-messaging

Optional Redis transport (FastStream + Redis streams):

uv add "approck-messaging[transport]"

Core install includes only Pydantic models — no broker runtime.

What is included

Usage

Message models

Message describes the payload (caption, media, buttons, video note). TransportMessage adds routing metadata (recipient, optional extra, valid_until).

from approck_messaging.models.message import (
    MessageMedia,
    MessageType,
    TransportMessage,
    TransportMessageRecipient,
)

message = TransportMessage(
    type=MessageType.GENERIC,
    caption="Hello",
    recipient=TransportMessageRecipient(telegram_id=123456789),
)

Supported MessageType values: generic, video_note.

Publish to a Redis stream

Requires the transport extra (faststream[redis]).

from faststream.redis import RedisBroker

from approck_messaging.models.message import MessageType, TransportMessage, TransportMessageRecipient
from approck_messaging.publisher import Publisher

broker = RedisBroker("redis://localhost:6379")
publisher = Publisher(broker=broker, stream="outbound-messages")

await publisher.send_message(
    TransportMessage(
        type=MessageType.GENERIC,
        caption="Hello",
        recipient=TransportMessageRecipient(telegram_id=123456789),
    )
)

Publisher.from_uri(redis_uri, stream=...) is a shortcut when you do not share a broker instance.

Subscribe to a Redis stream

from faststream import Context
from faststream.redis import RedisBroker, StreamSub

from approck_messaging.models.message import TransportMessage
from approck_messaging.subscriber import Subscriber

broker = RedisBroker("redis://localhost:6379")
subscriber = Subscriber(broker=broker, my_var=42)


@subscriber.message(StreamSub("outbound-messages", group="bot", consumer="1"))
async def handle(message: TransportMessage, my_var: int = Context()):
    assert my_var == 42
    ...

Extra keyword arguments passed to Subscriber(...) are registered as FastStream global context values (context.set_global).

Development

uv sync --group dev --extra transport
uv run ruff check .
uv run ruff format --check .
uv run mypy approck_messaging
uv run pytest

Related packages

• approck-aiogram-utils — send Message / TransportMessage payloads to Telegram via aiogram.

License

MIT — see LICENSE.