Кейс — телеграм-бот с внешним API

полный цикл, два настоящих бага

Что это

Первый кейс — и заодно образец того, как устроены остальные: идея, вопросы, мини-ТЗ, запрос, план, код, ревью, баг, починка. Ничего нового по приёмам, всё уже было в книге; новое — то, что они идут подряд, как в жизни.

Идея в одну фразу: бот, который присылает утром погоду в выбранных городах.

Шаг 1. Десять вопросов

Полчаса по списку из статьи 13.6. Коротко, только ответы:

Шестая строчка — та самая, ради которой список и нужен. Без неё бот однажды промолчит утром, и будет непонятно, он сломался или погода не изменилась.

Шаг 2. Запрос

Как не надо:

Напиши телеграм-бота на питоне, который показывает погоду.

Что придёт: один файл, синхронный requests внутри асинхронного обработчика, ключ API строкой в коде, список городов в глобальной переменной. Работать будет — ровно до перезапуска.

Как надо (это и есть мини-ТЗ из статьи 3.15):

Бот на aiogram 3, Python 3.12. Хранение — SQLite.
Функции: /add <город>, /list, /remove <город>, утренняя рассылка.

Структура:
bot/handlers/  только разбор ввода и ответ
bot/services/  логика и поход в API погоды
bot/storage.py весь доступ к базе, больше нигде
bot/config.py  настройки через pydantic-settings

Правила: handlers не обращаются к базе напрямую;
никаких новых библиотек без моего согласия;
ключ API только из переменной окружения.

Сначала напиши план: какие файлы и что в каждом.
Код после моего «да».

Последние две строки — приём из статьи 3.10. План читается минуту, а переписывать его дешевле, чем код.

Шаг 3. Хранилище и первый баг

Начинаем снизу, со слоя данных (статья 13.1). Вот что пришло:

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()

Код читается, работает и выглядит правильным. Пишем тесты — не для галочки, а по списку из статьи 3.9: обычный случай, границы, то, что забывают.

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"], "город не должен добавиться дважды"

Запуск:

E       AssertionError: город не должен добавиться дважды
E       assert ['Oslo', 'Oslo'] == ['Oslo']
E         Left contains one more item: 'Oslo'

1 failed, 3 passed in 0.05s

Третий тест упал. Добавил город дважды — получил две подписки, а значит, и два сообщения каждое утро. Руками это не находится: кто же добавляет один и тот же город дважды? Пользователь добавляет — через месяц, забыв, что уже добавлял.

Починка — там, где причина, а не там, где симптом. Не проверка в коде перед вставкой (её можно забыть вызвать из другого места), а правило в самой таблице:

            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.04s

Шаг 4. Внешний API и второй баг

Поход за погодой в первом варианте выглядел так:

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

Две проблемы в двух строках, и обе типичные (статья 2.14): разбор ответа без проверки — и нет таймаута.

Проверим второе. Поднимем местный сервер, который «подвис» — принимает запрос и молчит:

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

без timeout: всё ещё висит через 12 с  (прервали сами)
с timeout=2: requests.Timeout через 2.0 с

Двенадцать секунд — это сколько мы согласились ждать, прежде чем прервать проверку. Настоящий зависший сервер держит соединение сколько угодно, и вместе с ним висит вся утренняя рассылка: один недоступный город — и остальные подписчики не получают ничего.

У requests таймаута по умолчанию нет. Ни одного. Это свойство библиотеки, о котором ИИ вспоминает через раз.

Исправленная версия — с таймаутом, с проверкой и с ответом на вопрос «что делать, если не ответил» из мини-ТЗ:

def fetch(city: str) -> float | None:
    try:
        r = requests.get(API_URL, params={...}, timeout=5)
        r.raise_for_status()
    except requests.RequestException:
        return None                      # вызывающий покажет прошлые данные
    data = r.json()
    main = data.get("main")
    if not isinstance(main, dict) or "temp" not in main:
        return None
    return main["temp"]

Проверено на местном сервере — настоящий город, выдуманный город и выключенный сервер:

Oslo        -> 4.2
Выдуманск   -> None
сервера нет -> None

Ни одного падения: на все три случая у функции есть ответ.

Осталась одна деталь, и она та самая, с которой начинался шаг 2: fetch синхронная, а бот асинхронный. Вызванная напрямую из хендлера или рассылки, она на время запроса остановит весь бот — никто не получит ответа, пока не придёт погода. Поэтому вызывается она в отдельном потоке:

temp = await asyncio.to_thread(fetch, city)

Другой путь — переписать fetch на aiohttp. Оба честные; главное, чтобы синхронный запрос не оказался внутри async def без такой обёртки.

Шаг 5. Чего в этом кейсе нет

В кейсе прогонялись слой данных (четыре теста) и поведение при зависшем сервере (местный сервер, настоящие замеры). Обработчики и запуск против настоящего Telegram и настоящего API погоды здесь не показаны: это следующий шаг, и делается он по чек-листу из статьи 6.6, а не на веру.

Это ровно та честность, которой книга требует от ИИ (статья 1.6), и к себе она применяется так же.

Шпаргалка