client/client.py - master - tinypot/lab4

client/client.py
69 строк2 KB
Tim Polus
lab4: Task Management API через RabbitMQ
02 мар 2026, 06:17
02 мар 2026, 06:1752ab147
from __future__ import annotations

import json
import uuid

import pika


class RpcClient:

    def __init__(
        self,
        host: str = "localhost",
        api_key: str = "master-api-key",
        version: str = "v1",
    ):
        self.api_key = api_key
        self.version = version
        self._response: dict | None = None
        self._corr_id: str | None = None

        self._connection = pika.BlockingConnection(
            pika.ConnectionParameters(host=host)
        )
        self._channel = self._connection.channel()

        result = self._channel.queue_declare(queue="", exclusive=True)
        self._callback_queue = result.method.queue

        self._channel.basic_consume(
            queue=self._callback_queue,
            on_message_callback=self._on_response,
            auto_ack=True,
        )

    def _on_response(self, _ch, _method, properties, body):
        if properties.correlation_id == self._corr_id:
            self._response = json.loads(body)

    def call(self, action: str, data: dict | None = None, request_id: str | None = None) -> dict:
        self._response = None
        self._corr_id = request_id or str(uuid.uuid4())

        message = {
            "id": self._corr_id,
            "version": self.version,
            "action": action,
            "data": data or {},
            "auth": self.api_key,
        }

        self._channel.basic_publish(
            exchange="",
            routing_key="api.requests",
            body=json.dumps(message, ensure_ascii=False),
            properties=pika.BasicProperties(
                reply_to=self._callback_queue,
                correlation_id=self._corr_id,
                content_type="application/json",
            ),
        )

        while self._response is None:
            self._connection.process_data_events(time_limit=30)

        return self._response

    def close(self):
        self._connection.close()