Python Integration Guide (Django, Flask, FastAPI)

Overview

This guide explains how to integrate the AmpliServ Backend SDK with your Python application. The integration supports all major Python web frameworks including Django, Flask, FastAPI, and async frameworks.

Prerequisites

• Python 3.8 or higher
• pip package manager
• AmpliServ SDK binary

Step 1: Install Required Dependencies

# Core logging library with JSON support
pip install python-json-logger

# For correlation ID (optional)
pip install uuid6

# Framework-specific (choose one)
pip install flask          # For Flask
# OR
pip install django         # For Django
# OR
pip install fastapi uvicorn # For FastAPI

# For async support (FastAPI)
pip install asyncio

Step 2: Create Logger Configuration

logger_config.py (Shared across all frameworks)

import logging
import json
from pythonjsonlogger import jsonlogger
from datetime import datetime
from typing import Dict, Any
import uuid
import os

class CustomJsonFormatter(jsonlogger.JsonFormatter):
    """Custom JSON formatter for AmpliServ SDK compatibility"""
    
    def add_fields(self, log_record: Dict[str, Any], record: logging.LogRecord, message_dict: Dict[str, Any]) -> None:
        super().add_fields(log_record, record, message_dict)
        
        # Add timestamp if not present
        if not log_record.get('timestamp'):
            log_record['timestamp'] = datetime.utcnow().isoformat() + 'Z'
        
        # Ensure level is uppercase
        if log_record.get('level'):
            log_record['level'] = log_record['level'].upper()
        elif record.levelname:
            log_record['level'] = record.levelname.upper()
        
        # Add service info
        log_record['service'] = os.getenv('AMPLISERV_SERVICE_NAME', 'python-backend')
        log_record['environment'] = os.getenv('AMPLISERV_ENV', 'development')
        
        # Ensure message is present
        if not log_record.get('message') and record.getMessage():
            log_record['message'] = record.getMessage()

def setup_logging(log_file_path: str = "app.log") -> logging.Logger:
    """
    Setup JSON logging for AmpliServ SDK
    
    Args:
        log_file_path: Path to the log file (default: app.log)
    
    Returns:
        Configured logger instance
    """
    
    # Create logger
    logger = logging.getLogger('ampliserv')
    logger.setLevel(logging.DEBUG)
    
    # Remove existing handlers
    logger.handlers.clear()
    
    # File handler (for AmpliServ SDK)
    file_handler = logging.FileHandler(log_file_path)
    file_handler.setLevel(logging.DEBUG)
    
    # Console handler (for debugging)
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    
    # JSON formatter
    formatter = CustomJsonFormatter('%(timestamp)s %(level)s %(name)s %(message)s')
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)
    
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    
    return logger

# Global correlation ID context (thread-local)
import threading
_correlation_id_local = threading.local()

def set_correlation_id(correlation_id: str):
    """Set correlation ID for current thread/request"""
    _correlation_id_local.correlation_id = correlation_id

def get_correlation_id() -> str:
    """Get correlation ID for current thread/request"""
    return getattr(_correlation_id_local, 'correlation_id', 'unknown')

def generate_correlation_id() -> str:
    """Generate a new UUID for correlation"""
    return str(uuid.uuid4())

# Initialize logger
logger = setup_logging()

