Make telegram bot working async. Add async fixture and tests.

Author: leddcode

app/main.py

@@ -5,7 +5,7 @@
 from app.database.database import engine
 from app.dependencies import (
     MEDIA_PATH, STATIC_PATH, templates)
-from app.telegram.pylander import pylander
+from app.telegram.bot import telegram_bot
 from app.routers import agenda, email, event, invitation, profile, telegram
 
 models.Base.metadata.create_all(bind=engine)
@@ -21,7 +21,7 @@
 app.include_router(email.router)
 app.include_router(invitation.router)
 
-pylander.set_webhook()
+telegram_bot.set_webhook()
 
 
 @app.get("/")

app/routers/telegram.py

@@ -21,13 +21,13 @@ async def telegram(request: Request, session=Depends(get_db)):
 
 
 @router.post("/")
-def bot_client(req: dict = Body(...), session=Depends(get_db)):
+async def bot_client(req: dict = Body(...), session=Depends(get_db)):
     chat = Chat(req)
 
     # Check if current chatter is registered to use the bot
     user = session.query(User).filter_by(telegram_id=chat.user_id).first()
     if user is None:
-        return reply_unknown_user(chat)
+        return await reply_unknown_user(chat)
 
     message = MessageHandler(chat, user)
-    return message.process_callback()
+    return await message.process_callback()

app/telegram/pylander.py renamed to app/telegram/bot.py

@@ -8,4 +8,4 @@
 BOT_API = settings.bot_api
 WEBHOOK_URL = settings.webhook_url
 
-pylander = Bot(BOT_API, WEBHOOK_URL)
+telegram_bot = Bot(BOT_API, WEBHOOK_URL)

app/telegram/handlers.py

@@ -3,7 +3,7 @@
 from .keyboards import (
     DATE_FORMAT, gen_inline_keyboard, get_this_week_buttons, show_events_kb)
 from .models import Chat
-from .pylander import pylander
+from .bot import telegram_bot
 from app.database.models import User
 
 
@@ -22,31 +22,31 @@ def __init__(self, chat: Chat, user: User):
             for button in row:
                 self.handlers[button['text']] = self.chosen_day_handler
 
-    def process_callback(self):
+    async def process_callback(self):
         if self.chat.message in self.handlers:
-            return self.handlers[self.chat.message]()
-        return self.default_handler()
+            return await self.handlers[self.chat.message]()
+        return await self.default_handler()
 
-    def default_handler(self):
+    async def default_handler(self):
         answer = "Unknown command."
-        pylander.send_message(chat_id=self.chat.user_id, text=answer)
+        await telegram_bot.send_message(chat_id=self.chat.user_id, text=answer)
         return answer
 
-    def start_handler(self):
+    async def start_handler(self):
         answer = f'''Hello, {self.chat.first_name}!
 Welcome to Pylander telegram client!'''
-        pylander.send_message(chat_id=self.chat.user_id, text=answer)
+        await telegram_bot.send_message(chat_id=self.chat.user_id, text=answer)
         return answer
 
-    def show_events_handler(self):
+    async def show_events_handler(self):
         answer = 'Choose events day.'
