Fix pyright & Add pyright lint & Reformat all files

Author: coder8080

Dockerfile.lint

@@ -2,9 +2,9 @@ FROM pypy:3
 
 WORKDIR /usr/src/app
 
-RUN pip install ruff
+RUN pip install ruff pyright
 COPY requirements.txt .
 RUN pip install -r requirements.txt --no-cache-dir
 COPY . .
 
-CMD ["ruff", "check", "--extend-select", "I"]
+CMD ruff check --extend-select I && pyright

src/entities/__init__.py

@@ -4,6 +4,17 @@
 from .logs import logger
 from .utils import get_keyboard, get_update_user_info, log_error
 
-__all__ = ["TOKEN", "POSTGRES_USER", "POSTGRES_PASSWORD", "ADMINS", "TEXT",
-           "States", "get_update_user_info", "storage", "bot",
-           "log_error", "get_keyboard", "logger"]
+__all__ = [
+    "TOKEN",
+    "POSTGRES_USER",
+    "POSTGRES_PASSWORD",
+    "ADMINS",
+    "TEXT",
+    "States",
+    "get_update_user_info",
+    "storage",
+    "bot",
+    "log_error",
+    "get_keyboard",
+    "logger",
+]

src/entities/constants.py

@@ -1,9 +1,12 @@
 from os import getenv
+from typing import cast
+
+TOKEN = cast(str, getenv("TOKEN"))
+assert TOKEN
 
-TOKEN = getenv("TOKEN")
 POSTGRES_USER = getenv("POSTGRES_USER")
 POSTGRES_PASSWORD = getenv("POSTGRES_PASSWORD")
-ADMINS = list(map(int, getenv("ADMINS").split(',')))
+ADMINS = list(map(int, (getenv("ADMINS") or "").split(",")))
 BOT_NAME = getenv("BOT_NAME")
 
 TEXT = {
@@ -44,7 +47,7 @@
 сколько угодно сертификатов",
     "user-count": "У вас осталось сертификатов: {%COUNT%}",
     "cancel": "❌ Отмена",
-    "create-link-cancelled": '👌 Создание ссылки отменено',
+    "create-link-cancelled": "👌 Создание ссылки отменено",
     "create-cert-cancelled": "👌 Создание сертификата отменено",
-    "check-cert-cancelled": "👌 Проверка сертификата отменена"
+    "check-cert-cancelled": "👌 Проверка сертификата отменена",
 }

src/entities/logs.py

@@ -7,12 +7,15 @@
 debug_file = logging.FileHandler("logs/debug.log")
 error_file = logging.FileHandler("logs/error.log")
 
+
 console_handler.setLevel(logging.DEBUG)
 debug_file.setLevel(logging.DEBUG)
 error_file.setLevel(logging.ERROR)
 
-detailed_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - \
-[%(pathname)s:%(lineno)d] - %(message)s')
+detailed_format = logging.Formatter(
+    "%(asctime)s - %(name)s - %(levelname)s - \
+[%(pathname)s:%(lineno)d] - %(message)s"
+)
 
 console_handler.setFormatter(detailed_format)
 debug_file.setFormatter(detailed_format)

src/entities/utils.py

@@ -1,4 +1,7 @@
-from aiogram.types import ReplyKeyboardMarkup, Update
+from aiogram.types import Message, ReplyKeyboardMarkup
+from aiogram.types.base import TelegramObject
+from aiogram.types.callback_query import CallbackQuery
+from aiogram.types.pre_checkout_query import PreCheckoutQuery
 from aiogram.utils.keyboard import ReplyKeyboardBuilder
 
 from entities import ADMINS, TEXT
@@ -10,17 +13,17 @@
 
 def admin_markup() -> ReplyKeyboardMarkup:
     builder = ReplyKeyboardBuilder()
-    builder.button(text=TEXT['create-link'])
-    builder.button(text=TEXT['create-cert'])
-    builder.button(text=TEXT['check-cert'])
+    builder.button(text=TEXT["create-link"])
+    builder.button(text=TEXT["create-cert"])
+    builder.button(text=TEXT["check-cert"])
     builder.adjust(1)
     return builder.as_markup(resize_keyboard=True)
 
 
 def user_markup() -> ReplyKeyboardMarkup:
     builder = ReplyKeyboardBuilder()
