laneswap-client 0.2.0
Effortless LaneSwap Monitor integration for Python services
License: MIT
Requires Python: >=3.8
Keywords: monitoring, health-check, microservices, process-manager, laneswap

# LaneSwap Client

**Effortless LaneSwap Monitor integration for Python services**

`laneswap-client` is a lightweight Python library that provides zero-config integration with [LaneSwap Monitor](https://github.com/laneswap/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

```bash
# 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)

```python
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

```python
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)

```python
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

```python
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.

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

For FastAPI, you can use async health checks:

```python
@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.

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

For FastAPI with async cleanup:

```python
@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.

```python
laneswap.run()
```

## Health Check Response Format

The library automatically formats health check responses:

```json
{
    "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:

```json
{
    "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:

```bash
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)

```python
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)

```python
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()`:

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

### Signal handlers not working in tests

Disable automatic signal registration:

```python
laneswap = LaneSwap(app, setup_signals=False)
```

### Port conflicts

Override the port explicitly:

```python
laneswap = LaneSwap(app, port=8080)
```

Or set the environment variable:

```bash
export PORT=8080
```

### Async health checks not running

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

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

## Development

```bash
# 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](CONTRIBUTING.md) for development guidelines.

## Documentation

- [API Reference](docs/api-reference.md)
- [Flask Integration Guide](docs/flask-guide.md)
- [FastAPI Integration Guide](docs/fastapi-guide.md)
- [Vanilla Mode Guide](docs/vanilla-guide.md)
- [Migration Guide](docs/migration-guide.md)
- [Troubleshooting](docs/troubleshooting.md)

## License

MIT License - see [LICENSE](LICENSE) file for details.

## Links

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

## Changelog

See [CHANGELOG.md](CHANGELOG.md) for version history and release notes.

---

Made with precision by the LaneSwap Team