Sync/Async Callback Demo

A demonstration of synchronous and asynchronous request processing with callback delivery, retry mechanisms, and load testing capabilities.

Architecture

• FastAPI: Modern Python web framework for API endpoints
• RQ (Redis Queue): Lightweight task queue for async processing
• SQLite: Embedded database for request state tracking

Features

• Synchronous request processing with immediate responses
• Asynchronous request processing with callback delivery
• Automatic retry mechanisms with exponential backoff
• Circuit breaker pattern to prevent request storms
• Rate limiting per callback URL
• Request state tracking and observability
• Load testing tools with metrics collection

Setup

Prerequisites

• Python 3.9+
• Redis server

Installation

1. Install Redis:

   # macOS
   brew install redis
   redis-server
   

2. Install Python dependencies:

   # Create virtual environment 
   python3 -m venv venv 
   source venv/bin/activate  # On Windows: venv\Scripts\activate
   
   # Install dependencies
   pip install -r requirements.txt

3. Initialize database:

   python3 init_db.py

Running

Start FastAPI Server

uvicorn app.main:app --reload

The API will be available at http://localhost:8000

• API Documentation: http://localhost:8000/docs
• Health Check: http://localhost:8000/health

Start RQ Workers

In separate terminals, start RQ workers:

rq worker --worker-class rq.worker.SimpleWorker default callbacks

You can start multiple workers to scale processing.

Start callback server

python test_callback_server.py

Start load testing

python -m load_tester.generator

API Endpoints

Synchronous Requests

POST /sync/

• Process request immediately and return result
• Returns: {request_id, status, result}

GET /sync/{request_id}

• Get sync request status

Asynchronous Requests

POST /async/

• Accept request and process asynchronously
• Requires: callback_url in request body
• Returns: {request_id, status, message} (202 Accepted)

GET /async/{request_id}

• Get async request status and callback delivery info

Load Testing

The load_tester package provides tools for generating load and collecting metrics.

Example usage:

from .generator import LoadGenerator
from .reporter import Reporter

generator = LoadGenerator()

# Generate sync load
await generator.generate_sync_load(
   requests_per_second=50,
   duration_seconds=60,
   concurrency=100
)

# Generate async load
await generator.generate_async_load(
   requests_per_second=50,
   duration_seconds=60,
   callback_url="http://localhost:8001/callback",
   concurrency=100,
)

await asyncio.sleep(10)

generator.handle_callback("http://localhost:8001")

# Generate reports
reporter = Reporter(generator.metrics)
reporter.export_json("report_async.json")
reporter.export_csv("report_async.csv")

Configuration

Environment variables:

• REDIS_URL: Redis connection URL (default: redis://localhost:6379/0)
• DATABASE_URL: Database URL (default: sqlite:///./requests.db)
• MAX_RETRY_ATTEMPTS: Maximum retry attempts (default: 5)
• RETRY_BASE_DELAY: Base delay for retries in seconds (default: 1)
• RETRY_MAX_DELAY: Maximum delay for retries in seconds (default: 60)
• CALLBACK_TIMEOUT: HTTP timeout for callbacks in seconds (default: 10)

Project Structure

app/
├── api/              # API routes
├── services/         # Business logic
├── models/           # Database models
├── infrastructure/   # Database, queue, HTTP client
├── tasks.py          # RQ task definitions
└── main.py           # FastAPI app

load_tester/
├── generator.py      # Load generation
├── metrics.py        # Metrics collection
└── reporter.py       # Report generation