diff --git a/amqp/rabbitmq/user_management_sample.py b/amqp/rabbitmq/user_management_sample.py index b6a2c72..3f3d4ba 100644 --- a/amqp/rabbitmq/user_management_sample.py +++ b/amqp/rabbitmq/user_management_sample.py @@ -1,11 +1,11 @@ import asyncio import logging -from typing import Dict, Any +from typing import Any, Dict from amqp.config.amq_configuration import AMQConfiguration from amqp.rabbitmq.user_management_service_client import ( UserManagementServiceClient, - UserManagementServiceException + UserManagementServiceException, ) @@ -14,39 +14,39 @@ async def get_user_info_from_token( ) -> Dict[str, Any]: """ Get user information from a JWT token. - + Args: client: User Management Service client token: JWT token - + Returns: User information - + Raises: UserManagementServiceException: If an error occurs """ try: # Validate the token token_info = await client.validate_token(token) - + # Extract user ID from token info user_id = None for item in token_info: if isinstance(item, dict) and "sub" in item: user_id = item["sub"] break - + if not user_id: raise UserManagementServiceException( "cleverthis.clevermicro.auth.invalid_token", "Token does not contain user ID (sub claim)", - {} + {}, ) - + # Query user information user_info = await client.query_user_by_id(user_id) return user_info - + except UserManagementServiceException as e: logging.error(f"User Management Service error: {e.exception_type}: {e.message}") raise @@ -57,19 +57,19 @@ async def get_user_info_from_token( async def main(): # Initialize configuration - config = AMQConfiguration() - + config = AMQConfiguration("") + # Create client client = UserManagementServiceClient(config) - + try: # Sample JWT token (replace with actual token) token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c" - + # Get user information user_info = await get_user_info_from_token(client, token) print(f"User information: {user_info}") - + except UserManagementServiceException as e: print(f"Error: {e.exception_type}: {e.message}") if e.additional_info: diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 8dea0e4..bcc7030 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -160,6 +160,12 @@ class DataMessageHandler: async def reconstitute_data_message( self, message: AbstractIncomingMessage ) -> DataMessage | None: + """ + DataMessage may be serialized as string in separate queue (addressing == 1) or be a MultiPart message with + files streamed over individual queues, one per file. + This function uses 'addressing' mode from message headers to determine how to deserialize (reconstitute) + the original message. + """ amq_message: DataMessage | None = None delivery_tag = message.delivery_tag addressing: int = message.headers.get("clevermicro.addressing", 0) @@ -246,8 +252,8 @@ class DataMessageHandler: async def reply_received_callback(self, message: AbstractIncomingMessage): """ - The reply for DataMessage that we sent out earlier - must match the received response with - the original request and respond to the caller. + The handler for reply for DataMessage that we sent out earlier - + the code matches the received response with the original request and passes the response to the caller. :param message: :return: """