# Custom adapter to include correlation_id automatically
class CorrelationLoggerAdapter(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        correlation_id = get_correlation_id()
        if correlation_id and correlation_id != 'unknown':
            if 'extra' not in kwargs:
                kwargs['extra'] = {}
            kwargs['extra']['correlation_id'] = correlation_id
        return msg, kwargs

# Create adapter instance
correlation_logger = CorrelationLoggerAdapter(logger, {})

Step 3: Flask Integration

flask_app.py

from flask import Flask, request, jsonify, g
import uuid
from functools import wraps
import traceback
import sys

from logger_config import (
    correlation_logger as logger, 
    set_correlation_id, 
    get_correlation_id,
    generate_correlation_id
)

app = Flask(__name__)

# ============ Correlation ID Middleware ============
@app.before_request
def before_request():
    """Extract or generate correlation ID before each request"""
    
    # Get correlation ID from header or generate new one
    correlation_id = request.headers.get('X-Correlation-ID')
    if not correlation_id:
        correlation_id = generate_correlation_id()
    
    # Store in Flask's g object and thread-local
    g.correlation_id = correlation_id
    set_correlation_id(correlation_id)
    
    # Log incoming request
    logger.info(
        f"Incoming {request.method} {request.path}",
        extra={
            'method': request.method,
            'path': request.path,
            'ip': request.remote_addr,
            'user_agent': request.headers.get('User-Agent')
        }
    )

@app.after_request
def after_request(response):
    """Log response and add correlation header"""
    
    correlation_id = getattr(g, 'correlation_id', 'unknown')
    response.headers['X-Correlation-ID'] = correlation_id
    
    # Log response (will be picked up by SDK)
    logger.info(
        f"{request.method} {request.path} {response.status_code}",
        extra={
            'method': request.method,
            'path': request.path,
            'status_code': response.status_code,
            'correlation_id': correlation_id
        }
    )
    
    return response

# ============ Error Handler (Captures Full Stack Traces) ============
@app.errorhandler(Exception)
def handle_exception(e):
    """Global exception handler - captures full stack trace"""
    
    correlation_id = getattr(g, 'correlation_id', 'unknown')
    
    # Get full stack trace
    exc_type, exc_value, exc_traceback = sys.exc_info()
    stack_trace = ''.join(traceback.format_exception(exc_type, exc_value, exc_traceback))
    
    # Log with full stack trace
    logger.error(
        str(e),
        extra={
            'error_type': exc_type.__name__,
            'stack_trace': stack_trace,
            'correlation_id': correlation_id
        }
    )
    
    return jsonify({
        'message': str(e),
        'correlation_id': correlation_id
    }), 500

# ============ Test Endpoints ============
@app.route('/api/test/info', methods=['GET'])
def test_info():
    """Test endpoint for INFO level logs"""
    correlation_id = getattr(g, 'correlation_id', 'unknown')
    
    logger.info("Info endpoint called - This is a normal log")
    
    return jsonify({
        'message': 'Info logged successfully',
        'correlation_id': correlation_id
    })

@app.route('/api/test/warn', methods=['GET'])
def test_warn():
    """Test endpoint for WARNING level logs"""
    correlation_id = getattr(g, 'correlation_id', 'unknown')
    
    logger.warning("Warning endpoint called - This is a warning log")
    
    return jsonify({
        'message': 'Warning logged',
        'correlation_id': correlation_id
    })

@app.route('/api/test/error', methods=['GET'])
def test_error():
    """Test endpoint that generates an error with deep stack trace"""
    correlation_id = getattr(g, 'correlation_id', 'unknown')
    
    logger.info("Test error endpoint called")
    
    def level3():
        raise ValueError("This is a test error with deep stack trace from Python!")
    
    def level2():
        level3()
    
    def level1():
        level2()
    
    try:
        level1()
    except Exception as e:
        # This will be caught by global error handler
        raise e
    
    return jsonify({'message': 'No error', 'correlation_id': correlation_id})

@app.route('/api/dashboard', methods=['GET'])
def dashboard():
    """Dashboard endpoint with business data"""
    correlation_id = getattr(g, 'correlation_id', 'unknown')
    
    logger.info("Dashboard endpoint called")
    
    dashboard_data = {
        'totalProducts': 100,
        'totalOrders': 50,
        'totalCustomers': 75,
        'totalRevenue': 250000,
        'correlation_id': correlation_id
    }
    
    logger.debug(f"Dashboard data: {dashboard_data}")
    
    return jsonify(dashboard_data)

# ============ Auth Endpoint ============
@app.route('/api/auth/login', methods=['POST'])
def login():
    """Authentication endpoint"""
    correlation_id = getattr(g, 'correlation_id', 'unknown')
    
    data = request.get_json()
    username = data.get('username', '')
    password = data.get('password', '')
    
    logger.info(f"Login attempt for user: {username}")
    
    if username == 'admin' and password == 'admin123':
        logger.info(f"User logged in successfully: {username}")
        
        return jsonify({
            'token': f'mock-jwt-token-{uuid.uuid4()}',
            'user': {'username': username},
            'correlation_id': correlation_id
        })
    
    logger.warning(f"Authentication failed: Invalid credentials for user: {username}")
    
    return jsonify({
        'message': 'Invalid username or password',
        'correlation_id': correlation_id
    }), 401

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000, debug=True)

Step 4: Django Integration

django_app/settings.py

import os
from pathlib import Path

