approck-aiogram-utils 0.2.24

Utilities for aiogram bots: Approck messaging helpers, callbacks, and optional FastStream/Uprock integration.

approck-aiogram-utils

Small helpers for aiogram bots that use Approck Messaging: sending Message models to Telegram, safe callbacks, and optional FastStream integration (Redis streams).

Requirements

• Python 3.10+
• uv (recommended)

Install

From PyPI:

uv add approck-aiogram-utils

Or with pip:

pip install approck-aiogram-utils

Optional extras

Extra	Purpose
events	Approck Events SDK integration
sentry	Sentry reporting helpers
transport	approck-messaging with transport (e.g. FastStream)

Other optional dependency groups (e.g. for the stream consumer factory and caption sanitization) are defined in pyproject.toml under [project.optional-dependencies] — install them alongside transport when you use those code paths.

Example for messaging + transport:

uv add "approck-aiogram-utils[transport]"

Usage

Send a text or image from your handler

The smallest path is send_simple_message: plain text, optional cover image URL, optional keyboard.

from aiogram import Bot

from approck_aiogram_utils.message import send_simple_message


async def notify_user(bot: Bot, telegram_id: int) -> None:
    await send_simple_message(
        bot,
        chat_id=telegram_id,
        caption="Hello from Approck Messaging",
        cover="https://example.com/preview.png",  # optional
    )

Send a full Message model

Use send_message when you already have an approck_messaging Message (text, media, inline buttons, video note, etc.).

from aiogram import Bot
from approck_messaging.models.message import Message, MessageMedia, MessageType

from approck_aiogram_utils.message import send_message


async def deliver(bot: Bot, chat_id: int) -> None:
    msg = Message(
        type=MessageType.GENERIC,
        caption="Rich payload",
        media=[
            MessageMedia(
                id="1",
                type="image/jpeg",
                name="photo.jpg",
                url="https://example.com/photo.jpg",
                status="finished",
            )
        ],
    )
    await send_message(bot, chat_id=chat_id, message=msg)

Supported MessageType values are wired in send_message (see approck_aiogram_utils.message); unknown types raise NotImplementedError.

Optional lifecycle callbacks

Define async functions that match CallbackType (same signature as in approck_aiogram_utils.callback). Use callback_call if you need the same “log and never raise” behavior around your own code.

from approck_aiogram_utils.callback import callback_call

async def my_callback(
    message,
    message_channel=None,
    exc=None,
    extra=None,
) -> None:
    ...

await callback_call(
    message=message,
    callback=my_callback,
    message_channel="my-channel",
    extra={"source": "worker"},
)

When you use the integrated factory (create_app) or approck_messaging.subscriber.Subscriber yourself, pass before_send_callback, on_success_callback, on_forbidden_callback, and on_failure_callback — they are forwarded into send_message_handler and run around delivery (see the FastStream section below).

Bundled callback presets (install the matching extras):

• events — module next to handlers.py: Approck Events for success / forbidden paths.
• sentry — module next to handlers.py: Sentry for failures.

FastStream: Redis streams → Telegram

Idea in one line: a producer writes TransportMessage JSON to a Redis stream; FastStream runs send_message_handler; the handler uses aiogram.Bot from FastStream context and sends the message to Telegram.

Producer  --XADD-->  Redis stream  --consumer group-->  send_message_handler  --API-->  Telegram
                              ^                              |
                              |                              v
                         (channel name)         context.set_global("bot", …), callbacks

Data flow (one message)

1. A producer XADDs (or equivalent) to a Redis stream. The message body must deserialize into TransportMessage (see example at the end of this section).
2. send_message_handler (subscriber function) runs inside FastStream. It reads TransportMessage, optional valid_until, optionally sanitizes caption if the optional dependencies used by that handler are installed, then calls the internal send pipeline (same callbacks as in aiogram-only usage).
3. TelegramBadRequest is swallowed so the broken message is acked and skipped; other exceptions become NackMessage so the broker can retry (see FastStream docs).

How the bot and callbacks get into the handler

send_message_handler takes bot and the four callbacks from FastStream Context(), not from your function arguments. In practice that means either:

• Subscriber.from_uri(...) (used inside create_app) passes your Bot and callbacks when building the subscriber — you do not set globals yourself, or
• In tests / a custom broker, you set the same keys with context.set_global("<name>", value) before publishing (names match the handler parameters: bot, before_send_callback, on_success_callback, on_forbidden_callback, on_failure_callback).

context.set_global key	Value
bot	aiogram.Bot used to call Telegram
before_send_callback	optional CallbackType
on_success_callback	optional CallbackType
on_forbidden_callback	optional CallbackType
on_failure_callback	optional CallbackType

More detail: FastStream context.

Two ways to wire it

A — Integrated aiogram app (create_app)
create_app (in the integration package’s app.py, next to handlers.py) builds a TelegramDispatcher, optional Redis FSM storage, and — if you pass message_channels — an approck_messaging.subscriber.Subscriber: for each channel name it registers StreamSub(...) with consumer group telegram-bot and wraps send_message_handler. On dispatcher startup it starts the FastStream broker; on shutdown it closes the broker. You still pass your Router, optional BotCommand list, and the four callbacks above.
Requires transport plus the other optional dependencies from pyproject.toml that app.py imports.

B — Your own FastStream app (manual)
You own the RedisBroker / FastStream app: attach send_message_handler to the same channel names your producers use, call context.set_global for bot (and any callbacks), then start the broker. The tests in this repo (tests/annotation.py, tests/test_callbacks.py) follow this pattern with TestRedisBroker.

Minimal test-style wiring

from aiogram import Bot
from approck_aiogram_utils.integration.uprock.handlers import send_message_handler
from approck_messaging.models.message import MessageType, TransportMessage, TransportMessageRecipient
from faststream import FastStream, context
from faststream.redis import RedisBroker, TestRedisBroker

broker = RedisBroker("redis://localhost:6379")
FastStream(broker)
broker.subscriber("test-channel")(send_message_handler)

context.set_global("bot", Bot(token="123456:ABC-DEF"))
context.set_global("on_success_callback", None)  # or your CallbackType


async def run_once():
    async with TestRedisBroker(broker) as br:
        await br.publish(
            TransportMessage(
                recipient=TransportMessageRecipient(telegram_id=123456789),
                type=MessageType.GENERIC,
                caption="Hello",
            ).model_dump(),
            channel="test-channel",
        )

In production, replace TestRedisBroker with a real RedisBroker URL and the same context.set_global setup before the broker serves traffic. The same pattern is exercised under tests/ in this repository. The string passed to broker.subscriber(...) must match the channel= you use when publishing.

Payload shape (TransportMessage)

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

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

Publish model_dump() (or equivalent JSON) so the subscriber validates into TransportMessage. The message_channel passed into your callbacks is taken from the raw Redis message metadata (see send_message_handler and raw_message.raw_message["channel"]).

Development

uv sync              # project + dev tools (pytest, ruff, mypy)
uv run pytest
uv run ruff check .
uv run ruff format .
uv build             # sdist and wheel under dist/

License

MIT — see LICENSE.