cloud-logging-handler 0.1.0

A Python logging handler for Google Cloud Logging with FastAPI integration and request tracing support

Cloud Logging Handler

A Python logging handler for Google Cloud Logging with FastAPI integration and request tracing support.

Features

• Structured JSON Logging: Outputs logs in Google Cloud Logging's structured format
• Request Tracing: Automatic trace context propagation via X-Cloud-Trace-Context header
• Log Aggregation: Aggregates all logs within a single request into one log entry
• Severity Tracking: Automatically tracks the highest severity level per request
• FastAPI Integration: Built-in error handlers for seamless FastAPI integration
• Custom JSON Encoder: Support for high-performance JSON libraries (e.g., ujson)
• Zero Dependencies: Core handler has no external dependencies

Installation

# Using uv (recommended)
uv add cloud-logging-handler

# Using pip
pip install cloud-logging-handler

# With FastAPI support
uv add "cloud-logging-handler[fastapi]"

# With ujson for better performance
uv add "cloud-logging-handler[ujson]"

# With all optional dependencies
uv add "cloud-logging-handler[all]"

Quick Start

Basic Usage

import logging
from cloud_logging_handler import CloudLoggingHandler

# Create handler
handler = CloudLoggingHandler(
    trace_header_name="X-Cloud-Trace-Context",
    project="your-gcp-project-id"
)

# Configure logging
logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)

# Log messages
logger.info("Hello, Cloud Logging!")

FastAPI Integration

import logging
import os
from fastapi import FastAPI, Request
from cloud_logging_handler import CloudLoggingHandler, RequestLogs, add_error_handler

app = FastAPI()

# Initialize handler
handler = CloudLoggingHandler(
    trace_header_name="X-Cloud-Trace-Context",
    project=os.environ.get("GCP_PROJECT"),
)

# Configure logging
logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)

# Add middleware for request context
@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    request.state.token = handler.set_request(RequestLogs(request, {}))
    response = await call_next(request)
    handler.flush()
    return response

# Add error handlers
add_error_handler(app, logging_handler=handler)

@app.get("/")
async def root():
    logging.info("Processing request")
    return {"message": "Hello World"}

Using with ujson

For better JSON serialization performance:

import ujson
from cloud_logging_handler import CloudLoggingHandler

handler = CloudLoggingHandler(
    trace_header_name="X-Cloud-Trace-Context",
    json_impl=ujson,
    project="your-gcp-project-id"
)

Log Output Format

The handler outputs structured JSON logs compatible with Google Cloud Logging:

{
  "severity": "INFO",
  "name": "root",
  "process": 12345,
  "message": "https://example.com/api/endpoint",
  "logging.googleapis.com/trace": "projects/your-project/traces/abc123",
  "logging.googleapis.com/spanId": "def456",
  "lines": [
    {
      "pathname": "main.py",
      "lineno": 42,
      "message": "Processing request"
    },
    {
      "pathname": "main.py",
      "lineno": 45,
      "message": "Request completed"
    }
  ]
}

Configuration

CloudLoggingHandler Parameters

Parameter	Type	Description
trace_header_name	str	HTTP header name for trace context (e.g., X-Cloud-Trace-Context)
json_impl	module	Custom JSON encoder module (must have dumps method)
project	str	GCP project ID for trace URL construction

Environment Variables

Variable	Description
GCP_PROJECT	Google Cloud Project ID

How It Works

1. Request Start: Middleware creates a RequestLogs context
2. Log Accumulation: All log calls within the request are accumulated
3. Severity Tracking: The highest severity level is tracked
4. Trace Extraction: Trace context is extracted from request headers
5. Request End: flush() emits all accumulated logs as a single structured entry

This approach provides several benefits:

• Correlate all logs from a single request
• View logs grouped by trace in Cloud Console
• Reduce log volume while maintaining detail

Development

Setup

# Clone the repository
git clone https://github.com/yourusername/cloud-logging-handler.git
cd cloud-logging-handler

# Install with dev dependencies using uv
uv sync --all-extras

# Run tests
uv run pytest

# Run linting
uv run ruff check .
uv run ruff format .

# Run type checking
uv run mypy cloud_logging_handler

Running the Example

# Set environment variable
export GCP_PROJECT="your-project-id"

# Run example server
uv run python examples/fastapi_app.py

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add some amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

License

This project is licensed under the MIT License - see the LICENSE file for details.

Acknowledgments

• Inspired by Google Cloud Logging documentation
• Built for use with FastAPI