/

Flask Integration

Flask Integration

Integrate ByteHide Logs seamlessly with Flask to track requests, errors, and user actions throughout your application lifecycle.

Setup

Install the Flask integration:

Bash
pip install bytehide-logs[flask]

Import and initialize logging in your Flask app:

Python
from flask import Flask
from bytehide_logs import Log, LogSettings, AuthUser
from datetime import timedelta

app = Flask(__name__)

# Configure logging
settings = LogSettings(
    duplicate_suppression_window=timedelta(seconds=5),
    mask_sensitive_data=["password", "token", "api_key"]
)
Log.configure(settings)

# Set application metadata
Log.add_meta_context("service", "flask-app")
Log.add_meta_context("environment", app.config.get("ENV", "development"))

Request Tracking with before_request

Log incoming requests using the before_request decorator:

Python
from flask import Flask, request, g
import uuid

app = Flask(__name__)

@app.before_request
def log_request_start():
    """Log incoming request."""
    # Generate correlation ID
    correlation_id = request.headers.get(
        "X-Correlation-ID",
        str(uuid.uuid4())
    )
    g.correlation_id = correlation_id
    
    # Add request context
    Log.add_meta_context("request_id", correlation_id)
    Log.add_meta_context("method", request.method)
    Log.add_meta_context("path", request.path)
    Log.add_meta_context("client_ip", request.remote_addr)
    
    Log.with_tags("request", "start").info(
        f"{request.method} {request.path}"
    )

Response Tracking with after_request

Log responses using the after_request decorator:

Python
from flask import Flask
import time

app = Flask(__name__)

@app.before_request
def track_request_start():
    """Track request timing."""
    g.start_time = time.time()

@app.after_request
def log_request_end(response):
    """Log request completion."""
    if hasattr(g, 'start_time'):
        elapsed = time.time() - g.start_time
        
        Log.with_tags("request", "end") \
            .with_context("status_code", response.status_code) \
            .with_context("elapsed_ms", int(elapsed * 1000)) \
            .with_context("content_length", len(response.data)) \
            .info("Request completed")
    
    # Clear request context
    Log.clear_meta_context()
    
    return response

Error Handling with errorhandler

Log exceptions using the errorhandler decorator:

Python
from flask import Flask, jsonify

app = Flask(__name__)

@app.errorhandler(Exception)
def handle_exception(error):
    """Handle all exceptions with logging."""
    Log.with_tags("error", "unhandled") \
        .with_context("error_type", type(error).__name__) \
        .with_context("error_message", str(error)) \
        .error("Unhandled exception", exception=error)
    
    # Return error response
    return jsonify({
        "error": "Internal server error",
        "correlation_id": g.get("correlation_id")
    }), 500

@app.errorhandler(404)
def handle_not_found(error):
    """Handle 404 errors."""
    Log.with_tags("error", "not_found") \
        .with_context("path", request.path) \
        .warning("Resource not found")
    
    return jsonify({"error": "Not found"}), 404

@app.errorhandler(400)
def handle_bad_request(error):
    """Handle 400 errors."""
    Log.with_tags("error", "bad_request") \
        .with_context("path", request.path) \
        .warning("Bad request")
    
    return jsonify({"error": "Bad request"}), 400

User Authentication Tracking

Track authenticated users with Flask session:

Python
from flask import Flask, request, session, g
from bytehide_logs import Log, AuthUser

app = Flask(__name__)

@app.before_request
def identify_user():
    """Identify authenticated user."""
    if "user_id" in session:
        # Create AuthUser from session
        user = AuthUser(
            id=session["user_id"],
            email=session.get("email"),
            token=session.get("session_token")
        )
        
        # Identify user for logs
        Log.identify(user)
        g.current_user = user

@app.after_request
def logout_user(response):
    """Logout user after request."""
    if hasattr(g, 'current_user'):
        Log.logout()
    
    return response

Complete Flask Example

Python
from flask import Flask, request, session, g, jsonify
from bytehide_logs import Log, LogSettings, AuthUser
from datetime import timedelta
import uuid
import time

# Initialize Flask app
app = Flask(__name__)
app.secret_key = "your-secret-key"

# Configure logging
settings = LogSettings(
    duplicate_suppression_window=timedelta(seconds=5),
    mask_sensitive_data=["password", "token", "api_key"]
)
Log.configure(settings)

# Application metadata
Log.add_meta_context("service", "user-api")
Log.add_meta_context("version", "1.0.0")

# ============== Before Request ==============
@app.before_request
def before_request():
    """Setup request context and logging."""
    # Generate correlation ID
    correlation_id = request.headers.get(
        "X-Correlation-ID",
        str(uuid.uuid4())
    )
    g.correlation_id = correlation_id
    g.start_time = time.time()
    
    # Add request context
    Log.add_meta_context("correlation_id", correlation_id)
    Log.add_meta_context("method", request.method)
    Log.add_meta_context("path", request.path)
    Log.add_meta_context("client_ip", request.remote_addr)
    
    # Identify user if authenticated
    if "user_id" in session:
        user = AuthUser(
            id=session["user_id"],
            email=session.get("email"),
            token=session.get("session_token")
        )
        Log.identify(user)
    
    Log.with_tags("request", "received").info(
        f"{request.method} {request.path}"
    )