-    builder.button(text=TEXT['create-cert'])
-    builder.button(text=TEXT['check-count'])
+    builder.button(text=TEXT["create-cert"])
+    builder.button(text=TEXT["check-count"])
     builder.adjust(1)
     return builder.as_markup(resize_keyboard=True)
 
@@ -31,18 +34,17 @@ def get_keyboard(user: User):
     return user_markup()
 
 
-def get_update_user_info(update: Update):
+def get_update_user_info(update: TelegramObject):
     chat_id = 0
     username = ""
-    if update.event_type == "message":
-        chat_id = update.message.chat.id
-        username = update.message.from_user.username
-    elif update.event_type == "callback_query":
-        chat_id = update.callback_query.from_user.id
-        username = update.callback_query.from_user.username
-    elif update.event_type == "pre_checkout_query":
-        chat_id = update.pre_checkout_query.from_user.id
-        username = update.pre_checkout_query.from_user.username
+    if (
+        isinstance(update, Message) or isinstance(update, PreCheckoutQuery)
+    ) and update.from_user:
+        chat_id = update.from_user.id
+        username = update.from_user.username
+    elif isinstance(update, CallbackQuery):
+        chat_id = update.from_user.id
+        username = update.from_user.username
 
     if username is None:
         username = f"unknown:${chat_id}"
@@ -53,5 +55,4 @@ def get_update_user_info(update: Update):
 async def log_error(message: str):
     logger.error(message)
     for admin_chat_id in ADMINS:
-        await bot.send_message(chat_id=admin_chat_id,
-                               text=message)
+        await bot.send_message(chat_id=admin_chat_id, text=message)

src/models/cert.py

@@ -6,7 +6,7 @@
 
 
 def cert_code():
-    return ''.join([str(randint(1, 9)) for _ in range(8)])
+    return "".join([str(randint(1, 9)) for _ in range(8)])
 
 
 class Cert(BaseModel):

src/models/db.py

@@ -2,11 +2,13 @@
 
 from entities import POSTGRES_PASSWORD, POSTGRES_USER
 
-db = peewee_async.PsycopgDatabase('usmile',
-                                  user=POSTGRES_USER,
-                                  password=POSTGRES_PASSWORD,
-                                  host="postgres",
-                                  port=5432)
+db = peewee_async.PsycopgDatabase(
+    "usmile",
+    user=POSTGRES_USER,
+    password=POSTGRES_PASSWORD,
+    host="postgres",
+    port=5432,
+)
 
 
 class BaseModel(peewee_async.AioModel):

src/models/user.py

@@ -8,6 +8,9 @@ class User(BaseModel):
     username = TextField()
     count = IntegerField(default=0)
 
+    def info(self):
+        return f"{self.username}:{self.chat_id}"
+
 
 def init_user():
     db.create_tables([User])

src/routers/check_cert.py

@@ -1,8 +1,12 @@
+from typing import cast
+
 from aiogram import F, Router
 from aiogram.filters import Command, StateFilter
 from aiogram.fsm.context import FSMContext
 from aiogram.types import CallbackQuery, Message, ReplyKeyboardRemove
+from aiogram.types.inaccessible_message import InaccessibleMessage
 from aiogram.utils.keyboard import InlineKeyboardBuilder
+from playhouse.cockroachdb import BooleanField
 
 from entities import ADMINS, TEXT, States, get_keyboard
 from models import Cert, User
@@ -11,39 +15,43 @@
 
 
 @router.message(StateFilter(None), Command("check"), F.chat.id.in_(ADMINS))
-@router.message(StateFilter(None), F.text == TEXT['check-cert'],
-                F.chat.id.in_(ADMINS))
+@router.message(
+    StateFilter(None), F.text == TEXT["check-cert"], F.chat.id.in_(ADMINS)
+)
 async def check_start(message: Message, state: FSMContext):
     await state.set_state(States.TypingCert)
