Правки тестов и воркфлоу

.github/workflows/lint-and-test.yml

@@ -7,18 +7,8 @@ on:
     branches: [ main ]
     types: [ opened, synchronize, reopened ]
 
-permissions:
-  contents: read
-  pull-requests: write
-
 jobs:
   lint-and-test:
-    uses: esclient/tools/.github/workflows/lint-and-test-python.yml@v1.0.1
+    uses: esclient/tools/.github/workflows/lint-and-test-python.yml@v1.0.2
     with:
       python-version: "3.13.7"
-      source: "modservice"
-      sonar-inclusions: "src/**,Dockerfile"
-      sonar-exclusions: "**/grpc/**"
-      sonar-coverage-exclusions: "src/modservice/server.py,src/modservice/settings.py,src/modservice/s3_client.py"
-    secrets:
-      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}

pyproject.toml

@@ -157,6 +157,7 @@ dev = [
     "types-psycopg2==2.9.21.20250915",
     "types-aioboto3==15.4.0",
     "types-aiofiles==25.1.0.20251011",
+    "types-confluent-kafka==1.2.2",
     "botocore-stubs==1.40.55",
     "asyncpg-stubs==0.30.2",
 ]

src/modservice/handler/get_mods.py

@@ -1,6 +1,11 @@
 import grpc
 
-from modservice.constants import STATUS_BANNED, STATUS_HIDDEN, STATUS_ON_MODERATION, STATUS_UPLOADED
+from modservice.constants import (
+    STATUS_BANNED,
+    STATUS_HIDDEN,
+    STATUS_ON_MODERATION,
+    STATUS_UPLOADED,
+)
 from modservice.grpc import mod_pb2
 from modservice.service.service import ModService
 

src/modservice/kafka/config.py

@@ -1,37 +1,48 @@
 import os
+from typing import Any
+
+type KafkaClientConfig = dict[str, Any]
 
 
 class KafkaConfig:
 
-    def __init__(self):
-        self.brokers = os.getenv('KAFKA_BROKERS', 'localhost:9092')
-        self.request_topic = os.getenv('KAFKA_REQUEST_TOPIC', 'moderation-request')
-        self.result_topic = os.getenv('KAFKA_RESULT', 'moderation-result')
-        self.consumer_group_id = os.getenv('KAFKA_CONSUMER_GROUP_ID', 'mod-service-group')
-        self.max_retries = int(os.getenv('KAFKA_MAX_RETRIES', '3'))
-        self.retry_backoff_ms = int(os.getenv('KAFKA_RETRY_BACKOFF_MS', '1000'))
-        self.enable_ssl = os.getenv('KAFKA_ENABLE_SSL', 'false').lower() == 'true'
+    def __init__(self) -> None:
+        self.brokers = os.getenv("KAFKA_BROKERS", "localhost:9092")
+        self.request_topic = os.getenv(
+            "KAFKA_REQUEST_TOPIC", "moderation-request"
+        )
+        self.result_topic = os.getenv("KAFKA_RESULT", "moderation-result")
+        self.consumer_group_id = os.getenv(
+            "KAFKA_CONSUMER_GROUP_ID", "mod-service-group"
+        )
+        self.max_retries = int(os.getenv("KAFKA_MAX_RETRIES", "3"))
+        self.retry_backoff_ms = int(
+            os.getenv("KAFKA_RETRY_BACKOFF_MS", "1000")
+        )
+        self.enable_ssl = (
+            os.getenv("KAFKA_ENABLE_SSL", "false").lower() == "true"
+        )
 
