A template for chat bots that hold no business logic.
The second messenger is what tells you whether the first one was built right. If adding it means copying a validation rule, that rule now has two homes and will eventually disagree with itself.
client core
────── ────
in 10m walk the dog ──POST─────────▶ validate, store
/list ──GET──────────▶ read
/cancel 3 ──DELETE───────▶ check ownership
a long-polling client has no address the core can call
back on. so the core queues, and the client asks again:
anything for me? ──GET /outbox──▶ pending messages
sent it ──POST ────────▶ acknowledged
docker compose upThat builds the core and runs a client that needs no credentials, which walks one scenario and narrates it:
→ asking the core for a reminder three seconds out
created #1: take the bread out
→ asking for one in the past
refused: That time has already passed.
the client did not check the date. the core did.
→ what is pending for this chat
#1 at 10:18:07 — take the bread out
→ polling the outbox, the way a bot on long polling has to
delivered to demo-chat: take the bread out
→ nothing left to deliver
outbox: 0 pending
The API is on http://localhost:8000, with Swagger at /docs. If that port is
taken, use THINBOT_PORT=8099 docker compose up.
To run the real Telegram client alongside it:
THINBOT_TELEGRAM_TOKEN=... docker compose --profile telegram upA client may not import the core. It talks over HTTP or not at all.
This is not a style preference, it is what the structure rests on. A client that can import the core can read the database, skip a check, or grow its own copy of one — and the copy is the part that hurts, because it drifts silently until two channels answer the same question differently.
So it is a test, not a paragraph:
@pytest.mark.parametrize("path", client_modules(), ids=lambda p: p.name)
def test_a_client_never_imports_the_core(path: Path):
reached_past_the_api = {
name for name in imported_names(path) if name.startswith("thinbot.core")
}
assert not reached_past_the_apiA second test asserts the first one is really looking at every client module, so a rename cannot quietly empty it.
src/thinbot/
├── core/
│ ├── service.py every rule. the only file that decides anything
│ ├── api.py routes: parse, delegate, shape the answer
│ ├── models.py tables
│ ├── clock.py the one place time crosses the storage boundary
│ └── app.py wiring, and the scheduler that fills the outbox
├── sdk/ typed HTTP client — the wire format, in one place
└── clients/
├── delivery.py the outbox loop, shared by every channel
├── telegram.py aiogram. wording and buttons, nothing else
└── demo.py the same calls, with no credentials needed
service.py is where an argument gets settled: how long a reminder may be,
whether a date is allowed, who is permitted to cancel what. Routes and clients
only carry the answer.
Write a module in clients/, import the SDK, and hand deliver_forever a
function that knows how to send one message on your platform. That function is
the only new code allowed to be platform-specific.
async def send(chat_id: str, text: str) -> None:
await my_platform.post_message(chat_id, text)
await deliver_forever(core, send)Pick a channel name and pass it to CoreClient. The core separates chats,
reminders and outbox rows by it, so two channels cannot read or cancel each
other's things.
A message is acknowledged only after the send returns, so a client that dies mid-delivery retries instead of dropping it. The cost is that a message can arrive twice; the alternative loses it entirely, and for a reminder that is the worse failure.
queued_at is what makes the scheduler safe to run on a timer: a reminder
leaves the pool the moment it is queued, so an overlapping tick finds nothing
to send again.
Reminders are here to give the structure something to carry — replace them.
What is worth keeping is service.py holding every rule, the SDK being the
only thing that knows the wire format, the outbox, and the test that stops a
client reaching past the API.
pip install -e ".[dev]"
pytest