Asynchronous calls for long polling Telegram API - A better broken state

async_telegram_polling.py

import asyncio
from typing import Dict
from urllib.parse import urljoin

import aiohttp
from pydantic import HttpUrl
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    bot_token: str
    webhook: HttpUrl

    class Config:
        extra = "allow"
        env_file = ".env"


settings = Settings()

BASE_URL = f"https://api.telegram.org/bot{settings.bot_token}/"


async def fetch_updates(session: aiohttp.ClientSession, payload: Dict[str, int]):
    async with session.post(urljoin(BASE_URL, "getUpdates"), params=payload) as response:
        if response.ok:
            return await response.json()
        else:
            print(await response.json())
            return None


async def main():
    offset = 0
    async with aiohttp.ClientSession() as session:
        major, minor = session.version
        print(f"Polling with v{major}.{minor}")
        while True:
            response = await fetch_updates(session, {'offset': offset, 'timeout': 60})
            if response:
                for result in response.get('result', []):
                    message = result.get('message')
                    await respond(session, {'chat_id': message.get('chat').get('id'), 'text': message.get('text')})
                    offset = result['update_id'] + 1
            else:
                break


async def respond(session, payload):
    async with session.post(url=urljoin(BASE_URL, "sendMessage"), params=payload) as response:
        res = await response.json()
        print(res)


if __name__ == "__main__":
    asyncio.run(main())