Case — a Telegram bot with an external API

the full cycle, two real bugs

What it is

The first case, and at the same time a model for the rest: the idea, the questions, the mini-spec, the request, the plan, the code, the review, a bug, the fix. Nothing new in technique — all of it has appeared in the book already; what is new is that they come one after another, as they do in life.

The idea in one phrase: a bot that sends the morning weather for your chosen cities.

Step 1. The ten questions

Half an hour with the list from article 13.6. Briefly, the answers only:

The sixth row is the one the list exists for. Without it the bot stays silent one morning and it isn't clear whether it broke or the weather simply didn't change.

Step 2. The request

How not to do it:

Write a telegram bot in python that shows the weather.

What arrives: one file, a synchronous requests inside an async handler, the API key as a string in the code, the city list in a global variable. It will work — right up to the first restart.

How to do it (and this is the mini-spec from article 3.15):

A bot on aiogram 3, Python 3.12. Storage: SQLite.
Features: /add <city>, /list, /remove <city>, a morning digest.

Structure:
bot/handlers/  only input parsing and the reply
bot/services/  logic and the call to the weather API
bot/storage.py all database access, nowhere else
bot/config.py  settings via pydantic-settings

Rules: handlers don't talk to the database directly;
no new libraries without my agreement;
the API key only from an environment variable.

Write the plan first: which files and what's in each.
Code after I say yes.

Those last two lines are the technique from article 3.10. The plan reads in a minute, and rewriting it is cheaper than rewriting code.

Step 3. Storage and the first bug

Start at the bottom, with the data layer (article 13.1). Here is what arrived:

def connect():
    db = sqlite3.connect(DB)
    db.execute("""
        CREATE TABLE IF NOT EXISTS subscriptions (
            user_id INTEGER NOT NULL,
            city    TEXT    NOT NULL
        )
    """)
    db.commit()
    return db


def add_city(db, user_id: int, city: str) -> None:
    db.execute("INSERT INTO subscriptions VALUES (?, ?)", (user_id, city))
    db.commit()

The code reads well, works, and looks right. Write the tests — not for appearances but by the list from article 3.9: the ordinary case, the edges, the thing people forget.

def test_add_and_list(db):
    storage.add_city(db, 1, "Oslo")
    assert storage.list_cities(db, 1) == ["Oslo"]


def test_cities_are_per_user(db):
    storage.add_city(db, 1, "Oslo")
    storage.add_city(db, 2, "Bergen")
    assert storage.list_cities(db, 1) == ["Oslo"]


def test_same_city_twice(db):
    storage.add_city(db, 1, "Oslo")
    storage.add_city(db, 1, "Oslo")
    assert storage.list_cities(db, 1) == ["Oslo"], "the city shouldn't be added twice"

Running it:

E       AssertionError: the city shouldn't be added twice
E       assert ['Oslo', 'Oslo'] == ['Oslo']
E         Left contains one more item: 'Oslo'

1 failed, 3 passed in 0.03s

The third test failed. Add a city twice and you get two subscriptions, which means two messages every morning. You don't find that by hand: who would add the same city twice? A user would — a month later, having forgotten they already had.

The fix goes where the cause is, not where the symptom is. Not a check in the code before the insert (which you can forget to call from somewhere else) but a rule in the table itself:

            user_id INTEGER NOT NULL,
            city    TEXT    NOT NULL,
            UNIQUE (user_id, city)

    db.execute("INSERT OR IGNORE INTO subscriptions VALUES (?, ?)",
               (user_id, city))

4 passed in 0.02s

Step 4. The external API and the second bug

The first version of the weather call looked like this:

def fetch(city: str) -> float:
    r = requests.get(API_URL, params={"q": city, "appid": settings.api_key})
    return r.json()["main"]["temp"]

Two problems in two lines, and both typical (article 2.14): parsing the response without checking — and no timeout.

Let's check the second. Start a local server that has "hung" — it accepts the request and says nothing:

class Slow(BaseHTTPRequestHandler):
    def do_GET(self):
        time.sleep(30)
        ...

no timeout:     still hanging after 12 s  (we aborted it ourselves)
with timeout=2: requests.Timeout after 2.0 s

Twelve seconds is how long we agreed to wait before aborting the check. A real hung server holds the connection as long as it likes, and the whole morning digest hangs with it: one unreachable city and the other subscribers get nothing.

requests has no default timeout. None at all. That is a property of the library an AI remembers about half the time.

The corrected version — with a timeout, with a check, and with the answer to "what do we do if it doesn't respond" from the mini-spec:

def fetch(city: str) -> float | None:
    try:
        r = requests.get(API_URL, params={...}, timeout=5)
        r.raise_for_status()
    except requests.RequestException:
        return None                      # the caller shows the previous data
    data = r.json()
    main = data.get("main")
    if not isinstance(main, dict) or "temp" not in main:
        return None
    return main["temp"]

Checked against the local server — a real city, an invented city, and the server switched off:

Oslo            -> 4.2
Madeupville     -> None
server is down  -> None

Not one crash: the function has an answer for all three cases.

One detail is left, and it is the very one step 2 started with: fetch is synchronous and the bot is asynchronous. Called straight from a handler or the digest, it stops the whole bot for the length of the request — nobody gets a reply until the weather arrives. So it is called in a separate thread:

temp = await asyncio.to_thread(fetch, city)

The other route is to rewrite fetch with aiohttp. Both are honest; what matters is that a synchronous request never sits inside an async def without such a wrapper.

Step 5. What this case doesn't have

What was run in this case is the data layer (four tests) and the behaviour against a hung server (a local server, real measurements). The handlers and a run against the real Telegram and a real weather API are not shown here: that is the next step, and it is done with the checklist from article 6.6, not on trust.

That is exactly the honesty the book demands of an AI (article 1.6), and it applies to itself the same way.

Cheat sheet