Merge pull request #1 from sdb9696/add_typing

.pre-commit-config.yaml

@@ -1,29 +1,28 @@
 repos:
--   repo: https://github.com/python-poetry/poetry
-    rev: 1.6.0
-    hooks:
-    -   id: poetry-check
+- repo: https://github.com/python-poetry/poetry
+  rev: 1.6.0
+  hooks:
+    - id: poetry-check
 
-- repo: https://github.com/python/black
-  rev: 23.3.0
+- repo: https://github.com/astral-sh/ruff-pre-commit
+  rev: v0.1.3
   hooks:
-  - id: black
-    files: firebase_messaging
+    - id: ruff
+      args: [--fix, --exit-non-zero-on-fix]
+    - id: ruff-format  
 
-- repo: https://github.com/pycqa/flake8
-  rev: 6.0.0
+- repo: https://github.com/pre-commit/mirrors-mypy
+  rev: v1.9.0
   hooks:
-  - id: flake8
-    #files: firebase_messaging
+    - id: mypy
+      args: ["--install-types", "--non-interactive", "--ignore-missing-imports"]
+      additional_dependencies: [types-protobuf]
+      exclude: |
+        (?x)^(
+            docs/.*|
+            firebase_messaging/proto/.*py$
+        )$
+
 
-- repo: local
-  hooks:
-  - id: pylint
-    name: pylint
-    entry: pylint
-    language: system
-    types: [python]
-    require_serial: true
-    files: firebase_messaging
 
 

docs/source/conf.py

@@ -32,5 +32,5 @@
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
 
 html_theme = "sphinx_rtd_theme"
-autodoc_member_order = 'bysource'
+autodoc_member_order = "bysource"
 # html_static_path = ["_static"]

firebase_messaging/__init__.py

@@ -1,3 +1,3 @@
-from .fcmpushclient import FcmPushClientConfig, FcmPushClient, FcmPushClientRunState
+from .fcmpushclient import FcmPushClient, FcmPushClientConfig, FcmPushClientRunState
 
 __all__ = ["FcmPushClientConfig", "FcmPushClient", "FcmPushClientRunState"]

firebase_messaging/fcm.py

@@ -26,7 +26,7 @@ def fcm_register(sender_id, token, retries=5, log_debug_verbose=False):
     # first byte of public key is skipped for some reason
     # maybe it's always zero
 
-    private_key = ec.generate_private_key(ec.SECP256R1)
+    private_key = ec.generate_private_key(ec.SECP256R1())
     public_key = private_key.public_key()
 
     serialized_private = private_key.private_bytes(
@@ -48,7 +48,7 @@ def fcm_register(sender_id, token, retries=5, log_debug_verbose=False):
     }
     data = {
         "authorized_entity": sender_id,
-        "endpoint": "{}/{}".format(FCM_SEND_URL, token),
+        "endpoint": f"{FCM_SEND_URL}/{token}",
         "encryption_key": keys["public"],
         "encryption_auth": keys["secret"],
     }

firebase_messaging/fcmpushclient.py

@@ -6,11 +6,13 @@
 import time
 import traceback
 from base64 import urlsafe_b64decode
-from ssl import SSLError
-from threading import Thread
-from typing import Any, Callable, Optional, List
+from contextlib import suppress as contextlib_suppress
 from dataclasses import dataclass
 from enum import Enum
+from ssl import SSLError
+from threading import Thread
+from typing import Any, Callable, Dict, List, Optional
+
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.serialization import load_der_private_key
 from google.protobuf.json_format import MessageToJson
@@ -36,7 +38,6 @@
     SelectiveAck,
 )
 
-
 _logger = logging.getLogger(__name__)
 
 
@@ -117,7 +118,7 @@ def __init__(
         self,
         *,
         credentials: Optional[dict] = None,
-        credentials_updated_callback: Optional[Callable[[str], None]] = None,
+        credentials_updated_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
         received_persistent_ids: Optional[List[str]] = None,
         config: Optional[FcmPushClientConfig] = None,
     ):
