A lightweight framework for building Python background services with a Flask-like API, enabling seamless inter-service communication via Redis pub/sub.
Daebus provides a simple way to create and connect distributed background services. It offers a Flask-inspired API that makes it easy to:
- Create systemd daemons that communicate with each other
- Expose HTTP endpoints for frontend applications (e.g., kiosk browsers)
- Enable quick bootstrapping of new services with minimal boilerplate
Services communicate via Redis pub/sub channels, providing a lightweight and reliable messaging system ideal for local service orchestration.
Note: Daebus requires a Redis instance running on the system.
- Flask-inspired API: Simple, decorator-based API for defining service functionality
- Redis Pub/Sub: Direct channel-based communication between services
- Action Routing: Route messages to specific handlers based on action field
- Background tasks: Easy scheduling of periodic background tasks
- Client library:
DaebusCallerallows for easy communication with Daebus services - HTTP endpoints: Expose lightweight HTTP APIs from your daemon without additional dependencies
- Specialized APIs: Distinct and purpose-built APIs for HTTP and pub/sub contexts
pip install daebusfrom Daebus import Daebus, request, response, broadcast, cache, logger app = Daebus(__name__) # Define an action handler (like a Flask route) @app.action("get_status") def handle_status_request(): try: payload = request.payload logger.info(f"Received status request: {payload}") # Process the request result = {"status": "healthy", "uptime": 3600} # Send response back to the requester return response.success(result) except Exception as e: return response.error(e) # Define another action handler @app.action("restart") def handle_restart(): try: logger.info("Handling restart request") # Restart logic here return response.success({"restarted": True}) except Exception as e: return response.error(e) # Define a channel listener for broadcasts @app.listen("notifications") def handle_notification(payload): logger.info(f"Received notification: {payload}") # Process the notification # Define a background task that runs every 30 seconds @app.background("health_check", 30) def health_check(): logger.info("Running health check") # Perform health check # Broadcast status to any interested services broadcast.send("system_status", {"status": "healthy"}) # Run the service if __name__ == "__main__": app.run(service="my_service", debug=True)from Daebus import DaebusCaller # Create a caller for the target service caller = DaebusCaller("my_service") try: # Send a request to a specific action and wait for a response response = caller.send_request("get_status", { "detail_level": "full" }) print(f"Service status: {response}") # Send a restart request restart_response = caller.send_request("restart", { "mode": "graceful" }) print(f"Restart result: {restart_response}") # Send a message to a broadcast channel caller.send_message("notifications", { "type": "alert", "message": "System temperature high" }) # Send a message directly to the service with a specific action caller.send_to_service({ "timestamp": 1625176230, "info": "Direct message with action" }, action="log_info") finally: # Clean up resources when done caller.close()You can easily add HTTP endpoints to your daemon using the built-in HTTP server:
from Daebus import Daebus, DaebusHttp app = Daebus(__name__) # Create and attach an HTTP endpoint http = DaebusHttp(port=8080) app.attach(http) # Define HTTP routes (similar to Flask) @app.route("/status") def get_status(req): # Use the request object to access request data # req.payload contains JSON, form data, or query params # For HTTP responses, use response.send with data and status code return response.send({ "service": "my_service", "status": "running", "timestamp": time.time() }, 200) # 200 OK status code # Routes with parameters @app.route("/devices/<device_id>") def get_device(req, device_id): # The device_id parameter is extracted from the URL if device_id == "123": return response.send({ "device_id": device_id, "status": "online" }, 200) else: # Error responses with appropriate status codes return response.send({ "error": "Device not found" }, 404) # 404 Not Found # POST requests @app.route("/control", methods=["POST"]) def control_action(req): # Access JSON data from the request payload data = req.payload # Process the request return response.send({ "success": True, "message": "Command received" }, 201) # 201 Created # Run the service with HTTP enabled app.run(service="my_service")If your HTTP API needs to be accessed from web browsers, you can configure Cross-Origin Resource Sharing (CORS):
from Daebus import Daebus, DaebusHttp app = Daebus(__name__) # Configure CORS when creating the HTTP endpoint cors_config = { 'allowed_origins': ['http://localhost:3000', 'https://example.com'], # Specific origins 'allowed_methods': ['GET', 'POST', 'OPTIONS'], # Allowed HTTP methods 'allowed_headers': ['Content-Type', 'Authorization'], # Allowed headers 'expose_headers': ['X-Custom-Header'], # Headers exposed to client 'allow_credentials': True, # Allow cookies 'max_age': 3600 # Cache preflight for 1 hour } # Create HTTP endpoint with CORS support http = DaebusHttp(port=8080, cors_config=cors_config) app.attach(http) # Alternatively, configure CORS after initialization # http = DaebusHttp(port=8080) # http.configure_cors({ # 'allowed_origins': '*', # Allow all origins # 'allowed_methods': ['GET', 'POST'], # 'allowed_headers': '*', # Allow all headers # }) @app.route("/api/data") def get_data(req): # This endpoint will now include CORS headers in the response return response.send({"data": "example"}, 200)You can use these CORS configuration options:
allowed_origins: List of allowed origins or'*'for any originallowed_methods: List of allowed HTTP methods or'*'for any methodallowed_headers: List of allowed headers or'*'for any headerexpose_headers: List of headers to expose to the clientallow_credentials: Boolean for allowing credentials (cookies)max_age: Cache time (seconds) for preflight requests
Daebus provides distinct response methods optimized for each protocol:
from Daebus import Daebus, DaebusHttp, request, response app = Daebus(__name__) http = DaebusHttp(port=8080) app.attach(http) # Shared function for business logic def get_status_data(): return { "service": "my_service", "status": "healthy", "uptime": 3600 } # HTTP route with HTTP-specific response using response.send @app.route("/status") def http_status_handler(req): try: data = get_status_data() # Use response.send() for HTTP responses, with status code return response.send(data, 200) except Exception as e: # HTTP errors with appropriate status codes return response.send({"error": str(e)}, 500) # Redis action handler with pub/sub specific response methods @app.action("get_status") def redis_status_handler(): try: data = get_status_data() # Use response.success() for Redis responses return response.success(data) except Exception as e: # Redis-specific error handling return response.error(e)For advanced usage, you can access the protocol-specific classes directly:
from Daebus import ( Daebus, DaebusHttp, HttpRequest, HttpResponse, # HTTP-specific classes PubSubRequest, PubSubResponse # Pub/Sub-specific classes ) app = Daebus(__name__) http = DaebusHttp(port=8080) app.attach(http) # Use HTTP-specific classes directly for advanced customization @app.route("/advanced") def advanced_handler(req): # Check if we have the right request type if not isinstance(req, HttpRequest): raise TypeError("Expected an HttpRequest object") # Create a custom HTTP response custom_response = HttpResponse(None) return custom_response.send({ "message": "Custom HTTP response", "path": req.path }, 200)Benefits of protocol-specific methods:
response.send(data, status_code)- HTTP focused, with explicit status codesresponse.success(data)- Pub/sub focused, adds necessary pub/sub metadataresponse.error(exception)- Pub/sub focused, formats exceptions for pub/sub- Clear distinction between HTTP and pub/sub handling
- Direct access to protocol-specific classes for advanced customization
Daebus uses a simple action routing system:
- Each service automatically listens on its main channel (named after the service)
- Messages sent to this channel can include an
actionfield to route to specific handlers - Handlers are registered using
@app.action("action_name")
For example:
- When a message with
action: "get_status"is received, it's routed to the function decorated with@app.action("get_status") - When a message with
action: "restart"is received, it's routed to the function decorated with@app.action("restart")
Daebus uses two types of Redis pub/sub channels:
- Service Channels: Named after the service (e.g.,
my_service) and used for direct communication with action routing - Custom Channels: Any additional channels that services can publish to or subscribe to (e.g.,
notifications,system_status)
-
Request/Response:
- Client calls
send_request("action_name", payload) - DaebusCaller sends message to service's main channel with the action field
- Service routes to appropriate handler based on action
- Handler processes request and sends response back to caller's response channel
- Client calls
-
Broadcasts:
- Any service can broadcast to any channel with
broadcast.send(channel, payload) - Interested services subscribe to those channels with
@app.listen(channel)
- Any service can broadcast to any channel with
Daebus provides a lightweight HTTP server that:
- Uses Python's built-in
http.servermodule (no external dependencies) - Automatically converts Python dictionaries to JSON responses
- Parses JSON and form data from requests
- Supports URL parameters with the same syntax as Flask (
/path/<param>) - Runs in a background thread, allowing your service to handle both Redis and HTTP requests
Daebus provides specialized classes for each protocol:
-
For Pub/Sub (Redis):
PubSubRequest: Handles pub/sub message payloads, reply channels, and request IDsPubSubResponse: Handles pub/sub responses and error handling
from daebus import Daebus, response, request, logger app = Daebus(__name__) @app.action("hello_world") def hello_world(): name = request.payload.get("name", "World") logger.info(f"Received request from {name}") return response.success({"message": f"Hello, {name}!"}) if __name__ == "__main__": app.run(service='hello-service')Daebus provides two logger options:
logger: A proxy that references the service's logger. This works within main handlers.direct_logger: A direct reference to the base logger. This works in all contexts.
For background tasks, scheduled jobs, and other situations where the thread context might differ, use direct_logger:
from daebus import Daebus, direct_logger app = Daebus(__name__) @app.background("periodic_task", interval=60) def run_periodic_task(): try: # Do some work direct_logger.info("Periodic task completed") except Exception as e: # Always use direct_logger in background tasks direct_logger.error(f"Error in periodic task: {e}") if __name__ == "__main__": app.run(service='background-service')from daebus import Daebus, DaebusHttp, response, logger app = Daebus(__name__) http = DaebusHttp(port=8080) app.attach(http) @app.route("/hello") def hello(request): name = request.params.get("name", "World") logger.info(f"HTTP request from {name}") return response.send({"message": f"Hello, {name}!"}, 200) if __name__ == "__main__": app.run(service='web-service')from daebus import Daebus, DaebusCaller, direct_logger app = Daebus(__name__) # Create a caller for another service other_service = DaebusCaller("other-service") @app.on_start() def initialize(): direct_logger.info("Notifying other services that we've started") try: # Send a message to another service result = other_service.send_message("service_started", { "service_name": "my-service", "timestamp": time.time() }) direct_logger.info(f"Notification sent, response: {result}") except Exception as e: direct_logger.error(f"Failed to send notification: {e}") if __name__ == "__main__": app.run(service='my-service')- Background Tasks: Run tasks periodically with
@app.background(name, interval) - Threads: Run long-running tasks in dedicated threads with
@app.thread(name) - Lifecycle Hooks: Register functions to run at service startup with
@app.on_start() - HTTP: Add HTTP endpoints with
@app.route(path, methods)
When working with Daebus in different contexts (HTTP handlers, background tasks, service initialization), follow these guidelines:
- Main Handlers: In action handlers and HTTP route handlers, you can use
logger,request, andresponse. - Background Tasks: Always use
direct_loggerfor logging in background tasks and scheduled jobs. - Initialization: Use
direct_loggerbefore callingapp.run(). - On Start Handlers: For consistency, use
direct_loggerinon_starthandlers.
Daebus 0.0.16+ includes built-in concurrent message processing to handle multiple incoming messages simultaneously:
- Messages are processed in a thread pool (default: 10 worker threads)
- Each message gets its own dedicated thread-local context
- Multiple services can communicate with the same service without blocking each other
- Long-running message handlers won't block other messages from being processed
You can adjust the thread pool size when initializing Daebus:
app = Daebus(__name__, max_workers=20) # Use 20 worker threads instead of the default 10This ensures that your service can handle multiple concurrent requests efficiently, especially in environments where multiple services might be sending messages at the same time.
Daebus achieves thread safety through several mechanisms:
-
Thread-Local Storage:
- Each message handler thread gets its own isolated thread-local
requestandresponseobjects - Thread-local storage prevents data leakage between concurrent operations
- The framework automatically manages the thread-local context lifecycle
- Each message handler thread gets its own isolated thread-local
-
Proxy Pattern:
- The
requestandresponseglobal objects are actually proxies - These proxies automatically detect which thread is executing and retrieve the correct thread-local objects
- This approach maintains backward compatibility while enabling concurrency
- The
-
Thread Pool Management:
- A configurable
ThreadPoolExecutormanages concurrent message processing - The pool size can be tuned based on system resources and workload characteristics
- Protection against thread pool exhaustion with retry mechanisms and exponential backoff
- A configurable
-
Resource Cleanup:
- Thread-local storage is properly cleaned up after each message is processed
- This prevents memory leaks and ensures resources are released
Based on stress testing, the concurrent processing architecture can handle:
- Thousands of messages per second on modest hardware
- Burst patterns of traffic without message loss
- Long-running handlers without blocking other messages
For optimal performance:
- Thread Pool Size: For most applications, 10-20 workers is sufficient. CPU-bound workloads may benefit from fewer workers (matching CPU core count), while I/O-bound workloads may benefit from more workers.
- Message Processing: Keep message handlers efficient. While concurrency allows long-running handlers, it's still best practice to keep processing time minimal.
- Redis Connection: Ensure your Redis instance can handle the concurrent connections and throughput.
When a handler encounters an exception:
- The error is contained within that specific thread
- Other message processing continues unaffected
- The framework logs the error and properly cleans up thread-local resources
- For background tasks, always use
direct_loggerinstead of the proxylogger
from daebus import Daebus, direct_logger app = Daebus(__name__, max_workers=20) @app.thread("long_running") def background_processor(running): while running(): try: # Process something pass except Exception as e: # Always use direct_logger for thread-safe logging direct_logger.error(f"Error in background processor: {e}")If your service needs to share state between message handlers:
- Use thread-safe data structures from
queueorcollections.concurrent - Protect shared resources with locks from the
threadingmodule - Consider using atomic operations where possible
Example:
import threading from daebus import Daebus, request, response app = Daebus(__name__) # Thread-safe counter using a lock request_counter = 0 counter_lock = threading.Lock() @app.action("increment") def increment_counter(): global request_counter # Use a lock to safely update shared state with counter_lock: request_counter += 1 current_count = request_counter return response.success({"count": current_count})MIT