From 093f667894f1dccc01bd8a5d3f25c0cd88b647e0 Mon Sep 17 00:00:00 2001 From: Andrey Kataev Date: Wed, 1 Apr 2026 23:12:25 +0300 Subject: [PATCH 1/2] =?UTF-8?q?=D0=98=D0=BD=D1=82=D0=B5=D0=B3=D1=80=D0=B8?= =?UTF-8?q?=D1=80=D0=BE=D0=B2=D0=B0=D0=BB=20pypika?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pdm.lock | 26 +++++- pyproject.toml | 2 + src/commentservice/clients/kafka/config.py | 46 ++++++++++ .../{ => clients}/kafka/consumer.py | 87 ++++++++++++------- .../{ => clients}/kafka/producer.py | 53 +++++------ .../clients/moderation_service.py | 85 ++++++++++++++++++ src/commentservice/handler/get_comments.py | 12 ++- src/commentservice/kafka/config.py | 38 -------- .../kafka/moderaiton_service.py | 83 ------------------ .../repository/create_comment.py | 20 +++-- src/commentservice/repository/edit_comment.py | 24 ++--- src/commentservice/repository/get_comments.py | 24 +++-- src/commentservice/repository/model.py | 12 ++- src/commentservice/repository/repository.py | 10 ++- src/commentservice/repository/set_status.py | 21 +++-- src/commentservice/repository/statuses.py | 7 -- .../repository/update_comment_status.py | 41 +++++---- src/commentservice/server.py | 5 +- src/commentservice/service/service.py | 14 ++- tests/handler/test_get_comments.py | 4 +- tests/repository/test_create_comment.py | 9 +- tests/repository/test_edit_comment.py | 20 ++--- tests/repository/test_get_comments.py | 14 +-- tests/repository/test_set_status.py | 15 ++-- tests/service/test_create_comment.py | 4 +- tests/service/test_edit_comment.py | 3 +- tests/service/test_get_comments.py | 7 +- tests/service/test_set_status.py | 3 +- 28 files changed, 399 insertions(+), 290 deletions(-) create mode 100644 src/commentservice/clients/kafka/config.py rename src/commentservice/{ => clients}/kafka/consumer.py (61%) rename src/commentservice/{ => clients}/kafka/producer.py (65%) create mode 100644 src/commentservice/clients/moderation_service.py delete mode 100644 src/commentservice/kafka/config.py delete mode 100644 src/commentservice/kafka/moderaiton_service.py delete mode 100644 src/commentservice/repository/statuses.py diff --git a/pdm.lock b/pdm.lock index 29a650f..dec64a1 100644 --- a/pdm.lock +++ b/pdm.lock @@ -5,7 +5,7 @@ groups = ["default", "dev"] strategy = ["inherit_metadata"] lock_version = "4.5.0" -content_hash = "sha256:57eeea1d5e7bf9ee1ad0acf79609855f60ee838ccb6d83752803a21ae3fcd7fd" +content_hash = "sha256:b6dc4c9db030e1429bea9c1b3bdcc2f0149682afcd48470ca80d8e1d0e840b10" [[metadata.targets]] requires_python = "~=3.13" @@ -671,6 +671,19 @@ files = [ {file = "pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887"}, ] +[[package]] +name = "pypika" +version = "0.51.1" +summary = "A SQL query builder API for Python" +groups = ["default"] +dependencies = [ + "typing-extensions>=4.5.0; python_version < \"3.11\"", +] +files = [ + {file = "pypika-0.51.1-py2.py3-none-any.whl", hash = "sha256:77985b4d7ce71b9905255bf12468cf598349e98837c037541cfc240e528aec46"}, + {file = "pypika-0.51.1.tar.gz", hash = "sha256:c30c7c1048fbf056fd3920c5a2b88b0c29dd190a9b2bee971fd17e4abe4d0ebe"}, +] + [[package]] name = "pytest" version = "8.4.2" @@ -818,6 +831,17 @@ files = [ {file = "sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc"}, ] +[[package]] +name = "types-confluent-kafka" +version = "1.2.2" +requires_python = "<4.0,>=3.8" +summary = "" +groups = ["dev"] +files = [ + {file = "types_confluent_kafka-1.2.2-py3-none-any.whl", hash = "sha256:61dbcf5223d48593ec925f69ce39f63b5ec79902b3f27e496ff0e7bceeba4baf"}, + {file = "types_confluent_kafka-1.2.2.tar.gz", hash = "sha256:c1e4095ed8bd87b9f2b05f0b766ad0c2ade4e19da7e41c91542c805639d0bbef"}, +] + [[package]] name = "types-protobuf" version = "6.32.1.20250918" diff --git a/pyproject.toml b/pyproject.toml index cc316b4..886550b 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/src/commentservice/clients/kafka/config.py b/src/commentservice/clients/kafka/config.py new file mode 100644 index 0000000..fb61441 --- /dev/null +++ b/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 diff --git a/src/commentservice/kafka/consumer.py b/src/commentservice/clients/kafka/consumer.py similarity index 61% rename from src/commentservice/kafka/consumer.py rename to src/commentservice/clients/kafka/consumer.py index 0869afe..08c84d6 100644 --- a/src/commentservice/kafka/consumer.py +++ b/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 \ No newline at end of file + return self.running diff --git a/src/commentservice/kafka/producer.py b/src/commentservice/clients/kafka/producer.py similarity index 65% rename from src/commentservice/kafka/producer.py rename to src/commentservice/clients/kafka/producer.py index e5c4d1c..9e2cb9e 100644 --- a/src/commentservice/kafka/producer.py +++ b/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() - - \ No newline at end of file diff --git a/src/commentservice/clients/moderation_service.py b/src/commentservice/clients/moderation_service.py new file mode 100644 index 0000000..8cbb4ca --- /dev/null +++ b/src/commentservice/clients/moderation_service.py @@ -0,0 +1,85 @@ +import asyncio +import logging +from concurrent.futures import Future + +from commentservice.grpc import moderation_pb2 +from commentservice.repository.repository import CommentRepository + +from .kafka.config import KafkaConfig +from .kafka.consumer import ModerationResponseConsumer +from .kafka.producer import ModerationRequestProducer + +logger = logging.getLogger(__name__) + + +class ModerationService: + def __init__( + self, repo: CommentRepository, loop: asyncio.AbstractEventLoop + ): + self.config = KafkaConfig() + self._repo = repo + self._loop = loop + + self.producer = ModerationRequestProducer(self.config) + self.consumer = ModerationResponseConsumer( + self.config, callback=self._handle_moderation_response + ) + + self.consumer.start() + + def request_moderaiton(self, comment_id: int, comment_text: str) -> bool: + try: + request = moderation_pb2.ModerateObjectRequest( + id=comment_id, + text=comment_text, + type=moderation_pb2.OBJECT_TYPE_COMMENT_TEXT, + ) + + success = self.producer.send_moderation_request(request) + + if not success: + logger.error( + f"send moderation request error: comment_id={comment_id}" + ) + + return success + + except Exception as e: + logger.error(f"Error requesting moderation: {e}") + return False + + def _handle_moderation_response( + self, response: moderation_pb2.ModerateObjectResponse, request_id: int + ) -> None: + + try: + is_flagged = response.success + self._update_comment_status(request_id, is_flagged) + + except Exception as e: + logger.error(f"Error handling moderation response: {e}") + + def _update_comment_status( + self, comment_id: int, is_flagged: bool + ) -> None: + coroutine = self._repo.update_comment_status(comment_id, is_flagged) + + future: Future[bool] = asyncio.run_coroutine_threadsafe( + coroutine, self._loop + ) + + def _log_result(done_future: Future[bool]) -> None: + try: + success = done_future.result() + if not success: + logger.error( + f"update comment status error: comment_id={comment_id}" + ) + except Exception as e: + logger.error(f"Error updating comment status: {e}") + + future.add_done_callback(_log_result) + + def shutdown(self) -> None: + self.consumer.stop() + self.producer.flush() diff --git a/src/commentservice/handler/get_comments.py b/src/commentservice/handler/get_comments.py index faa96c6..3e178c8 100644 --- a/src/commentservice/handler/get_comments.py +++ b/src/commentservice/handler/get_comments.py @@ -27,18 +27,16 @@ async def GetComments( ) -> comment_pb2.GetCommentsResponse: comments = await service.get_comments(mod_id=request.mod_id) return comment_pb2.GetCommentsResponse( - mod_id=request.mod_id, comments=convertCommentsToProto(comments) + mod_id=request.mod_id, comments=convert_comments_to_proto(comments) ) -def convertCommentToProto(comment: Comment) -> comment_pb2.Comment: - created_at_ts = dt_to_ts(comment.created_at) - +def convert_comment_to_proto(comment: Comment) -> comment_pb2.Comment: comment_proto = comment_pb2.Comment( id=comment.id, author_id=comment.author_id, text=comment.text, - created_at=created_at_ts, + created_at=dt_to_ts(comment.created_at), ) if comment.edited_at is not None: @@ -48,7 +46,7 @@ def convertCommentToProto(comment: Comment) -> comment_pb2.Comment: return comment_proto -def convertCommentsToProto( +def convert_comments_to_proto( comments: list[Comment], ) -> list[comment_pb2.Comment]: - return [convertCommentToProto(i) for i in comments] + return [convert_comment_to_proto(i) for i in comments] diff --git a/src/commentservice/kafka/config.py b/src/commentservice/kafka/config.py deleted file mode 100644 index 12caa70..0000000 --- a/src/commentservice/kafka/config.py +++ /dev/null @@ -1,38 +0,0 @@ -import os - -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', '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): - - config = { - '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): - - 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 - } - return config \ No newline at end of file diff --git a/src/commentservice/kafka/moderaiton_service.py b/src/commentservice/kafka/moderaiton_service.py deleted file mode 100644 index 6e50882..0000000 --- a/src/commentservice/kafka/moderaiton_service.py +++ /dev/null @@ -1,83 +0,0 @@ -import logging -import asyncio -from commentservice.repository.repository import CommentRepository -from ..kafka.producer import ModerationRequestProducer -from ..kafka.consumer import ModerationResponseConsumer -from ..kafka.config import KafkaConfig -from commentservice.grpc import moderation_pb2 - -logger = logging.getLogger(__name__) - -class ModerationService: - - def __init__(self, repo: CommentRepository, loop: asyncio.AbstractEventLoop): - self.config = KafkaConfig() - self._repo = repo - self._loop = loop - - self.producer = ModerationRequestProducer(self.config) - self.consumer = ModerationResponseConsumer( - self.config, - callback=self._handle_moderation_response - ) - - self.consumer.start() - - logger.info("ModerationService initialized") - - def request_moderaiton(self, comment_id: int, comment_text: str) -> bool: - try: - request = moderation_pb2.ModerateObjectRequest() - request.id = comment_id - request.text = comment_text - request.type = moderation_pb2.OBJECT_TYPE_COMMENT_TEXT - - success = self.producer.send_moderation_request(request) - - if success: - logger.info(f"Moderation request sent: comment_id={comment_id}") - else: - logger.error(f"Failed to send moderation request: comment_id={comment_id}") - - return success - - except Exception as e: - logger.error(f"Error requesting moderation: {e}") - return False - - def _handle_moderation_response(self, response: moderation_pb2.ModerateObjectResponse, request_id: int): - - try: - is_flagged = response.success - - logger.info( - f"Moderation result received: comment_id={request_id}, " - f"flagged={is_flagged}" - ) - - self._update_comment_status(request_id, is_flagged) - - except Exception as e: - logger.error(f"Error handling moderation response: {e}") - - def _update_comment_status(self, comment_id: int, is_flagged: bool) -> None: - coroutine = self._repo.update_comment_status(comment_id, is_flagged) - - future = asyncio.run_coroutine_threadsafe(coroutine, self._loop) - - def _log_result(future: asyncio.Future) -> None: - try: - success = future.result() - if not success: - logger.error(f"Failed to update comment status: comment_id={comment_id}") - except Exception as e: - logger.error(f"Error updating comment status: {e}") - - future.add_done_callback(_log_result) - - def shutdown(self): - self.consumer.stop() - self.producer.flush() - logger.info("ModerationService shutdown complete") - - \ No newline at end of file diff --git a/src/commentservice/repository/create_comment.py b/src/commentservice/repository/create_comment.py index 2ced7ef..7bec253 100644 --- a/src/commentservice/repository/create_comment.py +++ b/src/commentservice/repository/create_comment.py @@ -1,19 +1,23 @@ from typing import cast + from asyncpg import Pool +from pypika import Parameter, PostgreSQLQuery, Table async def create_comment( db_pool: Pool, mod_id: int, author_id: int, text: str ) -> int: + comments = Table("comments") + + query = ( + PostgreSQLQuery.into(comments) # type: ignore[operator] + .columns(comments.mod_id, comments.author_id, comments.text) + .insert(Parameter("$1"), Parameter("$2"), Parameter("$3")) + .returning(comments.id) + ) + async with db_pool.acquire() as conn: comment_id = await conn.fetchval( - """ - INSERT INTO comments (mod_id, author_id, text) - VALUES ($1, $2, $3) - RETURNING id - """, - mod_id, - author_id, - text + query.get_sql(), mod_id, author_id, text ) return cast(int, comment_id) diff --git a/src/commentservice/repository/edit_comment.py b/src/commentservice/repository/edit_comment.py index 1e1a761..4e00c43 100644 --- a/src/commentservice/repository/edit_comment.py +++ b/src/commentservice/repository/edit_comment.py @@ -1,17 +1,21 @@ from asyncpg import Pool +from pypika import Parameter, PostgreSQLQuery, Table, functions async def edit_comment(db_pool: Pool, comment_id: int, text: str) -> bool: - async with db_pool.acquire() as conn: - updated_id = await conn.fetchval( - """ - UPDATE comments - SET text = $1, edited_at = NOW() - WHERE id = $2 - RETURNING id - """, - text, - comment_id, + comments = Table("comments") + + query = ( + PostgreSQLQuery.update(comments) # type: ignore[operator] + .set(comments.text, Parameter("$1")) + .set( + comments.edited_at, + functions.Now(), # type: ignore[no-untyped-call] ) + .where(comments.id == Parameter("$2")) + .returning(comments.id) + ) + async with db_pool.acquire() as conn: + updated_id = await conn.fetchval(query.get_sql(), text, comment_id) return updated_id is not None diff --git a/src/commentservice/repository/get_comments.py b/src/commentservice/repository/get_comments.py index 04e049b..95bb720 100644 --- a/src/commentservice/repository/get_comments.py +++ b/src/commentservice/repository/get_comments.py @@ -1,17 +1,25 @@ from asyncpg import Pool +from pypika import Parameter, PostgreSQLQuery, Table from commentservice.repository.model import Comment async def get_comments(db_pool: Pool, mod_id: int) -> list[Comment]: - async with db_pool.acquire() as conn: - rows = await conn.fetch( - """ - SELECT id, author_id, text, created_at, edited_at, status - FROM comments - WHERE mod_id = $1 - """, - mod_id, + comments = Table("comments") + + query = ( + PostgreSQLQuery.from_(comments) + .select( + comments.id, + comments.author_id, + comments.text, + comments.created_at, + comments.edited_at, + comments.status, ) + .where(comments.mod_id == Parameter("$1")) + ) + async with db_pool.acquire() as conn: + rows = await conn.fetch(query.get_sql(), mod_id) return [Comment(**row) for row in rows] diff --git a/src/commentservice/repository/model.py b/src/commentservice/repository/model.py index f830c55..e278de5 100644 --- a/src/commentservice/repository/model.py +++ b/src/commentservice/repository/model.py @@ -1,6 +1,14 @@ from dataclasses import dataclass from datetime import datetime -from commentservice.repository.statuses import CommentStatus +from enum import StrEnum + + +class CommentStatus(StrEnum): + DELETED = "DELETED" + HIDDEN = "HIDDEN" + APPROVED = "APPROVED" + ON_MODERATION = "ON_MODERATION" + @dataclass class Comment: @@ -9,4 +17,4 @@ class Comment: text: str created_at: datetime edited_at: datetime | None - status: CommentStatus \ No newline at end of file + status: CommentStatus diff --git a/src/commentservice/repository/repository.py b/src/commentservice/repository/repository.py index 0762d21..585a47a 100644 --- a/src/commentservice/repository/repository.py +++ b/src/commentservice/repository/repository.py @@ -11,11 +11,11 @@ ) from commentservice.repository.model import Comment from commentservice.repository.set_status import set_status as _set_status - from commentservice.repository.update_comment_status import ( update_comment_status as _update_comment_status, ) + class CommentRepository: def __init__(self, db_pool: asyncpg.Pool): self._db_pool: asyncpg.Pool = db_pool @@ -37,5 +37,9 @@ async def set_status(self, comment_id: int, status: str) -> bool: async def get_comments(self, mod_id: int) -> list[Comment]: return await _get_comments(self._db_pool, mod_id) - async def update_comment_status(self, comment_id: int, is_flagged: bool) -> bool: - return await _update_comment_status(self._db_pool, comment_id, is_flagged) \ No newline at end of file + async def update_comment_status( + self, comment_id: int, is_flagged: bool + ) -> bool: + return await _update_comment_status( + self._db_pool, comment_id, is_flagged + ) diff --git a/src/commentservice/repository/set_status.py b/src/commentservice/repository/set_status.py index 26374fa..6757e34 100644 --- a/src/commentservice/repository/set_status.py +++ b/src/commentservice/repository/set_status.py @@ -1,19 +1,24 @@ from asyncpg import Pool +from pypika import Parameter, PostgreSQLQuery, Table async def set_status(db_pool: Pool, comment_id: int, status: str) -> bool: + comments = Table("comments") + + query = ( + PostgreSQLQuery.update(comments) # type: ignore[operator] + .set(comments.status, Parameter("$1")) + .where(comments.id == Parameter("$2")) + .returning(comments.id) + ) + try: async with db_pool.acquire() as conn: - result = await conn.execute( - """ - UPDATE comments - SET status = $1 - WHERE id = $2 - """, + updated_id = await conn.fetchval( + query.get_sql(), status, comment_id, ) - rows_affected = int(result.split()[-1]) if result else 0 - return rows_affected > 0 + return updated_id is not None except Exception: return False diff --git a/src/commentservice/repository/statuses.py b/src/commentservice/repository/statuses.py deleted file mode 100644 index e378938..0000000 --- a/src/commentservice/repository/statuses.py +++ /dev/null @@ -1,7 +0,0 @@ -from enum import Enum - -class CommentStatus(str, Enum): - DELETED = "DELETED" - HIDDEN = "HIDDEN" - APPROVED = "APPROVED" - ON_MODERATION = "ON_MODERATION" \ No newline at end of file diff --git a/src/commentservice/repository/update_comment_status.py b/src/commentservice/repository/update_comment_status.py index 05e6438..0ceebba 100644 --- a/src/commentservice/repository/update_comment_status.py +++ b/src/commentservice/repository/update_comment_status.py @@ -1,21 +1,30 @@ from asyncpg import Pool -from commentservice.repository.statuses import CommentStatus +from pypika import Parameter, PostgreSQLQuery, Table + +from commentservice.repository.model import CommentStatus + + +async def update_comment_status( + db_pool: Pool, comment_id: int, is_flagged: bool +) -> bool: + comments = Table("comments") + status = ( + CommentStatus.HIDDEN.value + if is_flagged + else CommentStatus.APPROVED.value + ) + + query = ( + PostgreSQLQuery.update(comments) # type: ignore[operator] + .set(comments.status, Parameter("$1")) + .where(comments.id == Parameter("$2")) + .returning(comments.id) + ) -async def update_comment_status(db_pool: Pool, comment_id: int, is_flagged: bool) -> bool: - if is_flagged: - status = CommentStatus.HIDDEN.value - else: - status = CommentStatus.APPROVED.value - async with db_pool.acquire() as conn: updated_id = await conn.fetchval( - """ - UPDATE comments - SET status = $1::comment_status - WHERE id = $2 - RETURNING id - """, - status, - comment_id + query.get_sql(), + status, + comment_id, ) - return updated_id is not None \ No newline at end of file + return updated_id is not None diff --git a/src/commentservice/server.py b/src/commentservice/server.py index 29e2bcf..961d5e3 100644 --- a/src/commentservice/server.py +++ b/src/commentservice/server.py @@ -1,18 +1,17 @@ import asyncio import logging -import signal from concurrent import futures import asyncpg import grpc from grpc_reflection.v1alpha import reflection +from commentservice.clients.moderation_service import ModerationService from commentservice.grpc import comment_pb2, comment_pb2_grpc from commentservice.handler.handler import CommentHandler from commentservice.repository.repository import CommentRepository from commentservice.service.service import CommentService from commentservice.settings import Settings -from commentservice.kafka.moderaiton_service import ModerationService async def serve() -> None: @@ -27,7 +26,7 @@ async def serve() -> None: ) repo = CommentRepository(db_pool) - loop = asyncio.get_running_loop() + loop = asyncio.get_running_loop() moderation_service = ModerationService(repo=repo, loop=loop) service = CommentService(repo, moderation_service) diff --git a/src/commentservice/service/service.py b/src/commentservice/service/service.py index 70069ce..dbf84ed 100644 --- a/src/commentservice/service/service.py +++ b/src/commentservice/service/service.py @@ -1,3 +1,5 @@ +from typing import TYPE_CHECKING + from commentservice.repository.model import Comment from commentservice.repository.repository import CommentRepository from commentservice.service.create_comment import ( @@ -5,21 +7,25 @@ ) from commentservice.service.edit_comment import edit_comment as _edit_comment from commentservice.service.get_comments import get_comments as _get_comments -from commentservice.kafka.moderaiton_service import ModerationService from commentservice.service.set_status import set_status as _set_status +if TYPE_CHECKING: + from commentservice.clients.moderation_service import ModerationService + class CommentService: - def __init__(self, repo: CommentRepository, moderation_service: ModerationService): + def __init__( + self, repo: CommentRepository, moderation_service: "ModerationService" + ): self._repo = repo self.moderation_service = moderation_service async def create_comment( self, mod_id: int, author_id: int, text: str ) -> int: - + comment_id = await _create_comment(self._repo, mod_id, author_id, text) - + self.moderation_service.request_moderaiton(comment_id, text) return comment_id diff --git a/tests/handler/test_get_comments.py b/tests/handler/test_get_comments.py index 4d4937b..1b0d6f5 100644 --- a/tests/handler/test_get_comments.py +++ b/tests/handler/test_get_comments.py @@ -10,7 +10,7 @@ GetCommentsResponse, ) from commentservice.handler.get_comments import GetComments, ts_to_dt -from commentservice.repository.model import Comment +from commentservice.repository.model import Comment, CommentStatus from commentservice.service.service import CommentService @@ -30,6 +30,7 @@ async def test_get_comments_success( text=faker.sentence(), created_at=earlier, edited_at=None, + status=CommentStatus.ON_MODERATION, ) comment2 = Comment( id=faker.random_int(min=1, max=100000), @@ -37,6 +38,7 @@ async def test_get_comments_success( text=faker.sentence(), created_at=now, edited_at=now, + status=CommentStatus.APPROVED, ) comments = [comment1, comment2] fake_service.get_comments.return_value = comments diff --git a/tests/repository/test_create_comment.py b/tests/repository/test_create_comment.py index 564d5c5..41b035c 100644 --- a/tests/repository/test_create_comment.py +++ b/tests/repository/test_create_comment.py @@ -32,11 +32,10 @@ async def test_repo_create_comment_returns_id( assert new_id == comment_id assert conn.fetchval.await_count == 1 - expected_sql = """ - INSERT INTO comments (mod_id, author_id, text) - VALUES ($1, $2, $3) - RETURNING id - """ + expected_sql = ( + 'INSERT INTO "comments" ("mod_id","author_id","text") ' + 'VALUES ($1,$2,$3) RETURNING "id"' + ) actual_sql = conn.fetchval.await_args.args[0] assert ( textwrap.dedent(actual_sql).strip() diff --git a/tests/repository/test_edit_comment.py b/tests/repository/test_edit_comment.py index 198dea4..105c871 100644 --- a/tests/repository/test_edit_comment.py +++ b/tests/repository/test_edit_comment.py @@ -26,12 +26,10 @@ async def test_repo_edit_comment_success( conn.fetchval.return_value = cid ok = await repo.edit_comment(comment_id=cid, new_text=new_text) assert ok is True - expected_sql = """ - UPDATE comments - SET text = $1, edited_at = NOW() - WHERE id = $2 - RETURNING id - """ + expected_sql = ( + 'UPDATE "comments" SET "text"=$1,"edited_at"=NOW() ' + 'WHERE "id"=$2 RETURNING "comments"."id"' + ) actual_sql = conn.fetchval.await_args.args[0] assert ( textwrap.dedent(actual_sql).strip() @@ -57,12 +55,10 @@ async def test_repo_edit_comment_not_found( new_text = faker.sentence() ok = await repo.edit_comment(comment_id=cid, new_text=new_text) assert ok is False - expected_sql = """ - UPDATE comments - SET text = $1, edited_at = NOW() - WHERE id = $2 - RETURNING id - """ + expected_sql = ( + 'UPDATE "comments" SET "text"=$1,"edited_at"=NOW() ' + 'WHERE "id"=$2 RETURNING "comments"."id"' + ) actual_sql = conn.fetchval.await_args.args[0] assert ( textwrap.dedent(actual_sql).strip() diff --git a/tests/repository/test_get_comments.py b/tests/repository/test_get_comments.py index 03981b5..6451365 100644 --- a/tests/repository/test_get_comments.py +++ b/tests/repository/test_get_comments.py @@ -6,6 +6,7 @@ from faker import Faker from pytest_mock import MockerFixture +from commentservice.repository.model import CommentStatus from commentservice.repository.repository import CommentRepository @@ -20,6 +21,7 @@ async def test_repo_get_comments_maps_rows( "text": faker.sentence(), "created_at": faker.date_time(tzinfo=UTC), "edited_at": None, + "status": CommentStatus.ON_MODERATION, }, { "id": faker.random_int(), @@ -27,6 +29,7 @@ async def test_repo_get_comments_maps_rows( "text": faker.sentence(), "created_at": faker.date_time(tzinfo=UTC), "edited_at": faker.date_time(tzinfo=UTC), + "status": CommentStatus.APPROVED, }, ] @@ -46,13 +49,14 @@ async def test_repo_get_comments_maps_rows( assert len(out) == 2 assert out[0].id == rows[0]["id"] and out[0].text == rows[0]["text"] assert out[0].edited_at is None + assert out[0].status == rows[0]["status"] assert out[1].id == rows[1]["id"] and out[1].text == rows[1]["text"] assert out[1].edited_at is not None - expected_sql = """ - SELECT id, author_id, text, created_at, edited_at - FROM comments - WHERE mod_id = $1 - """ + assert out[1].status == rows[1]["status"] + expected_sql = ( + 'SELECT "id","author_id","text","created_at","edited_at","status" ' + 'FROM "comments" WHERE "mod_id"=$1' + ) actual_sql = conn.fetch.await_args.args[0] assert ( textwrap.dedent(actual_sql).strip() diff --git a/tests/repository/test_set_status.py b/tests/repository/test_set_status.py index 17774af..6ba7b70 100644 --- a/tests/repository/test_set_status.py +++ b/tests/repository/test_set_status.py @@ -12,7 +12,7 @@ async def test_set_status_returns_true_on_update( mocker: MockerFixture, faker: Faker ) -> None: conn = mocker.Mock() - conn.execute = mocker.AsyncMock(return_value="UPDATE 1") + conn.fetchval = mocker.AsyncMock(return_value=1) acquire_cm = mocker.AsyncMock() acquire_cm.__aenter__.return_value = conn acquire_cm.__aexit__.return_value = None @@ -25,17 +25,16 @@ async def test_set_status_returns_true_on_update( result = await set_status(pool, comment_id, status) assert result is True - expected_sql = """ - UPDATE comments - SET status = $1 - WHERE id = $2 - """ - actual_sql = conn.execute.await_args.args[0] + expected_sql = ( + 'UPDATE "comments" SET "status"=$1 ' + 'WHERE "id"=$2 RETURNING "comments"."id"' + ) + actual_sql = conn.fetchval.await_args.args[0] assert ( textwrap.dedent(actual_sql).strip() == textwrap.dedent(expected_sql).strip() ) - assert conn.execute.await_args.args[1:] == (status, comment_id) + assert conn.fetchval.await_args.args[1:] == (status, comment_id) @pytest.mark.asyncio diff --git a/tests/service/test_create_comment.py b/tests/service/test_create_comment.py index 4f6e216..165401b 100644 --- a/tests/service/test_create_comment.py +++ b/tests/service/test_create_comment.py @@ -13,6 +13,7 @@ async def test_service_create_comment( mocker: MockerFixture, faker: Faker ) -> None: fake_repo = mocker.Mock(spec=CommentRepository) + moderation_service = mocker.Mock() new_id = faker.random_int(min=1, max=100000) fake_repo.create_comment = AsyncMock(return_value=new_id) @@ -20,10 +21,11 @@ async def test_service_create_comment( author_id = faker.random_int(min=1, max=100000) text = faker.sentence() - service = CommentService(fake_repo) + service = CommentService(fake_repo, moderation_service) result = await service.create_comment( mod_id=mod_id, author_id=author_id, text=text ) assert result == new_id fake_repo.create_comment.assert_awaited_once_with(mod_id, author_id, text) + moderation_service.request_moderaiton.assert_called_once_with(new_id, text) diff --git a/tests/service/test_edit_comment.py b/tests/service/test_edit_comment.py index c9d750d..015cb95 100644 --- a/tests/service/test_edit_comment.py +++ b/tests/service/test_edit_comment.py @@ -13,12 +13,13 @@ async def test_service_edit_comment( mocker: MockerFixture, faker: Faker ) -> None: fake_repo = mocker.Mock(spec=CommentRepository) + moderation_service = mocker.Mock() fake_repo.edit_comment = AsyncMock(return_value=True) comment_id = faker.random_int(min=1, max=100000) new_text = faker.sentence() - service = CommentService(fake_repo) + service = CommentService(fake_repo, moderation_service) result = await service.edit_comment( comment_id=comment_id, new_text=new_text ) diff --git a/tests/service/test_get_comments.py b/tests/service/test_get_comments.py index 6ad2343..0001a89 100644 --- a/tests/service/test_get_comments.py +++ b/tests/service/test_get_comments.py @@ -5,7 +5,7 @@ from faker import Faker from pytest_mock import MockerFixture -from commentservice.repository.model import Comment +from commentservice.repository.model import Comment, CommentStatus from commentservice.repository.repository import CommentRepository from commentservice.service.service import CommentService @@ -15,6 +15,7 @@ async def test_service_get_comments( mocker: MockerFixture, faker: Faker ) -> None: fake_repo = mocker.Mock(spec=CommentRepository) + moderation_service = mocker.Mock() comments = [ Comment( id=faker.random_int(), @@ -22,6 +23,7 @@ async def test_service_get_comments( text=faker.sentence(), created_at=faker.date_time(tzinfo=UTC), edited_at=None, + status=CommentStatus.ON_MODERATION, ), Comment( id=faker.random_int(), @@ -29,12 +31,13 @@ async def test_service_get_comments( text=faker.sentence(), created_at=faker.date_time(tzinfo=UTC), edited_at=None, + status=CommentStatus.APPROVED, ), ] fake_repo.get_comments = AsyncMock(return_value=comments) mod_id = faker.random_int(min=1, max=100000) - service = CommentService(fake_repo) + service = CommentService(fake_repo, moderation_service) out = await service.get_comments(mod_id=mod_id) assert out == comments diff --git a/tests/service/test_set_status.py b/tests/service/test_set_status.py index 340f782..2733f21 100644 --- a/tests/service/test_set_status.py +++ b/tests/service/test_set_status.py @@ -13,10 +13,11 @@ async def test_service_set_status_uses_helper( mocker: MockerFixture, faker: Faker ) -> None: repo = mocker.Mock(spec=CommentRepository) + moderation_service = mocker.Mock() helper = AsyncMock(return_value=True) mocker.patch("commentservice.service.service._set_status", helper) - service = CommentService(repo) + service = CommentService(repo, moderation_service) comment_id = faker.random_int(min=1, max=100000) status = "DELETED" From 25cc8b88fca9c2d63cf63db8182c82a4f68e12d8 Mon Sep 17 00:00:00 2001 From: Andrey Kataev Date: Wed, 1 Apr 2026 23:31:26 +0300 Subject: [PATCH 2/2] =?UTF-8?q?=D0=9E=D0=B1=D0=BD=D0=BE=D0=B2=D0=B8=D0=BB?= =?UTF-8?q?=20=D0=B2=D0=BE=D1=80=D0=BA=D1=84=D0=BB=D0=BE=D1=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/lint-and-test.yml | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/.github/workflows/lint-and-test.yml b/.github/workflows/lint-and-test.yml index 293f975..a90c314 100644 --- a/.github/workflows/lint-and-test.yml +++ b/.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 }}