laneswap-client 0.2.0

Effortless LaneSwap Monitor integration for Python services

LaneSwap Client

Effortless LaneSwap Monitor integration for Python services

laneswap-client is a lightweight Python library that provides zero-config integration with LaneSwap Monitor, reducing integration boilerplate from 50-80 lines down to just 5-10 lines of code.

Features

• Auto-framework detection: Automatically detects Flask, FastAPI, and configures health endpoints
• Decorator-based API: Simple @health_check and @on_shutdown decorators
• Signal handlers: Auto-registers SIGTERM/SIGINT handlers for graceful shutdown
• Async support: Full support for async health checks and shutdown handlers (FastAPI)
• Zero dependencies: Core library has no dependencies; framework integrations are optional extras
• Type hints: Full type annotations with py.typed marker for IDE support
• Built-in server: Includes HTTP server for framework-free services
• Python 3.8+: Compatible with Python 3.8, 3.9, 3.10, 3.11, and 3.12

Installation

# Core library only (framework-free)
pip install laneswap-client

# With Flask support
pip install laneswap-client[flask]

# With FastAPI support
pip install laneswap-client[fastapi]

# All integrations
pip install laneswap-client[all]

Quick Start

Flask Example (8 lines)

from flask import Flask
from laneswap import LaneSwap

app = Flask(__name__)
laneswap = LaneSwap(app)  # Auto-configures everything!

@laneswap.health_check
def check_database():
    return {"database": "connected"}

@laneswap.on_shutdown
def cleanup():
    db.close()

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=laneswap.port)

FastAPI Example with Async

from fastapi import FastAPI
import uvicorn
from laneswap import LaneSwap

app = FastAPI()
laneswap = LaneSwap(app)

@laneswap.health_check
async def check_redis():
    await redis.ping()
    return {"redis": "connected"}

@laneswap.on_shutdown
async def cleanup():
    await redis.close()

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=laneswap.port)

Vanilla Python (No Framework)

from laneswap import LaneSwap

laneswap = LaneSwap()  # Uses built-in HTTP server

@laneswap.health_check
def check_app():
    return {"status": "ready"}

@laneswap.on_shutdown
def cleanup():
    # Cleanup logic
    pass

# Run with built-in server
laneswap.run()

What It Does Automatically

When you create a LaneSwap instance, it automatically:

1. Reads environment variables: SERVICE_NAME, PORT, LOG_LEVEL
2. Registers health endpoint: Auto-creates /health route in your framework
3. Sets up signal handlers: Registers SIGTERM/SIGINT for graceful shutdown
4. Configures logging: Sets up structured logging with timestamps
5. Detects framework: Automatically detects Flask vs FastAPI and configures appropriately
6. Handles async: Properly handles async health checks and shutdown handlers in FastAPI

API Reference

LaneSwap Class

LaneSwap(
    app=None,              # Flask/FastAPI app (None for vanilla mode)
    service_name=None,     # Override SERVICE_NAME env var
    port=None,             # Override PORT env var
    setup_signals=True,    # Auto-register SIGTERM/SIGINT handlers
    setup_logs=True        # Auto-configure logging
)

Decorators

@laneswap.health_check

Register a custom health check function. The function should return a dictionary that will be merged into the health response.

@laneswap.health_check
def check_service():
    # Check service health
    return {"service": "healthy", "connections": 42}

For FastAPI, you can use async health checks:

@laneswap.health_check
async def check_database():
    await db.ping()
    return {"database": "connected"}

@laneswap.on_shutdown

Register a cleanup function to run on graceful shutdown. Supports both sync and async functions.

@laneswap.on_shutdown
def cleanup():
    db.close()
    cache.flush()

For FastAPI with async cleanup:

@laneswap.on_shutdown
async def cleanup():
    await db.close()
    await cache.flush()

Properties

• laneswap.service_name - The service name (from env or default)
• laneswap.port - The service port (from env or default)
• laneswap.logger - Configured logger instance

Methods

laneswap.run()

Run the built-in HTTP server (vanilla mode only). Blocks until shutdown signal received.

laneswap.run()

Health Check Response Format

The library automatically formats health check responses:

{
    "status": "healthy",
    "service": "my-service",
    "timestamp": 1763369448.57,
    "custom_key": "custom_value"
}

Custom keys come from your @health_check decorated functions. If any health check fails (raises an exception), the endpoint returns HTTP 503 with:

{
    "status": "unhealthy",
    "error": "Health check failed: <error message>"
}

Configuration

The library reads the following environment variables:

