EventFlow 🚀

EventFlow is a production-grade event-driven backend system built with Python, providing real-time visibility into user actions and system events. It demonstrates modern event streaming architecture with support for high-throughput message processing, reliable delivery, and comprehensive observability.

🏗️ Architecture Overview

                                    ┌─────────────────────────────────────────────────────────────┐
                                    │                      EventFlow System                        │
                                    └─────────────────────────────────────────────────────────────┘
                                                              │
                    ┌─────────────────────────────────────────┼─────────────────────────────────────────┐
                    │                                         │                                         │
                    ▼                                         ▼                                         ▼
         ┌─────────────────────┐                   ┌─────────────────────┐                   ┌─────────────────────┐
         │   Event Producers   │                   │   Message Broker    │                   │   Event Consumers   │
         │   ────────────────  │                   │   ──────────────    │                   │   ────────────────  │
         │ • REST API          │                   │ • Redis Streams     │                   │ • Event Workers     │
         │ • Batch Processing  │      ─────────▶   │ • Consumer Groups   │   ─────────▶     │ • Retry Handler     │
         │ • Event Factory     │                   │ • Message Ordering  │                   │ • DLQ Handler       │
         │ • CloudEvents Spec  │                   │ • Persistence       │                   │ • Webhook Delivery  │
         └─────────────────────┘                   └─────────────────────┘                   └─────────────────────┘
                    │                                         │                                         │
                    │                                         │                                         │
                    └──────────────────┬──────────────────────┴──────────────────┬──────────────────────┘
                                       │                                         │
                                       ▼                                         ▼
                            ┌─────────────────────┐                   ┌─────────────────────┐
                            │  Persistence Layer  │                   │    Observability    │
                            │  ─────────────────  │                   │    ────────────     │
                            │ • PostgreSQL        │                   │ • Structured Logs   │
                            │ • Event Store       │                   │ • Prometheus Metrics│
                            │ • Query Patterns    │                   │ • Health Checks     │
                            │ • Data Retention    │                   │ • Grafana Dashboards│
                            └─────────────────────┘                   └─────────────────────┘

✨ Key Features

Event Processing

CloudEvents-compliant schema design for interoperability
Multiple event types: User, System, Business, Analytics events
Event batching with automatic flushing for optimized throughput
Producer pool for high-concurrency scenarios

Message Broker

Redis Streams for durable, ordered message delivery
Consumer groups for horizontal scaling
Message acknowledgment with automatic retry
Dead Letter Queue (DLQ) for failed message handling

Reliability

Exponential backoff retry mechanism with jitter
Circuit breaker pattern for fault tolerance
Graceful shutdown with in-flight message completion
Idempotent processing via unique event IDs

Observability

Structured logging with JSON output (structlog)
Prometheus metrics for real-time monitoring
Health check endpoints for container orchestration
Request tracing with correlation IDs

API

RESTful endpoints built with FastAPI
Auto-generated OpenAPI documentation
Event replay capabilities
Subscription management for webhooks

📁 Project Structure

EventFlow/
├── src/eventflow/
│   ├── __init__.py              # Package initialization
│   ├── app.py                   # FastAPI application entry
│   ├── cli.py                   # Command-line interface
│   ├── config.py                # Configuration management
│   │
│   ├── api/                     # REST API layer
│   │   ├── dependencies.py      # FastAPI dependencies
│   │   ├── schemas.py           # Request/response models
│   │   └── routes/
│   │       ├── events.py        # Event CRUD endpoints
│   │       ├── health.py        # Health check endpoints
│   │       ├── dlq.py           # Dead letter queue management
│   │       ├── subscriptions.py # Webhook subscriptions
│   │       └── metrics.py       # Prometheus metrics endpoint
│   │
│   ├── core/                    # Core business logic
│   │   ├── broker.py            # Redis Streams broker
│   │   └── producer.py          # Event producer with batching
│   │
│   ├── consumers/               # Event processing
│   │   ├── consumer.py          # Event consumer with retry
│   │   └── worker.py            # Background workers
│   │
│   ├── models/                  # Database models
│   │   └── database.py          # SQLAlchemy models
│   │
│   ├── persistence/             # Data access layer
│   │   ├── database.py          # Database connection
│   │   └── repository.py        # Repository pattern
│   │
│   ├── schemas/                 # Domain schemas
│   │   └── events.py            # CloudEvents schemas
│   │
│   └── observability/           # Monitoring & logging
│       ├── logging.py           # Structured logging
│       ├── metrics.py           # Prometheus metrics
│       └── retry.py             # Retry & circuit breaker
│
├── tests/                       # Test suite
│   ├── conftest.py              # Pytest fixtures
│   ├── test_api.py              # API integration tests
│   ├── test_schemas.py          # Schema validation tests
│   └── test_retry.py            # Retry mechanism tests
│
├── scripts/
│   └── generate_demo_data.py    # Demo data generator
│
├── docker/                      # Docker configuration
│   ├── prometheus.yml           # Prometheus config
│   └── grafana/                 # Grafana provisioning
│
├── Dockerfile                   # API server image
├── Dockerfile.worker            # Worker image
├── docker-compose.yml           # Full stack orchestration
├── pyproject.toml               # Project configuration
├── .env.example                 # Environment template
└── README.md                    # This file

