Request Hooks

Request hooks allow you to execute custom code before and after each request.

Overview

The client supports two hooks:

Hook	When Called	Use Cases
`pre_request_hook`	Before sending request	Logging, metrics, modification
`post_response_hook`	After receiving response	Logging, metrics, validation

Pre-Request Hook

Called before each request is sent:

from intapp_rest_client import RestClient
 
def log_request(method, url, headers, body):
    print(f"→ {method} {url}")
    if body:
        print(f"  Body size: {len(str(body))} chars")
 
client = RestClient(
    base_url="https://api.example.com",
    api_key="your-key",
    pre_request_hook=log_request
)
 
client.get("/api/resources")
# Output: → GET https://api.example.com/api/resources

Hook Signature

def pre_request_hook(
    method: str,      # HTTP method (GET, POST, etc.)
    url: str,         # Full URL
    headers: dict,    # Request headers
    body: Any         # Request body (for POST/PUT/PATCH)
) -> None:
    pass

⚠️

The hook receives headers after authentication headers are added. Sensitive values like Authorization are masked in logs but visible in hooks.

Post-Response Hook

Called after each response is received:

def log_response(response, duration_ms):
    print(f"← {response.status_code} ({duration_ms:.1f}ms)")
 
client = RestClient(
    base_url="https://api.example.com",
    api_key="your-key",
    post_response_hook=log_response
)
 
client.get("/api/resources")
# Output: ← 200 (123.4ms)

Hook Signature

def post_response_hook(
    response: httpx.Response,  # Full response object
    duration_ms: float         # Request duration in milliseconds
) -> None:
    pass

Common Use Cases

Metrics Collection

from prometheus_client import Counter, Histogram
 
REQUEST_COUNT = Counter(
    'api_requests_total',
    'Total API requests',
    ['method', 'endpoint', 'status']
)
 
REQUEST_LATENCY = Histogram(
    'api_request_latency_seconds',
    'Request latency',
    ['method', 'endpoint']
)
 
def collect_metrics(response, duration_ms):
    method = response.request.method
    # Extract endpoint path without query params
    endpoint = response.request.url.path
    status = response.status_code
    
    REQUEST_COUNT.labels(
        method=method,
        endpoint=endpoint,
        status=status
    ).inc()
    
    REQUEST_LATENCY.labels(
        method=method,
        endpoint=endpoint
    ).observe(duration_ms / 1000)
 
client = RestClient(
    base_url="https://api.example.com",
    api_key="key",
    post_response_hook=collect_metrics
)

Request Logging

import logging
import json
 
logger = logging.getLogger("api_client")
 
def log_request(method, url, headers, body):
    logger.info(json.dumps({
        "event": "request",
        "method": method,
        "url": url,
        "has_body": body is not None
    }))
 
def log_response(response, duration_ms):
    logger.info(json.dumps({
        "event": "response",
        "status": response.status_code,
        "duration_ms": round(duration_ms, 2),
        "content_length": len(response.content)
    }))
 
client = RestClient(
    base_url="https://api.example.com",
    api_key="key",
    pre_request_hook=log_request,
    post_response_hook=log_response
)

Request Auditing

import datetime
 
audit_log = []
 
def audit_request(method, url, headers, body):
    audit_log.append({
        "timestamp": datetime.datetime.now().isoformat(),
        "type": "request",
        "method": method,
        "url": url
    })
 
def audit_response(response, duration_ms):
    audit_log.append({
        "timestamp": datetime.datetime.now().isoformat(),
        "type": "response",
        "status": response.status_code,
        "duration_ms": duration_ms
    })
 
client = RestClient(
    base_url="https://api.example.com",
    api_key="key",
    pre_request_hook=audit_request,
    post_response_hook=audit_response
)
 
# Later: review audit log
for entry in audit_log:
    print(entry)

Rate Limit Monitoring

import time
 
class RateLimitMonitor:
    def __init__(self):
        self.remaining = None
        self.reset_at = None
    
    def check_limits(self, response, duration_ms):
        # Many APIs return rate limit info in headers
        self.remaining = response.headers.get("X-RateLimit-Remaining")
        reset = response.headers.get("X-RateLimit-Reset")
        if reset:
            self.reset_at = datetime.fromtimestamp(int(reset))
        
        if self.remaining and int(self.remaining) < 10:
            print(f"⚠️ Low rate limit: {self.remaining} remaining")
 
monitor = RateLimitMonitor()
 
client = RestClient(
    base_url="https://api.example.com",
    api_key="key",
    post_response_hook=monitor.check_limits
)

Timing Analysis

from statistics import mean, stdev
 
class TimingAnalyzer:
    def __init__(self):
        self.timings = {}
    
    def record_timing(self, response, duration_ms):
        endpoint = response.request.url.path
        if endpoint not in self.timings:
            self.timings[endpoint] = []
        self.timings[endpoint].append(duration_ms)
    
    def report(self):
        for endpoint, times in self.timings.items():
            avg = mean(times)
            std = stdev(times) if len(times) > 1 else 0
            print(f"{endpoint}: {avg:.1f}ms ± {std:.1f}ms ({len(times)} calls)")
 
analyzer = TimingAnalyzer()
 
client = RestClient(
    base_url="https://api.example.com",
    api_key="key",
    post_response_hook=analyzer.record_timing
)
 
# Make requests...
client.get("/api/users")
client.get("/api/companies")
client.get("/api/users")
 
# Print report
analyzer.report()
# Output:
# /api/users: 125.3ms ± 12.1ms (2 calls)
# /api/companies: 89.2ms ± 0.0ms (1 calls)

Error Handling in Hooks

Hooks should not raise exceptions. The client catches and logs hook errors:

def risky_hook(response, duration_ms):
    # If this fails, the request still succeeds
    save_to_database(response)  # Might fail
 
client = RestClient(
    base_url="https://api.example.com",
    api_key="key",
    post_response_hook=risky_hook
)
 
# If hook fails, you'll see:
# WARNING - Post-response hook failed: DatabaseError(...)
# But the request result is still returned

💡

Hooks are called even for failed requests, making them reliable for metrics collection. Check response.status_code to distinguish success from failure.

Combining with Built-in Logging

Hooks complement the built-in logging:

from intapp_rest_client import RestClient, LogLevel
 
# Built-in logging for standard info
# Custom hook for specific metrics
 
client = RestClient(
    base_url="https://api.example.com",
    api_key="key",
    log_level=LogLevel.INFO,  # Standard logging
    post_response_hook=custom_metrics_collector  # Custom metrics
)

Next Steps

Logging - Built-in logging
Tracing - OpenTelemetry integration
Error Handling - Handle hook errors

Tracing & Telemetry Configuration Models