🔍 Advanced Status Codes — Complex HTTP Scenarios

Professional APIs use advanced status codes and headers to optimize performance, ensure data consistency, and handle complex scenarios like range requests and conditional caching.

🔄 Conditional Requests

Save bandwidth by checking if data changed:

import requests
from datetime import datetime

# Get resource with ETag or Last-Modified
response = requests.get('https://api.example.com/users/1')
etag = response.headers.get('ETag')
last_modified = response.headers.get('Last-Modified')

# Later, check if it changed
conditional_headers = {}
if etag:
    conditional_headers['If-None-Match'] = etag
if last_modified:
    conditional_headers['If-Modified-Since'] = last_modified

response = requests.get(
    'https://api.example.com/users/1',
    headers=conditional_headers
)

if response.status_code == 304:  # Not Modified
    print("Data unchanged - use cached version")
else:
    print(f"Data updated - new data: {response.json()}")

Server-Side Conditional Implementation

from flask import Flask, jsonify, request
from datetime import datetime
import hashlib

app = Flask(__name__)

users_db = {
    1: {'id': 1, 'name': 'Alice', 'updated': datetime.now()},
    2: {'id': 2, 'name': 'Bob', 'updated': datetime.now()}
}

def generate_etag(data):
    """Generate ETag from data"""
    data_str = str(data).encode()
    return hashlib.md5(data_str).hexdigest()

@app.route('/users/<int:user_id>')
def get_user(user_id):
    """Handle conditional requests"""
    user = users_db.get(user_id)

    if not user:
        return jsonify({'error': 'Not found'}), 404

    # Generate ETag
    etag = generate_etag(user)

    # Check conditional headers
    if request.headers.get('If-None-Match') == etag:
        return '', 304  # Not Modified

    if request.headers.get('If-Modified-Since'):
        if_modified = datetime.fromisoformat(request.headers['If-Modified-Since'])
        if user['updated'] <= if_modified:
            return '', 304

    # Return with cache headers
    response = jsonify(user)
    response.headers['ETag'] = f'"{etag}"'
    response.headers['Last-Modified'] = user['updated'].strftime('%a, %d %b %Y %H:%M:%S GMT')
    response.headers['Cache-Control'] = 'public, max-age=3600'

    return response

📦 Range Requests (206 Partial Content)

Support partial content for large files:

import requests

# Request specific byte range
headers = {'Range': 'bytes=0-1023'}
response = requests.get(
    'https://example.com/large-file.bin',
    headers=headers
)

if response.status_code == 206:  # Partial Content
    content_range = response.headers.get('Content-Range')
    print(f"Received: {content_range}")  # bytes 0-1023/1048576

    # Can request next chunk
    next_range = 'bytes=1024-2047'
    headers = {'Range': next_range}
    response = requests.get(
        'https://example.com/large-file.bin',
        headers=headers
    )

Server-Side Range Implementation

from flask import Flask, jsonify, request
import os

app = Flask(__name__)

@app.route('/download/<filename>')
def download_file(filename):
    """Support range requests"""
    filepath = f'/files/{filename}'

    try:
        file_size = os.path.getsize(filepath)
    except OSError:
        return jsonify({'error': 'Not found'}), 404

    # Handle range request
    range_header = request.headers.get('Range')

    if range_header:
        # Parse Range header: Range: bytes=0-1023
        try:
            range_value = range_header.replace('bytes=', '')
            start, end = map(int, range_value.split('-'))

            if start >= file_size or (end and end >= file_size):
                return jsonify({'error': 'Invalid range'}), 416

            if not end:
                end = file_size - 1

            # Return partial content
            with open(filepath, 'rb') as f:
                f.seek(start)
                data = f.read(end - start + 1)

            response = make_response(data)
            response.headers['Content-Range'] = f'bytes {start}-{end}/{file_size}'
            response.headers['Content-Length'] = str(len(data))
            response.status_code = 206  # Partial Content

            return response

        except ValueError:
            pass  # Fall through to full file

    # Return full file
    with open(filepath, 'rb') as f:
        data = f.read()

    response = make_response(data)
    response.headers['Accept-Ranges'] = 'bytes'
    response.headers['Content-Length'] = str(file_size)
    response.status_code = 200

    return response

🔀 Content Negotiation Edge Cases

import requests

# Prefer JSON but accept XML if not available
headers = {
    'Accept': 'application/json, application/xml;q=0.9, */*;q=0.1'
}

response = requests.get('https://api.example.com/data', headers=headers)

# Server chooses based on quality factors
if 'application/json' in response.headers.get('Content-Type', ''):
    data = response.json()
elif 'application/xml' in response.headers.get('Content-Type', ''):
    # Parse XML
    pass

# Handle 406 Not Acceptable
if response.status_code == 406:
    print("Server cannot provide requested format")
    # Retry with fallback format
    headers['Accept'] = '*/*'
    response = requests.get('https://api.example.com/data', headers=headers)

🚨 Status Code Sequences

import requests
import time

