amq-adapter-python/AMQ_ADAPTER_DEVELOPER_GUIDE.md

AMQ Adapter Developer Guide

Overview

The AMQ Adapter library enables FastAPI-based Business Services to communicate with the CleverMicro platform via RabbitMQ. This guide demonstrates how to build your own service using the AMQ Adapter, following the pattern established by the OTDemo application.

Table of Contents

1. Architecture Overview
2. Prerequisites
3. Project Structure
4. Implementation Steps
5. Configuration
6. Creating Your Adapter
7. Frontend Integration
8. Backend Processing
9. Testing and Deployment
10. Advanced Features

Architecture Overview

The AMQ Adapter follows a layered architecture:

┌─────────────────────────────────────────────────────────────┐
│                    Business Service                         │
├─────────────────────────────────────────────────────────────┤
│  FastAPI Endpoints  │  Your Custom Adapter  │  Backend     │
│                     │                       │  Processing  │
└─────────────────────┼───────────────────────┼───────────────┘
                      │                       │
┌─────────────────────┼───────────────────────┼───────────────┐
│    AMQ Adapter Library                      │               │
├─────────────────────────────────────────────────────────────┤
│  CleverThisServiceAdapter  │  AMQService  │  Router        │
│  (Base Class)              │              │                │
└─────────────────────┼───────────────────────┼───────────────┘
                      │                       │
┌─────────────────────┼───────────────────────┼───────────────┐
│                    RabbitMQ                 │               │
│  Exchanges  │  Queues  │  Routing Keys      │               │
└─────────────────────────────────────────────────────────────┘

Key Components

• Your Custom Adapter: Extends CleverThisServiceAdapter to handle service-specific logic
• AMQService: Manages RabbitMQ connections and message routing
• Router: Routes incoming AMQP messages to appropriate endpoints
• Configuration: Defines service settings, routing, and RabbitMQ connection details

Prerequisites

Before building your service, ensure you have:

1. Python 3.8+ installed
2. RabbitMQ server running (or access to CleverMicro platform)
3. FastAPI knowledge
4. AMQ Adapter library installed in your project

Dependencies

Add the following to your requirements.txt:

fastapi>=0.111.0
uvicorn[standard]>=0.29.0
python-multipart>=0.0.20
aio-pika>=9.0.0
pydantic>=2.0.0
# Add your AMQ Adapter library dependency

Project Structure

Follow this recommended structure for your Business Service:

your-business-service/
├── main.py                 # FastAPI application entry point
├── requirements.txt        # Python dependencies
├── config/
│   ├── frontend.properties # Frontend service configuration
│   └── backend.properties  # Backend service configuration
├── adapter/
│   └── your_adapter.py     # Your custom AMQ adapter
├── backend/
│   └── backend_processor.py # Backend message processing
└── tests/
    └── test_your_service.py

Implementation Steps

Step 1: Create Your Configuration Files

Create configuration files for both frontend and backend services.

config/frontend.properties:

# ====================================================================
# CleverMicro AMQ Adapter settings
# ====================================================================
[CleverMicro-AMQ]
cm.amq-adapter.service-name=your-service
cm.amq-adapter.generator-id=1
cm.amq-adapter.route-mapping=your-service:your-exchange:your-queue-1:api.cm.dev.cleverthis.localhost.test.python.frontend.#:5:0
cm.amq-adapter.require-authenticated-user=false
cm.amq-adapter.callbacks-enabled=True

# CleverMicro AMQ Adapter dispatcher settings
cm.dispatch.use-dlq=true
cm.dispatch.amq-host=localhost
cm.dispatch.amq-port=5672
cm.dispatch.amq-port-tls=5671
cm.dispatch.download-dir=/tmp/downloaded_files

# Backpressure settings
cm.backpressure.threshold=1
cm.backpressure.time-window=150
cm.backpressure.cpu-overload-duration=300
cm.backpressure.cpu-idle-duration=300

config/backend.properties:

# ====================================================================
# CleverMicro AMQ Adapter settings
# ====================================================================
[CleverMicro-AMQ]
cm.amq-adapter.service-name=your-service-backend
cm.amq-adapter.generator-id=1
cm.amq-adapter.route-mapping=your-service-backend:your-exchange:your-queue-2:your-service-backend.#:5:0
cm.amq-adapter.require-authenticated-user=false
cm.amq-adapter.callbacks-enabled=False

