krista-artemis 1.0.0
Artemis python package with queue integration
Requires Python: >=3.8

# Krista Artemis

Python package with ActiveMQ Artemis integration for Krista infrastructure. This package provides simple and robust queue operations with automatic message handling, retry logic, and AMQP 1.0 protocol support.

## Installation


### From Source
```bash
git clone <repository-url>
cd krista_artemis
pip install -e .
```

## Configuration

### queue_config.json

The package uses a `queue_config.json` file for Artemis connection settings and queue definitions. This file should be placed in your project root or specified via environment variables.

**Example queue_config.json:**
```json
{
  "host": "artemis",
  "port": 5672,
  "username": "admin", 
  "password": "admin",
  "protocol": "amqp",
  "queues": {
    "default": {
      "name": "knowledge-agent.default",
      "durable": true,
      "auto_delete": false,
      "routing_type": "ANYCAST",
      "ack_mode": "client",
      "auto_consumer": false,
      "handler_timeout_seconds": 60,
      "message_ttl_seconds": -1,
      "description": "Default queue for general tasks"
    },
    "chat": {
      "name": "knowledge-agent.chat",
      "durable": true,
      "auto_delete": false,
      "routing_type": "ANYCAST",
      "ack_mode": "client",
      "auto_consumer": true,
      "handler_timeout_seconds": 90,
      "message_ttl_seconds": -1,
      "description": "Queue for chat messages"
    }
  }
}
```

**Configuration Parameters:**

### Server Connection
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `host` | string | `artemis` | Artemis server hostname |
| `port` | integer | `5672` | AMQP port |
| `username` | string | `admin` | Authentication username |
| `password` | string | `admin` | Authentication password |
| `protocol` | string | `amqp` | Protocol to use (AMQP 1.0) |

### Queue Configuration
Each queue in the `queues` object supports these parameters:

#### Basic Settings
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `name` | string | - | Full queue name in Artemis |
| `durable` | boolean | `true` | Queue survives broker restart |
| `auto_delete` | boolean | `false` | Delete queue when no consumers |
| `routing_type` | string | `ANYCAST` | ANYCAST (queue) or MULTICAST (topic) |
| `ack_mode` | string | `client` | Acknowledgment mode (auto, client) |
| `auto_consumer` | boolean | `false` | Enable automatic consumer |
| `handler_timeout_seconds` | integer | `60` | Message handler timeout |
| `message_ttl_seconds` | integer | `-1` | Message time-to-live (-1 = no expiry) |
| `description` | string | - | Human-readable queue description |

#### Acknowledgment Modes
- `auto`: Messages auto-acknowledged (at-most-once delivery, no retry)
- `client`: Manual acknowledgment (at-least-once delivery, **RECOMMENDED**)

#### Routing Types
- `ANYCAST`: Point-to-point messaging (queue behavior)
- `MULTICAST`: Publish-subscribe messaging (topic behavior)

### Environment Variables

You can override configuration using environment variables:

```bash
export ARTEMIS_HOST=prod-server.example.com
export ARTEMIS_AMQP_PORT=5672
export ARTEMIS_USER=myuser
export ARTEMIS_PASSWORD=mypassword
export ARTEMIS_MANAGEMENT_HOST=prod-server.example.com
export ARTEMIS_MANAGEMENT_PORT=8161
```

## Usage

### Basic Queue Operations

#### Producer - Sending Messages

```python
from krista_artemis import QueueProducer, QueueConfig

# Initialize with default config
config = QueueConfig()
producer = QueueProducer(config)

# Send dictionary message
message_data = {"user_id": 123, "action": "login", "timestamp": "2024-01-01T10:00:00Z"}
producer.send_message("default", message_data)

# Send string message
producer.send_message("chat", "Hello, World!")

# Send message with headers
headers = {"priority": "high", "source": "web-app"}
producer.send_message("default", message_data, headers)
```

#### Consumer - Receiving Messages

```python
from krista_artemis import QueueConsumer, QueueConfig, QueueMessage

def message_handler(message: QueueMessage):
    """Handle incoming messages."""
    print(f"Received: {message.body}")
    print(f"Message ID: {message.id}")
    print(f"Properties: {message.properties}")
    
    # Process the message
    if isinstance(message.body, dict):
        user_id = message.body.get("user_id")
        print(f"Processing for user: {user_id}")
    
    # If this function raises an exception:
    # - 'auto' mode: message is lost
    # - 'client' mode: message is rejected and redelivered

# Initialize consumer
config = QueueConfig()
consumer = QueueConsumer(config, "default", message_handler)

# Start consuming (blocking)
consumer.run()
```

### Advanced Configuration

```python
from krista_artemis import QueueConfig, QueueProducer, QueueConsumer

# Custom configuration
config = QueueConfig(config_file="custom_queue_config.json")

# Producer with custom retry settings
producer = QueueProducer(
    config=config,
    retry_delay=2.0,
    max_retry_delay=60.0
)

# Consumer with custom retry settings
consumer = QueueConsumer(
    config=config,
    queue_name="chat",
    handler=my_handler,
    retry_delay=1.0,
    max_retry_delay=30.0
)
```

### Consumer Manager - Multiple Queues

```python
from krista_artemis import QueueConsumerManager, QueueConfig

def default_handler(message):
    print(f"Default: {message.body}")

def chat_handler(message):
    print(f"Chat: {message.body}")

# Create manager
config = QueueConfig()
manager = QueueConsumerManager(config)

# Register consumers
manager.register_consumer('default', default_handler)
manager.register_consumer('chat', chat_handler)

# Start all consumers
manager.start()

# Wait for shutdown signal
try:
    manager.wait()
except KeyboardInterrupt:
    print("Shutting down...")
    manager.stop()
```