def handle_complex_api_scenario(url, method='GET', data=None, max_retries=3):
    """Handle complex status code scenarios"""

    for attempt in range(max_retries):
        try:
            if method == 'GET':
                response = requests.get(url, timeout=5)
            else:
                response = requests.post(url, json=data, timeout=5)

            # Handle various scenarios
            if response.status_code == 200:
                return response.json(), 'success'

            elif response.status_code == 201:
                # Created - resource available at Location header
                location = response.headers.get('Location')
                return {'created_at': location}, 'created'

            elif response.status_code == 204:
                # No content
                return None, 'no_content'

            elif response.status_code == 304:
                # Not modified - use cached
                return None, 'not_modified'

            elif response.status_code == 307 or response.status_code == 308:
                # Temporary/permanent redirect - may need to retry
                new_url = response.headers.get('Location')
                return handle_complex_api_scenario(new_url, method, data, attempt + 1)

            elif response.status_code == 400:
                # Bad request - check error details
                errors = response.json().get('errors', [])
                return {'errors': errors}, 'invalid'

            elif response.status_code == 409:
                # Conflict - data race or duplicate
                return {'conflict': response.json()}, 'conflict'

            elif response.status_code == 429:
                # Rate limited - wait and retry
                retry_after = int(response.headers.get('Retry-After', 60))
                if attempt < max_retries - 1:
                    time.sleep(retry_after)
                    continue

            elif response.status_code == 503:
                # Service unavailable - retry with backoff
                if attempt < max_retries - 1:
                    wait = 2 ** attempt
                    time.sleep(wait)
                    continue

            else:
                return None, f'unknown_{response.status_code}'

        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                return None, f'error: {e}'

            time.sleep(2 ** attempt)

    return None, 'max_retries_exceeded'

# Usage
data, status = handle_complex_api_scenario(
    'https://api.example.com/data',
    method='POST',
    data={'name': 'Alice'}
)

print(f"Result: {status} - {data}")

📊 Status Code Metrics

import requests
from collections import defaultdict
import time

class APIMetrics:
    """Track API status code distribution"""

    def __init__(self):
        self.status_counts = defaultdict(int)
        self.response_times = []

    def track_request(self, url, method='GET', **kwargs):
        """Track request and collect metrics"""
        start = time.time()

        try:
            if method == 'GET':
                response = requests.get(url, **kwargs)
            elif method == 'POST':
                response = requests.post(url, **kwargs)
            else:
                return None

            elapsed = time.time() - start

            # Track metrics
            self.status_counts[response.status_code] += 1
            self.response_times.append(elapsed)

            return response

        except Exception as e:
            self.status_counts['error'] += 1
            return None

    def get_report(self):
        """Generate metrics report"""
        total = sum(self.status_counts.values())

        report = {
            'total_requests': total,
            'status_distribution': dict(self.status_counts),
            'success_rate': (self.status_counts.get(200, 0) / total * 100) if total > 0 else 0,
            'average_response_time': sum(self.response_times) / len(self.response_times) if self.response_times else 0,
            'percentiles': {
                'p50': sorted(self.response_times)[len(self.response_times) // 2] if self.response_times else 0,
                'p95': sorted(self.response_times)[int(len(self.response_times) * 0.95)] if self.response_times else 0,
                'p99': sorted(self.response_times)[int(len(self.response_times) * 0.99)] if self.response_times else 0,
            }
        }

        return report

# Usage
metrics = APIMetrics()

for i in range(100):
    metrics.track_request('https://api.example.com/data')

report = metrics.get_report()
print(f"Success Rate: {report['success_rate']:.1f}%")
print(f"Avg Response Time: {report['average_response_time']:.3f}s")

🔐 Idempotency and Status Codes

from flask import Flask, jsonify, request
import uuid

app = Flask(__name__)

# Track processed requests
processed_requests = {}

@app.route('/api/transfer', methods=['POST'])
def transfer_money():
    """Idempotent endpoint - safe to retry"""

    # Get idempotency key
    idempotency_key = request.headers.get('Idempotency-Key')

    if not idempotency_key:
        return jsonify({'error': 'Idempotency-Key header required'}), 400

    # Check if already processed
    if idempotency_key in processed_requests:
        # Return cached response with same status
        cached = processed_requests[idempotency_key]
        return jsonify(cached['data']), cached['status']

    # Process request
    data = request.get_json()

    try:
        # Perform transfer
        result = {
            'id': str(uuid.uuid4()),
            'from': data['from'],
            'to': data['to'],
            'amount': data['amount'],
            'status': 'completed'
        }

        # Cache response
        processed_requests[idempotency_key] = {
            'data': result,
            'status': 201
        }

        return jsonify(result), 201

    except Exception as e:
        return jsonify({'error': str(e)}), 400

# Client usage
headers = {
    'Idempotency-Key': 'transfer-12345-67890'
}

# Safe to retry - same idempotency key returns same response
for i in range(3):
    response = requests.post(
        'https://api.example.com/transfer',
        json={'from': 'alice', 'to': 'bob', 'amount': 100},
        headers=headers
    )
    print(f"Attempt {i+1}: {response.status_code}")

✅ Key Takeaways

🔗 What's Next?

Explore advanced JSON serialization and validation patterns.

Next: Advanced JSON Serialization →

Ready for advanced challenges? Try advanced challenges