@@ -132,8 +133,8 @@ def __init__(
         self.reader = None
         self.writer = None
         self.do_listen = False
-        self.sequential_error_counters = {}
-        self.log_warn_counters = {}
+        self.sequential_error_counters: Dict[ErrorType, int] = {}
+        self.log_warn_counters: Dict[str, int] = {}
 
         # reset variables
         self.input_stream_id = 0
@@ -145,12 +146,12 @@ def __init__(
         self.run_state: FcmPushClientRunState = FcmPushClientRunState.CREATED
         self.tasks = None
 
-        self.listen_event_loop = None
-        self.callback_event_loop = None
-        self.fcm_thread = None
+        self.listen_event_loop: Optional[asyncio.AbstractEventLoop] = None
+        self.callback_event_loop: Optional[asyncio.AbstractEventLoop] = None
+        self.fcm_thread: Optional[Thread] = None
 
-        self.app_id = None
-        self.sender_id = None
+        self.app_id: Optional[str] = None
+        self.sender_id: Optional[int] = None
 
         self.reset_lock = None
         self.stopping_lock = None
@@ -250,10 +251,8 @@ def _encode_varint32(x):
     @staticmethod
     def _make_packet(msg, include_version):
         tag = MCS_MESSAGE_TAG[type(msg)]
-        if include_version:
-            header = bytearray([MCS_VERSION, tag])
-        else:
-            header = bytearray([tag])
+
+        header = bytearray([MCS_VERSION, tag]) if include_version else bytearray([tag])
 
         payload = msg.SerializeToString()
         buf = bytes(header) + FcmPushClient._encode_varint32(len(payload)) + payload
@@ -271,7 +270,7 @@ async def _receive_msg(self):
             r = await self.reader.readexactly(2)
             version, tag = struct.unpack("BB", r)
             if version < MCS_VERSION and version != 38:
-                raise RuntimeError("protocol version {} unsupported".format(version))
+                raise RuntimeError(f"protocol version {version} unsupported")
             self.first_message = False
         else:
             r = await self.reader.readexactly(1)
@@ -371,7 +370,7 @@ def _app_data_by_key(self, p, key):
             if x.key == key:
                 return x.value
 
-        raise RuntimeError("couldn't find in app_data {}".format(key))
+        raise RuntimeError(f"couldn't find in app_data {key}")
 
     def _handle_data_message(self, callback, msg, obj):
         _logger.debug(
@@ -394,10 +393,8 @@ def _handle_data_message(self, callback, msg, obj):
         decrypted = self._decrypt_raw_data(
             self.credentials, crypto_key, salt, msg.raw_data
         )
-        try:
+        with contextlib_suppress((json.JSONDecodeError, ValueError)):
             decrypted_json = json.loads(decrypted.decode("utf-8"))
-        except (json.JSONDecodeError, ValueError):
-            pass
 
         ret_val = decrypted_json if decrypted_json else decrypted
         self._log_verbose(
@@ -792,9 +789,9 @@ def start(
         callback: Optional[Callable[[dict, str, Optional[Any]], None]],
         obj: Any = None,
         *,
-        listen_event_loop: asyncio.AbstractEventLoop = None,
-        callback_event_loop: asyncio.AbstractEventLoop = None,
-    ):
+        listen_event_loop: Optional[asyncio.AbstractEventLoop] = None,
+        callback_event_loop: Optional[asyncio.AbstractEventLoop] = None,
+    ) -> None:
         """Connect to FCM and start listening for push
             messages on a seperate service thread.
 
@@ -860,10 +857,10 @@ async def _stop_connection(self):
                 self.fcm_thread = None
                 self.listen_event_loop = None
 
-    def is_started(self):
+    def is_started(self) -> bool:
         return self.run_state == FcmPushClientRunState.STARTED
 
-    def stop(self):
+    def stop(self) -> None:
         """Disconnects from FCM and shuts down the service thread."""
         if self.fcm_thread:
             if (
@@ -917,14 +914,15 @@ async def _send_data_message(self, raw_data, persistent_id):
         dms.data = raw_data
         # Not supported yet
 
-    def send_message(self, raw_data, persistent_id):
+    def send_message(self, raw_data, persistent_id) -> None:
         """Not implemented, does nothing atm."""
         if self.fcm_thread:
             asyncio.run_coroutine_threadsafe(
-                self._send_data_message(raw_data, persistent_id), self.listen_event_loop
+                self._send_data_message(raw_data, persistent_id),
+                self.listen_event_loop,  # type: ignore[arg-type]
             )
         else:
-            self.listen_event_loop.create_task(
+            self.listen_event_loop.create_task(  # type: ignore[union-attr]
                 self._send_data_message(raw_data, persistent_id)
             )
 

firebase_messaging/gcm.py

@@ -1,17 +1,17 @@
 import logging
 import time
-from typing import Optional
+from typing import Optional, Union
 
 import requests
-
 from google.protobuf.json_format import MessageToDict, MessageToJson
 
 from .const import GCM_CHECKIN_URL, GCM_REGISTER_URL, GCM_SERVER_KEY_B64
-from .proto.android_checkin_pb2 import (  # pylint: disable=no-name-in-module
+from .proto.android_checkin_pb2 import (
+    DEVICE_CHROME_BROWSER,
     AndroidCheckinProto,
     ChromeBuildProto,
 )
-from .proto.checkin_pb2 import (  # pylint: disable=no-name-in-module
+from .proto.checkin_pb2 import (
     AndroidCheckinRequest,
     AndroidCheckinResponse,
 )
@@ -23,12 +23,12 @@ def _get_checkin_payload(
     android_id: Optional[int] = None, security_token: Optional[int] = None
 ):
     chrome = ChromeBuildProto()
-    chrome.platform = 3
+    chrome.platform = ChromeBuildProto.Platform.PLATFORM_LINUX  # 3
     chrome.chrome_version = "63.0.3234.0"
-    chrome.channel = 1
+    chrome.channel = ChromeBuildProto.Channel.CHANNEL_STABLE  # 1
 
     checkin = AndroidCheckinProto()
-    checkin.type = 3
+    checkin.type = DEVICE_CHROME_BROWSER  # 3
     checkin.chrome_build.CopyFrom(chrome)
 
     payload = AndroidCheckinRequest()
@@ -73,6 +73,7 @@ def gcm_check_in(
             )
             if resp.status_code == 200:
                 acir = AndroidCheckinResponse()
+                break
             else:
                 _logger.warning(
                     "GCM checkin failed on attempt %s out of %s with status: %s, %s",
@@ -132,7 +133,7 @@ def gcm_register(app_id: str, retries=5, log_debug_verbose=False):
         _logger.debug("GCM Registration request: %s", body)
 
     auth = "AidLogin {}:{}".format(chk["androidId"], chk["securityToken"])
-    last_error = None
+    last_error: Optional[Union[str, Exception]] = None
     for try_num in range(retries):
         try:
             resp = requests.post(