Python SDK (BETA)

Official Python SDK for LogFlux agent communication with native Python integration

BETA SOFTWARE: This SDK is feature-complete for basic logging use cases but is marked as BETA while we gather community feedback and add additional features. The API is stable but may evolve based on user needs.

The LogFlux Python SDK provides a lightweight way to send logs to the LogFlux agent from your Python applications, with native Python features like async support and logging library integration.

View full documentation and examples on GitHub →

Installation

1

pip install logflux

Quick Start

Basic Usage

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


import logflux

# Create client - async mode enabled by default for non-blocking sends
client = logflux.new_unix_client("/tmp/logflux-agent.sock")
client.connect()

# Send log entry (non-blocking with circuit breaker protection)
entry = (logflux.new_log_entry("Hello, LogFlux!", "my-app")
         .with_log_level(logflux.LEVEL_INFO)
         .with_metadata("environment", "production"))

client.send_log_entry(entry)
client.close()

Python Logging Integration

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


import logging
import logflux
from logflux.integrations import LogFluxHandler

client = logflux.new_unix_client("/tmp/logflux-agent.sock")
client.connect()

handler = LogFluxHandler(client=client, source="my-app", json_format=True)
logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.INFO)

logger.info("Application started", extra={"version": "1.0.0"})
handler.close()

Key Features

Agent Communication: Lightweight communication with LogFlux agent via Unix socket or TCP
Circuit Breaker: Automatic failure detection and recovery
High Performance: Async processing with batch support
Python Native: Context managers, type hints, proper error handling
Logger Integration: Drop-in handler for Python’s standard logging
JSON Support: Structured and unstructured logging formats

Configuration

For comprehensive configuration examples and patterns, see Configuration Examples.

Connection Options

1
2
3
4
5


# Unix socket connection (recommended for local agent)
client = logflux.new_unix_client("/tmp/logflux-agent.sock")

# TCP connection (for remote agent)
client = logflux.new_tcp_client("localhost:9090")

Client Configuration

1
2
3
4
5
6
7
8
9


# Custom configuration
config = logflux.Config(
    timeout=5.0,
    retry_attempts=3,
    retry_delay=1.0,
    buffer_size=1000
)

client = logflux.new_unix_client("/tmp/logflux-agent.sock", config)

Batch Processing

Batch Client for High Throughput

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


# Configure batch processing
batch_config = logflux.BatchConfig(
    max_batch_size=10, 
    flush_interval=2.0,
    auto_flush=True
)

client = logflux.new_batch_unix_client("/tmp/logflux-agent.sock", batch_config)
client.connect()

# Send multiple entries - automatically batched
for i in range(25):
    entry = logflux.new_log_entry(f"Log entry {i}", "batch-app")
    client.send_log_entry(entry)

client.close()

Fluent API

1
2
3
4
5
6
7
8


# Chain methods for cleaner code
entry = (logflux.new_log_entry("User authenticated", "auth-service")
         .with_log_level(logflux.LEVEL_INFO)
         .with_source("basic-example")
         .with_metadata("user_id", "12345")
         .with_metadata("environment", "production"))

client.send_log_entry(entry)

Logger Integrations

Python Logging Handler

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32


import logging
import logflux
from logflux.integrations import LogFluxHandler

client = logflux.new_unix_client("/tmp/logflux-agent.sock")
client.connect()

# Create handler with customization
handler = LogFluxHandler(
    client=client,
    source="my-app",
    include_extra=True,
    include_stack_info=True,
    json_format=True,  # Send logs as JSON
)

# Set up logging
logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Use standard logging - goes to LogFlux
logging.info("User logged in", extra={"user_id": 12345, "username": "john.doe"})
logging.error("Database connection failed")

# Exception logging with stack traces
try:
    raise ValueError("This is a test exception")
except Exception:
    logger.exception("Caught an exception")

handler.close()

Batch Logging Integration

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


# Use batch client for high-performance logging
batch_config = logflux.BatchConfig(max_batch_size=5, flush_interval=2.0)
batch_client = logflux.new_batch_unix_client("/tmp/logflux-agent.sock", batch_config)