# Build paths
BASE_DIR = Path(__file__).resolve().parent.parent

# Security
SECRET_KEY = 'django-insecure-your-secret-key-here'
DEBUG = True
ALLOWED_HOSTS = ['*']

# Application definition
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',  # For API endpoints
    'corsheaders',     # For CORS support
    'api',             # Custom API app
]

MIDDLEWARE = [
    'corsheaders.middleware.CorsMiddleware',
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    'api.middleware.CorrelationIdMiddleware',  # Custom middleware
]

ROOT_URLCONF = 'django_app.urls'

TEMPLATES = [
    {
        'BACKEND': 'django.template.backends.django.DjangoTemplates',
        'DIRS': [],
        'APP_DIRS': True,
        'OPTIONS': {
            'context_processors': [
                'django.template.context_processors.debug',
                'django.template.context_processors.request',
                'django.contrib.auth.context_processors.auth',
                'django.contrib.messages.context_processors.messages',
            ],
        },
    },
]

WSGI_APPLICATION = 'django_app.wsgi.application'

# Database
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
    }
}

# Password validation
AUTH_PASSWORD_VALIDATORS = [
    {'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator'},
    {'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator'},
    {'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator'},
    {'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator'},
]

# Internationalization
LANGUAGE_CODE = 'en-us'
TIME_ZONE = 'UTC'
USE_I18N = True
USE_TZ = True

# Static files
STATIC_URL = 'static/'

# Default primary key field type
DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'

# CORS settings
CORS_ALLOW_ALL_ORIGINS = True

# Logging configuration for AmpliServ SDK
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'json': {
            '()': 'pythonjsonlogger.jsonlogger.JsonFormatter',
            'format': '%(timestamp)s %(level)s %(name)s %(message)s %(correlation_id)s',
        },
    },
    'handlers': {
        'file': {
            'level': 'DEBUG',
            'class': 'logging.FileHandler',
            'filename': BASE_DIR / 'app.log',
            'formatter': 'json',
        },
        'console': {
            'level': 'INFO',
            'class': 'logging.StreamHandler',
            'formatter': 'json',
        },
    },
    'loggers': {
        'ampliserv': {
            'handlers': ['file', 'console'],
            'level': 'DEBUG',
            'propagate': True,
        },
    },
}

django_app/api/middleware.py

import threading
import uuid
from django.utils.deprecation import MiddlewareMixin

# Thread-local storage for correlation ID
_local = threading.local()

def get_correlation_id():
    return getattr(_local, 'correlation_id', 'unknown')

def set_correlation_id(correlation_id):
    _local.correlation_id = correlation_id

class CorrelationIdMiddleware(MiddlewareMixin):
    """Middleware to handle correlation ID for each request"""
    
    def process_request(self, request):
        # Get correlation ID from header or generate new
        correlation_id = request.headers.get('X-Correlation-ID')
        if not correlation_id:
            correlation_id = str(uuid.uuid4())
        
        # Store in thread-local
        set_correlation_id(correlation_id)
        
        # Store in request for access in views
        request.correlation_id = correlation_id
        
        # Import logger here to avoid circular import
        from .logger import correlation_logger
        
        # Log incoming request
        correlation_logger.info(
            f"Incoming {request.method} {request.path}",
            extra={
                'method': request.method,
                'path': request.path,
                'ip': request.META.get('REMOTE_ADDR'),
                'user_agent': request.META.get('HTTP_USER_AGENT')
            }
        )
    
    def process_response(self, request, response):
        # Add correlation ID to response headers
        correlation_id = getattr(request, 'correlation_id', 'unknown')
        response['X-Correlation-ID'] = correlation_id
        
        # Import logger here to avoid circular import
        from .logger import correlation_logger
        
        # Log response
        correlation_logger.info(
            f"{request.method} {request.path} {response.status_code}",
            extra={
                'method': request.method,
                'path': request.path,
                'status_code': response.status_code,
                'correlation_id': correlation_id
            }
        )
        
        return response
    
    def process_exception(self, request, exception):
        """Handle exceptions and capture stack traces"""
        import traceback
        import sys
        
        correlation_id = getattr(request, 'correlation_id', 'unknown')
        
        # Get full stack trace
        exc_type, exc_value, exc_traceback = sys.exc_info()
        stack_trace = ''.join(traceback.format_exception(exc_type, exc_value, exc_traceback))
        
        # Import logger here to avoid circular import
        from .logger import correlation_logger
        
        # Log with full stack trace
        correlation_logger.error(
            str(exception),
            extra={
                'error_type': exc_type.__name__,
                'stack_trace': stack_trace,
                'correlation_id': correlation_id
            }
        )
        
        return None

