Problem with auditlog when request or response is stream

[henriquebastos]

We should find a way to ignore the request and or response body if they are stream.

The simplest solution would be to ignore and keep them empty. But this does not seem ideal.

[henriquebastos]

@claude analyse possible solutions and write the explanations here so I can choose one.

[claude]

Claude finished @henriquebastos's task —— View job

---

Analyzing audit log stream handling solutions

• Understand current audit logging implementation
• Identify stream handling issues
• Research solution approaches
• Analyze trade-offs of each solution
• Present recommendations with explanations

Problem Analysis

The issue occurs in src/requestspro/audit.py at lines 73 and 76:

Line 73 - Request body processing:

request_body=request.body if isinstance(request.body, str) else (request.body or b"").decode()

Line 76 - Response body processing:

response_body=response.text

Problems with streams:

1. Request streams: When request.body is a file-like object, generator, or stream, the current code tries to decode it as bytes, which will fail
2. Response streams: When stream=True is used in requests, response.text will try to load the entire response into memory, defeating the purpose of streaming
3. Memory issues: Large streams could cause memory exhaustion when loaded entirely for logging
4. Stream consumption: Reading a stream for logging consumes it, making it unavailable for the actual request/response processing

Solution Options

1. Skip/Empty Approach (Current suggestion)

# For request body
if hasattr(request.body, 'read') or hasattr(request.body, '__iter__'):
    request_body = "<stream>"
else:
    request_body = request.body if isinstance(request.body, str) else (request.body or b"").decode()

# For response body  
if hasattr(response, 'raw') and hasattr(response.raw, 'isclosed') and not response.raw.isclosed():
    response_body = "<stream>"
else:
    response_body = response.text

Pros:

• Simple and safe
• No memory issues
• No stream consumption
• Clear indication of streaming

Cons:

• Loss of audit information
• No insight into stream content
• May miss important debugging info

2. Partial Content Sampling

def safe_sample_content(content, max_bytes=1024):
    """Sample first N bytes/chars of potentially large content"""
    if hasattr(content, 'read'):
        # Stream-like object
        current_pos = getattr(content, 'tell', lambda: 0)()
        try:
            sample = content.read(max_bytes)
            content.seek(current_pos)  # Reset position
            return f"<stream_sample>{sample!r}...</stream_sample>"
        except:
            return "<stream_unreadable>"
    return content

# Usage
request_body = safe_sample_content(request.body) if not isinstance(request.body, str) else request.body

Pros:

• Provides some content visibility
• Limited memory usage
• Preserves stream for actual use

Cons:

• Complex implementation
• Risk of stream position issues
• May still fail with some stream types

3. Metadata-Only Logging

def get_stream_metadata(obj):
    """Extract metadata from stream objects without consuming them"""
    if hasattr(obj, 'read'):
        return {
            "type": "stream",
            "stream_type": type(obj).__name__,
            "size": getattr(obj, 'size', None) or getattr(obj, '__len__', lambda: None)(),
            "seekable": hasattr(obj, 'seek'),
            "readable": hasattr(obj, 'read'),
        }
    return obj

# For request
if hasattr(request.body, 'read'):
    request_body = json.dumps(get_stream_metadata(request.body))

Pros:

• Rich debugging information
• No content consumption
• Safe for all stream types

Cons:

• No actual content visibility
• More complex audit log structure
• JSON serialization needed

4. Content-Type Based Strategy

def smart_body_handling(request, response):
    """Handle body based on content type and size"""
    
    # For request
    content_type = request.headers.get('Content-Type', '')
    if 'application/octet-stream' in content_type or hasattr(request.body, 'read'):
        request_body = f"<{content_type or 'stream'}>"
    else:
        request_body = request.body if isinstance(request.body, str) else (request.body or b"").decode()
    
    # For response  
    if response.headers.get('Transfer-Encoding') == 'chunked' or \
       int(response.headers.get('Content-Length', 0)) > 10_000:
        response_body = f"<large_response:{response.headers.get('Content-Type', 'unknown')}>"
    else:
        response_body = response.text
        
    return request_body, response_body

Pros:

• Intelligent content-based decisions
• Preserves small content, skips large
• Respects HTTP semantics

Cons:

• Complex logic
• Magic numbers for size thresholds
• May miss edge cases

5. Configurable Strategy with Audit Filter

class StreamAuditConfig:
    def __init__(self, 
                 max_request_size=1024,
                 max_response_size=1024, 
                 stream_placeholder="<stream>",
                 sample_streams=False):
        self.max_request_size = max_request_size
        self.max_response_size = max_response_size  
        self.stream_placeholder = stream_placeholder
        self.sample_streams = sample_streams

def create_stream_aware_filter(config: StreamAuditConfig):
    def filter_streams(audit_dict):
        # Apply stream handling based on config
        # Can be customized per use case
        return audit_dict
    return filter_streams

# Usage
config = StreamAuditConfig(sample_streams=True, max_request_size=512)
audit = Audit.for_session(session, audit_filter=create_stream_aware_filter(config))

Pros:

• Highly configurable
• Uses existing filter mechanism
• Backward compatible
• Different strategies per use case

Cons:

• Most complex implementation
• Requires user configuration
• Learning curve for users

Recommendation

I recommend Option 5 (Configurable Strategy) with Option 1 (Skip/Empty) as the default behavior:

1. Immediate fix: Implement safe stream detection and use placeholder text
2. Future enhancement: Add configurable sampling and metadata options
3. Leverage existing: Use the audit filter system for customization

This approach:

• Fixes the immediate issue safely
• Provides a path for enhanced functionality
• Maintains backward compatibility
• Allows users to choose their preferred strategy

Would you like me to implement any of these solutions?

[henriquebastos]

@claude create a PR for option 1. Make sure you have a TDD approach with behavior tests. Follow the test current test strategies if possible.

[claude]

Claude finished @henriquebastos's task —— View job • claude/issue-4-20250915-2231 • Create PR ➔

---

Implementing stream handling for audit logs with TDD approach

• Read repository setup instructions (CLAUDE.md)
• Analyze current audit implementation in src/requestspro/audit.py
• Study existing test patterns and structure
• Write failing tests for stream handling scenarios (TDD red phase)
• Implement stream detection and placeholder logic (TDD green phase)
• Refactor and ensure all tests pass (TDD refactor phase)
• Run linting and testing commands
• Commit and push changes
• Create PR with proper description

Implementation Complete: Successfully implemented Option 1 (Skip/Empty approach) using TDD methodology. The solution safely handles all stream scenarios while preserving backward compatibility.

Key Changes:

• audit.py:73-76: Replaced direct body access with safe helper methods
• _safe_get_request_body(): Detects file-like objects, generators, and iterables
• _safe_get_response_body(): Detects streamed responses via raw socket status
• Comprehensive tests: 6 test cases covering all stream scenarios and backward compatibility

Stream Detection Logic:

• File-like objects: hasattr(body, 'read')
• Generators/iterables: hasattr(body, '__iter__') and not isinstance(body, (bytes, str))
• Streamed responses: response.raw.isclosed() == False
• Returns <stream> placeholder for all stream types
• Preserves existing behavior for strings and bytes