taskferry-servicebus

Run Taskferry tasks on Azure Service Bus — pull-based delivery with unrestricted scheduling.

flowchart LR
    APP["Application"] --> TP["Taskferry"] --> AD["taskferry-servicebus"] --> Q["Service Bus queue"] --> C["Azure Functions · Container Apps"]

pip install 'taskferry-servicebus[azure]'

Why it is a separate package from SQS

Both are pull-based queues, and they differ where it counts: scheduled_enqueue_time_utc accepts any future time, so a task deferred by a week is a first-class operation here and impossible on SQS, which caps at 900 seconds.

One "queue adapter with a flag" would have to pick a lowest common denominator. Two adapters with two honest capability sets do not. See ADR-0005.

Sending

runtime = Taskferry.from_mapping(
    {
        "backends": {
            "sb": {
                "factory": "servicebus",
                "connection_string": "Endpoint=sb://...",
                "queue_name": "tasks",
            }
        },
        "defaults": {"task": "sb"},
    }
)

runtime.tasks.submit("myapp.tasks:reindex", 42, delay=timedelta(days=7))

Managed Identity instead of a connection string:

{
    "factory": "servicebus",
    "namespace": "myns.servicebus.windows.net",
    "credential": DefaultAzureCredential(),
    "queue_name": "tasks",
}

Consuming

Azure Functions:

from taskferry import FunctionRegistry
from taskferry_servicebus import process_message

REGISTRY = FunctionRegistry(allowed_modules=["myapp"])


def main(msg: func.ServiceBusMessage) -> None:
    process_message(msg.get_body(), registry=REGISTRY)

Container Apps / a plain process:

from taskferry_servicebus import receive_forever

receive_forever(client, "tasks", registry=REGISTRY)

Completes on success, abandons on failure so Service Bus redelivers and eventually dead-letters per the queue's MaxDeliveryCount.

Capabilities

DEDUPLICATION is real but bounded by the queue's duplicate-detection window — a tool for idempotency, never an exactly-once promise. See ADR-0010.

License

Apache-2.0.