django_app/api/logger.py

import logging
from pythonjsonlogger import jsonlogger
from datetime import datetime
from typing import Dict, Any
import os

class CustomJsonFormatter(jsonlogger.JsonFormatter):
    """Custom JSON formatter for AmpliServ SDK"""
    
    def add_fields(self, log_record: Dict[str, Any], record: logging.LogRecord, message_dict: Dict[str, Any]) -> None:
        super().add_fields(log_record, record, message_dict)
        
        if not log_record.get('timestamp'):
            log_record['timestamp'] = datetime.utcnow().isoformat() + 'Z'
        
        if log_record.get('level'):
            log_record['level'] = log_record['level'].upper()
        
        log_record['service'] = os.getenv('AMPLISERV_SERVICE_NAME', 'django-backend')
        log_record['environment'] = os.getenv('AMPLISERV_ENV', 'development')

# Get logger instance
logger = logging.getLogger('ampliserv')

# Create adapter for correlation ID
class CorrelationLoggerAdapter(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        from .middleware import get_correlation_id
        correlation_id = get_correlation_id()
        if correlation_id and correlation_id != 'unknown':
            if 'extra' not in kwargs:
                kwargs['extra'] = {}
            kwargs['extra']['correlation_id'] = correlation_id
        return msg, kwargs

correlation_logger = CorrelationLoggerAdapter(logger, {})

django_app/api/views.py

from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework import status
import uuid
import traceback
import sys

from .logger import correlation_logger as logger

@api_view(['GET'])
def test_info(request):
    """Test endpoint for INFO level logs"""
    correlation_id = getattr(request, 'correlation_id', 'unknown')
    
    logger.info("Info endpoint called - This is a normal log")
    
    return Response({
        'message': 'Info logged successfully',
        'correlation_id': correlation_id
    })

@api_view(['GET'])
def test_warn(request):
    """Test endpoint for WARNING level logs"""
    correlation_id = getattr(request, 'correlation_id', 'unknown')
    
    logger.warning("Warning endpoint called - This is a warning log")
    
    return Response({
        'message': 'Warning logged',
        'correlation_id': correlation_id
    })

@api_view(['GET'])
def test_error(request):
    """Test endpoint that generates an error with deep stack trace"""
    correlation_id = getattr(request, 'correlation_id', 'unknown')
    
    logger.info("Test error endpoint called")
    
    def level3():
        raise ValueError("This is a test error with deep stack trace from Django!")
    
    def level2():
        level3()
    
    def level1():
        level2()
    
    try:
        level1()
    except Exception as e:
        # Re-raise to let middleware handle
        raise e
    
    return Response({'message': 'No error', 'correlation_id': correlation_id})

@api_view(['GET'])
def dashboard(request):
    """Dashboard endpoint"""
    correlation_id = getattr(request, 'correlation_id', 'unknown')
    
    logger.info("Dashboard endpoint called")
    
    dashboard_data = {
        'totalProducts': 100,
        'totalOrders': 50,
        'totalCustomers': 75,
        'totalRevenue': 250000,
        'correlation_id': correlation_id
    }
    
    logger.debug(f"Dashboard data: {dashboard_data}")
    
    return Response(dashboard_data)

@api_view(['POST'])
def login(request):
    """Authentication endpoint"""
    correlation_id = getattr(request, 'correlation_id', 'unknown')
    
    username = request.data.get('username', '')
    password = request.data.get('password', '')
    
    logger.info(f"Login attempt for user: {username}")
    
    if username == 'admin' and password == 'admin123':
        logger.info(f"User logged in successfully: {username}")
        
        return Response({
            'token': f'mock-jwt-token-{uuid.uuid4()}',
            'user': {'username': username},
            'correlation_id': correlation_id
        })
    
    logger.warning(f"Authentication failed: Invalid credentials for user: {username}")
    
    return Response({
        'message': 'Invalid username or password',
        'correlation_id': correlation_id
    }, status=status.HTTP_401_UNAUTHORIZED)

django_app/api/urls.py

from django.urls import path
from . import views

urlpatterns = [
    path('api/test/info', views.test_info, name='test_info'),
    path('api/test/warn', views.test_warn, name='test_warn'),
    path('api/test/error', views.test_error, name='test_error'),
    path('api/dashboard', views.dashboard, name='dashboard'),
    path('api/auth/login', views.login, name='login'),
]

django_app/urls.py

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('api.urls')),
]

