Интегрировал pypika

.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: "commentservice"
-      sonar-inclusions: "src/**,Dockerfile"
-      sonar-exclusions: "**/grpc/**"
-      sonar-coverage-exclusions: "src/commentservice/server.py,src/commentservice/settings.py"
-    secrets:
-      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}

pyproject.toml

@@ -126,6 +126,7 @@ dependencies = [
     "watchfiles==1.1.0",
     "asyncpg>=0.30.0",
     "confluent-kafka>=2.12.2",
+    "PyPika>=0.51.1",
 ]
 requires-python = ">=3.13"
 readme = "README.md"
@@ -154,4 +155,5 @@ dev = [
     "mypy==1.18.2",
     "types-protobuf==6.32.1.20250918",
     "types-psycopg2==2.9.21.20250915",
+    "types-confluent-kafka>=1.2.2",
 ]

src/commentservice/clients/kafka/config.py

@@ -0,0 +1,46 @@
+import os
+
+
+class KafkaConfig:
+    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", "comment-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) -> dict[str, str | int | bool]:
+        config: dict[str, str | int | bool] = {
+            "bootstrap.servers": self.brokers,
+            "client.id": "comment_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) -> dict[str, str | int | bool]:
+
+        config: dict[str, str | int | bool] = {
+            "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/commentservice/kafka/consumer.py → src/commentservice/clients/kafka/consumer.py

@@ -1,50 +1,69 @@
-from confluent_kafka import Consumer, KafkaException, KafkaError
 import logging
 import threading
-from .config import KafkaConfig
+from collections.abc import Callable
+from typing import Any
+
+from confluent_kafka import (
+    Consumer,
+    KafkaError,
+    KafkaException,
+)
+
 from commentservice.grpc import moderation_pb2
 
+from .config import KafkaConfig
+
 logger = logging.getLogger(__name__)
 
-class ModerationResponseConsumer:
+ModerationCallback = Callable[
+    [moderation_pb2.ModerateObjectResponse, int], None
+]
+
 
-    def __init__(self, config: KafkaConfig, callback):
+class ModerationResponseConsumer:
+    def __init__(
+        self, config: KafkaConfig, callback: ModerationCallback | 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.running: bool = False
+        self.consumer_thread: threading.Thread | None = None
+
         self.consumer.subscribe([self.topic])
-        logger.info(f"ModerationResponseConsumer subscribed to topic: {self.topic}")
+        logger.info(
+            f"ModerationResponseConsumer 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("ModerationResponseConsumer started")
 
-    def stop(self):
-        
+    def stop(self) -> None:
+
         if not self.running:
             return
-        
+
         self.running = False
 
         if self.consumer_thread and self.consumer_thread.is_alive():
             self.consumer_thread.join(timeout=5.0)
-        
+
         self.consumer.close()
         logger.info("ModerationResponseConsumer stopped")
-    
-    def _consume_loop(self):
+
+    def _consume_loop(self) -> None:
 
         logger.info("Consumer loop started")
 
@@ -54,34 +73,40 @@ 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)
 
             except KafkaException as e:
                 logger.error(f"Kafka exception in consume loop: {e}")
             except Exception as e:
                 logger.error(f"Unexpected error in consume loop: {e}")
-        
+
         logger.info("Consumer loop ended")
-    
-    def _process_message(self, msg):
+
+    def _process_message(self, msg: Any) -> None:
 
         try:
             request_id = 0
             if msg.key() is not None:
                 try:
-                    key_str = msg.key().decode('utf-8')
+                    key_str = msg.key().decode("utf-8")
                     request_id = int(key_str)
-                except (ValueError) as e:
-                    logger.error(f"Failed to parse message key to request ID: {e}")
+                except ValueError as e:
+                    logger.error(
+                        f"Failed to parse message key to request ID: {e}"
+                    )
                     return
 
             response = moderation_pb2.ModerateObjectResponse()
@@ -97,9 +122,9 @@ def _process_message(self, msg):
                     self.callback(response, request_id)
                 except Exception as e:
                     logger.error(f"Error in callback: {e}")
-            
+
         except Exception as e:
             logger.error(f"Failed to process message: {e}")
-        
+
     def is_running(self) -> bool:
-        return self.running
+        return self.running

src/commentservice/kafka/producer.py → src/commentservice/clients/kafka/producer.py

@@ -1,41 +1,45 @@
-from confluent_kafka import Producer
-from confluent_kafka import KafkaException
 import logging
-from .config import KafkaConfig
+from typing import Any
+
+from confluent_kafka import (
+    KafkaException,
+    Producer,
+)
+
 from commentservice.grpc import moderation_pb2
-import time
+
+from .config import KafkaConfig
 
 logger = logging.getLogger(__name__)
 
-class ModerationRequestProducer:
 
-    def __init__(self, config: KafkaConfig):
+class ModerationRequestProducer:
+    def __init__(self, config: KafkaConfig) -> None:
         self.config = config
         self.producer = Producer(config.get_producer_config())
         self.topic = config.request_topic
 
-        logger.info(f"ModerationRequestProducer 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)}")
                 return True
-            
+
             except BufferError as e:
                 logger.error(f"Producer buffer full, message not queued: {e}")
                 return False
@@ -46,9 +50,8 @@ def send_moderation_request(self, request: moderation_pb2.ModerateObjectRequest,
                 logger.error(f"Unexpected error producing message: {e}")
                 return False
         return False
-
-
-    def _delivery_callback(self, err, msg):
+
+    def _delivery_callback(self, err: object, msg: Any) -> None:
 
         if err is not None:
             logger.error(f"Message delivery failed: {err}")
@@ -57,18 +60,18 @@ def _delivery_callback(self, err, msg):
                 f"Message delivered: topic={msg.topic()}, "
                 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()
-
-