tg-delivery-client 0.1.0

AMQP publisher and a facade for sending commands to the Telegram delivery service.

tg-delivery-client

AMQP publisher and facade for sending commands to a Telegram delivery service.

Requirements

• Python ≥ 3.12
• RabbitMQ (or any AMQP broker). To use priorities your queue must support x-max-priority.

Installation

Install from PyPI:

pip install tg-delivery-client

Install the latest development version from GitHub:

pip install git+<github-url>

Or add to your pyproject.toml (Poetry):

[tool.poetry.dependencies]
tg-delivery-client = "^0.1.0"

Quick Start

import asyncio
from tg_delivery_client.facade import create_amqp_client

async def main():
    client = await create_amqp_client(
        "amqps://user:pass@host/vhost",  # AMQP URL
        "tg-delivery",                   # queue name
    )

    async with client:
        await client.send_text(123456789, "Hello!", priority=5)
        await client.send_image(123456789, "https://img/url.jpg", priority=3, text="Caption")
        await client.send_i18n_text(123456789, "greeting.hello", priority=2, i18n_args={"name": "John"})
        await client.send_image_i18n(
            123456789,
            image_url="https://img/url.jpg",
            i18n_key="promo.banner",
            priority=1,
            i18n_args={"discount": 50},
        )

asyncio.run(main())

---

Facade API

Creating a client

from tg_delivery_client.facade import TgDeliveryClient, create_amqp_client
from tg_delivery_client.options import AmqpPublisherConfig

# Helper:
client = await create_amqp_client("amqp://...", "queue_name", config=AmqpPublisherConfig(...))

# Or manually:
from tg_delivery_client.amqp_publisher import AmqpCommandPublisher
sender = AmqpCommandPublisher("amqp://...", "queue_name", cfg=AmqpPublisherConfig())
client = TgDeliveryClient(sender)

TgDeliveryClient methods

async def send_text(
    chat_id: int,
    text: str,
    priority: int,
    *,
    reply_markup: dict[str, object] | None = None,
) -> None

async def send_image(
    chat_id: int,
    image_url: str,
    priority: int,
    *,
    text: str | None = None,
    reply_markup: dict[str, object] | None = None,
) -> None

async def send_i18n_text(
    chat_id: int,
    i18n_key: str,
    priority: int,
    *,
    i18n_args: dict[str, object] | None = None,
    reply_markup: dict[str, object] | None = None,
) -> None

async def send_image_i18n(
    chat_id: int,
    image_url: str,
    i18n_key: str,
    priority: int,
    *,
    reply_markup: dict[str, object] | None = None,
    i18n_args: dict[str, object] | None = None,
) -> None

All methods build a SendCommand and call sender.send() with PublishOptions(priority=...).

---

Direct AMQP publisher usage

import asyncio
from tg_delivery_client.amqp_publisher import AmqpCommandPublisher
from tg_delivery_client.domain import SendCommand
from tg_delivery_client.options import PublishOptions, AmqpPublisherConfig

async def main():
    cfg = AmqpPublisherConfig(ensure_queue=True, max_priority=10)
    pub = AmqpCommandPublisher("amqps://user:pass@host/vhost", "tg-delivery", cfg=cfg)
    await pub.connect()

    cmd = SendCommand(chat_id=1, text="hi")
    await pub.send(cmd, opts=PublishOptions(priority=3, headers={"x-tag": "demo"}))

    await pub.close()

asyncio.run(main())

---

Configuration

AmqpPublisherConfig

from dataclasses import dataclass
from tg_delivery_client.serializers import JsonCommandSerializer
from tg_delivery_client.retry import RetryPolicy

@dataclass(slots=True)
class AmqpPublisherConfig:
    ensure_queue: bool = True                 # declare queue on connect
    max_priority: int = 10                    # x-max-priority argument
    serializer: JsonCommandSerializer = JsonCommandSerializer()
    retry_policy: RetryPolicy = RetryPolicy() # backoff on ConnectionClosed/ChannelClosed

PublishOptions

from dataclasses import dataclass
from typing import Mapping

@dataclass(slots=True)
class PublishOptions:
    priority: int
    headers: Mapping[str, object] | None = None

RetryPolicy

from tg_delivery_client.retry import RetryPolicy

rp = RetryPolicy(
    attempts=5,
    base_delay=0.2,
    max_delay=2.0,
    multiplier=2.0,
    jitter=True,
)
# pass it into AmqpPublisherConfig(retry_policy=rp)

---

Command format

from dataclasses import dataclass, field

@dataclass(slots=True)
class SendCommand:
    chat_id: int
    text: str | None = None
    i18n_key: str | None = None
    i18n_args: dict[str, object] = field(default_factory=dict)
    reply_markup: dict[str, object] | None = None
    image_url: str | None = None

---

Serialization

Default: JSON (JsonCommandSerializer).

from typing import Protocol

class CommandSerializer(Protocol):
    content_type: str
    content_encoding: str
    def serialize(self, cmd: SendCommand) -> bytes: ...

Implement your own and pass it to AmqpPublisherConfig.serializer if needed.

---

Tests & Lint

pytest -q
ruff check .

See pyproject.toml for tool settings.

---

License

MIT. See LICENSE.