Coverage for api/services/telegram.py: 32%

1import asyncio 

2import logging 

3 

4import aiohttp 

5from config import APP_NAME, ENVIRONMENT, TelegramConfig 

6 

7 

8class GROUP_TYPE: 

9 ALERT = "alert" 

10 INFO = "info" 

11 SUCCESS = "success" 

12 ERROR = "error" 

13 

14 

15def map_environment_header(mode: str): 

16 environment = ENVIRONMENT 

17 app_name = APP_NAME 

18 environment_emoji = { 

19 "prod": "☠🔥☠", 

20 "release": "🌊", 

21 "test": "🐞", 

22 "dev": "🟡" if mode == GROUP_TYPE.INFO else "🔴", 

23 }.get(environment, "🛋️") 

24 return f"{environment_emoji} {app_name} | {environment.upper()} {environment_emoji}\n\n" 

25 

26 

27# migrated from blocking code 

28def send_message(message: str, level: int, mode: str = GROUP_TYPE.ALERT): 

29 if ENVIRONMENT == "local": 

30 return 

31 if level < TelegramConfig.LOG_LEVEL: 

32 return 

33 if mode == GROUP_TYPE.INFO: 

34 chat_id = TelegramConfig.INFO_CHAT_ID 

35 else: 

36 chat_id = TelegramConfig.ALERT_CHAT_ID 

37 text_with_meta = map_environment_header(mode) + message 

38 asyncio.create_task(async_send_message(chat_id, text_with_meta)) 

39 

40 

41async def async_send_message(chat_id: int, message: str): 

42 async with aiohttp.ClientSession() as s: 

43 async with s.post( 

44 f"https://api.telegram.org/bot{TelegramConfig.BOT_TOKEN}/sendMessage", 

45 data={"chat_id": chat_id, "text": message[:4096]}, 

46 ) as resp: 

47 if not 200 <= resp.status < 400: 

48 logging.error( 

49 "Failed to send message [%s] to telegram chat", message 

50 ) 

51 

52 

53def group_notification( 

54 message: str, level: int = logging.INFO, mode=GROUP_TYPE.ALERT 

55): 

56 send_message(message, level, mode)