-    await message.answer(TEXT['input-cert'],
-                         reply_markup=ReplyKeyboardRemove())
+    await message.answer(TEXT["input-cert"], reply_markup=ReplyKeyboardRemove())
 
 
 @router.message(StateFilter(States.TypingCert), Command("cancel"))
 async def cancel(message: Message, state: FSMContext, user: User):
     await state.set_state(None)
-    await message.answer(TEXT['check-cert-cancelled'],
-                         reply_markup=get_keyboard(user))
+    await message.answer(
+        TEXT["check-cert-cancelled"], reply_markup=get_keyboard(user)
+    )
 
 
 @router.message(StateFilter(States.TypingCert), F.chat.id.in_(ADMINS))
 async def check_cert(message: Message, state: FSMContext, user: User):
+    if not message.text:
+        raise Exception(f"Message without text {user.info()}")
+
     await state.set_state(None)
 
     markup = get_keyboard(user)
     query = Cert.select().where(Cert.code == message.text.strip())
 
-    if not await query.aio_exists():
-        await message.answer(TEXT['no-cert'], reply_markup=markup)
+    if not await query.aio_exists():  # type: ignore
+        await message.answer(TEXT["no-cert"], reply_markup=markup)
         return
 
-    cert = await query.aio_first()
+    cert = await query.aio_first()  # type: ignore
 
     if cert.used:
-        await message.answer(TEXT['used-cert'], reply_markup=markup)
+        await message.answer(TEXT["used-cert"], reply_markup=markup)
         return
 
-    text = TEXT['ok-cert']
+    text = TEXT["ok-cert"]
     text = text.replace("{%NAME%}", cert.name)
 
     await message.answer(text, reply_markup=markup)
@@ -53,26 +61,35 @@ async def check_cert(message: Message, state: FSMContext, user: User):
     builder.button(text="Нет", callback_data="unused")
     builder.adjust(2)
 
-    await message.answer(TEXT['input-used'], reply_markup=builder.as_markup())
+    await message.answer(TEXT["input-used"], reply_markup=builder.as_markup())
 
 
-@router.callback_query(F.data.regexp("used\\|\\d+"),
-                       F.from_user.id.in_(ADMINS))
-async def cert_used(query: CallbackQuery):
+@router.callback_query(F.data.regexp("used\\|\\d+"), F.from_user.id.in_(ADMINS))
+async def cert_used(query: CallbackQuery, user: User):
     await query.answer()
+
+    if not query.message or isinstance(query.message, InaccessibleMessage):
+        raise Exception(f"Query without corresponding message {user.info()}")
+    if not query.data:
+        raise Exception(f"Query without data {user.info()}")
+
     await query.message.edit_reply_markup(reply_markup=None)
 
     _, code = query.data.split("|")
     cert: Cert = await Cert.aio_get(Cert.code == code)
-    cert.used = True
+    cert.used = cast(BooleanField, True)
     await cert.aio_save()
 
-    await query.message.edit_text(TEXT['ok-used'])
+    await query.message.edit_text(TEXT["ok-used"])
 
 
 @router.callback_query(F.data == "unused", F.from_user.id.in_(ADMINS))
-async def cert_unused(query: CallbackQuery):
+async def cert_unused(query: CallbackQuery, user: User):
     await query.answer()
+
+    if not query.message or isinstance(query.message, InaccessibleMessage):
+        raise Exception(f"Query without corresponding message {user.info()}")
+
     await query.message.edit_reply_markup(reply_markup=None)
 
-    await query.message.edit_text(TEXT['ok-unused'])
+    await query.message.edit_text(TEXT["ok-unused"])

src/routers/check_count.py

@@ -9,12 +9,12 @@
 
 
 @router.message(StateFilter(None), Command("check_count"))
-@router.message(StateFilter(None), F.text == TEXT['check-count'])
+@router.message(StateFilter(None), F.text == TEXT["check-count"])
 async def check_count(message: Message, user: User):
     if user.chat_id in ADMINS:
-        await message.answer(TEXT['admin-count'])
+        await message.answer(TEXT["admin-count"])
         return
-    text = TEXT['user-count']
+    text = TEXT["user-count"]
     text = text.replace("{%COUNT%}", str(user.count))
 
     await message.answer(text)