-        pylander.send_message(
+        await telegram_bot.send_message(
             chat_id=self.chat.user_id,
             text=answer,
             reply_markup=show_events_kb)
         return answer
 
-    def today_handler(self):
+    async def today_handler(self):
         today = datetime.date.today()
         events = [
             event for event in self.user.events
@@ -60,20 +60,20 @@ def today_handler(self):
         for event in events:
             answer += f'\n\n{event.title}: from {event.start} to {event.ends}.'
 
-        pylander.send_message(chat_id=self.chat.user_id, text=answer)
+        await telegram_bot.send_message(chat_id=self.chat.user_id, text=answer)
         return answer
 
-    def this_week_handler(self):
+    async def this_week_handler(self):
         answer = 'Choose a day.'
         this_week_kb = gen_inline_keyboard(get_this_week_buttons())
 
-        pylander.send_message(
+        await telegram_bot.send_message(
             chat_id=self.chat.user_id,
             text=answer,
             reply_markup=this_week_kb)
         return answer
 
-    def chosen_day_handler(self):
+    async def chosen_day_handler(self):
         # Convert chosen day (string) to datetime format
         chosen_date = datetime.datetime.strptime(
             self.chat.message, DATE_FORMAT)
@@ -85,17 +85,17 @@ def chosen_day_handler(self):
         answer = f"{chosen_date.strftime('%B %d')}, \
 {chosen_date.strftime('%A')} Events:\n"
 
-        if not len(events):
+        if not events:
             answer = f"There're no events on {chosen_date.strftime('%B %d')}."
 
         for event in events:
             answer += f'\n\n{event.title}: from {event.start} to {event.ends}.'
 
-        pylander.send_message(chat_id=self.chat.user_id, text=answer)
+        await telegram_bot.send_message(chat_id=self.chat.user_id, text=answer)
         return answer
 
 
-def reply_unknown_user(chat):
+async def reply_unknown_user(chat):
     answer = f'''
 Hello, {chat.first_name}!
 
@@ -107,5 +107,5 @@ def reply_unknown_user(chat):
 
 https://calendar.pythonic.guru/profile/
 '''
-    pylander.send_message(chat_id=chat.user_id, text=answer)
+    await telegram_bot.send_message(chat_id=chat.user_id, text=answer)
     return answer

app/telegram/keyboards.py

@@ -15,29 +15,27 @@
 
 def get_this_week_buttons() -> List[List[Any]]:
     today = datetime.date.today()
-    day1 = today + datetime.timedelta(days=1)
-    day2 = today + datetime.timedelta(days=2)
-    day3 = today + datetime.timedelta(days=3)
-    day4 = today + datetime.timedelta(days=4)
-    day5 = today + datetime.timedelta(days=5)
-    day6 = today + datetime.timedelta(days=6)
+    buttons = []
+    for day in range(1, 7):
+        day = today + datetime.timedelta(days=day)
+        buttons.append(day.strftime(DATE_FORMAT))
 
     return [
         [
-            {'text': day1.strftime(DATE_FORMAT),
-                'callback_data': day1.strftime(DATE_FORMAT)},
-            {'text': day2.strftime(DATE_FORMAT),
-                'callback_data': day2.strftime(DATE_FORMAT)},
-            {'text': day3.strftime(DATE_FORMAT),
-                'callback_data': day3.strftime(DATE_FORMAT)}
+            {'text': buttons[0],
+                'callback_data': buttons[0]},
+            {'text': buttons[1],
+                'callback_data': buttons[1]},
+            {'text': buttons[2],
+                'callback_data': buttons[2]}
         ],
         [
-            {'text': day4.strftime(DATE_FORMAT),
-                'callback_data': day4.strftime(DATE_FORMAT)},
-            {'text': day5.strftime(DATE_FORMAT),
-                'callback_data': day5.strftime(DATE_FORMAT)},
-            {'text': day6.strftime(DATE_FORMAT),
-                'callback_data': day6.strftime(DATE_FORMAT)}
+            {'text': buttons[3],
+                'callback_data': buttons[3]},
+            {'text': buttons[4],
+                'callback_data': buttons[4]},
+            {'text': buttons[5],
+                'callback_data': buttons[5]}
         ]
     ]
 

app/telegram/models.py

@@ -4,22 +4,22 @@
 
 
 class Chat:
-    def __init__(self, data):
+    def __init__(self, data: Dict):
         self.message = self._get_message_content(data)
         self.user_id = self._get_user_id(data)
         self.first_name = self._get_first_name(data)
 
-    def _get_message_content(self, data):
+    def _get_message_content(self, data: Dict) -> str:
         if 'callback_query' in data:
             return data['callback_query']['data']
         return data['message']['text']
 
-    def _get_user_id(self, data):
+    def _get_user_id(self, data: Dict) -> str:
         if 'callback_query' in data:
             return data['callback_query']['from']['id']
         return data['message']['from']['id']
 
-    def _get_first_name(self, data):
+    def _get_first_name(self, data: Dict) -> str:
         if 'callback_query' in data:
             return data['callback_query']['from']['first_name']
         return data['message']['from']['first_name']
@@ -30,23 +30,20 @@ def __init__(self, bot_api: str, webhook_url: str):
         self.base = self._set_base_url(bot_api)
         self.webhook_setter_url = self._set_webhook_setter_url(webhook_url)
 
-    def _set_base_url(self, bot_api):
+    def _set_base_url(self, bot_api: str) -> str:
         return f'https://api.telegram.org/bot{bot_api}/'
 
-    def _set_webhook_setter_url(self, webhook_url: str):
+    def _set_webhook_setter_url(self, webhook_url: str) -> str:
         return f'{self.base}setWebhook?url={webhook_url}/telegram/'
 
     def set_webhook(self):
         return requests.get(self.webhook_setter_url)
 
     def drop_webhook(self):
-        data = {
-            'drop_pending_updates': True
-        }
-        return requests.get(
-            url=f'{self.base}deleteWebhook', data=data)
+        data = {'drop_pending_updates': True}
+        return requests.get(url=f'{self.base}deleteWebhook', data=data)
 
-    def send_message(
+    async def send_message(
             self, chat_id: str,
             text: str,
             reply_markup: Optional[Dict[str, Any]] = None):

tests/asyncio_fixture.py

@@ -0,0 +1,17 @@
+from httpx import AsyncClient
+import pytest
+
+from app.database.database import Base
+from app.main import app
+from app.routers import telegram
+from tests.conftest import test_engine, get_test_db
+
+
+@pytest.fixture
+async def telegram_client():
+    Base.metadata.create_all(bind=test_engine)
+    app.dependency_overrides[telegram.get_db] = get_test_db
+    async with AsyncClient(app=app, base_url="http://test") as ac:
+        yield ac
+    app.dependency_overrides = {}
+    Base.metadata.drop_all(bind=test_engine)

tests/client_fixture.py

@@ -4,7 +4,7 @@
 from app.database.models import User
 from app.main import app
 from app.database.database import Base
-from app.routers import agenda, invitation, profile, telegram
+from app.routers import agenda, invitation, profile
 from tests.conftest import test_engine, get_test_db
 
 
@@ -59,15 +59,3 @@ def get_test_placeholder_user():
         full_name='FakeName',
         telegram_id='666666'
     )
-
-
-@pytest.fixture(scope="module")
-def telegram_client():
-    Base.metadata.create_all(bind=test_engine)
-    app.dependency_overrides[telegram.get_db] = get_test_db
-
-    with TestClient(app) as client:
-        yield client
-
-    app.dependency_overrides = {}
-    Base.metadata.drop_all(bind=test_engine)

tests/conftest.py

@@ -10,6 +10,7 @@
     'tests.invitation_fixture',
     'tests.association_fixture',
     'tests.client_fixture',
+    'tests.asyncio_fixture',
     'smtpdfix',
 ]