Step 5: FastAPI Integration (Async)

fastapi_app.py

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from contextvars import ContextVar
import uuid
import traceback
import sys
import asyncio
from typing import Dict, Any

from logger_config import logger, set_correlation_id, get_correlation_id, generate_correlation_id

# Context variable for correlation ID (async-safe)
_correlation_id_var: ContextVar[str] = ContextVar('correlation_id', default='unknown')

def set_correlation_id_async(correlation_id: str):
    _correlation_id_var.set(correlation_id)

def get_correlation_id_async() -> str:
    return _correlation_id_var.get()

# Patch the logger_config functions to use async context
import logger_config
logger_config.set_correlation_id = set_correlation_id_async
logger_config.get_correlation_id = get_correlation_id_async

app = FastAPI(title="ERP Backend API", version="1.0.0")

# ============ Correlation ID Middleware ============
@app.middleware("http")
async def correlation_middleware(request: Request, call_next):
    """Extract or generate correlation ID for each request"""
    
    # Get correlation ID from header or generate new
    correlation_id = request.headers.get('X-Correlation-ID')
    if not correlation_id:
        correlation_id = generate_correlation_id()
    
    # Set in context
    set_correlation_id_async(correlation_id)
    
    # Log incoming request
    logger.info(
        f"Incoming {request.method} {request.url.path}",
        extra={
            'method': request.method,
            'path': request.url.path,
            'ip': request.client.host if request.client else 'unknown',
            'user_agent': request.headers.get('User-Agent')
        }
    )
    
    # Process request
    start_time = asyncio.get_event_loop().time()
    
    try:
        response = await call_next(request)
        
        # Add correlation ID to response headers
        response.headers['X-Correlation-ID'] = correlation_id
        
        # Log response
        duration = (asyncio.get_event_loop().time() - start_time) * 1000
        logger.info(
            f"{request.method} {request.url.path} {response.status_code} - {duration:.0f}ms",
            extra={
                'method': request.method,
                'path': request.url.path,
                'status_code': response.status_code,
                'duration_ms': duration,
                'correlation_id': correlation_id
            }
        )
        
        return response
        
    except Exception as e:
        # Handle exception with full stack trace
        duration = (asyncio.get_event_loop().time() - start_time) * 1000
        
        # Get full stack trace
        exc_type, exc_value, exc_traceback = sys.exc_info()
        stack_trace = ''.join(traceback.format_exception(exc_type, exc_value, exc_traceback))
        
        logger.error(
            str(e),
            extra={
                'error_type': exc_type.__name__,
                'stack_trace': stack_trace,
                'method': request.method,
                'path': request.url.path,
                'duration_ms': duration,
                'correlation_id': correlation_id
            }
        )
        
        return JSONResponse(
            status_code=500,
            content={
                'message': str(e),
                'correlation_id': correlation_id
            }
        )