# CleverMicro AMQ Adapter dispatcher settings
cm.dispatch.use-dlq=true
cm.dispatch.amq-host=localhost
cm.dispatch.amq-port=5672
cm.dispatch.amq-port-tls=5671
cm.dispatch.download-dir=/tmp/downloaded_files

# Backpressure settings
cm.backpressure.threshold=1
cm.backpressure.time-window=150
cm.backpressure.cpu-overload-duration=300
cm.backpressure.cpu-idle-duration=300

Step 2: Create Your Custom Adapter

Create your custom adapter by extending CleverThisServiceAdapter:

"""
Your Business Service AMQP Adapter for handling API requests via AMQP.
This adapter is specific to your business service and handles service-specific logic.
"""

from threading import Thread
from typing import Any, List

from fastapi.routing import APIRoute

from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
from amqp.config.amq_configuration import AMQConfiguration
from amqp.service.amq_service import AMQService


class YourServiceAmqpAdapter(CleverThisServiceAdapter):
    """
    Your Business Service AMQP Adapter. Implements service-specific logic that overrides
    common logic defined by the AMQ Adapter library.

    Key responsibilities:
    1. Handle authentication and user management
    2. Process incoming AMQP messages
    3. Route messages to appropriate endpoints
    4. Manage service-specific business logic
    """

    def __init__(self, amq_configuration: AMQConfiguration, routes: List[APIRoute]):
        super().__init__(amq_configuration, routes=routes, session=None)
        self.amq_service: AMQService = AMQService(amq_configuration, self)
        self.thread = Thread(target=self.amq_service.run)
        self.thread.start()

    async def get_current_user(self, token: str) -> str:
        """
        Override the default get_current_user method to handle authentication.
        In this example, we simply return the token, but you should implement
        proper JWT token validation and user extraction.

        :param token: JWT token from the AMQP message
        :return: User identifier or user object
        """
        # TODO: Implement proper JWT token validation
        # Example: return await your_auth_service.validate_token(token)
        return token

    def get_endpoint(self, endpoint: Any) -> Any:
        """
        Override to provide custom endpoint selection logic.
        For example, you might want to use JSON wrapper endpoints
        or implement custom routing logic.

        :param endpoint: The original endpoint callable
        :return: The endpoint callable to use (original or modified)
        """
        return endpoint

Step 3: Create Your FastAPI Application

Create your main FastAPI application with endpoints:

"""
Your Business Service FastAPI Application
"""

import os
from fastapi import FastAPI, File, Form, UploadFile
from fastapi.routing import APIRoute

from amqp.config.amq_configuration import AMQConfiguration
from adapter.your_adapter import YourServiceAmqpAdapter


# Create FastAPI application
app = FastAPI(
    title="Your Business Service",
    description="A FastAPI application using AMQ Adapter for CleverMicro communication",
    version="1.0.0"
)

# Initialize your AMQ adapter
# This should be done after all routes are defined
your_adapter = None


@app.post("/api/v1/items")
async def create_item(
    name: str = Form(...),
    description: str = Form(...),
    file: UploadFile = File(...)
):
    """
    Create a new item with file upload.
    This endpoint will be available via both REST and AMQP.
    """
    # Your business logic here
    result = {
        "message": "Item created successfully",
        "name": name,
        "description": description,
        "filename": file.filename
    }

    # Example: Send a command to backend via AMQP
    if your_adapter:
        your_adapter.amq_service.command(
            "your-service-backend",
            "process_item",
            item_name=name,
            item_description=description
        )

    return result


@app.get("/api/v1/items")
async def get_items():
    """
    Retrieve all items.
    This endpoint will be available via both REST and AMQP.
    """
    # Your business logic here
    items = [
        {"id": "1", "name": "Item 1", "description": "Description 1"},
        {"id": "2", "name": "Item 2", "description": "Description 2"}
    ]

    return {"items": items}


@app.get("/health")
async def health_check():
    """
    Health check endpoint.
    """
    return {"status": "healthy", "service": "your-business-service"}