🚀 Quick Start

Prerequisites

Python 3.11+
Docker & Docker Compose
Redis 7.0+
PostgreSQL 15+

1. Clone and Setup

git clone https://github.com/yourusername/eventflow.git
cd eventflow

# Create environment file
cp .env.example .env

2. Start with Docker Compose

# Start all services (API, Worker, PostgreSQL, Redis, Monitoring)
docker-compose up -d

# View logs
docker-compose logs -f eventflow-api

# Check health
curl http://localhost:8000/health

3. Access Services

Service	URL	Credentials
EventFlow API	http://localhost:8000	-
API Documentation	http://localhost:8000/docs	-
Prometheus	http://localhost:9090	-
Grafana	http://localhost:3000	admin/admin
pgAdmin	http://localhost:5050	[email protected]/admin
Redis Commander	http://localhost:8081	-

4. Send Your First Event

# Create a user event
curl -X POST http://localhost:8000/api/v1/events \
  -H "Content-Type: application/json" \
  -d '{
    "type": "user.created",
    "source": "api",
    "data": {
      "user_id": "usr_12345",
      "email": "[email protected]",
      "username": "johndoe"
    }
  }'

💻 Local Development

Setup Virtual Environment

# Create and activate virtual environment
python -m venv .venv
.venv\Scripts\activate  # Windows
source .venv/bin/activate  # Linux/Mac

# Install dependencies
pip install -e ".[dev]"

Run Services Locally

# Start dependencies
docker-compose up -d postgres redis

# Run database migrations
python -m eventflow.cli db upgrade

# Start API server
uvicorn eventflow.app:create_app --factory --reload

# In another terminal - Start worker
python -m eventflow.cli worker start

Generate Demo Data

# Generate 500 realistic events
python scripts/generate_demo_data.py

📖 API Reference

Events API

Create Event

POST /api/v1/events
Content-Type: application/json

{
  "type": "user.login",
  "source": "web-app",
  "data": {
    "user_id": "usr_123",
    "ip_address": "192.168.1.1"
  },
  "metadata": {
    "priority": "high",
    "partition_key": "usr_123"
  }
}

Query Events

GET /api/v1/events?event_type=user.login&limit=50&offset=0

Get Event by ID

GET /api/v1/events/{event_id}

Replay Events

POST /api/v1/events/replay
Content-Type: application/json

{
  "stream": "events:user",
  "start_time": "2024-01-01T00:00:00Z",
  "end_time": "2024-01-02T00:00:00Z"
}

Dead Letter Queue API

List DLQ Messages

GET /api/v1/dlq?limit=100

Retry Failed Message

POST /api/v1/dlq/{message_id}/retry

Delete DLQ Message

DELETE /api/v1/dlq/{message_id}

Subscriptions API

Create Subscription

POST /api/v1/subscriptions
Content-Type: application/json

{
  "name": "order-notifications",
  "event_types": ["order.created", "order.completed"],
  "webhook_url": "https://your-service.com/webhook",
  "secret": "your-webhook-secret"
}

Health Check

GET /health           # Basic health
GET /health/ready     # Readiness probe
GET /health/live      # Liveness probe

Metrics

GET /metrics          # Prometheus metrics

📊 Event Types

User Events

Event Type	Description
`user.created`	New user registration
`user.updated`	User profile update
`user.deleted`	User account deletion
`user.login`	User login
`user.logout`	User logout

System Events

Event Type	Description
`system.health_check`	Service health report
`system.error`	System error occurred
`system.config_changed`	Configuration update

Business Events

Event Type	Description
`order.created`	New order placed
`order.updated`	Order status change
`payment.completed`	Payment processed
`notification.sent`	Notification delivered

Analytics Events

Event Type	Description
`page.view`	Page view tracked
`feature.used`	Feature usage logged
`conversion.tracked`	Conversion recorded

⚙️ Configuration

Environment Variables

Variable	Default	Description
`APP_NAME`	eventflow	Application name
`APP_ENV`	development	Environment (development/staging/production)
`DEBUG`	false	Enable debug mode
`API_HOST`	0.0.0.0	API server host
`API_PORT`	8000	API server port
`DATABASE_URL`	postgresql+asyncpg://...	Database connection URL
`REDIS_URL`	redis://localhost:6379	Redis connection URL
`LOG_LEVEL`	INFO	Logging level
`LOG_FORMAT`	json	Log format (json/console)

