tketool.core

tketool.core contains the shared configuration, file, cache, command, in-memory async messaging, and utility APIs used by the other tketool distributions.

pip install tketool.core

from tketool.core import ConfigManager, read_file, write_file
from tketool.core.cache.sqlite import init_db

Optional dependencies are grouped by feature:

pip install "tketool.core[storage]"
pip install "tketool.core[documents]"
pip install "tketool.core[console]"

In-memory async messaging

tketool.core.messaging provides a small replaceable publish/subscribe contract and a dependency-free in-memory adapter:

import asyncio

from tketool.core.messaging import create_message_bus


async def main() -> None:
    bus = create_message_bus()

    async def handle_order(message) -> None:
        print(message.message_id, message.payload)

    subscription = await bus.subscribe(
        "orders.created",
        handle_order,
        max_queue_size=100,
        handler_timeout=5.0,
    )

    receipt = await bus.publish("orders.created", {"order_id": "o-1"})
    assert receipt.subscriber_count == 1
    await bus.wait_idle()

    await subscription.close()
    await bus.close()


asyncio.run(main())

One subscription processes messages serially in FIFO order; different subscriptions run concurrently. A successful publish only confirms that the message was accepted by the subscriber queues. Handler failures are available through Subscription.last_error and an optional on_error callback.

This adapter is process-local and bound to one asyncio event loop. It does not persist messages, share work across processes, retry failures, or provide exactly-once delivery. Mutable payloads are passed by reference and should be treated as read-only. Use a durable broker adapter for work that must survive a restart.

Lifecycle and error handling

Use the bus as an async context manager when its lifetime matches one application scope:

import asyncio

from tketool.core.messaging import HandlerError, create_message_bus


async def report_error(error: HandlerError) -> None:
    print(error.message.message_id, error.exception)


async def process_document(message) -> None:
    print(message.payload["document_id"])


async def main() -> None:
    async with create_message_bus() as bus:
        await bus.subscribe(
            "documents.ready",
            process_document,
            handler_timeout=10.0,
            on_error=report_error,
        )
        await bus.publish("documents.ready", {"document_id": "d-1"})
        await bus.wait_idle()


asyncio.run(main())

publish() waits only for subscriber queues to accept the message. It returns a PublishReceipt; callback completion is observed with wait_idle(). Exceptions and async callback timeouts increment Subscription.failed_count, set Subscription.last_error, and invoke on_error. They are not sent back to the publisher and are not retried automatically.

close(drain=True) stops accepting new work and finishes accepted messages. close(drain=False) drops queued messages and cancels subscription workers. Synchronous callbacks are supported, but run on the event-loop thread and must not block; explicitly use asyncio.to_thread for blocking functions.

Delivery model

If several workers should compete for one message instead of all receiving it, that is a work-queue/competing-consumer contract and is intentionally separate from this publish/subscribe API.

Verification and design documents

bash scripts/test-package.sh core

• Architecture and reliability semantics
• Architecture decision record
• Open-source reuse audit