# Initialize the AMQ adapter after all routes are defined
def initialize_amq_adapter():
    global your_adapter
    config = AMQConfiguration("config/frontend.properties")
    your_adapter = YourServiceAmqpAdapter(config, app.routes)


# Initialize adapter when the application starts
@app.on_event("startup")
async def startup_event():
    initialize_amq_adapter()


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Step 4: Create Backend Processing

Create a backend processor for handling AMQP messages:

"""
Backend processor for your business service.
Handles AMQP messages and performs background processing.
"""

import asyncio
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import AMQRoute, DataMessage
from adapter.your_adapter import YourServiceAmqpAdapter


def process_item_static(item_name: str, item_description: str) -> str:
    """
    Static function to process an item.
    """
    print(f"STATIC Processing item: {item_name} - {item_description}")
    return f"STATIC PROCESSED: {item_name}"


class BackendProcessor:
    """
    Backend processor class for handling business logic.
    """

    def process_item(self, item_name: str, item_description: str) -> str:
        """
        Process an item using instance method.
        """
        print(f"CLASS Processing item: {item_name} - {item_description}")
        return f"CLASS PROCESSED: {item_name}"


async def processing_loop():
    """
    Main processing loop for handling AMQP messages.
    """
    # Wait for AMQ adapter to initialize
    while not otadapter.amq_service.future.done():
        await asyncio.sleep(0.1)

    # Find the route for backend processing
    route: AMQRoute = otadapter.amq_service.router.find_route_by_service("your-service-backend")
    generator = otadapter.amq_service.data_message_generator(route)

    print(f"Data message generator created: {generator}")

    # Start the generator
    message = await generator.asend(None)

    # Process messages
    while True:
        print(f"Received message: {message}")
        if message is None:
            break

        data_message: DataMessage = message

        # Extract message attributes
        print(f"Message attributes: {data_message.method()} {data_message.domain()} {data_message.path()}")

        # Process based on message path
        if data_message.path() == "process_item_static":
            item_name = data_message._base64body.decode("utf-8")
            process_item_static(item_name, "Static processing")
        elif data_message.path() == "process_item":
            item_name = data_message._base64body.decode("utf-8")
            processor = BackendProcessor()
            processor.process_item(item_name, "Class processing")

        # Get next message
        message = await generator.asend(True)


if __name__ == "__main__":
    # Initialize backend adapter
    otadapter = YourServiceAmqpAdapter(AMQConfiguration("config/backend.properties"), [])

    # Run the processing loop in the AMQ service event loop
    asyncio.run_coroutine_threadsafe(processing_loop(), otadapter.amq_service.loop)

    # Keep the program running
    otadapter.thread.join()

Configuration

Configuration Parameters

The configuration file contains several key sections:

AMQ Adapter Settings (cm.amq-adapter.*)

• service-name: Unique identifier for your service
• generator-id: Unique ID for message generation (usually 1)
• route-mapping: RabbitMQ routing configuration
• require-authenticated-user: Whether authentication is required
• callbacks-enabled: Whether to enable callback functionality

Dispatch Settings (cm.dispatch.*)

• amq-host: RabbitMQ host address
• amq-port: RabbitMQ port (5672 for non-TLS, 5671 for TLS)
• use-dlq: Whether to use Dead Letter Queue
• download-dir: Directory for downloaded files

Backpressure Settings (cm.backpressure.*)

• threshold: Backpressure threshold value
• time-window: Time window for backpressure calculation
• cpu-overload-duration: Duration for CPU overload detection
• cpu-idle-duration: Duration for CPU idle detection

Route Mapping Format

The route mapping follows this format:

service-name:exchange:queue:routing-key:prefetch:durable

Example:

your-service:your-exchange:your-queue-1:api.cm.dev.cleverthis.localhost.test.python.frontend.#:5:0

Frontend Integration

Key Integration Points

1. Adapter Initialization: Initialize your adapter after defining all FastAPI routes
2. Message Handling: The adapter automatically routes AMQP messages to your endpoints
3. Command Sending: Use amq_service.command() to send commands to backend services

Example Integration