Event Processing

Variable	Default	Description
`BATCH_SIZE`	100	Events per batch
`BATCH_TIMEOUT_MS`	1000	Batch flush timeout
`CONSUMER_GROUP`	eventflow-workers	Redis consumer group
`MAX_RETRIES`	3	Max retry attempts
`RETRY_BASE_DELAY`	1.0	Base retry delay (seconds)

🧪 Testing

Run Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=eventflow --cov-report=html

# Run specific test file
pytest tests/test_api.py -v

# Run tests matching pattern
pytest -k "test_event" -v

Test Categories

Unit Tests: Schema validation, utility functions
Integration Tests: API endpoints, database operations
Reliability Tests: Retry mechanisms, circuit breaker

🐳 Docker Deployment

Production Deployment

# Build images
docker-compose build

# Start in production mode
docker-compose -f docker-compose.yml up -d

# Scale workers
docker-compose up -d --scale eventflow-worker=3

Health Checks

The containers include health checks for orchestration:

healthcheck:
  test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
  interval: 30s
  timeout: 10s
  retries: 3

Resource Management

# View resource usage
docker stats

# View logs
docker-compose logs -f --tail=100

# Restart services
docker-compose restart eventflow-api eventflow-worker

📈 Monitoring

Prometheus Metrics

Key metrics exported:

Metric	Type	Description
`eventflow_events_published_total`	Counter	Total events published
`eventflow_events_consumed_total`	Counter	Total events consumed
`eventflow_events_failed_total`	Counter	Failed event processing
`eventflow_event_processing_seconds`	Histogram	Processing latency
`eventflow_dlq_size`	Gauge	Dead letter queue size
`eventflow_circuit_breaker_state`	Gauge	Circuit breaker status

Grafana Dashboards

Pre-configured dashboards for:

Event throughput and latency
Error rates and DLQ growth
Consumer lag and processing time
System health metrics

🔧 Reliability Patterns

Retry with Exponential Backoff

from eventflow.observability import retry_async, RetryConfig

config = RetryConfig(
    max_retries=3,
    base_delay=1.0,
    max_delay=60.0,
    exponential_base=2.0,
    jitter=True,
)

@retry_async(config)
async def process_event(event):
    # Processing logic
    pass

Circuit Breaker

from eventflow.observability import CircuitBreaker

breaker = CircuitBreaker(
    failure_threshold=5,
    recovery_timeout=30.0,
    half_open_max_calls=3,
)

async with breaker:
    await call_external_service()

🤝 Contributing

Fork the repository
Create a feature branch (git checkout -b feature/amazing-feature)
Commit your changes (git commit -m 'Add amazing feature')
Push to the branch (git push origin feature/amazing-feature)
Open a Pull Request

Development Guidelines

Follow PEP 8 style guide
Write tests for new features
Update documentation as needed
Use conventional commit messages

📄 License

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

🙏 Acknowledgments

FastAPI - Modern Python web framework
Redis - In-memory data store
CloudEvents - Event specification
Pydantic - Data validation
SQLAlchemy - SQL toolkit

Built with ❤️ for scalable event-driven systems

Name		Name	Last commit message	Last commit date
Latest commit History 1 Commit
docker		docker
scripts		scripts
src		src
tests		tests
.env.example		.env.example
Dockerfile		Dockerfile
Dockerfile.worker		Dockerfile.worker
LICENSE		LICENSE
README.md		README.md
docker-compose.yml		docker-compose.yml
pyproject.toml		pyproject.toml

License

cliff-de-tech/EventFlow

Folders and files

Latest commit

History

Repository files navigation

EventFlow 🚀

🏗️ Architecture Overview

✨ Key Features

Event Processing

Message Broker

Reliability

Observability

API

📁 Project Structure

🚀 Quick Start

Prerequisites

1. Clone and Setup

2. Start with Docker Compose

3. Access Services

4. Send Your First Event

💻 Local Development

Setup Virtual Environment

Run Services Locally

Generate Demo Data

📖 API Reference

Events API

Create Event

Query Events

Get Event by ID

Replay Events

Dead Letter Queue API

List DLQ Messages

Retry Failed Message

Delete DLQ Message

Subscriptions API

Create Subscription

Health Check

Metrics

📊 Event Types

User Events

System Events

Business Events

Analytics Events

⚙️ Configuration

Environment Variables

Event Processing

🧪 Testing

Run Tests

Test Categories

🐳 Docker Deployment

Production Deployment

Health Checks

Resource Management

📈 Monitoring

Prometheus Metrics

Grafana Dashboards

🔧 Reliability Patterns

Retry with Exponential Backoff

Circuit Breaker

🤝 Contributing

Development Guidelines

📄 License

🙏 Acknowledgments

About

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Languages

Packages