# ============ Exception Handler ============
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Global exception handler with full stack trace"""
    correlation_id = get_correlation_id_async()
    
    exc_type, exc_value, exc_traceback = sys.exc_info()
    stack_trace = ''.join(traceback.format_exception(exc_type, exc_value, exc_traceback))
    
    logger.error(
        str(exc),
        extra={
            'error_type': exc_type.__name__,
            'stack_trace': stack_trace,
            'correlation_id': correlation_id
        }
    )
    
    return JSONResponse(
        status_code=500,
        content={
            'message': str(exc),
            'correlation_id': correlation_id
        }
    )

# ============ Test Endpoints ============
@app.get("/api/test/info")
async def test_info(request: Request):
    """Test endpoint for INFO level logs"""
    correlation_id = get_correlation_id_async()
    
    logger.info("Info endpoint called - This is a normal log")
    
    return {
        'message': 'Info logged successfully',
        'correlation_id': correlation_id
    }

@app.get("/api/test/warn")
async def test_warn(request: Request):
    """Test endpoint for WARNING level logs"""
    correlation_id = get_correlation_id_async()
    
    logger.warning("Warning endpoint called - This is a warning log")
    
    return {
        'message': 'Warning logged',
        'correlation_id': correlation_id
    }

@app.get("/api/test/error")
async def test_error(request: Request):
    """Test endpoint that generates an error with deep stack trace"""
    correlation_id = get_correlation_id_async()
    
    logger.info("Test error endpoint called")
    
    async def level3():
        raise ValueError("This is a test error with deep stack trace from FastAPI!")
    
    async def level2():
        await level3()
    
    async def level1():
        await level2()
    
    try:
        await level1()
    except Exception as e:
        raise e
    
    return {'message': 'No error', 'correlation_id': correlation_id}

@app.get("/api/dashboard")
async def dashboard(request: Request):
    """Dashboard endpoint"""
    correlation_id = get_correlation_id_async()
    
    logger.info("Dashboard endpoint called")
    
    dashboard_data = {
        'totalProducts': 100,
        'totalOrders': 50,
        'totalCustomers': 75,
        'totalRevenue': 250000,
        'correlation_id': correlation_id
    }
    
    logger.debug(f"Dashboard data: {dashboard_data}")
    
    return dashboard_data

@app.post("/api/auth/login")
async def login(request: Request):
    """Authentication endpoint"""
    correlation_id = get_correlation_id_async()
    
    body = await request.json()
    username = body.get('username', '')
    password = body.get('password', '')
    
    logger.info(f"Login attempt for user: {username}")
    
    if username == 'admin' and password == 'admin123':
        logger.info(f"User logged in successfully: {username}")
        
        return {
            'token': f'mock-jwt-token-{uuid.uuid4()}',
            'user': {'username': username},
            'correlation_id': correlation_id
        }
    
    logger.warning(f"Authentication failed: Invalid credentials for user: {username}")
    
    return JSONResponse(
        status_code=401,
        content={
            'message': 'Invalid username or password',
            'correlation_id': correlation_id
        }
    )

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=3000)

Step 6: Configure AmpliServ SDK

Create run-agent.sh (Linux/Mac):

#!/bin/bash
export AMPLISERV_API_KEY="your-api-key-here"
export AMPLISERV_COLLECTOR_MODE="file"
export AMPLISERV_ENDPOINT="http://dev-backend.ampliserv.com/api/ingestion/v1/ingest"
export AMPLISERV_SERVICE_NAME="ingestion-service"
export AMPLISERV_ENV="development"
export AMPLISERV_LOG_PATH="/path/to/your/project/app.log"
export AMPLISERV_BATCH_SIZE="50"
export AMPLISERV_FLUSH_INTERVAL_SEC="5"
export AMPLISERV_INCIDENT_ENABLED="true"

echo "Starting AmpliServ Agent..."
./ampliserv-agent

Create run-agent.bat (Windows):

@echo off
set AMPLISERV_API_KEY=your-api-key-here
set AMPLISERV_COLLECTOR_MODE=file
set AMPLISERV_ENDPOINT=http://dev-backend.ampliserv.com/api/ingestion/v1/ingest
set AMPLISERV_SERVICE_NAME=ingestion-service
set AMPLISERV_ENV=development
set AMPLISERV_LOG_PATH=C:\path\to\your\project\app.log
set AMPLISERV_BATCH_SIZE=50
set AMPLISERV_FLUSH_INTERVAL_SEC=5
set AMPLISERV_INCIDENT_ENABLED=true

echo Starting AmpliServ Agent...
ampliserv-agent.exe

Step 7: Run the Integration

Terminal 1: Start Python Application

Flask:

export FLASK_APP=flask_app.py
flask run --port=3000
# OR
python flask_app.py

Django:

python manage.py runserver 0.0.0.0:3000

FastAPI:

python fastapi_app.py
# OR
uvicorn fastapi_app:app --reload --port 3000

Terminal 2: Start AmpliServ SDK

chmod +x run-agent.sh
./run-agent.sh

Terminal 3: Test the Integration

# Test info endpoint
curl http://localhost:3000/api/test/info

# Test warning endpoint
curl http://localhost:3000/api/test/warn

# Test error endpoint (generates stack trace)
curl http://localhost:3000/api/test/error

# Test dashboard
curl http://localhost:3000/api/dashboard

# Test login
curl -X POST http://localhost:3000/api/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username":"admin","password":"admin123"}'

Sample JSON Log Output

{
  "timestamp": "2026-03-23T10:30:45.123456Z",
  "level": "ERROR",
  "name": "ampliserv",
  "message": "Test error occurred",
  "correlation_id": "a1270278-7ef9-4e59-803e-e9862c257d73",
  "service": "python-backend",
  "environment": "development",
  "error_type": "ValueError",
  "stack_trace": "Traceback (most recent call last):\n  File \"/app/fastapi_app.py\", line 135, in test_error\n    await level1()\n  File \"/app/fastapi_app.py\", line 131, in level1\n    await level2()\n  File \"/app/fastapi_app.py\", line 128, in level2\n    await level3()\n  File \"/app/fastapi_app.py\", line 125, in level3\n    raise ValueError(\"This is a test error with deep stack trace from Python!\")\nValueError: This is a test error with deep stack trace from Python!"
}

Docker Deployment

Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Create log directory
RUN mkdir -p /app/logs

# Expose port
EXPOSE 3000

# Run application
CMD ["python", "flask_app.py"]

docker-compose.yml

version: '3.8'

services:
  app:
    build: .
    ports:
      - "3000:3000"
    volumes:
      - ./app.log:/app/app.log
    environment:
      - AMPLISERV_SERVICE_NAME=python-backend
      - AMPLISERV_ENV=production
    restart: unless-stopped

  ampliserv-agent:
    image: ampliserv/agent:latest
    volumes:
      - ./app.log:/app/app.log:ro
    environment:
      - AMPLISERV_API_KEY=${AMPLISERV_API_KEY}
      - AMPLISERV_COLLECTOR_MODE=file
      - AMPLISERV_LOG_PATH=/app/app.log
      - AMPLISERV_SERVICE_NAME=python-backend
      - AMPLISERV_ENV=production
    restart: unless-stopped

requirements.txt

# Core logging
python-json-logger==2.0.7
uuid6==2023.12.1

# Flask
Flask==3.0.0

# Django
Django==5.0
djangorestframework==3.14.0
django-cors-headers==4.3.1

# FastAPI
fastapi==0.104.1
uvicorn==0.24.0

# Async support
httpx==0.25.1

Verification Checklist

• Python application starts without errors
• app.log file is created with JSON content
• Logs are in valid JSON format
• AmpliServ SDK shows "Starting AmpliServ Agent"
• SDK displays "Mode: file"
• Test endpoints produce logs
• Correlation ID is present in all logs
• Full stack traces are captured in error logs

Common Issues and Solutions

Issue	Solution
Module not found errors	Run pip install -r requirements.txt
JSON parsing errors	Ensure python-json-logger is installed
File permission denied	Run chmod 666 app.log or use absolute path
Correlation ID missing in async	Use ContextVar instead of threading.local
Stack traces truncated	Python includes full stack traces by default

Production Considerations

Environment Variables for Production

export AMPLISERV_ENV="production"
export AMPLISERV_LOG_LEVEL="WARNING"  # Reduce logging in production
export PYTHONUNBUFFERED=1  # Ensure logs are flushed immediately

Gunicorn Configuration (Flask/Django)

pip install gunicorn

# Run with gunicorn
gunicorn flask_app:app --workers 4 --bind 0.0.0.0:3000 --access-logfile -

Log Rotation

from logging.handlers import RotatingFileHandler

# Use RotatingFileHandler instead of FileHandler
handler = RotatingFileHandler(
    'app.log',
    maxBytes=10485760,  # 10MB
    backupCount=5
)

Performance Tuning

• Use async logging for high-throughput applications
• Set appropriate log levels in production (INFO or WARNING)
• Use buffered logging to reduce I/O
• Monitor log file size and implement rotation

# Async logging example
import asyncio
from logging.handlers import QueueHandler, QueueListener
from queue import Queue

log_queue = Queue(-1)
queue_handler = QueueHandler(log_queue)
listener = QueueListener(log_queue, file_handler, console_handler)
listener.start()

Framework-Specific Notes

Framework	Key Points
Flask	Use g object for request-scoped data; @app.before_request and @app.after_request
Django	Use middleware for correlation ID; thread-local storage works
FastAPI	Use ContextVar for async-safe correlation ID; dependency injection for request context

This comprehensive guide provides complete integration for Python applications with the AmpliServ SDK, including full stack trace capture, correlation ID propagation, and support for all major frameworks.