### Queue Management

```python
from krista_artemis import QueueCreator, QueueConfig

# Create queue creator
config = QueueConfig()
creator = QueueCreator(config)

# Check if queue exists
if creator.queue_exists("my-queue", "ANYCAST"):
    print("Queue exists")

# Create queue manually
success = creator.create_queue(
    queue_name="my-custom-queue",
    durable=True,
    auto_delete=False,
    routing_type="ANYCAST"
)

if success:
    print("Queue created successfully")
```

### Message Serialization

The package automatically handles message serialization:

```python
# All Python data types are supported
producer.send_message("default", "Simple string")
producer.send_message("default", {"key": "value", "number": 42})
producer.send_message("default", [1, 2, 3, 4, 5])
producer.send_message("default", 42)
producer.send_message("default", True)

# Complex nested structures
complex_data = {
    "users": [
        {"id": 1, "name": "Alice", "active": True},
        {"id": 2, "name": "Bob", "active": False}
    ],
    "metadata": {
        "created": "2024-01-01T10:00:00Z",
        "version": "1.0"
    }
}
producer.send_message("default", complex_data)
```

### Error Handling and Retry Logic

```python
from krista_artemis import QueueConsumer, QueueConfig
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)

def robust_handler(message):
    """Message handler with error handling."""
    try:
        # Process message
        data = message.body
        
        # Simulate processing
        if data.get("should_fail"):
            raise ValueError("Simulated processing error")
            
        print(f"Successfully processed: {data}")
        
    except ValueError as e:
        # Log error - message will be redelivered in 'client' mode
        logging.error(f"Processing failed: {e}")
        raise  # Re-raise to trigger redelivery
    except Exception as e:
        # Unexpected error
        logging.error(f"Unexpected error: {e}")
        raise

# Consumer with retry logic
config = QueueConfig()
consumer = QueueConsumer(
    config=config,
    queue_name="default",
    handler=robust_handler,
    retry_delay=1.0,        # Start with 1 second delay
    max_retry_delay=30.0    # Cap at 30 seconds
)

consumer.run()
```

### Connection Management

The package includes automatic reconnection with exponential backoff:

```python
# Consumer automatically reconnects on connection loss
consumer = QueueConsumer(config, "default", handler)

# Producer retries failed sends
producer = QueueProducer(config)

# Both handle:
# - Network interruptions
# - Broker restarts
# - Authentication failures
# - Queue creation issues
```

## AMQP 1.0 Protocol

This package uses AMQP 1.0 protocol via `qpid-proton`, providing:

- **Better Performance**: More efficient than STOMP
- **Reliable Delivery**: Proper transaction support
- **Industry Standard**: AMQP 1.0 is widely supported
- **Advanced Features**: Message properties, headers, and routing

### Message Format

Messages are automatically serialized to JSON and wrapped in AMQP format:

```json
{
  "body": "your-serialized-data",
  "properties": {
    "message_id": "unique-id",
    "creation_time": 1704110400000,
    "content_type": "application/json"
  },
  "application_properties": {
    "custom_header": "value"
  }
}
```

## Authentication

The package supports SASL authentication:

```json
{
  "username": "your-username",
  "password": "your-password"
}
```

For production environments, consider using:
- Environment variables for credentials
- Encrypted configuration files
- Secret management systems

## Management API

Queue creation uses Artemis Jolokia management API:

- **Default Port**: 8161
- **Endpoint**: `/console/jolokia`
- **Authentication**: HTTP Basic Auth

## Logging

The package uses Python's standard logging:

```python
import logging

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)

# Or configure specific logger
logger = logging.getLogger('krista_artemis')
logger.setLevel(logging.INFO)
```

## Error Handling

Common exceptions and how to handle them:

```python
from krista_artemis import QueueProducer, QueueConsumer
from requests.exceptions import ConnectionError, Timeout

try:
    producer = QueueProducer(config)
    producer.send_message("default", {"test": "data"})
except ConnectionError:
    print("Cannot connect to Artemis server")
except Timeout:
    print("Request timed out")
except Exception as e:
    print(f"Unexpected error: {e}")
```

## Performance Tips

1. **Connection Reuse**: Producers and consumers reuse connections
2. **Batch Processing**: Process multiple messages in handlers when possible
3. **Acknowledgment Mode**: Use 'client' mode for reliability, 'auto' for speed
4. **Queue Configuration**: Set appropriate TTL and durability settings
5. **Handler Timeout**: Configure reasonable timeout values for message processing

## Docker Integration

Example `docker-compose.yml` for Artemis:

```yaml
services:
  artemis:
    image: apache/activemq-artemis:latest
    container_name: artemis
    environment:
      ARTEMIS_USER: admin
      ARTEMIS_PASSWORD: admin
      ANONYMOUS_LOGIN: "false"
    ports:
      - "8161:8161"   # Web Console
      - "61616:61616" # Core Protocol
      - "5672:5672"   # AMQP
      - "61613:61613" # STOMP
    volumes:
      - artemis_data:/var/lib/artemis-instance/data

volumes:
  artemis_data:
```

## Testing

See [TESTING.md](TESTING.md) for comprehensive testing instructions.

```bash
# Run all tests
pytest

# Run with coverage
pytest --cov=package --cov-report=term-missing

# Run specific test file
pytest tests/test_producer.py
```

## Contributing

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure all tests pass
5. Submit a pull request

## License

MIT License - see [LICENSE](../LICENSE) file for details.