# ============== After Request ==============
@app.after_request
def after_request(response):
    """Log request completion and cleanup."""
    # Calculate request duration
    if hasattr(g, 'start_time'):
        elapsed = time.time() - g.start_time
        
        Log.with_tags("request", "completed") \
            .with_context("status_code", response.status_code) \
            .with_context("elapsed_ms", int(elapsed * 1000)) \
            .with_context("response_size", len(response.data)) \
            .info("Request processed")
    
    # Logout user
    Log.logout()
    
    # Clear request context
    Log.clear_meta_context()
    
    return response

# ============== Error Handlers ==============
@app.errorhandler(Exception)
def handle_exception(error):
    """Handle unhandled exceptions."""
    Log.with_tags("error", "exception") \
        .with_context("error_type", type(error).__name__) \
        .error("Unhandled exception", exception=error)
    
    return jsonify({
        "error": "Internal server error",
        "correlation_id": g.correlation_id
    }), 500

@app.errorhandler(404)
def handle_not_found(error):
    """Handle 404 errors."""
    Log.with_tags("error", "not_found").warning("Resource not found")
    return jsonify({"error": "Not found"}), 404

# ============== Routes ==============
@app.route("/login", methods=["POST"])
def login():
    """User login endpoint."""
    try:
        data = request.get_json()
        email = data.get("email")
        password = data.get("password")
        
        Log.with_tags("auth", "login") \
            .with_context("email", email) \
            .info("Login attempt")
        
        # Authenticate user (example)
        user_data = authenticate_user(email, password)
        
        # Set session
        session["user_id"] = user_data["id"]
        session["email"] = user_data["email"]
        session["session_token"] = generate_token()
        
        Log.with_tags("auth", "login", "success") \
            .with_context("user_id", user_data["id"]) \
            .info("User login successful")
        
        return jsonify({
            "user_id": user_data["id"],
            "email": user_data["email"]
        })
        
    except Exception as e:
        Log.with_tags("auth", "login", "error") \
            .error("Login failed", exception=e)
        return jsonify({"error": "Login failed"}), 401

@app.route("/logout", methods=["POST"])
def logout():
    """User logout endpoint."""
    Log.with_tags("auth", "logout").info("User logout initiated")
    
    # Clear session
    session.clear()
    
    Log.with_tags("auth", "logout", "success").info("User logout completed")
    
    return jsonify({"message": "Logged out"})

@app.route("/users/<user_id>", methods=["GET"])
def get_user(user_id):
    """Get user profile."""
    Log.with_tags("user", "profile", "fetch") \
        .with_context("user_id", user_id) \
        .info("Fetching user profile")
    
    try:
        user = fetch_user_from_db(user_id)
        
        Log.with_tags("user", "profile", "success") \
            .with_context("user_id", user_id) \
            .info("User profile retrieved")
        
        return jsonify(user)
        
    except Exception as e:
        Log.with_tags("user", "profile", "error") \
            .with_context("user_id", user_id) \
            .error("Failed to fetch user profile", exception=e)
        return jsonify({"error": "User not found"}), 404

@app.route("/users/<user_id>", methods=["PUT"])
def update_user(user_id):
    """Update user profile."""
    try:
        data = request.get_json()
        
        Log.with_tags("user", "profile", "update") \
            .with_context("user_id", user_id) \
            .with_context("fields", list(data.keys())) \
            .info("Updating user profile")
        
        updated_user = update_user_in_db(user_id, data)
        
        Log.with_tags("user", "profile", "update", "success") \
            .with_context("user_id", user_id) \
            .info("User profile updated")
        
        return jsonify(updated_user)
        
    except Exception as e:
        Log.with_tags("user", "profile", "update", "error") \
            .with_context("user_id", user_id) \
            .error("Failed to update user profile", exception=e)
        return jsonify({"error": "Update failed"}), 500

if __name__ == "__main__":
    Log.info("Flask application starting")
    app.run(debug=True)

Best Practices

Always clean up context after request:

Python
@app.after_request
def cleanup(response):
    """Cleanup request context."""
    Log.logout()  # Logout user
    Log.clear_meta_context()  # Clear request context
    return response

Use correlation IDs for tracking:

Python
correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
g.correlation_id = correlation_id
Log.add_meta_context("correlation_id", correlation_id)

Handle all error types:

Python
@app.errorhandler(Exception)
def handle_all_errors(error):
    """Catch and log all exceptions."""
    Log.error("Unhandled exception", exception=error)
    return error_response(str(error))

Next Steps

Previous
Security Settings
Next
Django