mypy fix

Author: p1p1daster

pybotx/async_buffer.py

@@ -1,3 +1,4 @@
+import abc
 import os
 from typing import Optional
 
@@ -18,10 +19,12 @@ async def tell(self) -> int: ...  # noqa: WPS428
 
 
 class AsyncBufferWritable(AsyncBufferBase):
+    @abc.abstractmethod
     async def write(self, content: bytes) -> int: ...  # noqa: WPS428
 
 
 class AsyncBufferReadable(AsyncBufferBase):
+    @abc.abstractmethod
     async def read(
         self,
         bytes_to_read: Optional[int] = None,

pybotx/bot/handler.py

@@ -1,6 +1,15 @@
 from dataclasses import dataclass
 from functools import partial
-from typing import TYPE_CHECKING, Awaitable, Callable, List, Literal, TypeVar, Union
+from typing import (
+    TYPE_CHECKING,
+    Awaitable,
+    Callable,
+    List,
+    Literal,
+    TypeVar,
+    Union,
+    cast,
+)
 
 from pybotx.models.commands import BotCommand
 from pybotx.models.message.incoming_message import IncomingMessage
@@ -69,7 +78,7 @@ async def __call__(self, message: IncomingMessage, bot: "Bot") -> None:
         handler_func = self.handler_func
 
         for middleware in self.middlewares[::-1]:
-            handler_func = partial(middleware, call_next=handler_func)
+            handler_func = cast(HandlerFunc, partial(middleware, call_next=handler_func))  # type: ignore[type-arg, call-arg]
 
         await handler_func(message, bot)
 

pybotx/models/async_files.py

@@ -52,7 +52,6 @@ class Image(AsyncFileBase):
 @dataclass
 class Video(AsyncFileBase):
     type: Literal[AttachmentTypes.VIDEO]
-
     duration: int
 
 
@@ -64,7 +63,6 @@ class Document(AsyncFileBase):
 @dataclass
 class Voice(AsyncFileBase):
     type: Literal[AttachmentTypes.VOICE]
-
     duration: int
 
 
@@ -78,12 +76,6 @@ class APIAsyncFileBase(VerifiedPayloadBaseModel):
     file_hash: str
 
     class Config:
-        """BotX sends extra fields which are used by client only.
-
-        We skip their validation, but extra fields will be saved during
-        serialization/deserialization.
-        """
-
         extra = "allow"
 
 
@@ -93,7 +85,6 @@ class ApiAsyncFileImage(APIAsyncFileBase):
 
 class ApiAsyncFileVideo(APIAsyncFileBase):
     type: Literal[APIAttachmentTypes.VIDEO]
-
     duration: int
 
 
@@ -103,7 +94,6 @@ class ApiAsyncFileDocument(APIAsyncFileBase):
 
 class ApiAsyncFileVoice(APIAsyncFileBase):
     type: Literal[APIAttachmentTypes.VOICE]
-
     duration: int
 
 
@@ -121,7 +111,6 @@ def convert_async_file_from_domain(file: File) -> APIAsyncFile:
     attachment_type = convert_attachment_type_from_domain(file.type)
 
     if attachment_type == APIAttachmentTypes.IMAGE:
-        attachment_type = cast(Literal[APIAttachmentTypes.IMAGE], attachment_type)
         file = cast(Image, file)
 
         return ApiAsyncFileImage(
@@ -135,7 +124,6 @@ def convert_async_file_from_domain(file: File) -> APIAsyncFile:
         )
 
     if attachment_type == APIAttachmentTypes.VIDEO:
-        attachment_type = cast(Literal[APIAttachmentTypes.VIDEO], attachment_type)
         file = cast(Video, file)
 
         return ApiAsyncFileVideo(
@@ -150,7 +138,6 @@ def convert_async_file_from_domain(file: File) -> APIAsyncFile:
         )
 
     if attachment_type == APIAttachmentTypes.DOCUMENT:
-        attachment_type = cast(Literal[APIAttachmentTypes.DOCUMENT], attachment_type)
         file = cast(Document, file)
 
         return ApiAsyncFileDocument(
@@ -164,7 +151,6 @@ def convert_async_file_from_domain(file: File) -> APIAsyncFile:
         )
 
     if attachment_type == APIAttachmentTypes.VOICE:
-        attachment_type = cast(Literal[APIAttachmentTypes.VOICE], attachment_type)
         file = cast(Voice, file)
 
         return ApiAsyncFileVoice(
@@ -185,7 +171,6 @@ def convert_async_file_to_domain(async_file: APIAsyncFile) -> File:
     attachment_type = convert_attachment_type_to_domain(async_file.type)
 
     if attachment_type == AttachmentTypes.IMAGE:
-        attachment_type = cast(Literal[AttachmentTypes.IMAGE], attachment_type)
         async_file = cast(ApiAsyncFileImage, async_file)
 
         return Image(
@@ -200,7 +185,6 @@ def convert_async_file_to_domain(async_file: APIAsyncFile) -> File:
         )
 
     if attachment_type == AttachmentTypes.VIDEO:
-        attachment_type = cast(Literal[AttachmentTypes.VIDEO], attachment_type)
         async_file = cast(ApiAsyncFileVideo, async_file)
 
         return Video(
@@ -216,7 +200,6 @@ def convert_async_file_to_domain(async_file: APIAsyncFile) -> File:
         )
 
     if attachment_type == AttachmentTypes.DOCUMENT:
-        attachment_type = cast(Literal[AttachmentTypes.DOCUMENT], attachment_type)
         async_file = cast(ApiAsyncFileDocument, async_file)
 
         return Document(
@@ -231,7 +214,6 @@ def convert_async_file_to_domain(async_file: APIAsyncFile) -> File:
         )
 
     if attachment_type == AttachmentTypes.VOICE:
-        attachment_type = cast(Literal[AttachmentTypes.VOICE], attachment_type)
         async_file = cast(ApiAsyncFileVoice, async_file)
 
         return Voice(

pybotx/models/attachments.py

@@ -220,7 +220,7 @@ def convert_api_attachment_to_domain(  # noqa: WPS212
     attachment_type = convert_attachment_type_to_domain(api_attachment.type)
 
     if attachment_type == AttachmentTypes.IMAGE:
-        attachment_type = cast(Literal[AttachmentTypes.IMAGE], attachment_type)
+        attachment_type = cast(Literal[AttachmentTypes.IMAGE], attachment_type)  # type: ignore[redundant-cast]
         api_attachment = cast(BotAPIAttachmentImage, api_attachment)
         content = decode_rfc2397(api_attachment.data.content)
 
@@ -233,7 +233,7 @@ def convert_api_attachment_to_domain(  # noqa: WPS212
         )
 
     if attachment_type == AttachmentTypes.VIDEO:
-        attachment_type = cast(Literal[AttachmentTypes.VIDEO], attachment_type)
+        attachment_type = cast(Literal[AttachmentTypes.VIDEO], attachment_type)  # type: ignore[redundant-cast]
         api_attachment = cast(BotAPIAttachmentVideo, api_attachment)
         content = decode_rfc2397(api_attachment.data.content)
 
@@ -247,7 +247,7 @@ def convert_api_attachment_to_domain(  # noqa: WPS212
         )
 
     if attachment_type == AttachmentTypes.DOCUMENT:
-        attachment_type = cast(Literal[AttachmentTypes.DOCUMENT], attachment_type)
+        attachment_type = cast(Literal[AttachmentTypes.DOCUMENT], attachment_type)  # type: ignore[redundant-cast]
         api_attachment = cast(BotAPIAttachmentDocument, api_attachment)
         content = decode_rfc2397(api_attachment.data.content)
 
@@ -260,7 +260,7 @@ def convert_api_attachment_to_domain(  # noqa: WPS212
         )
 
     if attachment_type == AttachmentTypes.VOICE:
-        attachment_type = cast(Literal[AttachmentTypes.VOICE], attachment_type)
+        attachment_type = cast(Literal[AttachmentTypes.VOICE], attachment_type)  # type: ignore[redundant-cast]
         api_attachment = cast(BotAPIAttachmentVoice, api_attachment)
         content = decode_rfc2397(api_attachment.data.content)
         attachment_extension = get_attachment_extension_from_encoded_content(
@@ -277,7 +277,7 @@ def convert_api_attachment_to_domain(  # noqa: WPS212
         )
 
     if attachment_type == AttachmentTypes.LOCATION:
-        attachment_type = cast(Literal[AttachmentTypes.LOCATION], attachment_type)
+        attachment_type = cast(Literal[AttachmentTypes.LOCATION], attachment_type)  # type: ignore[redundant-cast]
         api_attachment = cast(BotAPIAttachmentLocation, api_attachment)
 
         return Location(
@@ -288,15 +288,14 @@ def convert_api_attachment_to_domain(  # noqa: WPS212
         )
 
     if attachment_type == AttachmentTypes.CONTACT:
-        attachment_type = cast(Literal[AttachmentTypes.CONTACT], attachment_type)
+        attachment_type = cast(Literal[AttachmentTypes.CONTACT], attachment_type)  # type: ignore[redundant-cast]
         api_attachment = cast(BotAPIAttachmentContact, api_attachment)
 
         return Contact(
             name=api_attachment.data.contact_name,
         )
 
     if attachment_type == AttachmentTypes.LINK:
-        attachment_type = cast(Literal[AttachmentTypes.LINK], attachment_type)
         api_attachment = cast(BotAPIAttachmentLink, api_attachment)
 
         return Link(
@@ -307,7 +306,6 @@ def convert_api_attachment_to_domain(  # noqa: WPS212
         )
 
     if attachment_type == AttachmentTypes.STICKER:
-        attachment_type = cast(Literal[AttachmentTypes.STICKER], attachment_type)
         api_attachment = cast(BotAPIAttachmentSticker, api_attachment)
 
         return Sticker(

pybotx/models/message/incoming_message.py

@@ -154,11 +154,9 @@ def _convert_bot_api_mention_to_domain(api_mention_data: BotAPIMentionData) -> M
 
 def convert_bot_api_entity_to_domain(api_entity: BotAPIEntity) -> Entity:
     if api_entity.type == BotAPIEntityTypes.MENTION:
-        api_entity = cast(BotAPIMention, api_entity)
         return _convert_bot_api_mention_to_domain(api_entity.data)
 
     if api_entity.type == BotAPIEntityTypes.FORWARD:
-        api_entity = cast(BotAPIForward, api_entity)
 
         return Forward(
             chat_id=api_entity.data.group_chat_id,
@@ -170,7 +168,6 @@ def convert_bot_api_entity_to_domain(api_entity: BotAPIEntity) -> Entity:
         )
 
     if api_entity.type == BotAPIEntityTypes.REPLY:
-        api_entity = cast(BotAPIReply, api_entity)
 
         mentions = MentionList()
         for api_mention_data in api_entity.data.mentions:

tests/client/test_botx_method_callback.py

@@ -1,4 +1,4 @@
-# type: ignore [attr-defined]
+# mypy: disable-error-code=attr-defined
 
 import asyncio
 import time

tests/system_events/factories.py

@@ -1,18 +1,17 @@
 import uuid
 from typing import Any, Dict, List, Optional
 
-from factory.base import DictFactory  # type: ignore
-from factory.declarations import SubFactory  # type: ignore
+from factory.base import DictFactory
+from factory.declarations import SubFactory
 
 
-class DeviceMetaFactory(DictFactory):  # type: ignore[misc]
-
+class DeviceMetaFactory(DictFactory):
     permissions: Optional[str] = None
     pushes: Optional[str] = None
     timezone: Optional[str] = None
 
 
-class FromFactory(DictFactory):  # type: ignore[misc]
+class FromFactory(DictFactory):
     user_huid: Optional[str] = None
     group_chat_id: str = "8dada2c8-67a6-4434-9dec-570d244e78ee"
     ad_login: Optional[str] = None
@@ -22,7 +21,7 @@ class FromFactory(DictFactory):  # type: ignore[misc]
     manufacturer: Optional[str] = None
     device: Optional[str] = None
     device_software: Optional[str] = None
-    device_meta: Dict[str, Any] = SubFactory(DeviceMetaFactory)  # noqa: F821
+    device_meta: Any = SubFactory(DeviceMetaFactory)  # type: ignore[no-untyped-call]
     platform: Optional[str] = None
     platform_package_id: Optional[str] = None
     is_admin: Optional[bool] = None
@@ -32,29 +31,24 @@ class FromFactory(DictFactory):  # type: ignore[misc]
     host: str = "cts.ccteam.ru"
 
 
-class CommandDataFactory(DictFactory):  # type: ignore[misc]
-
+class CommandDataFactory(DictFactory):
     added_members: List[str] = [uuid.uuid4().hex, uuid.uuid4().hex]
 
 
-class CommandFactory(DictFactory):  # type: ignore[misc]
-
+class CommandFactory(DictFactory):
     body: str = "system:user_joined_to_chat"
     command_type: str = "system"
-    data: Dict[str, Any] = SubFactory(CommandDataFactory)  # noqa: F821
+    data: Any = SubFactory(CommandDataFactory)  # type: ignore[no-untyped-call]
     metadata: Dict[str, Any] = {}
 
 
-class BotAPIJoinToChatFactory(DictFactory):  # type: ignore[misc]
-
+class BotAPIJoinToChatFactory(DictFactory):
     sync_id: str = uuid.uuid4().hex
-    command: Dict[str, Any] = SubFactory(CommandFactory)  # noqa: F821
+    command: Any = SubFactory(CommandFactory)  # type: ignore[no-untyped-call]
     async_files: List[str] = []
     attachments: List[str] = []
     entities: List[str] = []
-    from_: Dict[str, Any] = SubFactory(
-        FromFactory,
-    )  # noqa: F821
+    from_: Any = SubFactory(FromFactory)  # type: ignore[no-untyped-call]
     bot_id: str = uuid.uuid4().hex
     proto_version: int = 4
     source_sync_id: Optional[str] = None
@@ -63,8 +57,7 @@ class Meta:
         rename = {"from_": "from"}
 
 
-class ConferenceChangedDataFactory(DictFactory):  # type: ignore[misc]
-
+class ConferenceChangedDataFactory(DictFactory):
     access_code = None
     actor = None
     added_users = ["5c053f2a-0bdf-4ab1-9bc9-256fee9db7ba"]

tests/system_events/test_conference_changed.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Callable, Dict, Optional
+from typing import Any, Callable, Dict, Optional, cast
 from uuid import UUID
 
 import pytest
@@ -14,7 +14,7 @@
 )
 from pybotx.models.enums import ConferenceLinkTypes
 from pybotx.models.system_events.conference_changed import ConferenceChangedEvent
-from tests.system_events.factories import ConferenceChangedDataFactory  # type: ignore
+from tests.system_events.factories import ConferenceChangedDataFactory
 
 pytestmark = [
     pytest.mark.asyncio,
@@ -32,9 +32,12 @@ async def test__conference_changed_succeed(
     api_incoming_message_factory: Callable[..., Dict[str, Any]],
 ) -> None:
     # - Arrange -
-    conference_change_data = ConferenceChangedDataFactory(
-        call_id=str(call_id),
-        link_type="public",
+    conference_change_data = cast(
+        Dict[str, Any],
+        ConferenceChangedDataFactory(
+            call_id=str(call_id),
+            link_type="public",
+        ),  # type: ignore[no-untyped-call]
     )
     payload = api_incoming_message_factory(
         body="system:conference_changed",

tests/system_events/test_join_to_chat.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional
+from typing import Any, Dict, Optional, cast
 from uuid import UUID
 
 import pytest
@@ -13,7 +13,7 @@
     lifespan_wrapper,
 )
 from pybotx.models.system_events.user_joined_to_chat import JoinToChatEvent
-from tests.system_events.factories import BotAPIJoinToChatFactory  # type: ignore
+from tests.system_events.factories import BotAPIJoinToChatFactory
 
 pytestmark = [
     pytest.mark.asyncio,
@@ -33,7 +33,10 @@ async def test__join_to_chat__succeed(
     3. The registered user_joined_to_chat handler is called with this event
     """
 
-    payload: dict[str, Any] = BotAPIJoinToChatFactory(bot_id=bot_account.id.hex)
+    payload: Dict[str, Any] = cast(
+        Dict[str, Any],
+        BotAPIJoinToChatFactory(bot_id=bot_account.id.hex),  # type: ignore[no-untyped-call]
+    )
 
     collector = HandlerCollector()
     join_to_chat: Optional[JoinToChatEvent] = None