• SERVICE_NAME - Service identifier (default: "python-service")
• PORT - HTTP port for health endpoint (default: 5000)
• LOG_LEVEL - Logging level (default: "INFO")

Example:

export SERVICE_NAME=my-api
export PORT=8080
export LOG_LEVEL=DEBUG
python app.py

Framework-Specific Behavior

Flask Integration

• Registers /health endpoint using @app.route
• Returns jsonify() response
• Registers shutdown handler with @app.teardown_appcontext
• Supports only sync health checks

FastAPI Integration

• Registers /health endpoint using @app.get
• Returns JSONResponse
• Registers shutdown handler with @app.on_event("shutdown")
• Supports both sync and async health checks
• All health checks run concurrently

Vanilla Mode (No Framework)

• Starts built-in HTTP server using http.server.HTTPServer
• Runs in separate thread
• Handles only /health endpoint
• Returns 404 for all other paths
• Call laneswap.run() to start server

Best Practices

1. Keep health checks fast - They run on every health check interval (typically 5-10s)
2. Use async for I/O - If checking Redis/DB in FastAPI, use async health checks
3. Clean up resources - Always use @on_shutdown for cleanup (close connections, etc.)
4. Return dicts - Health checks should return dictionaries that get merged into response
5. Handle errors - Health check failures return 503 automatically, no need to handle errors

Examples

See the examples/ directory for complete working examples:

• flask_example.py - Flask service with database health check
• fastapi_example.py - FastAPI service with async Redis check
• vanilla_example.py - Framework-free service with built-in server
• advanced_async.py - Advanced async patterns and multiple health checks

Comparison: Before vs After

Before (50+ lines)

from flask import Flask, jsonify
import signal
import sys
import os
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

SERVICE_NAME = os.getenv('SERVICE_NAME', 'python-service')
PORT = int(os.getenv('PORT', 5000))

health_checks = []
shutdown_handlers = []

def register_health_check(func):
    health_checks.append(func)
    return func

def register_shutdown(func):
    shutdown_handlers.append(func)
    return func

@app.route('/health')
def health():
    try:
        status = {"status": "healthy", "service": SERVICE_NAME}
        for check in health_checks:
            status.update(check())
        return jsonify(status), 200
    except Exception as e:
        return jsonify({"status": "unhealthy", "error": str(e)}), 503

def signal_handler(sig, frame):
    logger.info(f"Received signal {sig}, shutting down...")
    for handler in shutdown_handlers:
        try:
            handler()
        except Exception as e:
            logger.error(f"Shutdown handler failed: {e}")
    sys.exit(0)

signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)

@register_health_check
def check_database():
    return {"database": "connected"}

@register_shutdown
def cleanup():
    db.close()

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=PORT)

After (12 lines)

from flask import Flask
from laneswap import LaneSwap

app = Flask(__name__)
laneswap = LaneSwap(app)

@laneswap.health_check
def check_database():
    return {"database": "connected"}

@laneswap.on_shutdown
def cleanup():
    db.close()

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=laneswap.port)

Troubleshooting

Health endpoint not registered

Make sure you're passing your Flask/FastAPI app to LaneSwap():

laneswap = LaneSwap(app)  # Correct
laneswap = LaneSwap()     # Wrong for Flask/FastAPI (this is vanilla mode)

Signal handlers not working in tests

Disable automatic signal registration:

laneswap = LaneSwap(app, setup_signals=False)

Port conflicts

Override the port explicitly:

laneswap = LaneSwap(app, port=8080)

Or set the environment variable:

export PORT=8080

Async health checks not running

Make sure you're using FastAPI, not Flask (Flask doesn't support async):

from fastapi import FastAPI  # Supports async
from flask import Flask      # Sync only

Development

# Clone the repository
git clone https://github.com/laneswap/laneswap-client.git
cd laneswap-client

# Install in development mode with all extras
pip install -e .[all,dev]

# Run tests
pytest

# Run tests with coverage
pytest --cov=laneswap --cov-report=html

# Type checking
mypy src/laneswap

# Code formatting
black src tests
ruff check src tests

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for development guidelines.

Documentation

• API Reference
• Flask Integration Guide
• FastAPI Integration Guide
• Vanilla Mode Guide
• Migration Guide
• Troubleshooting

License

MIT License - see LICENSE file for details.

Links

• PyPI: https://pypi.org/project/laneswap-client/
• LaneSwap Monitor: https://github.com/laneswap/laneswap-monitor

Changelog

See CHANGELOG.md for version history and release notes.

---

Made with precision by the LaneSwap Team