# In your FastAPI application
@app.post("/api/v1/process")
async def process_data(data: dict):
    # Your business logic
    result = {"status": "processing", "data": data}

    # Send command to backend
    your_adapter.amq_service.command(
        "your-service-backend",
        "process_data",
        **data
    )

    return result

Backend Processing

Message Processing Pattern

1. Initialize Adapter: Create adapter with backend configuration
2. Wait for Connection: Wait for RabbitMQ connection to be established
3. Find Route: Locate the route for your backend service
4. Create Generator: Create message generator for the route
5. Process Messages: Continuously process incoming messages

Example Processing

async def process_messages():
    # Wait for connection
    while not adapter.amq_service.future.done():
        await asyncio.sleep(0.1)

    # Get route and generator
    route = adapter.amq_service.router.find_route_by_service("your-service-backend")
    generator = adapter.amq_service.data_message_generator(route)

    # Process messages
    message = await generator.asend(None)
    while message:
        # Process the message
        await handle_message(message)
        message = await generator.asend(True)

Testing and Deployment

Local Testing

1. Start RabbitMQ: Ensure RabbitMQ is running locally
2. Run Frontend: Start your FastAPI application
3. Run Backend: Start your backend processor
4. Test Endpoints: Use tools like curl or Postman to test REST endpoints
5. Monitor AMQP: Check RabbitMQ management interface for message flow

Docker Deployment

Create a Dockerfile for your service:

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker Compose

Create a docker-compose.yml for local development:

version: '3.8'

services:
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

  your-service-frontend:
    build: .
    ports:
      - "8000:8000"
    environment:
      - RABBITMQ_HOST=rabbitmq
    depends_on:
      - rabbitmq

  your-service-backend:
    build: .
    command: ["python", "backend/backend_processor.py"]
    environment:
      - RABBITMQ_HOST=rabbitmq
    depends_on:
      - rabbitmq

Advanced Features

Authentication

Implement proper JWT token validation in your adapter:

async def get_current_user(self, token: str) -> User:
    """
    Validate JWT token and extract user information.
    """
    try:
        # Decode and validate JWT token
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")

        # Fetch user from database
        user = await get_user_by_id(user_id)
        return user
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

Error Handling

Implement comprehensive error handling:

async def handle_message(self, message: DataMessage):
    """
    Handle incoming messages with proper error handling.
    """
    try:
        # Process message
        result = await self.process_message(message)
        return result
    except ValidationError as e:
        logging.error(f"Validation error: {e}")
        # Handle validation errors
    except ProcessingError as e:
        logging.error(f"Processing error: {e}")
        # Handle processing errors
    except Exception as e:
        logging.error(f"Unexpected error: {e}")
        # Handle unexpected errors

Monitoring and Logging

Implement proper logging and monitoring:

import logging
from opentelemetry import trace, metrics

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configure tracing
tracer = trace.get_tracer(__name__)

# Configure metrics
meter = metrics.get_meter(__name__)
message_counter = meter.create_counter("messages_processed_total")

async def process_message(self, message: DataMessage):
    """
    Process message with tracing and metrics.
    """
    with tracer.start_as_current_span("process_message"):
        logger.info(f"Processing message: {message.id()}")

        # Your processing logic here

        message_counter.add(1, {"service": "your-service"})
        logger.info(f"Message processed successfully: {message.id()}")

Troubleshooting

Common Issues

1. Connection Errors: Ensure RabbitMQ is running and accessible
2. Route Not Found: Check route mapping configuration
3. Message Not Processed: Verify endpoint method signatures match expected parameters
4. Authentication Failures: Check JWT token validation logic

Debugging Tips

1. Enable Debug Logging: Set logging level to DEBUG
2. Check RabbitMQ Management: Use RabbitMQ management interface to monitor queues
3. Trace Message Flow: Use OpenTelemetry tracing to follow message flow
4. Validate Configuration: Ensure all configuration parameters are correct

Conclusion

This guide provides a comprehensive overview of building a Business Service using the AMQ Adapter library. By following the patterns established in the OTDemo application, you can create robust, scalable services that integrate seamlessly with the CleverMicro platform.

For more detailed information, refer to the AMQ Adapter library documentation and the OTDemo source code examples.