Redis Message Queue Backend for Abstract Backend

A production-ready Redis Streams-based message queue backend implementation for the Abstract Backend project. This backend extends message queue capabilities to support Redis, enabling reliable, scalable, and high-performance Abstract Backend event processing.

Status & Quality

CI/CD & Testing

Code Coverage & Quality

Code Style & Standards

Package Info

Downloads

---

Overview

abe-redis is a Redis Streams-based message queue backend that integrates seamlessly with the Abstract Backend's component loading mechanism. It provides a reliable, scalable solution for handling Abstract Backend events using Redis as the message queue infrastructure.

Key Features

• 🔌 Plug-and-Play: Install via pip and configure with environment variables
• ⚡ Redis Streams: Modern stream-based message processing with consumer groups
• 🚀 Production Ready: Connection pooling, error handling, and retry logic built-in
• 🔄 Async-First: Built for modern Python async/await patterns
• 📦 Universal Compatibility: Works with any project using the same component loading mechanism
• 🧪 Well Tested: Comprehensive unit and integration tests with high coverage
• 📚 Fully Documented: Complete API reference and usage examples

Python Version Support

Python 3.12, 3.13

Quick Start

Installation

Install the package via pip:

pip install abe-redis

Configuration

Configure your environment to use Redis as the message queue backend:

export QUEUE_BACKEND=redis
export REDIS_URL=redis://localhost:6379/0

Start Redis

Using Docker (recommended):

docker run -d --name test-redis -p 6379:6379 redis:7-alpine

Basic Usage

import asyncio
from abe_plugin.backends.message_queue import RedisMessageQueueBackend


async def main():
    # Create backend from environment variables
    backend = RedisMessageQueueBackend.from_env()

    # Publish a message
    await backend.publish("slack:events", {"type": "message", "text": "Hello Redis!"})

    # Consume messages
    async for message in backend.consume():
        print(f"Received: {message}")
        break

    await backend.close()


asyncio.run(main())

Architecture

The Redis backend uses Redis Streams for reliable message queueing:

• Persistent Storage: Messages stored in Redis Streams with configurable retention
• Consumer Groups: Distributed consumption across multiple workers
• Automatic Acknowledgment: Messages acknowledged after successful processing
• Stream Pattern Matching: Automatically discovers and consumes from slack:* streams
• Connection Pooling: Efficient connection management with configurable pool size

Configuration Options

Environment Variable	Required	Default	Description
QUEUE_BACKEND	Yes	-	Must be set to redis
REDIS_URL	Yes	redis://localhost:6379/0	Redis connection URL
REDIS_PASSWORD	No	None	Redis authentication password
REDIS_SSL	No	false	Enable SSL/TLS connection
REDIS_MAX_CONNECTIONS	No	10	Maximum connection pool size
REDIS_STREAM_MAXLEN	No	10000	Maximum stream length for trimming

Documentation

For comprehensive documentation, including API references, examples, and development guides:

📚 Full Documentation

Quick Links

• Introduction
• Quick Start Guide
• Installation Guide
• API Reference
• Development Guide
• CI/CD Documentation

Development

Prerequisites

• Python 3.12 or 3.13
• Redis 6.0+ (7.0+ recommended)
• uv or pip package manager

Setup Development Environment

# Clone the repository
git clone https://github.com/Chisanan232/abe-redis.git
cd abe-redis

# Install dependencies
uv sync

# Start Redis for testing
docker run -d --name redis-dev -p 6379:6379 redis:7-alpine

# Run tests
uv run pytest

# Run tests with coverage
uv run pytest --cov=abe_plugin --cov-report=html

Code Quality

The project follows strict code quality standards:

• Code Formatting: black
• Linting: pylint
• Import Sorting: isort
• Type Checking: mypy

Run quality checks:

# Format code
uv run black abe_plugin/ test/

# Lint code
uv run pylint abe_plugin/

# Sort imports
uv run isort abe_plugin/ test/

# Type check
uv run mypy abe_plugin/

Contributing

We welcome contributions! Please see our Contributing Guide for details.

How to Contribute

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes
4. Run tests and quality checks
5. Commit your changes (git commit -m 'feat: add amazing feature')
6. Push to the branch (git push origin feature/amazing-feature)
7. Open a Pull Request

License

This project is licensed under the MIT License - see the LICENSE file for details.

Links

• Documentation: https://chisanan232.github.io/abe-redis/
• PyPI Package: https://pypi.org/project/abe-redis/
• Source Code: https://github.com/Chisanan232/abe-redis
• Issue Tracker: https://github.com/Chisanan232/abe-redis/issues
• Abstract Backend: https://github.com/Chisanan232/abstract-backend

Support

If you encounter any issues or have questions:

• 📖 Check the Documentation
• 🐛 Report bugs via GitHub Issues
• 💬 Ask questions in GitHub Discussions

---

Made with ❤️ for the Abstract Backend ecosystem