handler = LogFluxHandler(
    client=batch_client,
    source="batch-app",
    json_format=True
)

# Send multiple log messages efficiently
for i in range(100):
    logging.info(f"Processing item {i}", extra={"item_id": i, "batch": "001"})

Log Levels

For complete log level reference including usage guidelines and examples, see Log Levels Reference.

Python Constants

1
2
3
4
5
6
7
8


logflux.LEVEL_EMERGENCY  # 0 - System is unusable
logflux.LEVEL_ALERT      # 1 - Action must be taken immediately  
logflux.LEVEL_CRITICAL   # 2 - Critical conditions
logflux.LEVEL_ERROR      # 3 - Error conditions
logflux.LEVEL_WARNING    # 4 - Warning conditions
logflux.LEVEL_NOTICE     # 5 - Normal but significant conditions
logflux.LEVEL_INFO       # 6 - Informational messages
logflux.LEVEL_DEBUG      # 7 - Debug-level messages

Best Practices

Always Close Connections

1
2
3
4
5
6
7
8


# Proper resource management
client = logflux.new_unix_client("/tmp/logflux-agent.sock")
try:
    client.connect()
    # Use client
    client.send_log("Hello, LogFlux!", "my-app")
finally:
    client.close()

Use Batch Processing for High Volume

1
2
3
4
5
6
7
8


# Better performance for multiple log entries
batch_config = logflux.BatchConfig(
    max_batch_size=50,
    flush_interval=5.0,
    auto_flush=True
)

client = logflux.new_batch_unix_client("/tmp/logflux-agent.sock", batch_config)

Handle Connection Errors Gracefully

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


try:
    client = logflux.new_unix_client("/tmp/logflux-agent.sock")
    client.connect()
    client.send_log("Application started", "my-app")
except Exception as e:
    # Fallback to local logging
    print(f"LogFlux unavailable: {e}")
finally:
    if client:
        client.close()

Common Use Cases

Web Application Logging

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


# Flask/FastAPI application setup
import logging
import logflux
from logflux.integrations import LogFluxHandler

# Set up LogFlux handler for the application
client = logflux.new_unix_client("/tmp/logflux-agent.sock")
client.connect()

handler = LogFluxHandler(client=client, source="web-app", json_format=True)
app_logger = logging.getLogger("my_app")
app_logger.addHandler(handler)

# Application logs now go to LogFlux
@app.route("/login")
def login():
    app_logger.info("User login attempt", extra={"ip": request.remote_addr})
    return "Login page"

Background Task Processing

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


# High-throughput batch logging for workers
batch_config = logflux.BatchConfig(max_batch_size=100, auto_flush=True)
client = logflux.new_batch_unix_client("/tmp/logflux-agent.sock", batch_config)
client.connect()

def process_task(task_id):
    client.send_log(f"Started processing task {task_id}", "worker")
    
    # Process task...
    
    client.send_log(f"Completed task {task_id}", "worker")

Microservice Communication

1
2
3
4
5
6
7
8


# Service-to-service logging with metadata
entry = (logflux.new_log_entry("API call to user service", "api-gateway")
         .with_log_level(logflux.LEVEL_INFO)
         .with_metadata("service", "user-service")
         .with_metadata("endpoint", "/users/123")
         .with_metadata("response_time", "45ms"))

client.send_log_entry(entry)

Current Status & Roadmap

Stable API for core logging functionality
Production quality code and testing
Ready for evaluation and non-critical use cases
Additional features (metrics, traces, events) coming soon
Gathering feedback for API refinements

Requirements

Python: 3.8 or later
Dependencies: None (lightweight, no external dependencies)
LogFlux Agent: Requires running LogFlux agent for communication

Support

Full Documentation: GitHub Repository
Issues: GitHub Issues
LogFlux Website: logflux.io

Disclaimer

The Python logo and trademarks are the property of the Python Software Foundation. LogFlux is not affiliated with, endorsed by, or sponsored by the Python Software Foundation. The Python logo is used solely for identification purposes to indicate compatibility and integration capabilities.