-    def get_producer_config(self):
-        config = {
-            'bootstrap.servers': self.brokers,
-            'client.id': 'mod_service_producer',
-            'acks': 'all',
-            'retries': self.max_retries,
-            'retry.backoff.ms': self.retry_backoff_ms,
-            'enable.idempotence': True,
-            'compression.type': 'none'
+    def get_producer_config(self) -> KafkaClientConfig:
+        config: KafkaClientConfig = {
+            "bootstrap.servers": self.brokers,
+            "client.id": "mod_service_producer",
+            "acks": "all",
+            "retries": self.max_retries,
+            "retry.backoff.ms": self.retry_backoff_ms,
+            "enable.idempotence": True,
+            "compression.type": "none",
         }
         return config
 
-    def get_consumer_config(self):
-        config = {
-            'bootstrap.servers': self.brokers,
-            'group.id': self.consumer_group_id,
-            'auto.offset.reset': 'earliest',
-            'enable.auto.commit': True,
-            'auto.commit.interval.ms': 1000,
-            'session.timeout.ms': 30000,
-            'heartbeat.interval.ms': 10000
+    def get_consumer_config(self) -> KafkaClientConfig:
+        config: KafkaClientConfig = {
+            "bootstrap.servers": self.brokers,
+            "group.id": self.consumer_group_id,
+            "auto.offset.reset": "earliest",
+            "enable.auto.commit": True,
+            "auto.commit.interval.ms": 1000,
+            "session.timeout.ms": 30000,
+            "heartbeat.interval.ms": 10000,
         }
         return config

src/modservice/kafka/consumer.py

@@ -1,40 +1,52 @@
 """Kafka consumer for receiving moderation responses."""
 
-from confluent_kafka import Consumer, KafkaException, KafkaError
 import logging
 import threading
-from .config import KafkaConfig
+from collections.abc import Callable
+
+from confluent_kafka import Consumer, KafkaError, KafkaException, Message
+
 from modservice.grpc import moderation_pb2
 
+from .config import KafkaConfig
+
 logger = logging.getLogger(__name__)
 
 
 class KafkaConsumer:
 
-    def __init__(self, config: KafkaConfig, callback):
+    def __init__(
+        self,
+        config: KafkaConfig,
+        callback: (
+            Callable[[moderation_pb2.ModerateObjectResponse, int], None] | None
+        ),
+    ) -> None:
 
         self.config = config
         self.callback = callback
         self.consumer = Consumer(config.get_consumer_config())
         self.topic = config.result_topic
         self.running = False
-        self.consumer_thread = None
+        self.consumer_thread: threading.Thread | None = None
 
         self.consumer.subscribe([self.topic])
         logger.info(f"KafkaConsumer subscribed to topic: {self.topic}")
 
-    def start(self):
+    def start(self) -> None:
 
         if self.running:
             logger.warning("Consumer already running")
             return
 
         self.running = True
-        self.consumer_thread = threading.Thread(target=self._consume_loop, daemon=True)
+        self.consumer_thread = threading.Thread(
+            target=self._consume_loop, daemon=True
+        )
         self.consumer_thread.start()
         logger.info("KafkaConsumer started")
 
-    def stop(self):
+    def stop(self) -> None:
 
         if not self.running:
             return
@@ -47,7 +59,7 @@ def stop(self):
         self.consumer.close()
         logger.info("KafkaConsumer stopped")
 
-    def _consume_loop(self):
+    def _consume_loop(self) -> None:
 
         logger.info("Consumer loop started")
 
@@ -57,13 +69,16 @@ def _consume_loop(self):
 
                 if msg is None:
                     continue
-                if msg.error():
-                    if msg.error().code() == KafkaError._PARTITION_EOF:
-                        logger.debug(f"Reached end of partition for topic {msg.topic()}")
-                    elif msg.error().code() == KafkaError._TIMED_OUT:
+                error = msg.error()
+                if error is not None:
+                    if error.code() == KafkaError._PARTITION_EOF:
+                        logger.debug(
+                            f"Reached end of partition for topic {msg.topic()}"
+                        )
+                    elif error.code() == KafkaError._TIMED_OUT:
                         logger.debug("Timeout")
                     else:
-                        logger.error(f"Consumer error: {msg.error()}")
+                        logger.error(f"Consumer error: {error}")
                     continue
 
                 self._process_message(msg)
@@ -75,20 +90,35 @@ def _consume_loop(self):
 
         logger.info("Consumer loop ended")
 
-    def _process_message(self, msg):
+    def _process_message(self, msg: Message) -> None:
 
         try:
             object_id = 0
-            if msg.key() is not None:
+            message_key = msg.key()
+            if message_key is not None:
                 try:
-                    key_str = msg.key().decode('utf-8')
+                    key_str = (
+                        message_key.decode("utf-8")
+                        if isinstance(message_key, bytes)
+                        else message_key
+                    )
                     object_id = int(key_str)
                 except ValueError as e:
-                    logger.error(f"Failed to parse message key to request ID: {e}")
+                    logger.error(
+                        f"Failed to parse message key to request ID: {e}"
+                    )
                     return
 
+            message_value = msg.value()
+            if message_value is None:
+                logger.error("Received message without payload")
+                return
+            if not isinstance(message_value, bytes):
+                logger.error("Received message payload with unexpected type")
+                return
+
             response = moderation_pb2.ModerateObjectResponse()
-            response.ParseFromString(msg.value())
+            response.ParseFromString(message_value)
 
             logger.info(
                 f"Received moderation response: object_id={object_id}, "

src/modservice/kafka/producer.py

@@ -1,40 +1,47 @@
 """Kafka producer for sending moderation requests."""
 
-from confluent_kafka import Producer, KafkaException
 import logging
-from .config import KafkaConfig
+
+from confluent_kafka import KafkaError, KafkaException, Message, Producer
+
 from modservice.grpc import moderation_pb2
 
+from .config import KafkaConfig
+
 logger = logging.getLogger(__name__)
 
 
 class KafkaProducer:
 
-    def __init__(self, config: KafkaConfig):
+    def __init__(self, config: KafkaConfig) -> None:
         self.config = config
         self.producer = Producer(config.get_producer_config())
         self.topic = config.request_topic
 
         logger.info(f"KafkaProducer initialized for topic: {self.topic}")
 
-    def send_moderation_request(self, request: moderation_pb2.ModerateObjectRequest, retries: int = 3) -> bool:
+    def send_moderation_request(
+        self, request: moderation_pb2.ModerateObjectRequest, retries: int = 3
+    ) -> bool:
         for _ in range(retries):
             try:
                 serialized_request = request.SerializeToString()
 
-                key = str(request.id).encode('utf-8')
+                key = str(request.id).encode("utf-8")
 
                 self.producer.produce(
                     topic=self.topic,
                     value=serialized_request,
                     key=key,
                     partition=-1,
-                    callback=self._delivery_callback
+                    on_delivery=self._delivery_callback,
                 )
 
                 self.producer.poll(0)
 
-                logger.debug(f"Moderation request queued: ID={request.id}, text_length={len(request.text)}")
+                logger.debug(
+                    f"Moderation request queued: ID={request.id}, text_length={len(request.text)}"
+                )
                 return True
 
             except BufferError as e:
@@ -48,7 +55,7 @@ def send_moderation_request(self, request: moderation_pb2.ModerateObjectRequest,
                 return False
         return False
 
-    def _delivery_callback(self, err, msg):
+    def _delivery_callback(self, err: KafkaError | None, msg: Message) -> None:
 
         if err is not None:
             logger.error(f"Message delivery failed: {err}")
@@ -58,15 +65,17 @@ def _delivery_callback(self, err, msg):
                 f"partition={msg.partition()}, offset={msg.offset()}"
             )
 
-    def flush(self, timeout: float = 10.0):
+    def flush(self, timeout: float = 10.0) -> None:
 
         remaining = self.producer.flush(timeout)
 
         if remaining > 0:
-            logger.warning(f"Flush incomplete: {remaining} messages still pending")
+            logger.warning(
+                f"Flush incomplete: {remaining} messages still pending"
+            )
         else:
             logger.debug("All messages flushed successfully")
 
-    def __del__(self):
-        if hasattr(self, 'producer'):
+    def __del__(self) -> None:
+        if hasattr(self, "producer"):
             self.flush()